Manual Push to Gitlab

pipeline/ingest.py

+#!/usr/bin/env python3
+import argparse, sys, json, os
+from pathlib import Path
+import pandas as pd
+import yaml
+
+def load_config():
+    with open("config.yaml", "r") as f:
+        return yaml.safe_load(f)
+
+def coerce_timestamp(df, cfg):
+    # More robust timestamp parsing
+    ts_cols = cfg["data"]["timestamp_columns"]
+    date_col = cfg["data"]["date_column"]
+    time_col = cfg["data"]["time_column"]
+    
+    # Try the datetime_combined column first as it has the correct format
+    if "datetime_combined" in df.columns:
+        try:
+            df["timestamp"] = pd.to_datetime(df["datetime_combined"], errors="coerce", utc=True)
+            print(f"Successfully parsed datetime_combined: {df['timestamp'].notna().sum()} valid timestamps")
+            return df
+        except Exception as e:
+            print(f"Error parsing datetime_combined: {e}")
+    
+    # Try each configured timestamp column
+    for c in ts_cols:
+        if c in df.columns:
+            try:
+                ts = pd.to_datetime(df[c], errors="coerce", utc=True)
+                if ts.notna().sum() > 0:
+                    df["timestamp"] = ts
+                    print(f"Successfully parsed {c}: {ts.notna().sum()} valid timestamps")
+                    return df
+            except Exception as e:
+                print(f"Error parsing {c}: {e}")
+    
+    # Try date + time columns
+    if date_col in df.columns and time_col in df.columns:
+        try:
+            dt = pd.to_datetime(df[date_col].astype(str) + " " + df[time_col].astype(str),
+                            errors="coerce", utc=True)
+            df["timestamp"] = dt
+            print(f"Successfully parsed {date_col}+{time_col}: {dt.notna().sum()} valid timestamps")
+            return df
+        except Exception as e:
+            print(f"Error parsing {date_col}+{time_col}: {e}")
+    
+    if "timestamp" not in df.columns or df["timestamp"].isna().all():
+        raise ValueError("Could not parse a timestamp column. Check config.yaml and CSV headers.")
+    
+    return df
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--csv", type=str, default=None, help="Path to raw CSV")
+    parser.add_argument("--debug", action="store_true", help="Print debug info")
+    parser.add_argument("--phase", type=str, choices=["train", "predict"], default="train", 
+                       help="Whether this is for training or prediction phase")
+    parser.add_argument("--out", type=str, default=None, help="Custom output filename")
+    args = parser.parse_args()
+    cfg = load_config()
+
+    # Debug output
+    if args.debug:
+        print(f"Config: {json.dumps(cfg, indent=2)}")
+    
+    # Select CSV based on phase
+    if args.csv:
+        csv_path = args.csv
+    elif args.phase == "train" and "train_csv" in cfg["data"]:
+        csv_path = cfg["data"]["train_csv"]
+    elif args.phase == "predict" and "predict_csv" in cfg["data"]:
+        csv_path = cfg["data"]["predict_csv"]
+    else:
+        csv_path = cfg["data"]["raw_csv"]
+    
+    print(f"Running in {args.phase} phase with CSV: {csv_path}")
+    csv_path = Path(csv_path)
+    if not csv_path.exists():
+        raise FileNotFoundError(f"CSV not found at {csv_path}. Update config.yaml or pass --csv.")
+
+    print(f"Loading CSV from: {csv_path}")
+    df = pd.read_csv(csv_path, engine="python")
+    
+    if args.debug:
+        print(f"CSV columns: {df.columns.tolist()}")
+        print(f"CSV sample:\n{df.head(3)}")
+    
+    # Make sure we're using the configured room_id_column
+    room_id_col = cfg["data"]["room_id_column"]
+    
+    # Fix: Handle room_id correctly
+    if room_id_col not in df.columns:
+        # If the configured room_id column doesn't exist, create it from device_id or default to 1
+        df[room_id_col] = df.get("device_id", 1)
+        print(f"Created {room_id_col} column")
+    
+    # Parse timestamps
+    df = coerce_timestamp(df, cfg)
+    
+    # Use the configured room_id_column for sorting and deduplication
+    df = df.sort_values([room_id_col, "timestamp"]).drop_duplicates(subset=[room_id_col, "timestamp"])
+    
+    # Select columns to keep
+    keep = set([room_id_col, "timestamp", cfg["data"]["label_column"]] + cfg["data"]["sensors"])
+    existing = [c for c in df.columns if c in keep]
+    
+    if args.debug:
+        print(f"Keeping columns: {existing}")
+        missing = keep - set(existing)
+        if missing:
+            print(f"Warning: Missing columns: {missing}")
+    
+    base = df[existing].copy()
+    
+    # Save processed data
+    artifacts = Path(cfg["data"]["artifacts_dir"])
+    artifacts.mkdir(parents=True, exist_ok=True)
+    
+    # Use different output paths for training and prediction
+    if args.out:
+        out_filename = args.out
+    else:
+        out_filename = f"base_{args.phase}.parquet"
+    
+    out_path = artifacts / out_filename
+    base.to_parquet(out_path, index=False)
+    
+    print(json.dumps({
+        "rows": len(base), 
+        "cols": list(base.columns), 
+        "out": str(out_path),
+        "timestamp_range": [
+            base["timestamp"].min().strftime("%Y-%m-%d %H:%M:%S") if not base["timestamp"].empty else None,
+            base["timestamp"].max().strftime("%Y-%m-%d %H:%M:%S") if not base["timestamp"].empty else None
+        ]
+    }, indent=2))
+
+if __name__ == "__main__":
+    main()
\ No newline at end of file

pipeline/optimize_energy.py

