Spaces:
Sleeping
Sleeping
| """ | |
| Explainability Engine — 100% Offline | |
| Generates human-readable forecast explanations using deterministic templates. | |
| No LLMs. No external APIs. Fully air-gappable. | |
| Why templates over LLMs for this use case: | |
| 1. Deterministic: same input → same explanation (auditable, testable) | |
| 2. Faster: zero API latency (~0ms vs ~500ms for LLM call) | |
| 3. Cheaper: zero cost per explanation | |
| 4. More operationally useful: includes exact MW values, physics reasons, and clamp details | |
| 5. Fully compliant: no hosted LLM usage whatsoever | |
| A grid operator cares about: "Why is Pavagada forecast 64 MW at noon?" | |
| Template answer: "Solar plant at 12:00 IST: Baseline forecast 82 MW. Downward adjustment of 18 MW due to cloud cover (37%). Output capped by clear-sky GHI limit." | |
| LLM answer: "At 12:00 IST, expect moderate solar generation under partly cloudy skies." | |
| The template is objectively more useful for operational decision-making. | |
| """ | |
| import pandas as pd | |
| SUPPORTED_LANGUAGES = {"en", "kn"} | |
| def _normalize_language(language: str | None) -> str: | |
| value = (language or "en").strip().lower() | |
| return value if value in SUPPORTED_LANGUAGES else "en" | |
| def generate_explanation(row): | |
| """Generate operationally useful explanation from forecast row.""" | |
| language = _normalize_language(row.get("language")) | |
| parts = [] | |
| plant_type = "Solar" if "SOL_" in str(row.get("plant_id", "")) else "Wind" | |
| base = row.get("predictions", row.get("baseline_MW", 0)) | |
| res = row.get("residual_MW", 0) | |
| final = row.get("final_forecast_MW", 0) | |
| clamp = row.get("clamp_reason", "") | |
| cloud = row.get("cloud_fraction", 0) | |
| ws = row.get("wind_speed_10m", 0) | |
| hour = pd.to_datetime(row["timestamp"]).hour | |
| is_wind = plant_type == "Wind" | |
| if language == "kn": | |
| parts.append(f"{'ಸೌರ' if not is_wind else 'ಗಾಳಿ'} ಘಟಕ {hour:02d}:00 IST:") | |
| parts.append(f"ಮೂಲ ಅಂದಾಜು {base:.1f} MW.") | |
| if abs(res) > 1: | |
| if is_wind and pd.notna(ws): | |
| parts.append(f"ಗಾಳಿ ವೇಗ {float(ws):.1f} m/s ಆಧಾರವಾಗಿ {final:.1f} MW ಗೆ ಸರಿಪಡಿಸಲಾಗಿದೆ.") | |
| elif res < 0: | |
| parts.append(f"ಮೋಡಾವರಣ ({cloud*100:.0f}%) ಕಾರಣ {abs(res):.1f} MW ಇಳಿಕೆ.") | |
| else: | |
| parts.append(f"{res:.1f} MW ಹೆಚ್ಚುವರಿ ಸರಿಪಡింపు.") | |
| if clamp: | |
| parts.append(f"ಭೌತಿಕ ಮಿತಿ ಅನ್ವಯಿಸಲಾಗಿದೆ: {clamp}") | |
| if final < base * 0.8: | |
| parts.append("ಹವಾಮಾನ ಅಸ್ಥಿರತೆಯಿಂದ ವಿಶ್ವಾಸ ವ್ಯಾಪ್ತಿ ವಿಸ್ತರಿಸಲಾಗಿದೆ.") | |
| return " ".join(parts) | |
| parts.append(f"{plant_type} plant at {hour:02d}:00 IST:") | |
| parts.append(f"Baseline forecast {base:.1f} MW.") | |
| if abs(res) > 1: | |
| if is_wind and pd.notna(ws): | |
| parts.append( | |
| f"Weather-driven adjustment to {final:.1f} MW based on {float(ws):.1f} m/s wind speed." | |
| ) | |
| elif res < 0: | |
| parts.append(f"Downward adjustment of {abs(res):.1f} MW due to cloud cover ({cloud*100:.0f}%).") | |
| else: | |
| parts.append(f"Upward adjustment of {res:.1f} MW.") | |
| if clamp: | |
| parts.append(f"Output capped: {clamp}") | |
| if final < base * 0.8: | |
| parts.append("Confidence band widened due to high weather volatility.") | |
| return " ".join(parts) | |
| def generate_explanations(df, language: str | None = None): | |
| """Add explanation column to forecast DataFrame.""" | |
| df = df.copy() | |
| if language is not None: | |
| df["language"] = _normalize_language(language) | |
| elif "language" in df.columns: | |
| df["language"] = df["language"].map(_normalize_language) | |
| else: | |
| df["language"] = "en" | |
| df["explanation"] = df.apply(generate_explanation, axis=1) | |
| return df | |
| if __name__ == "__main__": | |
| from data.scada_generator import PLANTS | |
| from pipeline.residual_adjuster import apply_residual_layer | |
| from pipeline.physics_constraints import apply_physics_constraints | |
| bf = pd.read_parquet("models/baseline/baseline_forecasts.parquet") | |
| wf = pd.read_parquet("data/weather/weather_all.parquet") | |
| res = apply_residual_layer(bf, wf, PLANTS) | |
| phy = apply_physics_constraints(res, wf, PLANTS) | |
| out = generate_explanations(phy) | |
| print(out[["plant_id", "timestamp", "final_forecast_MW", "explanation"]].head(5)) | |