# ============================================ # YORK CHILLER OPTIMIZER API # Compatible with NumPy 2.0.2 and pandas 2.2.3 # For 4-Chiller Plants (1000+ TR) # ============================================ import os import sys import warnings warnings.filterwarnings('ignore') # Import libraries import numpy as np print(f"NumPy version: {np.__version__}") import pandas as pd print(f"Pandas version: {pd.__version__}") from datetime import datetime from typing import List, Optional, Dict, Any from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field import joblib import pickle app = FastAPI( title="York Chiller Energy Optimizer - 4 Chiller Plant Edition", description="Random Forest Model for Chiller Energy Efficiency - 12 Features (1000+ TR Plants)", version="2.0.0" ) # ============================================ # LOAD MODEL FILES # ============================================ model = None scaler = None feature_names = None # Exact feature names from your model EXPECTED_FEATURES = [ 'total_building_load', 'avg_chilled_water_rate', 'avg_cooling_water_temp', 'avg_outside_temp', 'avg_dew_point', 'avg_humidity', 'avg_wind_speed', 'avg_pressure', 'hour', 'day_of_week', 'month', 'day_of_year' ] def load_model_files(): """Load the existing model files with NumPy 2.0.2""" global model, scaler, feature_names print("\n📂 Checking for model files...") # Check for production_model.pkl if os.path.exists("production_model.pkl"): file_size = os.path.getsize("production_model.pkl") / 1024 print(f" ✅ Found: production_model.pkl ({file_size:.1f} KB)") else: print(f" ❌ Missing: production_model.pkl") return False # Load model with joblib try: model = joblib.load("production_model.pkl") print(f"\n✅ Model loaded successfully with joblib") print(f" Type: {type(model).__name__}") if hasattr(model, 'n_features_in_'): print(f" Features expected: {model.n_features_in_}") if hasattr(model, 'n_estimators'): print(f" Number of trees: {model.n_estimators}") if hasattr(model, 'max_depth'): print(f" Max depth: {model.max_depth}") except Exception as e: print(f"⚠️ Error loading model with joblib: {e}") return False # Load scaler if exists if os.path.exists("scaler.pkl"): try: scaler = joblib.load("scaler.pkl") print(f"✅ Scaler loaded") except Exception as e: print(f"⚠️ Could not load scaler: {e}") # Load feature names from features.pkl if os.path.exists("features.pkl"): try: feature_data = joblib.load("features.pkl") # Convert to list if it's a pandas Series or other type if hasattr(feature_data, 'tolist'): feature_names = feature_data.tolist() elif isinstance(feature_data, (list, tuple)): feature_names = list(feature_data) elif hasattr(feature_data, 'values'): feature_names = list(feature_data.values) else: feature_names = list(feature_data) print(f"✅ Features loaded from features.pkl: {len(feature_names)} features") if len(feature_names) > 0: print(f" First 3 features: {feature_names[:3]}") except Exception as e: print(f"⚠️ Could not load features.pkl: {e}") feature_names = None # Use default feature names if none loaded if feature_names is None: feature_names = EXPECTED_FEATURES print(f"✅ Using default feature names: {len(feature_names)} features") # Ensure feature_names is a list if not isinstance(feature_names, list): if hasattr(feature_names, 'tolist'): feature_names = feature_names.tolist() else: feature_names = list(feature_names) return True # Load model on startup load_success = load_model_files() if model: print(f"\n📊 Model Status: ✅ ONLINE") print(f" NumPy version: {np.__version__}") print(f" Pandas version: {pd.__version__}") print(f" Features: {len(feature_names) if feature_names else 0}") print(f" Scaler: {'✅' if scaler else '❌'}") else: print(f"\n📊 Model Status: ❌ OFFLINE") # ============================================ # REQUEST MODELS # ============================================ class ChillerInput(BaseModel): """12 input features - optimized for 1000+ TR plants""" total_building_load: float = Field(..., ge=400, le=2500, description="Total building cooling load (RT) - For 4 chiller plant: 400-2500 TR") avg_chilled_water_rate: float = Field(..., ge=200, le=2000, description="Average chilled water flow rate (L/sec)") avg_cooling_water_temp: float = Field(..., ge=15, le=35, description="Average cooling water temperature (°C)") avg_outside_temp: float = Field(..., ge=32, le=120, description="Average outside air temperature (°F)") avg_dew_point: float = Field(..., ge=20, le=80, description="Average dew point temperature (°F)") avg_humidity: float = Field(..., ge=20, le=100, description="Average relative humidity (%)") avg_wind_speed: float = Field(..., ge=0, le=30, description="Average wind speed (mph)") avg_pressure: float = Field(..., ge=28, le=31, description="Average atmospheric pressure (in Hg)") hour: int = Field(..., ge=0, le=23, description="Hour of the day (0-23)") day_of_week: int = Field(..., ge=0, le=6, description="Day of week (0=Monday, 6=Sunday)") month: int = Field(..., ge=1, le=12, description="Month of the year (1-12)") day_of_year: int = Field(..., ge=1, le=366, description="Day of the year (1-365)") # Optional parameters for 4-chiller plant current_chw_setpoint_c: Optional[float] = Field(8.0, ge=5, le=10, description="Current CHW setpoint (°C)") num_chillers_operating: Optional[int] = Field(4, ge=1, le=4, description="Number of chillers currently operating") electricity_rate_usd_per_kwh: Optional[float] = Field(0.12, ge=0.05, le=0.50, description="Electricity rate ($/kWh)") class PredictionResponse(BaseModel): status: str kw_per_tr: float total_power_kw: float power_per_chiller_kw: float cooling_load_tr: float load_per_chiller_tr: float cop: float efficiency_rating: str plant_summary: Dict[str, Any] annual_cost_estimate: Dict[str, Any] features_used: int model_type: str timestamp: str class OptimizationRecommendation(BaseModel): action: str current_value: str recommended_value: str expected_savings: str priority: str operator_action: str class OptimizeResponse(BaseModel): timestamp: str current_kw_per_tr: float current_total_power_kw: float optimal_kw_per_tr: float optimal_total_power_kw: float efficiency_improvement_pct: float power_savings_kw: float power_savings_pct: float annual_savings_usd: float co2_reduction_kg_per_year: float recommendations: List[OptimizationRecommendation] summary: Dict[str, str] # ============================================ # PREDICTION FUNCTIONS # ============================================ def predict_kw_per_tr(input_data: ChillerInput) -> float: """Predict Combined_Kw_per_TR using the loaded model""" if model is None: raise ValueError("Model not loaded") # Create feature array with exact order features = np.array([[ input_data.total_building_load, input_data.avg_chilled_water_rate, input_data.avg_cooling_water_temp, input_data.avg_outside_temp, input_data.avg_dew_point, input_data.avg_humidity, input_data.avg_wind_speed, input_data.avg_pressure, input_data.hour, input_data.day_of_week, input_data.month, input_data.day_of_year ]], dtype=np.float64) # Apply scaler if available if scaler is not None: try: features = scaler.transform(features) except Exception as e: print(f" ⚠️ Scaler transform failed: {e}") # Predict prediction = model.predict(features)[0] # Typical range for large centrifugal chillers (1000+ TR) # Modern efficient: 0.45-0.55 kW/TR # Older units: 0.60-0.80 kW/TR prediction = np.clip(prediction, 0.40, 0.90) return float(prediction) def calculate_total_power(kw_per_tr: float, cooling_load_tr: float) -> float: """ Calculate total chiller plant power consumption for all chillers Formula: Total Power (kW) = kW/TR × Cooling Load (TR) """ return kw_per_tr * cooling_load_tr def calculate_annual_energy_cost(kw_per_tr: float, avg_load_tr: float, operating_hours: int = 3000, electricity_rate: float = 0.12) -> dict: """ Calculate annual energy cost and consumption """ avg_power_kw = kw_per_tr * avg_load_tr annual_kwh = avg_power_kw * operating_hours annual_cost_usd = annual_kwh * electricity_rate return { "avg_power_kw": round(avg_power_kw, 1), "annual_kwh": round(annual_kwh / 1000, 1), "annual_cost_usd": round(annual_cost_usd, 0), "operating_hours": operating_hours, "electricity_rate": electricity_rate } def calculate_cop_from_kw_per_tr(kw_per_tr: float) -> float: """ Convert kW/TR to COP (Coefficient of Performance) Formula: COP = 3.516 / kW/TR """ if kw_per_tr <= 0: return 0 return 3.516 / kw_per_tr def get_efficiency_rating(kw_per_tr: float) -> str: """Get efficiency rating based on kW/TR value""" if kw_per_tr < 0.55: return "Excellent" elif kw_per_tr < 0.65: return "Good" elif kw_per_tr < 0.75: return "Fair" else: return "Poor" def optimize_chw_setpoint(input_data: ChillerInput) -> tuple: """Find optimal CHW setpoint by testing different values""" current_sp = input_data.current_chw_setpoint_c or 8.0 # Test different setpoints test_setpoints = [6.0, 6.5, 7.0, 7.5, 8.0, 8.5, 9.0, 9.5, 10.0] best_kw = float('inf') best_sp = current_sp for sp in test_setpoints: # Create copy with new setpoint using dict input_dict = input_data.dict() input_dict['current_chw_setpoint_c'] = sp test_input = ChillerInput(**input_dict) try: kw = predict_kw_per_tr(test_input) if kw < best_kw: best_kw = kw best_sp = sp except Exception as e: continue return best_sp, best_kw # ============================================ # API ENDPOINTS # ============================================ @app.get("/") async def root(): """Root endpoint with API information for large chiller plants""" feature_count = len(feature_names) if feature_names and isinstance(feature_names, list) else len(EXPECTED_FEATURES) return { "service": "York Chiller Energy Optimizer - 4 Chiller Plant Edition", "model_type": "Random Forest Regressor", "version": "2.0.0", "plant_size": "1000+ TR (4 centrifugal chillers)", "status": "online" if model is not None else "model_not_loaded", "model_info": { "loaded": model is not None, "features": feature_count, "scaler_loaded": scaler is not None, "numpy_version": np.__version__, "pandas_version": pd.__version__ }, "endpoints": { "/": "This information", "/health": "Health check", "/predict": "POST - Predict efficiency and power for 4-chiller plant", "/optimize": "POST - Get optimization with annual savings ($ and CO2)", "/plant-analysis": "POST - Detailed 4-chiller plant analysis" }, "sample_4_chiller_input": { "total_building_load": 1200, "avg_chilled_water_rate": 800, "avg_cooling_water_temp": 28, "avg_outside_temp": 95, "avg_dew_point": 65, "avg_humidity": 60, "avg_wind_speed": 8, "avg_pressure": 29.9, "hour": 14, "day_of_week": 2, "month": 7, "day_of_year": 200, "num_chillers_operating": 4, "electricity_rate_usd_per_kwh": 0.12 }, "expected_output_sample": { "total_power_kw": "~700-800 kW total", "power_per_chiller_kw": "~175-200 kW each", "annual_energy_cost": "$250,000-$350,000 at $0.12/kWh" }, "interpretation": { "kw_per_tr": "Lower is better for 1000+ TR plants: <0.55 = excellent", "total_power_kw": "Actual plant power for all chillers combined", "typical_power_for_1000tr": "500-700 kW at full load", "potential_savings": "50-150 kW possible through optimization = $15k-45k/year" } } @app.get("/health") async def health(): """Health check endpoint""" feature_count = len(feature_names) if feature_names and isinstance(feature_names, list) else len(EXPECTED_FEATURES) return { "status": "healthy" if model is not None else "degraded", "model_loaded": model is not None, "model_type": type(model).__name__ if model else None, "feature_count": feature_count, "scaler_loaded": scaler is not None, "numpy_version": np.__version__, "pandas_version": pd.__version__ } @app.post("/predict", response_model=PredictionResponse) async def predict_endpoint(input_data: ChillerInput): """Predict efficiency and power for 4-chiller plant (1000+ TR)""" try: if model is None: raise HTTPException(status_code=503, detail="Model not loaded - check logs") # Get predicted kW/TR kw_per_tr = predict_kw_per_tr(input_data) # Calculate total plant power total_power_kw = calculate_total_power(kw_per_tr, input_data.total_building_load) # Get number of chillers (default 4) num_chillers = input_data.num_chillers_operating or 4 # Per-chiller metrics power_per_chiller = total_power_kw / num_chillers load_per_chiller = input_data.total_building_load / num_chillers # Calculate COP cop = calculate_cop_from_kw_per_tr(kw_per_tr) # Get efficiency rating rating = get_efficiency_rating(kw_per_tr) # Calculate annual cost estimate electricity_rate = input_data.electricity_rate_usd_per_kwh or 0.12 annual_cost = calculate_annual_energy_cost( kw_per_tr, input_data.total_building_load, operating_hours=3000, electricity_rate=electricity_rate ) # Plant summary plant_summary = { "num_chillers": num_chillers, "total_capacity_tr": round(input_data.total_building_load, 0), "load_per_chiller_tr": round(load_per_chiller, 1), "power_density_kw_per_100tr": round(kw_per_tr * 100, 2), "chiller_type": "Centrifugal (1000+ TR scale)" } feature_count = len(feature_names) if feature_names and isinstance(feature_names, list) else len(EXPECTED_FEATURES) return PredictionResponse( status="success", kw_per_tr=round(kw_per_tr, 4), total_power_kw=round(total_power_kw, 1), power_per_chiller_kw=round(power_per_chiller, 1), cooling_load_tr=round(input_data.total_building_load, 1), load_per_chiller_tr=round(load_per_chiller, 1), cop=round(cop, 2), efficiency_rating=rating, plant_summary=plant_summary, annual_cost_estimate=annual_cost, features_used=feature_count, model_type="RandomForestRegressor", timestamp=datetime.now().isoformat() ) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/optimize", response_model=OptimizeResponse) async def optimize_endpoint(input_data: ChillerInput): """Get optimization recommendations for large chiller plant""" try: if model is None: raise HTTPException(status_code=503, detail="Model not loaded - check logs") num_chillers = input_data.num_chillers_operating or 4 electricity_rate = input_data.electricity_rate_usd_per_kwh or 0.12 # Current efficiency and power current_kw_per_tr = predict_kw_per_tr(input_data) current_power_kw = calculate_total_power(current_kw_per_tr, input_data.total_building_load) # Find optimal setpoint optimal_sp, optimal_kw_per_tr = optimize_chw_setpoint(input_data) optimal_power_kw = calculate_total_power(optimal_kw_per_tr, input_data.total_building_load) # Calculate savings savings_pct = ((current_kw_per_tr - optimal_kw_per_tr) / current_kw_per_tr) * 100 if current_kw_per_tr > 0 else 0 savings_pct = max(0, savings_pct) # Power savings power_savings_kw = current_power_kw - optimal_power_kw power_savings_pct = (power_savings_kw / current_power_kw) * 100 if current_power_kw > 0 else 0 # Annual savings annual_operating_hours = 3000 annual_savings_kwh = power_savings_kw * annual_operating_hours annual_savings_usd = annual_savings_kwh * electricity_rate # CO2 reduction (assuming 0.4 kg CO2 per kWh - US average) co2_reduction_kg = annual_savings_kwh * 0.4 # Build recommendations for 4-chiller plant recommendations = [] # Setpoint optimization current_sp = input_data.current_chw_setpoint_c or 8.0 if optimal_sp != current_sp and savings_pct > 1: recommendations.append(OptimizationRecommendation( action="CHW Setpoint Optimization", current_value=f"{current_sp:.1f}°C", recommended_value=f"{optimal_sp:.1f}°C", expected_savings=f"{savings_pct:.1f}% ({power_savings_kw:.0f} kW, ${annual_savings_usd:,.0f}/year)", priority="HIGH" if savings_pct > 5 else "MEDIUM", operator_action=f"Raise CHW setpoint to {optimal_sp:.1f}°C across all {num_chillers} chillers" )) # Chiller sequencing for large plants load_per_chiller = input_data.total_building_load / num_chillers if input_data.total_building_load < 800: recommendations.append(OptimizationRecommendation( action="Chiller Sequencing", current_value=f"{num_chillers} chillers operating", recommended_value="3 chillers (or less)", expected_savings="20-30% power reduction at low load", priority="HIGH", operator_action=f"Sequentially shut down one chiller, adjust load on remaining {num_chillers-1} chillers" )) elif input_data.total_building_load > 1800: recommendations.append(OptimizationRecommendation( action="Load Balancing", current_value=f"{load_per_chiller:.0f} TR per chiller", recommended_value="Verify all chillers are contributing equally", expected_savings="5-10% efficiency improvement", priority="MEDIUM", operator_action="Check operating logs for load sharing between chillers" )) # Condenser water temperature optimization if input_data.avg_cooling_water_temp < 24: recommendations.append(OptimizationRecommendation( action="Condenser Water Reset", current_value=f"{input_data.avg_cooling_water_temp:.1f}°C", recommended_value="Allow cooling tower to float higher", expected_savings="3-8% pump energy reduction", priority="MEDIUM", operator_action="Reduce cooling tower fan speed or disable some cells" )) # Free cooling recommendation if input_data.avg_outside_temp < 50 and input_data.avg_humidity < 60: estimated_savings_kw = current_power_kw * 0.35 estimated_savings_usd = estimated_savings_kw * annual_operating_hours * electricity_rate recommendations.append(OptimizationRecommendation( action="Free Cooling Mode", current_value="Mechanical cooling only", recommended_value="Enable waterside economizer", expected_savings=f"35% (approx {estimated_savings_kw:.0f} kW, ${estimated_savings_usd:,.0f}/year)", priority="HIGH", operator_action="Open bypass valve for cooling tower to provide directly chilled water" )) # Efficiency rating rating = get_efficiency_rating(current_kw_per_tr) current_cop = calculate_cop_from_kw_per_tr(current_kw_per_tr) optimal_cop = calculate_cop_from_kw_per_tr(optimal_kw_per_tr) summary = { "plant_summary": f"{num_chillers} chillers, {input_data.total_building_load:.0f} TR total", "current_efficiency": f"{current_kw_per_tr:.3f} kW/TR (COP: {current_cop:.2f})", "current_power": f"{current_power_kw:.0f} kW ({current_power_kw/num_chillers:.0f} kW per chiller)", "optimal_efficiency": f"{optimal_kw_per_tr:.3f} kW/TR (COP: {optimal_cop:.2f})", "optimal_power": f"{optimal_power_kw:.0f} kW", "power_savings": f"{power_savings_kw:.0f} kW ({savings_pct:.1f}%)", "annual_savings_usd": f"${annual_savings_usd:,.0f}", "co2_reduction": f"{co2_reduction_kg/1000:.1f} metric tons CO2/year", "efficiency_rating": rating, "load_per_chiller": f"{load_per_chiller:.0f} TR", "recommended_setpoint": f"{optimal_sp:.1f}°C", "electricity_rate": f"${electricity_rate:.2f}/kWh" } return OptimizeResponse( timestamp=datetime.now().isoformat(), current_kw_per_tr=round(current_kw_per_tr, 4), current_total_power_kw=round(current_power_kw, 1), optimal_kw_per_tr=round(optimal_kw_per_tr, 4), optimal_total_power_kw=round(optimal_power_kw, 1), efficiency_improvement_pct=round(savings_pct, 2), power_savings_kw=round(power_savings_kw, 1), power_savings_pct=round(power_savings_pct, 2), annual_savings_usd=round(annual_savings_usd, 0), co2_reduction_kg_per_year=round(co2_reduction_kg, 0), recommendations=recommendations, summary=summary ) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/plant-analysis") async def plant_analysis(input_data: ChillerInput): """Detailed analysis for 4-chiller plant""" try: if model is None: raise HTTPException(status_code=503, detail="Model not loaded") kw_per_tr = predict_kw_per_tr(input_data) num_chillers = input_data.num_chillers_operating or 4 # Calculate various metrics total_power = kw_per_tr * input_data.total_building_load power_per_chiller = total_power / num_chillers # Compare to industry benchmarks benchmarks = { "excellent": 0.50, "good": 0.60, "average": 0.65, "poor": 0.75 } # Energy cost at different loads load_levels = [50, 75, 100] cost_analysis = [] for load_pct in load_levels: load_tr = input_data.total_building_load * load_pct / 100 power_kw = kw_per_tr * load_tr cost_analysis.append({ "load_pct": load_pct, "load_tr": round(load_tr, 0), "power_kw": round(power_kw, 0), "power_per_chiller_kw": round(power_kw / num_chillers, 0) }) return { "timestamp": datetime.now().isoformat(), "plant_configuration": { "num_chillers": num_chillers, "total_capacity_tr": input_data.total_building_load, "current_load_tr": input_data.total_building_load, "load_factor": "100%" }, "performance_metrics": { "kw_per_tr": round(kw_per_tr, 3), "total_power_kw": round(total_power, 0), "power_per_chiller_kw": round(power_per_chiller, 0), "cop": round(calculate_cop_from_kw_per_tr(kw_per_tr), 2), "vs_benchmark_excellent": f"{((kw_per_tr - benchmarks['excellent'])/benchmarks['excellent']*100):+.1f}%" }, "cost_analysis": cost_analysis, "recommendations": [ f"Target kW/TR < 0.60 for this plant size (currently {kw_per_tr:.3f})", f"At full load, each chiller draws ~{power_per_chiller:.0f} kW", "Review if all 4 chillers are needed at current load" ] } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)