+#!/usr/bin/env python3
+import pandas as pd
+import numpy as np
+from pathlib import Path
+import yaml
+import json
+import logging
+import time
+
+# Set up logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    handlers=[
+        logging.FileHandler("optimize_energy.log"),
+        logging.StreamHandler()
+    ]
+)
+logger = logging.getLogger("energy_optimization")
+
+def load_config():
+    logger.info("Loading configuration from config.yaml")
+    with open("config.yaml", "r") as f:
+        cfg = yaml.safe_load(f)
+        
+        # Log key energy optimization parameters
+        energy_params = {
+            'comfort_setpoint': cfg["schedule"].get("comfort_setpoint", 21.0),
+            'setback_setpoint': cfg["schedule"].get("setback_setpoint", 16.0),
+            'preheat_minutes': cfg["schedule"].get("preheat_minutes_default", 30)
+        }
+        logger.info(f"Default energy parameters: {energy_params}")
+        return cfg
+
+def calculate_heating_plan(levels_df, room_settings, config):
+    """
+    Create optimized heating plan based on occupancy levels
+    """
+    room_id_col = config["data"]["room_id_column"]
+    comfort_temp = config["schedule"]["comfort_setpoint"]
+    setback_temp = config["schedule"].get("setback_setpoint", 16.0)
+    preheat_minutes = config["schedule"].get("preheat_minutes_default", 30)
+    
+    logger.info(f"Calculating heating plan for {levels_df[room_id_col].nunique()} rooms")
+    logger.info(f"Default settings: comfort={comfort_temp}°C, setback={setback_temp}°C, preheat={preheat_minutes}min")
+    
+    # Log room-specific settings
+    for room_id, settings in room_settings.items():
+        logger.info(f"Room {room_id} custom settings: {settings}")
+    
+    # Need to include block_start in our aggregation
+    logger.info(f"Grouping data by {room_id_col} and block_index")
+    grouped = levels_df.groupby([room_id_col, 'block_index'])
+    
+    # Calculate mean occupancy and get the first block_start for each group
+    logger.info("Aggregating occupancy probabilities and timestamps")
+    agg_result = grouped.agg({
+        'is_occ': 'mean',
+        'block_start': 'first'  # Keep the block_start column
+    }).reset_index()
+    
+    logger.info(f"Aggregated data: {len(agg_result)} unique time blocks across all rooms")
+    
+    # Calculate optimal setpoints
+    heating_plan = {}
+    total_comfort_blocks = 0
+    total_setback_blocks = 0
+    
+    for room_id, room_data in agg_result.groupby(room_id_col):
+        room_plan = []
+        room_prob = room_data.sort_values('block_index')
+        
+        logger.info(f"Processing Room {room_id}: {len(room_prob)} time blocks")
+        
+        # Room-specific settings
+        room_comfort = room_settings.get(str(room_id), {}).get('comfort_temp', comfort_temp)
+        room_setback = room_settings.get(str(room_id), {}).get('setback_temp', setback_temp)
+        room_preheat = room_settings.get(str(room_id), {}).get('preheat_minutes', preheat_minutes)
+        threshold = room_settings.get(str(room_id), {}).get('threshold', 0.5)
+        
+        logger.info(f"Room {room_id} effective settings: comfort={room_comfort}°C, setback={room_setback}°C, "
+                   f"preheat={room_preheat}min, threshold={threshold}")
+        
+        # Track room-specific metrics
+        room_comfort_blocks = 0
+        room_setback_blocks = 0
+        
+        # For each time block, decide on temperature
+        for _, block in room_prob.iterrows():
+            block_time = block['block_start'].strftime("%Y-%m-%d %H:%M")
+            block_probability = float(block['is_occ'])
+            
+            logger.debug(f"Room {room_id}, Block {block['block_index']} ({block_time}): probability={block_probability:.2f}")
+            
+            if block_probability >= threshold:
+                # High probability of occupancy - comfort temperature
+                # Include pre-heating period
+                start_time = block['block_start'] - pd.Timedelta(minutes=room_preheat)
+                end_time = block['block_start'] + pd.Timedelta(minutes=120)  # 2-hour blocks
+                
+                room_plan.append({
+                    'start': start_time.isoformat(),
+                    'end': end_time.isoformat(),
+                    'setpoint': room_comfort,
+                    'probability': block_probability
+                })
+                
+                logger.debug(f"  → Comfort period: {start_time.strftime('%H:%M')} to {end_time.strftime('%H:%M')}, "
+                           f"{room_comfort}°C (with {room_preheat}min preheat)")
+                room_comfort_blocks += 1
+                
+            else:
+                # Low probability - use setback temperature
+                room_plan.append({
+                    'start': block['block_start'].isoformat(),
+                    'end': (block['block_start'] + pd.Timedelta(minutes=120)).isoformat(),
+                    'setpoint': room_setback,
+                    'probability': block_probability
+                })
+                
+                logger.debug(f"  → Setback period: {block['block_start'].strftime('%H:%M')} to "
+                           f"{(block['block_start'] + pd.Timedelta(minutes=120)).strftime('%H:%M')}, {room_setback}°C")
+                room_setback_blocks += 1
+        
+        # Optimize the schedule by combining adjacent comfort periods
+        logger.info(f"Room {room_id} initial plan: {room_comfort_blocks} comfort periods, {room_setback_blocks} setback periods")
+        optimized_plan = optimize_adjacent_periods(room_plan)
+        
+        logger.info(f"Room {room_id} optimized plan: {len(optimized_plan)} total heating periods "
+                   f"(reduced from {len(room_plan)})")
+        
+        heating_plan[str(room_id)] = optimized_plan
+        total_comfort_blocks += room_comfort_blocks
+        total_setback_blocks += room_setback_blocks
+    
+    # Log overall statistics
+    total_blocks = total_comfort_blocks + total_setback_blocks
+    comfort_percentage = (total_comfort_blocks / total_blocks * 100) if total_blocks > 0 else 0
+    
+    logger.info(f"Overall heating plan statistics:")
+    logger.info(f"  - Total blocks: {total_blocks}")
+    logger.info(f"  - Comfort blocks: {total_comfort_blocks} ({comfort_percentage:.1f}%)")
+    logger.info(f"  - Setback blocks: {total_setback_blocks} ({100-comfort_percentage:.1f}%)")
+    
+    return heating_plan
+
+def optimize_adjacent_periods(schedule):
+    """Merge adjacent comfort periods to avoid frequent temperature changes"""
+    if not schedule:
+        return []
+    
+    logger.info(f"Optimizing schedule with {len(schedule)} periods")
+    
+    # Sort by start time
+    sorted_schedule = sorted(schedule, key=lambda x: x['start'])
+    optimized = [sorted_schedule[0]]
+    
+    merged_count = 0
+    
+    for current in sorted_schedule[1:]:
+        previous = optimized[-1]
+        
+        # If current period starts when previous ends and has same setpoint
+        if (current['start'] == previous['end'] and 
+            abs(current['setpoint'] - previous['setpoint']) < 0.5):
+            # Extend previous period
+            previous['end'] = current['end']
+            previous['probability'] = max(previous['probability'], current['probability'])
+            merged_count += 1
+            logger.debug(f"Merged period: {previous['start']} to {previous['end']}, {previous['setpoint']}°C")
+        else:
+            optimized.append(current)
+    
+    reduction = len(schedule) - len(optimized)
+    logger.info(f"Optimization complete: {len(optimized)} periods after merging {merged_count} adjacent periods")
+    logger.info(f"Reduced heating transitions by {reduction} ({reduction/len(schedule)*100:.1f}%)")
+    
+    return optimized
+
+def main():
+    start_time = time.time()
+    logger.info("=== ENERGY OPTIMIZATION PROCESS STARTED ===")
+    
+    try:
+        config = load_config()
+        artifacts_dir = Path(config["data"]["artifacts_dir"])
+        
+        # Load occupancy levels
+        levels_path = artifacts_dir / "levels_2h.parquet"
+        logger.info(f"Loading occupancy levels from {levels_path}")
+        levels_df = pd.read_parquet(levels_path)
+        
+        # Convert timestamps
+        logger.info("Converting timestamps to datetime objects")
+        levels_df["block_start"] = pd.to_datetime(levels_df["block_start"], utc=True)
+        
+        # Check date range
+        date_range = levels_df["block_start"].agg(["min", "max"])
+        logger.info(f"Data spans from {date_range['min']} to {date_range['max']}")
+        logger.info(f"Loaded {len(levels_df)} occupancy records for {levels_df[config['data']['room_id_column']].nunique()} rooms")
+        
+        # Add these two lines to create required columns
+        block_minutes = config["levels"]["block_minutes"]
+        logger.info(f"Creating {block_minutes}-minute block indices")
+        levels_df["block_index"] = (levels_df["block_start"].dt.hour*60 + levels_df["block_start"].dt.minute)//block_minutes
+        
+        logger.info("Converting level values to binary occupancy indicators")
+        levels_df["is_occ"] = (levels_df["level"] > 0).astype(int)
+        
+        # Count occupancy distribution
+        occ_counts = levels_df["is_occ"].value_counts()
+        logger.info(f"Occupancy distribution: {occ_counts.to_dict()}")
+        
+        # Load room settings
+        logger.info("Configuring room-specific settings")
+        room_settings = {
+            "1": {
+                "comfort_temp": 21.5,
+                "setback_temp": 16.0,
+                "preheat_minutes": 30,
+                "threshold": 0.4  # More likely to heat (lower threshold)
+            }
+        }
+        
+        # Calculate heating plan
+        logger.info("Calculating optimized heating plan")
+        heating_plan = calculate_heating_plan(levels_df, room_settings, config)
+        
+        # Count total heating periods
+        total_periods = sum(len(periods) for periods in heating_plan.values())
+        logger.info(f"Generated heating plan with {len(heating_plan)} rooms and {total_periods} total periods")
+        
+        # Calculate potential energy savings
+        comfort_hours = sum(
+            (pd.Timestamp(period['end']) - pd.Timestamp(period['start'])).total_seconds() / 3600
+            for room_periods in heating_plan.values()
+            for period in room_periods
+            if period['setpoint'] > 20  # Assuming comfort temp > 20°C
+        )
+        
+        setback_hours = sum(
+            (pd.Timestamp(period['end']) - pd.Timestamp(period['start'])).total_seconds() / 3600
+            for room_periods in heating_plan.values()
+            for period in room_periods
+            if period['setpoint'] <= 20  # Assuming setback temp <= 20°C
+        )
+        
+        total_hours = comfort_hours + setback_hours
+        # Old calculation (incorrect):
+        # savings_percentage = (setback_hours / total_hours * 100) if total_hours > 0 else 0
+
+        # New calculation: compare energy use to baseline
+        comfort_temp = config["schedule"]["comfort_setpoint"]
+        setback_temp = config["schedule"]["setback_setpoint"]
+        outside_temp = 0  # Assume 0°C for simplicity
+
+        # Baseline: all hours at comfort temp
+        E_base = total_hours * (comfort_temp - outside_temp)
+        # Optimized: comfort and setback hours
+        E_opt = comfort_hours * (comfort_temp - outside_temp) + setback_hours * (setback_temp - outside_temp)
+
+        savings_percentage = 100 * (E_base - E_opt) / E_base if E_base > 0 else 0
+        
+        logger.info(f"Energy analysis:")
+        logger.info(f"  - Comfort temperature hours: {comfort_hours:.1f}")
+        logger.info(f"  - Setback temperature hours: {setback_hours:.1f}")
+        logger.info(f"  - Estimated energy savings: {savings_percentage:.1f}% (compared to constant comfort temperature)")
+        
+        # Save the optimized heating plan
+        output_path = artifacts_dir / "optimized_heating_plan.json"
+        logger.info(f"Saving optimized heating plan to {output_path}")
+        
+        with open(output_path, "w") as f:
+            json.dump(heating_plan, f, indent=2)
+        
+        # Calculate execution time
+        execution_time = time.time() - start_time
+        logger.info(f"Total execution time: {execution_time:.2f} seconds")
+        
+        # Print summary statistics
+        summary = {
+            "rooms": len(heating_plan),
+            "total_periods": total_periods,
+            "comfort_hours": round(comfort_hours, 1),
+            "setback_hours": round(setback_hours, 1),
+            "estimated_savings_pct": round(savings_percentage, 1),
+            "execution_time_seconds": round(execution_time, 2),
+            "out": str(output_path)
+        }
+        print(json.dumps(summary, indent=2))
+        
+        logger.info("=== ENERGY OPTIMIZATION COMPLETED SUCCESSFULLY ===")
+        
+    except Exception as e:
+        logger.error(f"Energy optimization failed: {e}", exc_info=True)
+        print(json.dumps({"error": str(e)}, indent=2))
+        raise
+
+if __name__ == "__main__":
+    main()
\ No newline at end of file

