from fastapi import FastAPI, HTTPException from fastapi.responses import HTMLResponse, JSONResponse from pydantic import BaseModel from typing import List import numpy as np import os import joblib from datetime import datetime app = FastAPI(title="York Chiller Energy Optimizer") # ============================================ # LOAD YOUR TRAINED MODEL (if exists) # ============================================ model = None feature_cols = None MODEL_READY = False try: if os.path.exists("retrained_best_model.pkl"): model = joblib.load("retrained_best_model.pkl") feature_cols = joblib.load("retrained_feature_cols.pkl") MODEL_READY = True print(f"✅ Model loaded: {type(model).__name__}") print(f"✅ Features: {len(feature_cols)}") else: print("⚠️ Model files not found, using fallback mode") except Exception as e: print(f"⚠️ Could not load model: {e}") # ============================================ # FEATURE ENGINEERING (if model exists) # ============================================ def create_features(hour: int, minute: int, day_of_week: int, month: int, recent_power: List[float]) -> dict: """Create features for model prediction""" features = {} # Time features features['hour'] = hour features['minute'] = minute features['dayofweek'] = day_of_week features['day'] = 1 features['month'] = month features['year'] = 2025 # Cyclical features features['sin_hour'] = np.sin(2 * np.pi * hour / 24) features['cos_hour'] = np.cos(2 * np.pi * hour / 24) features['sin_month'] = np.sin(2 * np.pi * month / 12) features['cos_month'] = np.cos(2 * np.pi * month / 12) # Binary features features['is_weekend'] = 1 if day_of_week >= 5 else 0 features['is_business_hours'] = 1 if 8 <= hour <= 18 else 0 # Pad power history if needed if len(recent_power) < 150: last_val = recent_power[-1] if recent_power else 50 recent_power = recent_power + [last_val] * (150 - len(recent_power)) # Lag features lags = [1, 2, 3, 6, 12, 24, 48, 96, 144] for lag in lags: if len(recent_power) > lag: features[f'power_lag_{lag}'] = float(recent_power[-lag]) else: features[f'power_lag_{lag}'] = float(recent_power[-1]) if recent_power else 50.0 # Rolling statistics windows = [6, 12, 24, 48, 96, 144] for window in windows: if len(recent_power) >= window: window_data = recent_power[-window:] features[f'power_rolling_mean_{window}'] = float(np.mean(window_data)) features[f'power_rolling_std_{window}'] = float(np.std(window_data)) features[f'power_rolling_min_{window}'] = float(np.min(window_data)) features[f'power_rolling_max_{window}'] = float(np.max(window_data)) else: default = float(recent_power[-1]) if recent_power else 50.0 features[f'power_rolling_mean_{window}'] = default features[f'power_rolling_std_{window}'] = 10.0 features[f'power_rolling_min_{window}'] = default * 0.8 features[f'power_rolling_max_{window}'] = default * 1.2 # Difference features if len(recent_power) >= 2: features['power_change_1min'] = float(recent_power[-1] - recent_power[-2]) features['power_change_5min'] = float(recent_power[-1] - recent_power[-6]) if len(recent_power) >= 6 else 0 features['power_change_15min'] = float(recent_power[-1] - recent_power[-16]) if len(recent_power) >= 16 else 0 else: features['power_change_1min'] = 0.0 features['power_change_5min'] = 0.0 features['power_change_15min'] = 0.0 features['power_acceleration'] = 0.0 features['power_ratio_to_mean_24'] = 1.0 return features # ============================================ # PREDICTION FUNCTION # ============================================ def predict_power(hour: int, minute: int, day_of_week: int, month: int, recent_power: List[float]) -> float: """Predict power using model or fallback""" if MODEL_READY and feature_cols: try: features = create_features(hour, minute, day_of_week, month, recent_power) input_array = [] for feat in feature_cols: input_array.append(features.get(feat, 0.0)) X_input = np.array(input_array).reshape(1, -1) prediction = float(model.predict(X_input)[0]) return max(0, min(500, prediction)) except Exception as e: print(f"Model prediction error: {e}") # Fallback prediction if 8 <= hour <= 18: base = 120 + 30 * np.sin((hour - 12) * np.pi / 12) elif hour < 6 or hour > 22: base = 30 else: base = 70 if len(recent_power) >= 6: trend = np.mean(recent_power[-3:]) - np.mean(recent_power[-6:-3]) base += trend * 0.3 return max(10, min(400, base)) # ============================================ # OPTIMIZATION RECOMMENDATIONS # ============================================ def get_recommendations(predicted_power: float, hour: int, day_of_week: int) -> List[dict]: recommendations = [] is_weekend = day_of_week >= 5 # Chiller staging if predicted_power < 45: recommendations.append({ "variable": "Chiller Staging", "current": "2+ chillers running", "recommended": "🔴 TURN OFF all chillers", "saving_kw": round(predicted_power * 0.3, 1), "priority": "HIGH" }) elif predicted_power < 120: recommendations.append({ "variable": "Chiller Staging", "current": "2+ chillers running", "recommended": "🟢 RUN 1 chiller", "saving_kw": round(predicted_power * 0.12, 1), "priority": "HIGH" }) elif predicted_power < 250: recommendations.append({ "variable": "Chiller Staging", "current": "1 chiller (overloaded)", "recommended": "🟡 RUN 2 chillers", "saving_kw": round(predicted_power * 0.08, 1), "priority": "MEDIUM" }) else: recommendations.append({ "variable": "Chiller Staging", "current": "2 chillers (overloaded)", "recommended": "🟡 RUN 3+ chillers", "saving_kw": round(predicted_power * 0.05, 1), "priority": "MEDIUM" }) # Pump speed (VFD) optimal_speed = 35 + (predicted_power / 300) * 50 optimal_speed = min(90, max(30, optimal_speed)) if optimal_speed < 60: action = f"🔵 REDUCE to {optimal_speed:.0f}% speed" saving = 8 else: action = f"🔵 INCREASE to {optimal_speed:.0f}% speed" saving = 5 recommendations.append({ "variable": "Chilled Water Pump (VFD)", "current": "85% speed", "recommended": action, "saving_kw": saving, "priority": "HIGH" if abs(optimal_speed - 85) > 20 else "MEDIUM" }) # Cooling tower fans if predicted_power < 50: recommendations.append({ "variable": "Cooling Tower Fans", "current": "100% speed", "recommended": "🔴 TURN OFF", "saving_kw": 12, "priority": "HIGH" }) elif predicted_power < 120: recommendations.append({ "variable": "Cooling Tower Fans", "current": "100% speed", "recommended": "🟡 RUN at 40% speed", "saving_kw": 7, "priority": "MEDIUM" }) else: recommendations.append({ "variable": "Cooling Tower Fans", "current": "Fixed speed", "recommended": "🟡 RUN at 65-85% speed", "saving_kw": 4, "priority": "MEDIUM" }) # Schedule optimization if (hour < 6 or hour > 22 or is_weekend) and predicted_power < 60: recommendations.append({ "variable": "Equipment Schedule", "current": "Running", "recommended": "🔴 CONSIDER SHUTDOWN", "saving_kw": round(predicted_power, 1), "priority": "HIGH" }) return recommendations # ============================================ # API REQUEST MODEL # ============================================ class PredictionRequest(BaseModel): hour: int minute: int day_of_week: int month: int recent_power: List[float] # ============================================ # EMBEDDED HTML DASHBOARD # ============================================ HTML_DASHBOARD = """ York Chiller Energy Optimizer