pipeline/predict.py

+#!/usr/bin/env python3
+import json, time, logging, argparse
+from pathlib import Path
+import pandas as pd
+import numpy as np
+import yaml
+import joblib
+
+# Set up logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    handlers=[
+        logging.FileHandler("predict.log"),
+        logging.StreamHandler()
+    ]
+)
+logger = logging.getLogger("prediction")
+
+def load_config():
+    logger.info("Loading configuration from config.yaml")
+    with open("config.yaml", "r") as f:
+        cfg = yaml.safe_load(f)
+        logger.info(f"Config loaded - Room ID column: {cfg['data']['room_id_column']}, "
+                   f"Artifacts dir: {cfg['data']['artifacts_dir']}")
+        return cfg
+
+def select_X(df, feat_names):
+    logger.info(f"Preparing feature matrix with {len(feat_names)} required features")
+    
+    # Check which features are available in the dataframe
+    available = [c for c in feat_names if c in df.columns]
+    missing = [c for c in feat_names if c not in df.columns]
+    
+    logger.info(f"Features available: {len(available)}/{len(feat_names)} ({len(missing)} missing)")
+    if missing:
+        logger.warning(f"Missing features (will be filled with zeros): {missing[:10]}{'...' if len(missing) > 10 else ''}")
+    
+    # Create feature matrix
+    X = df[available].copy()
+    
+    # Add missing features with zeros
+    for c in missing:
+        X[c] = 0.0
+        
+    # Check for NaN values
+    nan_counts = X.isna().sum()
+    nan_features = nan_counts[nan_counts > 0]
+    if not nan_features.empty:
+        logger.warning(f"NaN values found in {len(nan_features)} features, will be filled with zeros")
+        logger.debug(f"Features with NaNs: {nan_features.to_dict()}")
+    
+    # Fill NaN values
+    X_filled = X.fillna(0.0)
+    logger.info(f"Final feature matrix shape: {X_filled.shape}")
+    
+    return X_filled.values
+
+def main():
+    logger.info("=== PREDICTION PROCESS STARTED ===")
+    start_time = time.time()
+    
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--input", type=str, default=None, 
+                       help="Custom input filename (default: feature_store_predict.parquet or feature_store.parquet)")
+    parser.add_argument("--output", type=str, default=None,
+                       help="Custom output filename (default: preds.parquet)")
+    args = parser.parse_args()
+    
+    try:
+        # Load configuration
+        cfg = load_config()
+        room_id_col = cfg["data"]["room_id_column"]
+        artifacts = Path(cfg["data"]["artifacts_dir"])
+        
+        # Load feature data - try predict-specific file first, then fallback to general one
+        if args.input:
+            feat_path = artifacts / args.input
+        else:
+            feat_path = artifacts / "feature_store_predict.parquet"
+            if not feat_path.exists():
+                fallback_path = artifacts / "feature_store.parquet"
+                if fallback_path.exists():
+                    logger.warning(f"{feat_path} not found, falling back to {fallback_path}")
+                    feat_path = fallback_path
+                    
+        logger.info(f"Loading features from {feat_path}")
+        feat = pd.read_parquet(feat_path)
+        logger.info(f"Loaded {len(feat)} samples with {len(feat.columns)} columns")
+        
+        # Summarize dataset
+        rooms = feat[room_id_col].nunique()
+        time_range = pd.to_datetime(feat["timestamp"]).agg(["min", "max"])
+        logger.info(f"Data covers {rooms} unique rooms from {time_range['min']} to {time_range['max']}")
+        
+        # Load model
+        model_path = artifacts / "model.pkl"
+        logger.info(f"Loading model from {model_path}")
+        pkg = joblib.load(model_path)
+        model = pkg["model"]
+        feat_names = pkg["features"]
+        logger.info(f"Loaded model with {len(feat_names)} features")
+        
+        # Log model info if available
+        if hasattr(model, 'get_params'):
+            model_params = model.get_params()
+            logger.info(f"Model type: {type(model).__name__}")
+            if hasattr(model, 'steps'):
+                logger.info(f"Model pipeline steps: {[s[0] for s in model.steps]}")
+        
+        # Prepare features
+        logger.info("Preparing features for prediction")
+        X = select_X(feat, feat_names)
+        
+        # Make predictions
+        logger.info(f"Making predictions on {len(X)} samples")
+        pred_start_time = time.time()
+        y_pred = model.predict(X)
+        y_prob = model.predict_proba(X)
+        pred_time = time.time() - pred_start_time
+        
+        logger.info(f"Predictions completed in {pred_time:.2f} seconds ({len(X)/pred_time:.1f} samples/sec)")
+        
+        # Create prediction dataframe
+        classes = list(model.classes_)
+        logger.info(f"Model has {len(classes)} classes: {classes}")
+        
+        # Calculate prediction distribution
+        pred_counts = pd.Series(y_pred).value_counts().to_dict()
+        logger.info(f"Prediction distribution: {pred_counts}")
+        
+        # Create probability columns
+        prob_df = pd.DataFrame(y_prob, columns=[f"p{int(c)}" for c in classes])
+        logger.info(f"Added {len(classes)} probability columns")
+        
+        # Create output dataframe
+        logger.info("Creating final prediction dataframe")
+        out = pd.concat([
+            feat[[room_id_col, "timestamp"]].reset_index(drop=True),
+            pd.Series(y_pred, name="y_pred"),
+            prob_df
+        ], axis=1)
+        
+        # Add compatibility room_id column if needed
+        if "room_id" not in out.columns and room_id_col != "room_id":
+            logger.info(f"Adding 'room_id' column as alias for '{room_id_col}' for compatibility")
+            out["room_id"] = out[room_id_col]
+        
+        # Save predictions
+        if args.output:
+            out_path = artifacts / args.output
+        else:
+            out_path = artifacts / "preds.parquet"
+        logger.info(f"Saving {len(out)} predictions to {out_path}")
+        out.to_parquet(out_path, index=False)
+        
+        # Calculate average probabilities
+        avg_probs = {f"avg_p{c}": prob_df[f"p{c}"].mean() for c in classes}
+        logger.info(f"Average probabilities: {avg_probs}")
+        
+        # Calculate execution time
+        total_time = time.time() - start_time
+        logger.info(f"Total execution time: {total_time:.2f} seconds")
+        
+        # Print summary
+        summary = {
+            "rows": len(out),
+            "rooms": int(rooms),
+            "classes": len(classes), 
+            "pred_distribution": pred_counts,
+            "avg_probabilities": {k: round(v, 3) for k, v in avg_probs.items()},
+            "execution_time": round(total_time, 2),
+            "out": str(out_path)
+        }
+        print(json.dumps(summary, indent=2))
+        
+        logger.info("=== PREDICTION PROCESS COMPLETED SUCCESSFULLY ===")
+        
+    except Exception as e:
+        logger.error(f"Prediction process failed: {e}", exc_info=True)
+        print(json.dumps({"error": str(e)}, indent=2))
+        raise
+
+if __name__ == "__main__":
+    main()
\ No newline at end of file

pipeline/rule_based_data_generation.py

+import pandas as pd
+import numpy as np
+from datetime import datetime
+
+def main():
+    # File paths
+    input_file = "./data/raw/combined_sensor_data_train.csv"
+    output_file = "./data/processed/combined_sensor_data_train.csv"
+    
+    print(f"Loading data from {input_file}...")
+    
+    try:
+        df = pd.read_csv(input_file)
+        print(f"Successfully loaded data with {len(df)} rows and {len(df.columns)} columns.")
+        print(f"Columns found: {', '.join(df.columns)}")
+    except Exception as e:
+        print(f"Error loading data: {e}")
+        return
+    
+    # Ensure datetime is in proper format
+    if 'datetime' not in df.columns:
+        if 'datetime_combined' in df.columns:
+            df['datetime'] = pd.to_datetime(df['datetime_combined'])
+        elif 'timestamp' in df.columns:
+            df['datetime'] = pd.to_datetime(df['timestamp'], unit='s')
+        elif 'date' in df.columns and 'time' in df.columns:
+            # Create datetime from date and time columns if needed
+            df['datetime'] = pd.to_datetime(df['date'] + ' ' + df['time'])
+        else:
+            raise ValueError("Could not find datetime column in CSV file")
+    else:
+        # Ensure datetime is in datetime format
+        df['datetime'] = pd.to_datetime(df['datetime'])
+    
+    # Sort by datetime
+    df = df.sort_values(by='datetime')
+    
+    # Add time-based features
+    df['hour_of_day'] = df['datetime'].dt.hour
+    df['day_of_week'] = df['datetime'].dt.dayofweek  # 0 = Monday, 6 = Sunday
+    df['is_weekend'] = df['day_of_week'].apply(lambda x: 1 if x >= 5 else 0)
+    df['is_business_hours'] = df['hour_of_day'].apply(lambda x: 1 if 8 <= x <= 18 else 0)
+    
+    # Calculate rolling features
+    window_size = 3  # Adjust based on sampling frequency
+    df['co2_rolling_mean'] = df['co2'].rolling(window=window_size, min_periods=1).mean()
+    df['temp_rolling_mean'] = df['temperature'].rolling(window=window_size, min_periods=1).mean()
+    df['humidity_rolling_mean'] = df['humidity'].rolling(window=window_size, min_periods=1).mean()
+    
+    # Calculate rate of change for CO2
+    df['co2_change'] = df['co2'].diff().fillna(0)
+    
+    # Get baseline values for each day (early morning values when room is typically empty)
+    df['date_only'] = df['datetime'].dt.date
+    baselines = df[df['hour_of_day'].between(4, 6)].groupby('date_only').agg({
+        'co2': 'mean',
+        'temperature': 'mean',
+        'humidity': 'mean'
+    }).reset_index()
+    
+    baselines.columns = ['date_only', 'co2_baseline', 'temp_baseline', 'humidity_baseline']
+    
+    # Merge baselines back to main dataframe
+    df = pd.merge(df, baselines, on='date_only', how='left')
+    
+    # Fill missing baselines with reasonable defaults
+    df['co2_baseline'] = df['co2_baseline'].fillna(450)  # Typical outdoor CO2
+    df['temp_baseline'] = df['temp_baseline'].fillna(df['temperature'].median())
+    df['humidity_baseline'] = df['humidity_baseline'].fillna(df['humidity'].median())
+    
+    # Calculate differences from baseline
+    df['co2_diff_from_baseline'] = df['co2'] - df['co2_baseline']
+    df['temp_diff_from_baseline'] = df['temperature'] - df['temp_baseline']
+    df['humidity_diff_from_baseline'] = df['humidity'] - df['humidity_baseline']
+    
+    print("Applying occupancy rules...")
+    # Apply rule-based algorithm to determine occupancy
+    df['calculated_occupied'] = df.apply(determine_occupancy_level, axis=1)
+    
+    # Select columns to save
+    output_columns = [col for col in df.columns if col not in [
+        'calculated_occupied', 'date_only', 'is_weekend', 'is_business_hours',
+        'co2_rolling_mean', 'temp_rolling_mean', 'humidity_rolling_mean',
+        'co2_change', 'co2_baseline', 'temp_baseline', 'humidity_baseline',
+        'co2_diff_from_baseline', 'temp_diff_from_baseline', 'humidity_diff_from_baseline'
+    ]]
+    
+    output_df = df[output_columns + ['calculated_occupied']]
+    
+    # Compare with original occupancy if available
+    if 'occupied' in output_df.columns:
+        matches = (output_df['occupied'] == output_df['calculated_occupied']).sum()
+        total = len(output_df)
+        match_percentage = (matches / total) * 100
+        print(f"Calculated occupancy matches original: {matches}/{total} ({match_percentage:.2f}%)")
+        
+        # Confusion matrix
+        from sklearn.metrics import confusion_matrix
+        cm = confusion_matrix(output_df['occupied'], output_df['calculated_occupied'])
+        print("Confusion Matrix:")
+        print(cm)
+    else:
+        print("No original 'occupied' column found. Using only calculated occupancy.")
+    
+    # Save output
+    output_df.to_csv(output_file, index=False)
+    print(f"Output saved to {output_file}")
+
+def determine_occupancy_level(row):
+    """
+    Rule-based algorithm to determine occupancy level:
+    0 - No occupancy
+    1 - Low occupancy (<50% of capacity)
+    2 - Standard occupancy (≥50% of capacity)
+    """
+    # Initialize score
+    score = 0
+    
+    # ----- CO2-based rules (strongest indicator) -----
+    
+    # Absolute CO2 level - adjusted for new occupancy thresholds
+    if row['co2'] >= 1000:
+        score += 7  # Strong indicator of high occupancy (≥50%)
+    elif row['co2'] >= 800:
+        score += 5   # Moderate to high occupancy
+    elif row['co2'] >= 600:
+        score += 3   # Low to moderate occupancy (<50%)
+    elif row['co2'] >= 450:
+        score += 2   # Possible low occupancy
+        
+    # CO2 above baseline - adjusted for new occupancy thresholds
+    if row['co2_diff_from_baseline'] >= 350:  # Higher threshold for ≥50% occupancy
+        score += 5
+    elif row['co2_diff_from_baseline'] >= 200:  # Medium threshold
+        score += 3
+    elif row['co2_diff_from_baseline'] >= 100:  # Lower threshold
+        score += 1
+    
+    # CO2 rate of change
+    if row['co2_change'] >= 25:
+        score += 2   # Fast increase suggests people entering
+    elif row['co2_change'] <= -25:
+        score -= 1   # Fast decrease suggests people leaving
+    
+    # ----- Temperature-based rules -----
+    
+    # Higher temperatures often indicate occupancy
+    if row['temp_diff_from_baseline'] >= 1.5:
+        score += 2
+    elif row['temp_diff_from_baseline'] >= 0.8:
+        score += 1
+    
+    # ----- Humidity-based rules -----
+    
+    # People increase humidity through respiration and transpiration
+    if row['humidity_diff_from_baseline'] >= 10:
+        score += 2
+    elif row['humidity_diff_from_baseline'] >= 5:
+        score += 1
+        
+    # ----- Combined signals -----
+    
+    # Strong signals when multiple factors align
+    if row['co2'] >= 650 and row['temp_diff_from_baseline'] >= 0.5:
+        score += 1
+        
+    if row['co2'] >= 700 and row['humidity_diff_from_baseline'] >= 3:
+        score += 1
+    
+    # ----- Business hours adjustment -----
+    
+    # Lower threshold during business hours when occupancy is expected
+    base_threshold_level1 = 3 if row['is_business_hours'] == 1 else 4
+    base_threshold_level2 = 10 if row['is_business_hours'] == 1 else 11  # Increased threshold for ≥50% occupancy
+    
+    # Weekend adjustment - higher threshold on weekends when occupancy is less common
+    if row['is_weekend'] == 1:
+        base_threshold_level1 += 1
+        base_threshold_level2 += 1
+    
+    # ----- Determine final occupancy level -----
+    
+    if score >= base_threshold_level2:
+        return 2  # Standard occupancy (≥50%)
+    elif score >= base_threshold_level1:
+        return 1  # Low occupancy (<50%)
+    else:
+        return 0  # No occupancy
+
+if __name__ == "__main__":
+    main()
\ No newline at end of file