❄️ York Chiller Energy Optimizer

AI-powered prediction and optimization for York chiller systems

Checking API...
Loading...

Input Parameters

Last 10-30 readings at 1-minute intervals (comma-separated)

Powered by Random Forest AI

""" # ============================================ # FASTAPI ENDPOINTS # ============================================ @app.get("/") async def root(): """Serve the dashboard""" return HTMLResponse(content=HTML_DASHBOARD) @app.post("/api/predict") async def predict(request: PredictionRequest): """Make prediction and return JSON""" try: predicted_power = predict_power( request.hour, request.minute, request.day_of_week, request.month, request.recent_power ) recommendations = get_recommendations(predicted_power, request.hour, request.day_of_week) total_saving = sum(r["saving_kw"] for r in recommendations) return { "success": True, "predicted_power_kw": round(predicted_power, 1), "total_saving_kw": round(total_saving, 1), "recommendations": recommendations, "model_info": { "loaded": MODEL_READY, "type": "Random Forest" if MODEL_READY else "Fallback Predictor", "features": len(feature_cols) if feature_cols else 0 } } except Exception as e: return {"success": False, "error": str(e)} @app.get("/health") async def health(): """Health check""" return { "status": "healthy", "model_type": "Random Forest" if MODEL_READY else "Fallback", "timestamp": datetime.now().isoformat() } if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)