pipeline/schedule.py

+#!/usr/bin/env python3
+import json
+import pandas as pd
+from pathlib import Path
+import yaml
+import matplotlib.pyplot as plt
+import seaborn as sns
+import numpy as np
+from datetime import datetime, timedelta
+
+# Add this class at the beginning of your file, after the imports
+class NumpyEncoder(json.JSONEncoder):
+    """Custom encoder for numpy data types"""
+    def default(self, obj):
+        if isinstance(obj, np.integer):
+            return int(obj)
+        elif isinstance(obj, np.floating):
+            return float(obj)
+        elif isinstance(obj, np.ndarray):
+            return obj.tolist()
+        elif isinstance(obj, pd.Timestamp):
+            return obj.isoformat()
+        return json.JSONEncoder.default(self, obj)
+
+def generate_schedule():
+    """Generate a heating schedule based on occupancy data."""
+    # Load configuration
+    with open("config.yaml", "r") as f:
+        cfg = yaml.safe_load(f)
+    
+    artifacts_dir = Path(cfg["data"]["artifacts_dir"])
+    artifacts_dir.mkdir(exist_ok=True)
+    
+    # Load occupancy levels data
+    try:
+        levels_df = pd.read_parquet(artifacts_dir / "levels_2h.parquet")
+        levels_df["block_start"] = pd.to_datetime(levels_df["block_start"], utc=True)
+        print(f"Loaded {len(levels_df)} level records")
+    except Exception as e:
+        print(f"Error loading levels data: {e}")
+        return False
+    
+    # Determine current week
+    current_date = datetime.now()
+    start_of_week = current_date - timedelta(days=current_date.weekday())
+    week_id = f"{start_of_week.year}-W{start_of_week.isocalendar()[1]:02d}"
+    
+    # Filter data for recent weeks if needed
+    # This would be more sophisticated in production
+    
+    # Get unique rooms
+    rooms = sorted(levels_df["room_id"].unique())
+    
+    # Create schedule structure
+    schedule = {
+        "week": week_id,
+        "entries": {}
+    }
+    
+    # Process each room
+    for room in rooms:
+        room_id = str(room)
+        room_data = levels_df[levels_df["room_id"] == room].copy()
+        
+        if room_data.empty:
+            continue
+            
+        # Find periods of high occupancy (level 2)
+        # Group consecutive blocks with high occupancy
+        room_data = room_data.sort_values("block_start")
+        room_data["high_occ"] = room_data["level"] >= 2  # Consider level 2 as high occupancy
+        
+        # Group by date to process each day separately
+        room_data["date"] = room_data["block_start"].dt.date
+        schedule["entries"][room_id] = []
+        
+        for date, day_data in room_data.groupby("date"):
+            # Skip days with no high occupancy
+            if not day_data["high_occ"].any():
+                continue
+                
+            # Identify blocks with high occupancy
+            day_data["group"] = (day_data["high_occ"] != day_data["high_occ"].shift()).cumsum()
+            high_groups = day_data[day_data["high_occ"]].groupby("group")
+            
+            for _, group in high_groups:
+                if len(group) > 0:
+                    # Create a comfort period starting 1 hour before and ending 30 min after
+                    # the high occupancy period
+                    start_time = group["block_start"].min() - pd.Timedelta(hours=1)
+                    end_time = group["block_start"].max() + pd.Timedelta(hours=2.5)  # 2h block + 30min
+                    
+                    # Add probability based on consistency of occupancy
+                    prob = min(0.95, 0.7 + (len(group) / 10))  # Scale probability based on # of blocks
+                    
+                    schedule["entries"][room_id].append({
+                        "start": start_time.isoformat(),
+                        "end": end_time.isoformat(),
+                        "setpoint": cfg["schedule"]["comfort_setpoint"],
+                        "probability": round(prob, 2)
+                    })
+    
+    # Save the generated schedule
+    with open(artifacts_dir / "schedule_week.json", "w") as f:
+        json.dump(schedule, f, indent=2, cls=NumpyEncoder)
+    
+    print(f"Generated schedule saved to {artifacts_dir}/schedule_week.json")
+    return True
+
+def enhance_schedule():
+    """Enhance the heating schedule output with descriptive information and visualizations."""
+    
+    # Generate the schedule first
+    if not generate_schedule():
+        print("Failed to generate schedule. Exiting enhancement process.")
+        return
+    
+    # Load configuration
+    with open("config.yaml", "r") as f:
+        cfg = yaml.safe_load(f)
+    
+    artifacts_dir = Path(cfg["data"]["artifacts_dir"])
+    setback_temp = cfg["schedule"]["setback_setpoint"]
+    comfort_temp = cfg["schedule"]["comfort_setpoint"]
+    
+    # Load the schedule
+    with open(artifacts_dir / "schedule_week.json", "r") as f:
+        schedule = json.load(f)
+    
+    week = schedule["week"]
+    entries = schedule["entries"]
+    
+    # Create enhanced schedule output
+    enhanced = {
+        "week": week,
+        "summary": {
+            "total_rooms": len(entries),
+            "date_range": "",  # Will be filled below
+            "energy_saving_hours": 0,
+            "comfort_hours": 0,
+            "energy_saving_percentage": 0,
+            "estimated_energy_savings": 0
+        },
+        "rooms": {}
+    }
+    
+    # Process each room
+    all_start_times = []
+    all_end_times = []
+    total_comfort_hours = 0
+    
+    for room_id, room_entries in entries.items():
+        # Convert to DataFrame for easier analysis
+        if not room_entries:
+            continue
+            
+        df = pd.DataFrame(room_entries)
+        df["start"] = pd.to_datetime(df["start"])
+        df["end"] = pd.to_datetime(df["end"])
+        df["duration_hours"] = (df["end"] - df["start"]).dt.total_seconds() / 3600
+        
+        # Record all times for overall analysis
+        all_start_times.extend(df["start"].tolist())
+        all_end_times.extend(df["end"].tolist())
+        
+        # Calculate comfort hours
+        total_comfort_hours_room = df["duration_hours"].sum()
+        total_comfort_hours += total_comfort_hours_room
+        
+        # Calculate daily pattern
+        df["day"] = df["start"].dt.day_name()
+        by_day = df.groupby("day")["duration_hours"].sum().to_dict()
+        
+        # Find peak usage time
+        df["hour"] = df["start"].dt.hour
+        peak_hour = df.groupby("hour")["duration_hours"].sum().idxmax()
+        
+        # Find gaps (times when setback temp is used)
+        df = df.sort_values("start")
+        
+        # Calculate room-specific metrics for energy analysis
+        # Instead of assuming a week, let's calculate the actual timespan
+        if df["start"].min() and df["end"].max():
+            schedule_start = df["start"].min()
+            schedule_end = df["end"].max()
+            # Calculate total hours in the schedule timespan
+            room_total_hours = (schedule_end - schedule_start).total_seconds() / 3600
+        else:
+            room_total_hours = 24 * 7  # Default to one week
+            
+        # Calculate comfort vs. setback hours
+        room_comfort_hours = total_comfort_hours_room
+        room_setback_hours = room_total_hours - room_comfort_hours
+        
+        # Calculate energy savings correctly for each room
+        assumed_outside_temp = 5  # °C
+        # Baseline: everything at comfort temperature
+        baseline_energy_room = room_total_hours * (comfort_temp - assumed_outside_temp)
+        # Optimized: mix of comfort and setback temperatures
+        optimized_energy_room = room_comfort_hours * (comfort_temp - assumed_outside_temp) + \
+                               room_setback_hours * (setback_temp - assumed_outside_temp)
+        
+        energy_savings_room = baseline_energy_room - optimized_energy_room
+        room_energy_saving_pct = (energy_savings_room / baseline_energy_room) * 100 if baseline_energy_room > 0 else 0
+        
+        # Add room summary
+        enhanced["rooms"][room_id] = {
+            "comfort_periods_count": len(room_entries),
+            "comfort_hours": round(total_comfort_hours_room, 1),
+            "setback_hours": round(room_setback_hours, 1),
+            "energy_saving_percentage": round(room_energy_saving_pct, 1),
+            "peak_usage_hour": peak_hour,
+            "daily_pattern": by_day,
+            "recommended_actions": [
+                f"Maintain {comfort_temp}°C during scheduled comfort periods",
+                f"Reduce to {setback_temp}°C during all other times",
+                "Consider scheduling maintenance during consistently unoccupied periods"
+            ],
+            "schedule": room_entries  # Keep original schedule
+        }
+    
+    # Rest of your enhance_schedule function remains the same
+    # Calculate overall date range
+    if all_start_times and all_end_times:
+        min_date = min(all_start_times).strftime("%Y-%m-%d")
+        max_date = max(all_end_times).strftime("%Y-%m-%d")
+        enhanced["summary"]["date_range"] = f"{min_date} to {max_date}"
+    
+    # Calculate overall metrics based on actual schedule period
+    # Find the overall span of the schedules
+    if all_start_times and all_end_times:
+        overall_start = min(all_start_times)
+        overall_end = max(all_end_times)
+        # Calculate total hours across all rooms in the schedule period
+        total_hours = (overall_end - overall_start).total_seconds() / 3600 * len(entries)
+    else:
+        total_hours = 24 * 7 * len(entries)  # Default to one week per room
+        
+    # Calculate comfort vs. setback hours
+    total_setback_hours = total_hours - total_comfort_hours
+    
+    # Calculate energy savings using the correct approach (similar to evaluate.py)
+    assumed_outside_temp = 5  # °C
+    # Baseline: everything at comfort temperature
+    baseline_energy = total_hours * (comfort_temp - assumed_outside_temp)
+    # Optimized: mix of comfort and setback temperatures
+    optimized_energy = total_comfort_hours * (comfort_temp - assumed_outside_temp) + \
+                       total_setback_hours * (setback_temp - assumed_outside_temp)
+    
+    energy_savings = baseline_energy - optimized_energy
+    energy_saving_pct = (energy_savings / baseline_energy) * 100 if baseline_energy > 0 else 0
+    
+    # Estimate energy savings (simplified model)
+    estimated_energy_savings = energy_savings * 0.1  # kWh per degree-hour (simplified)
+    
+    # Update summary
+    enhanced["summary"]["comfort_hours"] = round(total_comfort_hours, 1)
+    enhanced["summary"]["energy_saving_hours"] = round(total_setback_hours, 1)
+    enhanced["summary"]["energy_saving_percentage"] = round(energy_saving_pct, 1)
+    enhanced["summary"]["estimated_energy_savings"] = round(estimated_energy_savings, 1)
+    
+    # Add recommended actions to the overall summary
+    enhanced["summary"]["recommendations"] = [
+        f"Implementation of this schedule provides {round(energy_saving_pct, 1)}% energy savings potential",
+        f"Expected to save approximately {round(estimated_energy_savings, 1)} kWh per week",
+        "Consider additional savings by reviewing rooms with low occupancy rates",
+        "Monitor thermal comfort feedback during the first week of implementation"
+    ]
+    
+    # Create visualization directory
+    vis_dir = artifacts_dir / "visualizations"
+    vis_dir.mkdir(exist_ok=True)
+    
+    # Create weekly schedule visualization for each room
+    for room_id, room_entries in entries.items():
+        if not room_entries:
+            continue
+            
+        df = pd.DataFrame(room_entries)
+        df["start"] = pd.to_datetime(df["start"])
+        df["end"] = pd.to_datetime(df["end"])
+        df["day"] = df["start"].dt.day_name()
+        df["hour"] = df["start"].dt.hour
+        
+        # Create a 24x7 grid (hours x days)
+        plt.figure(figsize=(12, 6))
+        
+        # Create a blank schedule grid
+        schedule_grid = pd.DataFrame(0, 
+                                   index=range(24), 
+                                   columns=["Monday", "Tuesday", "Wednesday", 
+                                            "Thursday", "Friday", "Saturday", "Sunday"])
+        
+        # Fill in comfort periods
+        for _, row in df.iterrows():
+            day = row["day"]
+            start_hour = row["start"].hour
+            end_hour = row["end"].hour if row["end"].hour > start_hour else 24
+            
+            for h in range(start_hour, end_hour):
+                if h < 24:  # Stay within bounds
+                    schedule_grid.at[h, day] = 1
+        
+        # Plot heatmap
+        sns.heatmap(schedule_grid, cmap=["lightblue", "red"], 
+                   cbar_kws={'label': 'Temperature Mode'},
+                   xticklabels=schedule_grid.columns,
+                   yticklabels=[f"{h:02d}:00" for h in range(24)])
+                   
+        plt.title(f"Room {room_id} - Weekly Heating Schedule")
+        plt.ylabel("Hour of Day")
+        plt.xlabel("Day of Week")
+        
+        # Add custom color bar labels
+        colorbar = plt.gcf().axes[-1]
+        ticks = colorbar.get_yticks()
+        if len(ticks) >= 2:
+            # Use first and last tick positions
+            colorbar.set_yticks([ticks[0], ticks[-1]])
+            colorbar.set_yticklabels([f"{setback_temp}°C (Setback)", f"{comfort_temp}°C (Comfort)"])
+        
+        # Save visualization
+        plt.tight_layout()
+        plt.savefig(vis_dir / f"schedule_room_{room_id}.png", dpi=300)
+        plt.close()
+        
+        # Add visualization path to enhanced output
+        enhanced["rooms"][room_id]["visualization"] = f"visualizations/schedule_room_{room_id}.png"
+    
+    # Save enhanced schedule
+    with open(artifacts_dir / "enhanced_schedule.json", "w") as f:
+        json.dump(enhanced, f, indent=2, cls=NumpyEncoder)
+    
+    print(f"Enhanced schedule saved to {artifacts_dir}/enhanced_schedule.json")
+    print(f"Visualizations saved to {vis_dir}")
+    
+    # Create a human-readable summary report
+    report = f"""# Heating Schedule Summary for Week {week}
+
+## Overview
+- **Date Range:** {enhanced["summary"]["date_range"]}
+- **Total Rooms:** {enhanced["summary"]["total_rooms"]}
+- **Energy Saving Potential:** {enhanced["summary"]["energy_saving_percentage"]}%
+- **Estimated Energy Savings:** {enhanced["summary"]["estimated_energy_savings"]} kWh per week
+- **Comfort Hours:** {enhanced["summary"]["comfort_hours"]} hours (total across all rooms)
+- **Setback Hours:** {enhanced["summary"]["energy_saving_hours"]} hours (energy saving mode)
+
+## Recommendations
+"""
+
+    for rec in enhanced["summary"]["recommendations"]:
+        report += f"- {rec}\n"
+        
+    report += "\n## Room Details\n\n"
+    
+    for room_id, room_data in enhanced["rooms"].items():
+        report += f"### Room {room_id}\n"
+        report += f"- **Comfort Periods:** {room_data['comfort_periods_count']}\n"
+        report += f"- **Energy Saving:** {room_data['energy_saving_percentage']}% ({room_data['setback_hours']} hours at {setback_temp}°C)\n"
+        report += f"- **Peak Usage Hour:** {room_data['peak_usage_hour']:02d}:00\n"
+        report += f"- **Daily Pattern:** "
+        
+        for day, hours in room_data['daily_pattern'].items():
+            report += f"{day[:3]}:{round(hours,1)}h "
+            
+        report += "\n\n**Actions:**\n"
+        
+        for action in room_data['recommended_actions']:
+            report += f"- {action}\n"
+            
+        report += f"\n![Room {room_id} Schedule](visualizations/schedule_room_{room_id}.png)\n\n"
+        report += "---\n\n"
+    
+    # Save the report
+    with open(artifacts_dir / "heating_schedule_report.md", "w") as f:
+        f.write(report)
+        
+    print(f"Human-readable report saved to {artifacts_dir}/heating_schedule_report.md")
+
+if __name__ == "__main__":
+    enhance_schedule()
\ No newline at end of file

pipeline/train.py

+#!/usr/bin/env python3
+import json, argparse, logging, time
+from pathlib import Path
+import pandas as pd
+import numpy as np
+from sklearn.model_selection import train_test_split
+import yaml
+from sklearn.linear_model import LogisticRegression
+from sklearn.metrics import f1_score, confusion_matrix, classification_report, precision_score, recall_score
+import joblib
+from sklearn.preprocessing import StandardScaler
+from sklearn.pipeline import Pipeline
+
+# Set up logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    handlers=[
+        logging.FileHandler("train.log"),
+        logging.StreamHandler()
+    ]
+)
+logger = logging.getLogger("model_training")
+
+def load_config():
+    logger.info("Loading configuration from config.yaml")
+    with open("config.yaml", "r") as f:
+        cfg = yaml.safe_load(f)
+        logger.info(f"Model configuration: class_weight={cfg['model']['class_weight']}, "
+                   f"test_fraction={cfg['model']['test_fraction_time']}")
+        return cfg
+
+def split_time(df, test_frac):
+    logger.info(f"Splitting data using time-based approach (test fraction: {test_frac})")
+    t = pd.to_datetime(df["timestamp"], utc=True)
+    cutoff = t.quantile(1 - test_frac)
+    logger.info(f"Time cutoff for train/test split: {cutoff}")
+    
+    train = df[t <= cutoff]
+    test = df[t > cutoff]
+    
+    logger.info(f"Training period: {t[t <= cutoff].min()} to {t[t <= cutoff].max()}")
+    logger.info(f"Testing period: {t[t > cutoff].min()} to {t[t > cutoff].max()}")
+    logger.info(f"Split result: {len(train)} training samples, {len(test)} test samples")
+    
+    return train, test
+
+# Replace the existing split_time function with this random split function
+def split_random(df, test_frac, random_state=42):
+    """
+    Split data randomly instead of by time
+    
+    Parameters:
+    - df: DataFrame with data
+    - test_frac: Fraction for test set (e.g., 0.25)
+    - random_state: Random seed for reproducibility
+    
+    Returns:
+    - train, test DataFrames
+    """
+    logger.info(f"Splitting data using random approach (test fraction: {test_frac}, random_state: {random_state})")
+    
+    train_df, test_df = train_test_split(
+        df, 
+        test_size=test_frac, 
+        random_state=random_state,
+        shuffle=True
+    )
+    
+    logger.info(f"Split result: {len(train_df)} training samples ({(1-test_frac)*100:.1f}%), {len(test_df)} test samples ({test_frac*100:.1f}%)")
+    
+    # Log some statistics to verify the split is representative
+    for col in ["hour", "dow"]:
+        if col in df.columns:
+            train_dist = train_df[col].value_counts(normalize=True).to_dict()
+            test_dist = test_df[col].value_counts(normalize=True).to_dict()
+            logger.info(f"Distribution of {col}: similar in train/test = {all(abs(train_dist.get(k, 0) - test_dist.get(k, 0)) < 0.05 for k in set(train_dist) | set(test_dist))}")
+    
+    # Log label distribution to check for stratification
+    label_col = "occupied"  # Adjust as needed
+    if label_col in df.columns:
+        train_labels = train_df[label_col].value_counts(normalize=True).to_dict()
+        test_labels = test_df[label_col].value_counts(normalize=True).to_dict()
+        logger.info(f"Label distribution - Training: {train_labels}")
+        logger.info(f"Label distribution - Testing: {test_labels}")
+    
+    return train_df, test_df
+
+def select_xy(df, cfg):
+    ycol = cfg["data"]["label_column"]
+    room_id_col = cfg["data"]["room_id_column"]
+    
+    logger.info(f"Selecting features and target. Label column: {ycol}, ID column: {room_id_col}")
+    
+    Xcols = []
+    for c in df.columns:
+        if c in ["timestamp", room_id_col, ycol]:
+            continue
+        if pd.api.types.is_numeric_dtype(df[c]):
+            Xcols.append(c)
+    
+    logger.info(f"Selected {len(Xcols)} numeric features")
+    
+    # Log feature information
+    sensor_features = {sensor: [col for col in Xcols if sensor in col] for sensor in cfg["data"]["sensors"]}
+    for sensor, cols in sensor_features.items():
+        if cols:
+            logger.info(f"  - {sensor}: {len(cols)} features")
+    
+    # Log label distribution
+    label_counts = df[ycol].value_counts().to_dict()
+    logger.info(f"Label distribution: {label_counts}")
+    
+    X = df[Xcols].fillna(0.0).values
+    y = df[ycol].values
+    
+    # Log basic statistics about features
+    logger.info(f"Feature matrix shape: {X.shape}")
+    logger.info(f"Missing values filled: {df[Xcols].isna().sum().sum()}")
+    logger.info(f"Feature value ranges: min={X.min():.2f}, max={X.max():.2f}, mean={X.mean():.2f}")
+    
+    return X, y, Xcols
+
+def main():
+    logger.info("=== MODEL TRAINING PROCESS STARTED ===")
+    start_time = time.time()
+    
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--input", type=str, default=None, 
+                       help="Custom input filename (default: feature_store_train.parquet or feature_store.parquet)")
+    args = parser.parse_args()
+    
+    try:
+        cfg = load_config()
+        artifacts_dir = Path(cfg["data"]["artifacts_dir"])
+        artifacts_dir.mkdir(parents=True, exist_ok=True)
+        
+        # Load feature data - try train-specific file first, then fallback to general one
+        if args.input:
+            feature_path = artifacts_dir / args.input
+        else:
+            feature_path = artifacts_dir / "feature_store_train.parquet"
+            if not feature_path.exists():
+                fallback_path = artifacts_dir / "feature_store.parquet"
+                if fallback_path.exists():
+                    logger.warning(f"{feature_path} not found, falling back to {fallback_path}")
+                    feature_path = fallback_path
+                    
+        logger.info(f"Loading features from {feature_path}")
+        feat = pd.read_parquet(feature_path)
+        initial_rows = len(feat)
+        
+        # Check for missing labels
+        label_col = cfg["data"]["label_column"]
+        missing_labels = feat[label_col].isna().sum()
+        logger.info(f"Initial data: {initial_rows} rows, {len(feat.columns)} columns")
+        logger.info(f"Missing labels: {missing_labels} ({missing_labels/initial_rows*100:.1f}%)")
+        
+        # Drop rows with missing labels
+        feat = feat.dropna(subset=[label_col])
+        logger.info(f"After dropping missing labels: {len(feat)} rows remaining")
+        
+        # # Split data
+        # train_df, test_df = split_time(feat, cfg["model"]["test_fraction_time"])
+        # For random split, uncomment the following line and comment the above line
+        train_df, test_df = split_random(
+            feat,
+            cfg["model"]["test_fraction_time"],
+            random_state=cfg["model"].get("random_state", 42)
+        )
+        
+        # Prepare features and labels
+        logger.info("Preparing training features and labels")
+        Xtr, ytr, cols = select_xy(train_df, cfg)
+        logger.info("Preparing test features and labels")
+        Xte, yte, _ = select_xy(test_df, cfg)
+        
+        # Create and train model pipeline
+        logger.info("Creating model pipeline with StandardScaler and LogisticRegression")
+        pipeline = Pipeline([
+            ('scaler', StandardScaler()),
+            ('classifier', LogisticRegression(
+                multi_class="ovr",
+                max_iter=25000,
+                class_weight=cfg["model"]["class_weight"],
+                random_state=cfg["model"]["random_state"]
+            ))
+        ])
+        
+        logger.info(f"Training model with {len(Xtr)} samples")
+        training_start = time.time()
+        pipeline.fit(Xtr, ytr)
+        training_time = time.time() - training_start
+        logger.info(f"Model training completed in {training_time:.2f} seconds")
+        
+        # Evaluate model
+        logger.info("Evaluating model on test data")
+        ypred = pipeline.predict(Xte)
+        
+        # Calculate metrics
+        f1 = f1_score(yte, ypred, average="macro")
+        precision = precision_score(yte, ypred, average="macro")
+        recall = recall_score(yte, ypred, average="macro")
+        cm = confusion_matrix(yte, ypred).tolist()
+        report = classification_report(yte, ypred, output_dict=True)
+        
+        logger.info(f"Model performance:")
+        logger.info(f"  - Macro F1 Score: {f1:.4f}")
+        logger.info(f"  - Macro Precision: {precision:.4f}")
+        logger.info(f"  - Macro Recall: {recall:.4f}")
+        
+        # Log confusion matrix
+        logger.info("Confusion matrix:")
+        cm_str = "\n".join([str(row) for row in cm])
+        logger.info(f"\n{cm_str}")
+        
+        # Log per-class metrics
+        logger.info("Per-class performance:")
+        for cls, metrics in report.items():
+            if isinstance(metrics, dict):
+                logger.info(f"  - Class {cls}: precision={metrics['precision']:.4f}, "
+                           f"recall={metrics['recall']:.4f}, f1={metrics['f1-score']:.4f}")
+        
+        # Save model and metrics
+        logger.info(f"Saving model to {artifacts_dir / 'model.pkl'}")
+        model_info = {
+            "model": pipeline, 
+            "features": cols,
+            "training_samples": len(Xtr),
+            "test_samples": len(Xte),
+            "feature_count": len(cols)
+        }
+        joblib.dump(model_info, artifacts_dir / "model.pkl")
+        
+        logger.info(f"Saving metrics to {artifacts_dir / 'metrics.json'}")
+        metrics = {
+            "macro_f1": f1,
+            "macro_precision": precision,
+            "macro_recall": recall, 
+            "confusion_matrix": cm,
+            "report": report,
+            "training_time_seconds": training_time
+        }
+        with open(artifacts_dir / "metrics.json", "w") as f:
+            json.dump(metrics, f, indent=2)
+        
+        # Feature importance if available
+        try:
+            if hasattr(pipeline['classifier'], 'coef_'):
+                # Get feature importance
+                coefs = pipeline['classifier'].coef_
+                if len(coefs.shape) > 1 and coefs.shape[0] > 1:
+                    # For multi-class, use the norm of coefficients
+                    importance = np.sqrt((coefs ** 2).sum(axis=0))
+                else:
+                    importance = np.abs(coefs.ravel())
+                
+                # Get top 10 features
+                top_indices = importance.argsort()[-10:][::-1]
+                top_features = [(cols[i], importance[i]) for i in top_indices]
+                
+                logger.info("Top 10 most important features:")
+                for feature, imp in top_features:
+                    logger.info(f"  - {feature}: {imp:.4f}")
+        except Exception as e:
+            logger.warning(f"Could not extract feature importance: {e}")
+        
+        # Final summary
+        total_time = time.time() - start_time
+        logger.info(f"Total processing time: {total_time:.2f} seconds")
+        print(json.dumps({
+            "macro_f1": f1,
+            "n_train": len(train_df),
+            "n_test": len(test_df),
+            "training_time": round(training_time, 2),
+            "total_time": round(total_time, 2)
+        }, indent=2))
+        
+        logger.info("=== MODEL TRAINING COMPLETED SUCCESSFULLY ===")
+        
+    except Exception as e:
+        logger.error(f"Model training failed: {e}", exc_info=True)
+        raise
+
+if __name__ == "__main__":
+    main()
\ No newline at end of file

run_pipeline.ps1

+#!/usr/bin/env pwsh
+# Run the complete pipeline with different datasets for training and prediction
+# This script demonstrates how to use the updated pipeline with different datasets
+
+$ErrorActionPreference = "Stop"
+$scriptDir = Split-Path -Parent $MyInvocation.MyCommand.Path
+Set-Location $scriptDir
+
+Write-Host "Starting pipeline with different datasets for training and prediction..." -ForegroundColor Green
+
+# Step 1: Data Ingestion - Training
+Write-Host "Step 1a: Ingesting training data..." -ForegroundColor Yellow
+python pipeline/ingest.py --phase train
+if ($LASTEXITCODE -ne 0) { Write-Error "Training data ingestion failed"; exit $LASTEXITCODE }
+
+# Step 1b: Data Ingestion - Prediction
+Write-Host "Step 1b: Ingesting prediction data..." -ForegroundColor Yellow
+python pipeline/ingest.py --phase predict
+if ($LASTEXITCODE -ne 0) { Write-Error "Prediction data ingestion failed"; exit $LASTEXITCODE }
+
+# Step 2: Feature Engineering - Training
+Write-Host "Step 2a: Feature engineering for training data..." -ForegroundColor Yellow
+python pipeline/features.py --phase train
+if ($LASTEXITCODE -ne 0) { Write-Error "Training feature engineering failed"; exit $LASTEXITCODE }
+
+# Step 2b: Feature Engineering - Prediction
+Write-Host "Step 2b: Feature engineering for prediction data..." -ForegroundColor Yellow
+python pipeline/features.py --phase predict
+if ($LASTEXITCODE -ne 0) { Write-Error "Prediction feature engineering failed"; exit $LASTEXITCODE }
+
+# Step 3: Model Training (using training data)
+Write-Host "Step 3: Training model..." -ForegroundColor Yellow
+python pipeline/train.py
+if ($LASTEXITCODE -ne 0) { Write-Error "Model training failed"; exit $LASTEXITCODE }
+
+# Step 4: Prediction (using prediction data)
+Write-Host "Step 4: Making predictions..." -ForegroundColor Yellow
+python pipeline/predict.py
+if ($LASTEXITCODE -ne 0) { Write-Error "Prediction failed"; exit $LASTEXITCODE }
+
+# Step 5: Schedule Generation
+Write-Host "Step 5: Generating schedule..." -ForegroundColor Yellow
+python pipeline/schedule.py
+if ($LASTEXITCODE -ne 0) { Write-Error "Schedule generation failed"; exit $LASTEXITCODE }
+
+# Step 6: Energy Optimization
+Write-Host "Step 6: Optimizing energy..." -ForegroundColor Yellow
+python pipeline/optimize_energy.py
+if ($LASTEXITCODE -ne 0) { Write-Error "Energy optimization failed"; exit $LASTEXITCODE }
+
+# Step 7: Evaluation
+Write-Host "Step 7: Evaluating results..." -ForegroundColor Yellow
+python pipeline/evaluate.py
+if ($LASTEXITCODE -ne 0) { Write-Error "Evaluation failed"; exit $LASTEXITCODE }
+
+Write-Host "Pipeline completed successfully!" -ForegroundColor Green
\ No newline at end of file