From b606ad572cca1178ce060e9b720d6d1145a21fa0 Mon Sep 17 00:00:00 2001 From: Oklyagina Date: Fri, 25 Sep 2026 16:09:04 +0100 Subject: [PATCH 1/8] RL failure forecast implementation --- usecases_examples/PowerGrid/.gitignore | 5 +- usecases_examples/PowerGrid/Dockerfile.app | 5 +- usecases_examples/PowerGrid/README.md | 20 +++++ .../PowerGrid/app/models/Communicate.py | 31 ++++++- .../PowerGrid/app/models/Simulator.py | 82 +++++++++++++++++-- .../PowerGrid/config/CONFIG.toml | 1 + .../PowerGrid/requirements-forecast.txt | 2 + 7 files changed, 133 insertions(+), 13 deletions(-) create mode 100644 usecases_examples/PowerGrid/requirements-forecast.txt diff --git a/usecases_examples/PowerGrid/.gitignore b/usecases_examples/PowerGrid/.gitignore index 7e90a88b..c13ef32f 100644 --- a/usecases_examples/PowerGrid/.gitignore +++ b/usecases_examples/PowerGrid/.gitignore @@ -11,6 +11,9 @@ full_scn_jan_28_1.txt # Scenario folder #env_ICAPS_input_data_test/ +# Trained failure-forecast deployment artifacts (distributed separately) +Ressources/failure_forecast/*.pkl + # Byte-compiled / optimized / DLL files __pycache__/ *.py[cod] @@ -171,4 +174,4 @@ cython_debug/ # be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore # and can be added to the global gitignore or merged into this file. For a more nuclear # option (not recommended) you can uncomment the following to ignore the entire idea folder. -#.idea/ \ No newline at end of file +#.idea/ diff --git a/usecases_examples/PowerGrid/Dockerfile.app b/usecases_examples/PowerGrid/Dockerfile.app index 782ea872..f3a49439 100644 --- a/usecases_examples/PowerGrid/Dockerfile.app +++ b/usecases_examples/PowerGrid/Dockerfile.app @@ -1,6 +1,6 @@ # syntax=docker/dockerfile:1 -FROM python:3.9-slim-bullseye +FROM python:3.10-slim-bullseye EXPOSE 5000 RUN mkdir /code @@ -9,5 +9,6 @@ WORKDIR /code RUN apt-get update && apt-get install ffmpeg libsm6 libxext6 -y RUN pip3 install -r requirements-app.txt +RUN pip3 install -r requirements-forecast.txt -CMD ["python3", "PowerGrid_poc_simulator_app.py"] \ No newline at end of file +CMD ["python3", "PowerGrid_poc_simulator_app.py"] diff --git a/usecases_examples/PowerGrid/README.md b/usecases_examples/PowerGrid/README.md index 3cbdb440..308bd358 100644 --- a/usecases_examples/PowerGrid/README.md +++ b/usecases_examples/PowerGrid/README.md @@ -64,8 +64,28 @@ When using Docker, dependencies are installed inside the image automatically. ```commandline cd InteractiveAI/usecases_examples/PowerGrid pip install -r requirements-app.txt +pip install -r requirements-forecast.txt ``` +The overload failure forecast also requires these trained artifacts in +`Ressources/failure_forecast/`: + +```text +mean_forecaster.pkl +aleatoric_forecaster.pkl +failure_classifier.pkl +failure_classifier_metadata.json +``` + +For a local validation with artifacts stored elsewhere, set +`FAILURE_FORECAST_ARTIFACT_DIR` to that directory before starting the +simulator. The default remains `Ressources/failure_forecast`. + +The simulator keeps the required one-hour, one-day, and one-week observation +history in memory. When an overload event is accepted by InteractiveAI, the +simulator forecasts failure for that overloaded line and publishes a second +event whose `parent_event_id` is the overload event ID. + # 2 Run the simulator The simulator is a web application. Two Docker Compose files are provided: diff --git a/usecases_examples/PowerGrid/app/models/Communicate.py b/usecases_examples/PowerGrid/app/models/Communicate.py index 9e779eb8..248e0825 100644 --- a/usecases_examples/PowerGrid/app/models/Communicate.py +++ b/usecases_examples/PowerGrid/app/models/Communicate.py @@ -7,6 +7,7 @@ import numpy as np from config.config import logging, set_pause, get_pause_status from app.models.recommendation_store import store as recommendation_store +from app.models.failure_forecast import build_failure_forecast_event class Communicate: """ @@ -273,6 +274,7 @@ def send_payload_and_store_it(self, payload, obs, scn_first_step): obs: The current observation of the network. scn_first_step: First step of the scenario. """ + response_payload = None try: # Payload updating for same events detected try: @@ -301,6 +303,7 @@ def send_payload_and_store_it(self, payload, obs, scn_first_step): # logging.info("FULL EVENT : %s", self.payload) # print(response.text) response.raise_for_status() + response_payload = response.json() text = self.payload["title"] if self.cab_api_on is True: @@ -325,6 +328,7 @@ def send_payload_and_store_it(self, payload, obs, scn_first_step): except Exception as e: logging.error(e) logging.info("The event's follow-up is not working") + return response_payload def send_event_online(self, context_date, @@ -357,6 +361,7 @@ def send_event_online(self, case_anticip: Flag for anticipation case (default is False). case_line_lost: Flag for line lost case (default is False). """ + event_response = None if zone is None: zone = [] if line is None: @@ -388,7 +393,9 @@ def send_event_online(self, payload = json.dumps(payload_dict) # print(f"Overload description: {payload}") - self.send_payload_and_store_it(payload, obs, scn_first_step) + event_response = self.send_payload_and_store_it( + payload, obs, scn_first_step + ) except Exception as e: logging.error(e) @@ -503,6 +510,28 @@ def send_event_online(self, except Exception as e: logging.error(e) + return event_response + + def send_failure_forecast_event(self, + context_date, + scn_first_step, + forecast, + parent_event_id, + obs, + line_name, + img_b64=None): + """Send a failure forecast linked to its overload parent event.""" + payload_dict = build_failure_forecast_event( + context_date, + forecast, + parent_event_id, + line_name, + img_b64, + ) + return self.send_payload_and_store_it( + json.dumps(payload_dict), obs, scn_first_step + ) + def issues_follow_up(self): """ Updates the list of ongoing issues in the simulator. diff --git a/usecases_examples/PowerGrid/app/models/Simulator.py b/usecases_examples/PowerGrid/app/models/Simulator.py index fd1bb734..9c85401a 100644 --- a/usecases_examples/PowerGrid/app/models/Simulator.py +++ b/usecases_examples/PowerGrid/app/models/Simulator.py @@ -1,4 +1,5 @@ import time +import os from datetime import datetime, timedelta, timezone import grid2op from grid2op.Chronics.handlers import PerfectForecastHandler, CSVHandler @@ -8,9 +9,11 @@ import numpy as np import toml import json +from pathlib import Path import matplotlib matplotlib.use('agg') from app.models.Listener import Listener +from app.models.failure_forecast import FailureForecastService from config.config import logging, set_pause, get_pause_status from app.models.utils import (create_observation_image, get_alert_lines, search_chronic_num_from_name, get_curent_lines_in_bad_kpi, get_curent_lines_lost, @@ -39,6 +42,7 @@ def __init__(self, socketio): self.local_assistant = None self.com = None self.agent_reco = None + self.failure_forecast = None self.socketio = socketio def load_and_edit_config(self, params=None): @@ -95,6 +99,23 @@ def initialize_simulation(self, com, session): self.env.chronics_handler.get_name()) session['message'].append(f"Loaded scenario: {self.env.chronics_handler.get_name()}") + artifact_dir = Path(os.environ.get( + 'FAILURE_FORECAST_ARTIFACT_DIR', + self.config.get( + 'failure_forecast_artifact_dir', + 'Ressources/failure_forecast' + ) + )) + try: + self.failure_forecast = FailureForecastService(artifact_dir) + self.failure_forecast.record_observation(self.obs) + logging.info( + "Failure forecast artifacts loaded from %s", artifact_dir + ) + except Exception as exc: + self.failure_forecast = None + logging.error("Failure forecast is unavailable: %s", exc) + assistant_path = self.config['assistant_path'] assistant_seed = int(self.config['assistant_seed']) self.local_assistant = load_assistant( @@ -175,6 +196,8 @@ def run_simulator(self, com): # Beginning of step: observation update self.obs, _, done, info = self.env.step(act) + if self.failure_forecast is not None: + self.failure_forecast.record_observation(self.obs) # Confirm whether a recommendation received from InteractiveAI was # actually applied to the simulation on this step. @@ -328,15 +351,56 @@ def run_simulator(self, com): if not img_b64_current: img_b64_current = create_observation_image(self.env, self.obs) - com.send_event_online(context_date, - self.config['scenario_first_step'], - self.listen.trigger_kpis( - self.obs, act), - self.obs, self.listen.current_issues, - img_b64_current, - line_name=get_curent_lines_in_bad_kpi( - self.obs), - case_overload=True) + overloaded_line = get_curent_lines_in_bad_kpi(self.obs) + forecast_line = overloaded_line.rsplit(":", 1)[-1] + overload_response = com.send_event_online( + context_date, + self.config['scenario_first_step'], + self.listen.trigger_kpis(self.obs, act), + self.obs, + self.listen.current_issues, + img_b64_current, + line_name=overloaded_line, + case_overload=True + ) + parent_event_id = None + if isinstance(overload_response, dict): + parent_event_id = ( + overload_response.get("id_event") + or overload_response.get("id") + ) + if parent_event_id and self.failure_forecast is not None: + try: + forecast = self.failure_forecast.predict( + self.env, + self.obs, + forecast_line + ) + com.send_failure_forecast_event( + context_date, + self.config['scenario_first_step'], + forecast, + parent_event_id, + self.obs, + overloaded_line, + img_b64_current + ) + logging.info( + "Failure forecast for line %s: %s", + overloaded_line, + forecast + ) + except Exception as exc: + logging.error( + "Failure forecast failed for line %s: %s", + overloaded_line, + exc + ) + elif self.failure_forecast is not None: + logging.error( + "Failure forecast event was not sent because " + "the overload event ID was unavailable" + ) if (self.obs.current_step < self.config['scenario_first_step']) or (com.cab_api_on is False): # Use cached XD_Silly (local) diff --git a/usecases_examples/PowerGrid/config/CONFIG.toml b/usecases_examples/PowerGrid/config/CONFIG.toml index 126ba790..8f39ac94 100644 --- a/usecases_examples/PowerGrid/config/CONFIG.toml +++ b/usecases_examples/PowerGrid/config/CONFIG.toml @@ -10,3 +10,4 @@ assistant_path = "Ressources/XD_silly_repo" assistant_seed = 1227139268 stepDuration_s = 0.5 scenario_first_step = 752 +failure_forecast_artifact_dir = "failure_forecaster" diff --git a/usecases_examples/PowerGrid/requirements-forecast.txt b/usecases_examples/PowerGrid/requirements-forecast.txt new file mode 100644 index 00000000..c807a949 --- /dev/null +++ b/usecases_examples/PowerGrid/requirements-forecast.txt @@ -0,0 +1,2 @@ +joblib==1.5.1 +scikit-learn==1.7.2 From 42db55aa479776bcaf770ce2ffa640930af4fc8c Mon Sep 17 00:00:00 2001 From: Oklyagina Date: Fri, 25 Sep 2026 17:18:18 +0100 Subject: [PATCH 2/8] RL failure forecast implementation --- .../PowerGrid/app/models/failure_forecast.py | 346 ++++++++++++++++++ 1 file changed, 346 insertions(+) create mode 100644 usecases_examples/PowerGrid/app/models/failure_forecast.py diff --git a/usecases_examples/PowerGrid/app/models/failure_forecast.py b/usecases_examples/PowerGrid/app/models/failure_forecast.py new file mode 100644 index 00000000..6d165711 --- /dev/null +++ b/usecases_examples/PowerGrid/app/models/failure_forecast.py @@ -0,0 +1,346 @@ +"""Runtime failure forecasting for PowerGrid overload events. + +This is the inference-only subset of the failure forecast implemented in the +RL_agent_failure_forecast project. It deliberately owns its history and model +loading inside the simulator so InteractiveAI does not need a forecasting API. +""" +from __future__ import annotations + +import datetime as dt +import json +from collections import deque +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Deque, Dict, Optional, Sequence, Tuple + +import numpy as np +import pandas as pd + + +HORIZON_STEPS = 12 +STEP_MINUTES = 5 +HISTORY_STEPS = 2016 + +CLASSIFIER_FEATURES = [ + "line_id_encoded", + "sum_load_p", + "sum_load_q", + "sum_gen_p", + "var_line_rho", + "avg_line_rho", + "max_line_rho", + "nb_rho_ge_0.95", + "load_gen_ratio", + "fcast_sum_load_p", + "fcast_sum_load_q", + "fcast_sum_gen_p", + "fcast_var_line_rho", + "fcast_avg_line_rho", + "fcast_max_line_rho", + "fcast_nb_rho_ge_0.95", + "aleatoric_load_p_mean", + "aleatoric_load_q_mean", + "aleatoric_gen_p_mean", + "epistemic_before", + "epistemic_after", +] + + +@dataclass(frozen=True) +class ObservationHistoryEntry: + """Small immutable subset of an observation needed by the forecasters.""" + + load_p: np.ndarray + load_q: np.ndarray + gen_p: np.ndarray + + @classmethod + def from_observation(cls, obs: Any) -> "ObservationHistoryEntry": + return cls( + load_p=np.asarray(obs.load_p, dtype=float).copy(), + load_q=np.asarray(obs.load_q, dtype=float).copy(), + gen_p=np.asarray(obs.gen_p, dtype=float).copy(), + ) + + +def _cyclic(value: float, period: float) -> Tuple[float, float]: + return ( + float(np.cos(2 * np.pi * value / period)), + float(np.sin(2 * np.pi * value / period)), + ) + + +def get_features_with_history( + history: Sequence[ObservationHistoryEntry], obs: Any +) -> np.ndarray: + """Build current, 1-hour, 1-day and 1-week injection features.""" + sizes = (len(obs.load_p), len(obs.load_q), len(obs.gen_p)) + + def values(lag_steps: int, attr: str, size: int) -> np.ndarray: + if len(history) <= lag_steps: + return np.full(size, np.nan, dtype=float) + return np.asarray(getattr(history[-1 - lag_steps], attr), dtype=float) + + injection_features = np.concatenate([ + np.asarray(obs.load_p, dtype=float), + np.asarray(obs.load_q, dtype=float), + np.asarray(obs.gen_p, dtype=float), + values(12, "load_p", sizes[0]), + values(12, "load_q", sizes[1]), + values(12, "gen_p", sizes[2]), + values(288, "load_p", sizes[0]), + values(288, "load_q", sizes[1]), + values(288, "gen_p", sizes[2]), + values(2016, "load_p", sizes[0]), + values(2016, "load_q", sizes[1]), + values(2016, "gen_p", sizes[2]), + ]) + hour_cos, hour_sin = _cyclic(getattr(obs, "hour_of_day", 0), 23) + minute_cos, minute_sin = _cyclic(getattr(obs, "minute_of_hour", 0), 59) + dow_cos, dow_sin = _cyclic(getattr(obs, "day_of_week", 0), 6) + temporal = np.asarray([ + getattr(obs, "day", 1), + hour_cos, + hour_sin, + minute_cos, + minute_sin, + dow_cos, + dow_sin, + ]) + return np.concatenate([injection_features, temporal]).astype(float) + + +def _grid_stats(obs: Any) -> Dict[str, float]: + rho = np.asarray(obs.rho, dtype=float) + return { + "sum_load_p": float(np.sum(obs.load_p)), + "sum_load_q": float(np.sum(obs.load_q)), + "sum_gen_p": float(np.sum(obs.gen_p)), + "var_line_rho": float(np.var(rho)), + "avg_line_rho": float(np.mean(rho)), + "max_line_rho": float(np.max(rho)), + "nb_rho_ge_0.95": int(np.sum(rho >= 0.95)), + } + + +def _no_op(env: Any, obs: Any) -> Any: + try: + return env.action_space({}) + except Exception: + return obs._obs_env._helper_action_env({}) + + +def _forecast_observation( + env: Any, + obs: Any, + load_p: np.ndarray, + load_q: np.ndarray, + gen_p: np.ndarray, +) -> Any: + simulator_factory = getattr(obs, "get_simulator", None) + if callable(simulator_factory): + simulator = simulator_factory().predict( + _no_op(env, obs), + new_gen_p=gen_p, + new_gen_v=obs.gen_v, + new_load_p=load_p, + new_load_q=load_q, + ) + if not simulator.converged: + raise RuntimeError(f"Forecast power flow did not converge: {simulator.error}") + return simulator.current_obs + + now = obs.get_time_stamp() + future = now + dt.timedelta(minutes=HORIZON_STEPS * STEP_MINUTES) + obs_copy = obs.copy() + obs_copy._forecasted_inj = [ + (now, {"injection": { + "load_p": obs.load_p, + "load_q": obs.load_q, + "prod_p": obs.gen_p, + "prod_v": obs.gen_v, + }}), + (future, {"injection": { + "load_p": load_p, + "load_q": load_q, + "prod_p": gen_p, + "prod_v": obs.gen_v, + }}), + ] + forecast_obs, _, done, info = obs_copy.simulate(_no_op(env, obs)) + if done and info.get("exception"): + raise RuntimeError(f"Forecast power flow failed: {info['exception']}") + return forecast_obs + + +class FailureForecastService: + """Load trained artifacts, retain simulator history, and score overloads.""" + + REQUIRED_FILES = ( + "mean_forecaster.pkl", + "aleatoric_forecaster.pkl", + "failure_classifier.pkl", + "failure_classifier_metadata.json", + ) + + def __init__(self, artifact_dir: str | Path): + import joblib + + self.artifact_dir = Path(artifact_dir) + missing = [ + name for name in self.REQUIRED_FILES + if not (self.artifact_dir / name).is_file() + ] + if missing: + raise FileNotFoundError( + "Missing failure forecast artifacts in " + f"{self.artifact_dir}: {', '.join(missing)}" + ) + self.mean_model = joblib.load(self.artifact_dir / "mean_forecaster.pkl") + self.aleatoric_model = joblib.load( + self.artifact_dir / "aleatoric_forecaster.pkl" + ) + self.classifier = joblib.load( + self.artifact_dir / "failure_classifier.pkl" + ) + self.metadata = json.loads( + (self.artifact_dir / "failure_classifier_metadata.json").read_text( + encoding="utf-8" + ) + ) + self.features = list(self.metadata.get("features", CLASSIFIER_FEATURES)) + self.line_map = { + str(name).replace("line_", ""): int(encoded) + for name, encoded in self.metadata.get("line_map", {}).items() + } + self.threshold = float(self.metadata.get("decision_threshold", 0.5)) + self.history: Deque[ObservationHistoryEntry] = deque( + maxlen=HISTORY_STEPS + 1 + ) + self.last_recorded_step: Optional[int] = None + + def record_observation(self, obs: Any) -> None: + """Record each simulator step once, including the warm-up period.""" + step = int(getattr(obs, "current_step", -1)) + if step == self.last_recorded_step: + return + self.history.append(ObservationHistoryEntry.from_observation(obs)) + self.last_recorded_step = step + + def _failure_probability(self, features: pd.DataFrame) -> float: + probabilities = np.asarray(self.classifier.predict_proba(features)) + classes = np.asarray(getattr(self.classifier, "classes_", [0, 1])) + matches = np.where(classes == 1)[0] + if len(matches) != 1: + raise ValueError("Failure classifier does not expose class 1") + return float(probabilities[0, matches[0]]) + + def predict(self, env: Any, obs: Any, line_name: str) -> Dict[str, Any]: + """Forecast failure for the line involved in the current overload.""" + self.record_observation(obs) + normalized_line = str(line_name).replace("line_", "") + if normalized_line not in self.line_map: + raise ValueError(f"Unknown line for failure classifier: {line_name!r}") + + x_t12 = get_features_with_history(list(self.history), obs) + mean_prediction = np.asarray( + self.mean_model.predict([x_t12])[0], dtype=float + ) + variance_prediction = np.asarray( + self.aleatoric_model.predict([x_t12])[0], dtype=float + ) + n_load = int(env.n_load) + n_gen = int(env.n_gen) + expected_outputs = 2 * n_load + n_gen + if mean_prediction.size != expected_outputs: + raise ValueError( + "Failure forecast artifact is incompatible with this grid: " + f"expected {expected_outputs} outputs, got {mean_prediction.size}" + ) + + load_p = mean_prediction[:n_load] + load_q = mean_prediction[n_load:2 * n_load] + gen_p = np.clip( + mean_prediction[2 * n_load:], + np.asarray(env.gen_pmin, dtype=float), + np.asarray(env.gen_pmax, dtype=float), + ) + sigma = np.sqrt( + np.expm1(np.clip(variance_prediction, 0.0, None)) + 1e-12 + ) + forecast_obs = _forecast_observation( + env, obs, load_p, load_q, gen_p + ) + current = _grid_stats(obs) + forecast = _grid_stats(forecast_obs) + sum_gen = current["sum_gen_p"] + row = { + "line_id_encoded": self.line_map[normalized_line], + **current, + "load_gen_ratio": current["sum_load_p"] / (sum_gen + 1e-6), + **{f"fcast_{key}": value for key, value in forecast.items()}, + "aleatoric_load_p_mean": float(np.mean(sigma[:n_load])), + "aleatoric_load_q_mean": float(np.mean(sigma[n_load:2 * n_load])), + "aleatoric_gen_p_mean": float(np.mean(sigma[2 * n_load:])), + # The deployed classifier was trained to accept missing optional ENN + # values, so the simulator does not need to load a second model. + "epistemic_before": float("nan"), + "epistemic_after": float("nan"), + } + classifier_input = pd.DataFrame([ + {feature: row.get(feature, np.nan) for feature in self.features} + ]) + probability = self._failure_probability(classifier_input) + return { + "failure_prediction": int(probability >= self.threshold), + "failure_probability": round(probability, 3), + "failure_probability_pct": round(probability * 100.0, 1), + "forecast_horizon_steps": int(HORIZON_STEPS), + "forecast_horizon_minutes": int(HORIZON_STEPS * STEP_MINUTES), + "line": normalized_line, + } + + +def build_failure_forecast_event( + context_date: dt.datetime, + forecast: Dict[str, Any], + parent_event_id: str, + line_name: str, + img_b64: Optional[str] = None, +) -> Dict[str, Any]: + """Build a schema-compatible event linked to an overload event.""" + probability_pct = float(forecast["failure_probability_pct"]) + will_fail = bool(forecast["failure_prediction"]) + criticality = "HIGH" # as there's already an overload in the line + if will_fail: + description = ( + "The agent is likely unable to provide a solution " + f"for overload on line {line_name} within one hour." + ) + else: + description = ( + "The agent is likely able to provide a solution for overload " + f"on line {line_name} within one hour." + ) + + event_data = { + "event_type": "anticipation", + # InteractiveAI currently de-duplicates PowerGrid events by data.line. + # Keep the forecast distinct from its overload parent. + "line": f"failure_forecast:{line_name}", + "kpis": dict(forecast), + } + if img_b64 is not None: + event_data["event_context"] = img_b64 + + return { + "criticality": criticality, + "title": f"Failure prediction {probability_pct:.1f}%", + "description": description, + "start_date": f"{context_date}", + "end_date": f"{context_date + dt.timedelta(minutes=5)}", + "parent_event_id": str(parent_event_id), + "data": event_data, + "use_case": "PowerGrid", + "is_active": False, + } From a57eb33932d7336855677e7fd869e8c84a42a14a Mon Sep 17 00:00:00 2001 From: Oklyagina Date: Thu, 22 Oct 2026 18:03:02 +0100 Subject: [PATCH 3/8] minor changes --- usecases_examples/PowerGrid/README.md | 6 +- .../PowerGrid/app/models/failure_forecast.py | 92 +++++++++++++++--- .../PowerGrid/config/CONFIG.toml | 2 +- .../PowerGrid/requirements-app.txt | Bin 3712 -> 3712 bytes 4 files changed, 83 insertions(+), 17 deletions(-) diff --git a/usecases_examples/PowerGrid/README.md b/usecases_examples/PowerGrid/README.md index 308bd358..1f83bbe3 100644 --- a/usecases_examples/PowerGrid/README.md +++ b/usecases_examples/PowerGrid/README.md @@ -83,8 +83,10 @@ simulator. The default remains `Ressources/failure_forecast`. The simulator keeps the required one-hour, one-day, and one-week observation history in memory. When an overload event is accepted by InteractiveAI, the -simulator forecasts failure for that overloaded line and publishes a second -event whose `parent_event_id` is the overload event ID. +simulator builds the one-hour-ahead grid state, disconnects that overloaded line, +derives the HGB features from the resulting contingency state, and publishes a +second event whose `parent_event_id` is the overload event ID. The event KPIs +retain both `failure_probability` and `failure_prediction`. # 2 Run the simulator diff --git a/usecases_examples/PowerGrid/app/models/failure_forecast.py b/usecases_examples/PowerGrid/app/models/failure_forecast.py index 6d165711..1a47a56c 100644 --- a/usecases_examples/PowerGrid/app/models/failure_forecast.py +++ b/usecases_examples/PowerGrid/app/models/failure_forecast.py @@ -136,7 +136,7 @@ def _forecast_observation( load_p: np.ndarray, load_q: np.ndarray, gen_p: np.ndarray, -) -> Any: +) -> Tuple[Any, Optional[Any]]: simulator_factory = getattr(obs, "get_simulator", None) if callable(simulator_factory): simulator = simulator_factory().predict( @@ -148,7 +148,7 @@ def _forecast_observation( ) if not simulator.converged: raise RuntimeError(f"Forecast power flow did not converge: {simulator.error}") - return simulator.current_obs + return simulator.current_obs, simulator now = obs.get_time_stamp() future = now + dt.timedelta(minutes=HORIZON_STEPS * STEP_MINUTES) @@ -170,7 +170,34 @@ def _forecast_observation( forecast_obs, _, done, info = obs_copy.simulate(_no_op(env, obs)) if done and info.get("exception"): raise RuntimeError(f"Forecast power flow failed: {info['exception']}") - return forecast_obs + return forecast_obs, None + + +def _line_id(env: Any, line_name: str) -> int: + normalized = str(line_name).replace("line_", "") + for index, name in enumerate(getattr(env, "name_line", ())): + if str(name).replace("line_", "") == normalized: + return int(index) + try: + return int(line_name) + except (TypeError, ValueError) as exc: + raise ValueError(f"Unknown Grid2Op line: {line_name!r}") from exc + + +def _contingency_observation( + env: Any, + forecast_obs: Any, + simulator: Any, + line_name: str, +) -> Any: + """Disconnect one line from the forecast state and return that scenario.""" + action = env.action_space({ + "set_line_status": [(_line_id(env, line_name), -1)] + }) + if simulator is not None: + return simulator.predict(action).current_obs + contingency_obs, _, _, _ = forecast_obs.simulate(action) + return contingency_obs class FailureForecastService: @@ -235,13 +262,9 @@ def _failure_probability(self, features: pd.DataFrame) -> float: raise ValueError("Failure classifier does not expose class 1") return float(probabilities[0, matches[0]]) - def predict(self, env: Any, obs: Any, line_name: str) -> Dict[str, Any]: - """Forecast failure for the line involved in the current overload.""" + def _forecast_context(self, env: Any, obs: Any) -> Dict[str, Any]: + """Build the common t+12 operating state used by all contingencies.""" self.record_observation(obs) - normalized_line = str(line_name).replace("line_", "") - if normalized_line not in self.line_map: - raise ValueError(f"Unknown line for failure classifier: {line_name!r}") - x_t12 = get_features_with_history(list(self.history), obs) mean_prediction = np.asarray( self.mean_model.predict([x_t12])[0], dtype=float @@ -268,17 +291,43 @@ def predict(self, env: Any, obs: Any, line_name: str) -> Dict[str, Any]: sigma = np.sqrt( np.expm1(np.clip(variance_prediction, 0.0, None)) + 1e-12 ) - forecast_obs = _forecast_observation( + forecast_obs, simulator = _forecast_observation( env, obs, load_p, load_q, gen_p ) - current = _grid_stats(obs) - forecast = _grid_stats(forecast_obs) + return { + "current": _grid_stats(obs), + "forecast_obs": forecast_obs, + "simulator": simulator, + "sigma": sigma, + "n_load": n_load, + } + + def _predict_line( + self, + env: Any, + line_name: str, + context: Dict[str, Any], + ) -> Dict[str, Any]: + normalized_line = str(line_name).replace("line_", "") + if normalized_line not in self.line_map: + raise ValueError(f"Unknown line for failure classifier: {line_name!r}") + + contingency_obs = _contingency_observation( + env, + context["forecast_obs"], + context["simulator"], + line_name, + ) + current = context["current"] + scenario = _grid_stats(contingency_obs) + sigma = context["sigma"] + n_load = context["n_load"] sum_gen = current["sum_gen_p"] row = { "line_id_encoded": self.line_map[normalized_line], **current, "load_gen_ratio": current["sum_load_p"] / (sum_gen + 1e-6), - **{f"fcast_{key}": value for key, value in forecast.items()}, + **{f"fcast_{key}": value for key, value in scenario.items()}, "aleatoric_load_p_mean": float(np.mean(sigma[:n_load])), "aleatoric_load_q_mean": float(np.mean(sigma[n_load:2 * n_load])), "aleatoric_gen_p_mean": float(np.mean(sigma[2 * n_load:])), @@ -300,6 +349,21 @@ def predict(self, env: Any, obs: Any, line_name: str) -> Dict[str, Any]: "line": normalized_line, } + def predict(self, env: Any, obs: Any, line_name: str) -> Dict[str, Any]: + """Forecast failure for the line involved in the current overload.""" + return self._predict_line(env, line_name, self._forecast_context(env, obs)) + + def predict_many( + self, + env: Any, + obs: Any, + line_names: Optional[Sequence[str]] = None, + ) -> list[Dict[str, Any]]: + """Return one HGB result per line from one common forecast state.""" + lines = list(line_names) if line_names is not None else list(self.line_map) + context = self._forecast_context(env, obs) + return [self._predict_line(env, line, context) for line in lines] + def build_failure_forecast_event( context_date: dt.datetime, @@ -335,7 +399,7 @@ def build_failure_forecast_event( return { "criticality": criticality, - "title": f"Failure prediction {probability_pct:.1f}%", + "title": f"Predicted probability of failure: {probability_pct:.1f}%", "description": description, "start_date": f"{context_date}", "end_date": f"{context_date + dt.timedelta(minutes=5)}", diff --git a/usecases_examples/PowerGrid/config/CONFIG.toml b/usecases_examples/PowerGrid/config/CONFIG.toml index 8f39ac94..c7fdd7ea 100644 --- a/usecases_examples/PowerGrid/config/CONFIG.toml +++ b/usecases_examples/PowerGrid/config/CONFIG.toml @@ -10,4 +10,4 @@ assistant_path = "Ressources/XD_silly_repo" assistant_seed = 1227139268 stepDuration_s = 0.5 scenario_first_step = 752 -failure_forecast_artifact_dir = "failure_forecaster" +failure_forecast_artifact_dir = "Ressources/failure_forecast" diff --git a/usecases_examples/PowerGrid/requirements-app.txt b/usecases_examples/PowerGrid/requirements-app.txt index 81a01aa1fd2a5b532e0a2d667ece717f6f296d68..da834858daf6583c415451b478914164fdbf5e1b 100644 GIT binary patch delta 65 zcmZpWZIIni$Hr{JV6eH7&6rW$oI#Jlkb#$hiy@Dplp&X)fT5DX76=W2;zkT6K++f_ OJGqfrbn^%H32XqNi41fA delta 65 zcmZpWZIIni$Hr{TV7a-G&6rW$j6sjVgn^fViy@Dplp&X)fT5DX76=W2;zkU{K++T> OJGqfrbn^%H32XqPoeX~f From 8459388fbbcaeff464e19fdeaf2dba93bcc969fc Mon Sep 17 00:00:00 2001 From: Oklyagina Date: Fri, 23 Oct 2026 20:30:30 +0100 Subject: [PATCH 4/8] deploy updated failure forecast models and align PowerGrid runtime --- usecases_examples/PowerGrid/README.md | 6 ++ .../PowerGrid/app/models/Simulator.py | 58 +++++++++++-------- .../PowerGrid/app/models/failure_forecast.py | 35 ++++++++++- 3 files changed, 71 insertions(+), 28 deletions(-) diff --git a/usecases_examples/PowerGrid/README.md b/usecases_examples/PowerGrid/README.md index 1f83bbe3..c46eb0c0 100644 --- a/usecases_examples/PowerGrid/README.md +++ b/usecases_examples/PowerGrid/README.md @@ -77,6 +77,12 @@ failure_classifier.pkl failure_classifier_metadata.json ``` +The current classifier covers the 10 selected transmission lines recorded in +`failure_classifier_metadata.json`. Overloads on other lines are still sent to +InteractiveAI, but no failure-forecast child event is produced for them. Keep +all four files from the same training run together; mixing model generations is +rejected when the simulator starts. + For a local validation with artifacts stored elsewhere, set `FAILURE_FORECAST_ARTIFACT_DIR` to that directory before starting the simulator. The default remains `Ressources/failure_forecast`. diff --git a/usecases_examples/PowerGrid/app/models/Simulator.py b/usecases_examples/PowerGrid/app/models/Simulator.py index 9c85401a..2da28a31 100644 --- a/usecases_examples/PowerGrid/app/models/Simulator.py +++ b/usecases_examples/PowerGrid/app/models/Simulator.py @@ -370,32 +370,40 @@ def run_simulator(self, com): or overload_response.get("id") ) if parent_event_id and self.failure_forecast is not None: - try: - forecast = self.failure_forecast.predict( - self.env, - self.obs, - forecast_line - ) - com.send_failure_forecast_event( - context_date, - self.config['scenario_first_step'], - forecast, - parent_event_id, - self.obs, - overloaded_line, - img_b64_current - ) - logging.info( - "Failure forecast for line %s: %s", - overloaded_line, - forecast - ) - except Exception as exc: - logging.error( - "Failure forecast failed for line %s: %s", - overloaded_line, - exc + if not self.failure_forecast.supports_line(forecast_line): + logging.warning( + "No failure forecast is available for line %s; " + "the overload event will remain without a " + "failure-forecast child event", + overloaded_line ) + else: + try: + forecast = self.failure_forecast.predict( + self.env, + self.obs, + forecast_line + ) + com.send_failure_forecast_event( + context_date, + self.config['scenario_first_step'], + forecast, + parent_event_id, + self.obs, + overloaded_line, + img_b64_current + ) + logging.info( + "Failure forecast for line %s: %s", + overloaded_line, + forecast + ) + except Exception as exc: + logging.error( + "Failure forecast failed for line %s: %s", + overloaded_line, + exc + ) elif self.failure_forecast is not None: logging.error( "Failure forecast event was not sent because " diff --git a/usecases_examples/PowerGrid/app/models/failure_forecast.py b/usecases_examples/PowerGrid/app/models/failure_forecast.py index 1a47a56c..f1c9741a 100644 --- a/usecases_examples/PowerGrid/app/models/failure_forecast.py +++ b/usecases_examples/PowerGrid/app/models/failure_forecast.py @@ -95,9 +95,9 @@ def values(lag_steps: int, attr: str, size: int) -> np.ndarray: values(2016, "load_q", sizes[1]), values(2016, "gen_p", sizes[2]), ]) - hour_cos, hour_sin = _cyclic(getattr(obs, "hour_of_day", 0), 23) - minute_cos, minute_sin = _cyclic(getattr(obs, "minute_of_hour", 0), 59) - dow_cos, dow_sin = _cyclic(getattr(obs, "day_of_week", 0), 6) + hour_cos, hour_sin = _cyclic(getattr(obs, "hour_of_day", 0), 24) + minute_cos, minute_sin = _cyclic(getattr(obs, "minute_of_hour", 0), 60) + dow_cos, dow_sin = _cyclic(getattr(obs, "day_of_week", 0), 7) temporal = np.asarray([ getattr(obs, "day", 1), hour_cos, @@ -241,11 +241,40 @@ def __init__(self, artifact_dir: str | Path): for name, encoded in self.metadata.get("line_map", {}).items() } self.threshold = float(self.metadata.get("decision_threshold", 0.5)) + self._validate_bundle() self.history: Deque[ObservationHistoryEntry] = deque( maxlen=HISTORY_STEPS + 1 ) self.last_recorded_step: Optional[int] = None + def _validate_bundle(self) -> None: + """Reject incomplete or mixed-generation deployment artifacts.""" + if self.features != CLASSIFIER_FEATURES: + raise ValueError( + "Failure forecast classifier features do not match this " + "PowerGrid runtime. Deploy a compatible metadata/model bundle." + ) + if not self.line_map: + raise ValueError("Failure forecast metadata has an empty line mapping") + horizon_steps = self.metadata.get("forecast_horizon_steps") + if horizon_steps != HORIZON_STEPS: + raise ValueError( + "Failure forecast metadata has an incompatible horizon: " + f"expected {HORIZON_STEPS} steps, got {horizon_steps!r}" + ) + classifier_width = getattr(self.classifier, "n_features_in_", None) + if classifier_width is not None and int(classifier_width) != len(self.features): + raise ValueError( + "Failure classifier input width does not match its metadata: " + f"model expects {classifier_width}, metadata defines " + f"{len(self.features)} features" + ) + + def supports_line(self, line_name: str) -> bool: + """Return whether the deployed classifier was trained for a line.""" + normalized_line = str(line_name).replace("line_", "") + return normalized_line in self.line_map + def record_observation(self, obs: Any) -> None: """Record each simulator step once, including the warm-up period.""" step = int(getattr(obs, "current_step", -1)) From d7f1774821ae769caa56270dc6a58c1bfc429092 Mon Sep 17 00:00:00 2001 From: Oklyagina Date: Fri, 23 Oct 2026 20:48:23 +0100 Subject: [PATCH 5/8] add per-line failure forecast probabilities to PowerGrid events --- usecases_examples/PowerGrid/README.md | 16 +++--- .../PowerGrid/app/models/Simulator.py | 57 ++++++++----------- .../PowerGrid/app/models/failure_forecast.py | 41 ++++++------- 3 files changed, 54 insertions(+), 60 deletions(-) diff --git a/usecases_examples/PowerGrid/README.md b/usecases_examples/PowerGrid/README.md index c46eb0c0..6aa08fbe 100644 --- a/usecases_examples/PowerGrid/README.md +++ b/usecases_examples/PowerGrid/README.md @@ -78,10 +78,9 @@ failure_classifier_metadata.json ``` The current classifier covers the 10 selected transmission lines recorded in -`failure_classifier_metadata.json`. Overloads on other lines are still sent to -InteractiveAI, but no failure-forecast child event is produced for them. Keep -all four files from the same training run together; mixing model generations is -rejected when the simulator starts. +`failure_classifier_metadata.json`. Each overload triggers one child event with +predictions for all 10 lines. Keep all four files from the same training run +together; mixing model generations is rejected when the simulator starts. For a local validation with artifacts stored elsewhere, set `FAILURE_FORECAST_ARTIFACT_DIR` to that directory before starting the @@ -89,10 +88,11 @@ simulator. The default remains `Ressources/failure_forecast`. The simulator keeps the required one-hour, one-day, and one-week observation history in memory. When an overload event is accepted by InteractiveAI, the -simulator builds the one-hour-ahead grid state, disconnects that overloaded line, -derives the HGB features from the resulting contingency state, and publishes a -second event whose `parent_event_id` is the overload event ID. The event KPIs -retain both `failure_probability` and `failure_prediction`. +simulator builds the one-hour-ahead grid state, disconnects each of the 10 +trained lines independently, and publishes a second event whose +`parent_event_id` is the overload event ID. The event KPIs contain +`failure_probability`, `failure_probability_pct`, and `failure_prediction` for +each trained line. # 2 Run the simulator diff --git a/usecases_examples/PowerGrid/app/models/Simulator.py b/usecases_examples/PowerGrid/app/models/Simulator.py index 2da28a31..48199db3 100644 --- a/usecases_examples/PowerGrid/app/models/Simulator.py +++ b/usecases_examples/PowerGrid/app/models/Simulator.py @@ -370,40 +370,31 @@ def run_simulator(self, com): or overload_response.get("id") ) if parent_event_id and self.failure_forecast is not None: - if not self.failure_forecast.supports_line(forecast_line): - logging.warning( - "No failure forecast is available for line %s; " - "the overload event will remain without a " - "failure-forecast child event", - overloaded_line + try: + forecasts = self.failure_forecast.predict_many( + self.env, + self.obs + ) + com.send_failure_forecast_event( + context_date, + self.config['scenario_first_step'], + forecasts, + parent_event_id, + self.obs, + forecast_line, + img_b64_current + ) + logging.info( + "Failure forecasts triggered by line %s: %s", + overloaded_line, + forecasts + ) + except Exception as exc: + logging.error( + "Failure forecast failed for line %s: %s", + overloaded_line, + exc ) - else: - try: - forecast = self.failure_forecast.predict( - self.env, - self.obs, - forecast_line - ) - com.send_failure_forecast_event( - context_date, - self.config['scenario_first_step'], - forecast, - parent_event_id, - self.obs, - overloaded_line, - img_b64_current - ) - logging.info( - "Failure forecast for line %s: %s", - overloaded_line, - forecast - ) - except Exception as exc: - logging.error( - "Failure forecast failed for line %s: %s", - overloaded_line, - exc - ) elif self.failure_forecast is not None: logging.error( "Failure forecast event was not sent because " diff --git a/usecases_examples/PowerGrid/app/models/failure_forecast.py b/usecases_examples/PowerGrid/app/models/failure_forecast.py index f1c9741a..4586a01f 100644 --- a/usecases_examples/PowerGrid/app/models/failure_forecast.py +++ b/usecases_examples/PowerGrid/app/models/failure_forecast.py @@ -396,40 +396,43 @@ def predict_many( def build_failure_forecast_event( context_date: dt.datetime, - forecast: Dict[str, Any], + forecasts: Sequence[Dict[str, Any]], parent_event_id: str, line_name: str, img_b64: Optional[str] = None, ) -> Dict[str, Any]: - """Build a schema-compatible event linked to an overload event.""" - probability_pct = float(forecast["failure_probability_pct"]) - will_fail = bool(forecast["failure_prediction"]) - criticality = "HIGH" # as there's already an overload in the line - if will_fail: - description = ( - "The agent is likely unable to provide a solution " - f"for overload on line {line_name} within one hour." - ) - else: - description = ( - "The agent is likely able to provide a solution for overload " - f"on line {line_name} within one hour." - ) + """Build one event containing failure probabilities for every line.""" + failure_forecasts = { + str(forecast["line"]).replace("line_", ""): { + "failure_probability": float(forecast["failure_probability"]), + "failure_probability_pct": float(forecast["failure_probability_pct"]), + "failure_prediction": int(forecast["failure_prediction"]), + } + for forecast in forecasts + } event_data = { "event_type": "anticipation", # InteractiveAI currently de-duplicates PowerGrid events by data.line. # Keep the forecast distinct from its overload parent. "line": f"failure_forecast:{line_name}", - "kpis": dict(forecast), + "kpis": { + "trigger_line": str(line_name).replace("line_", ""), + "forecast_horizon_steps": int(HORIZON_STEPS), + "forecast_horizon_minutes": int(HORIZON_STEPS * STEP_MINUTES), + "failure_forecasts": failure_forecasts, + }, } if img_b64 is not None: event_data["event_context"] = img_b64 return { - "criticality": criticality, - "title": f"Predicted probability of failure: {probability_pct:.1f}%", - "description": description, + "criticality": "HIGH", + "title": "Predicted failure probabilities by line", + "description": ( + f"One-hour-ahead failure predictions triggered by overload on " + f"line {line_name}." + ), "start_date": f"{context_date}", "end_date": f"{context_date + dt.timedelta(minutes=5)}", "parent_event_id": str(parent_event_id), From 05a4117dca6e53130e49f2f2c14fb9b1e8f6364d Mon Sep 17 00:00:00 2001 From: Oklyagina Date: Fri, 30 Oct 2026 18:46:54 +0000 Subject: [PATCH 6/8] added explainability + 10 lines --- usecases_examples/PowerGrid/Dockerfile.app | 7 ++- usecases_examples/PowerGrid/README.md | 17 +++++- .../PowerGrid/app/models/failure_forecast.py | 54 ++++++++++++++++++- 3 files changed, 73 insertions(+), 5 deletions(-) diff --git a/usecases_examples/PowerGrid/Dockerfile.app b/usecases_examples/PowerGrid/Dockerfile.app index f3a49439..e13e8809 100644 --- a/usecases_examples/PowerGrid/Dockerfile.app +++ b/usecases_examples/PowerGrid/Dockerfile.app @@ -1,13 +1,16 @@ # syntax=docker/dockerfile:1 -FROM python:3.10-slim-bullseye +FROM python:3.10-slim-bookworm EXPOSE 5000 RUN mkdir /code COPY . /code/ WORKDIR /code -RUN apt-get update && apt-get install ffmpeg libsm6 libxext6 -y +RUN apt-get update \ + && apt-get install -y --no-install-recommends ffmpeg libsm6 libxext6 \ + && rm -rf /var/lib/apt/lists/* + RUN pip3 install -r requirements-app.txt RUN pip3 install -r requirements-forecast.txt diff --git a/usecases_examples/PowerGrid/README.md b/usecases_examples/PowerGrid/README.md index 6aa08fbe..28c7b421 100644 --- a/usecases_examples/PowerGrid/README.md +++ b/usecases_examples/PowerGrid/README.md @@ -75,12 +75,20 @@ mean_forecaster.pkl aleatoric_forecaster.pkl failure_classifier.pkl failure_classifier_metadata.json +explainability/ +`-- rules/ + `-- line_/best_rule.py ``` The current classifier covers the 10 selected transmission lines recorded in `failure_classifier_metadata.json`. Each overload triggers one child event with predictions for all 10 lines. Keep all four files from the same training run -together; mixing model generations is rejected when the simulator starts. +together and deploy the matching explanatory rules from that run; mixing model +generations is rejected when the simulator starts. + +The explanatory rules are distilled offline in the failure-forecast project. +At runtime, PowerGrid validates and evaluates the local `best_rule.py` files; +it does not require an LLM connection, API key, or network request. For a local validation with artifacts stored elsewhere, set `FAILURE_FORECAST_ARTIFACT_DIR` to that directory before starting the @@ -92,7 +100,12 @@ simulator builds the one-hour-ahead grid state, disconnects each of the 10 trained lines independently, and publishes a second event whose `parent_event_id` is the overload event ID. The event KPIs contain `failure_probability`, `failure_probability_pct`, and `failure_prediction` for -each trained line. +each trained line. Every per-line result also contains `explanation`, +`rule_prediction`, and `rule_agrees_with_forecast`. The HGB +`failure_prediction` remains authoritative. `rule_prediction` is the result of +the simplified explanatory rule and can disagree with the HGB. If a rule is +missing or cannot be evaluated, the three explainability fields are `null` and +the HGB forecast is still published. # 2 Run the simulator diff --git a/usecases_examples/PowerGrid/app/models/failure_forecast.py b/usecases_examples/PowerGrid/app/models/failure_forecast.py index 4586a01f..3f7ccb98 100644 --- a/usecases_examples/PowerGrid/app/models/failure_forecast.py +++ b/usecases_examples/PowerGrid/app/models/failure_forecast.py @@ -8,6 +8,7 @@ import datetime as dt import json +import logging from collections import deque from dataclasses import dataclass from pathlib import Path @@ -16,11 +17,15 @@ import numpy as np import pandas as pd +from app.models.explainability import RuleExplainer + HORIZON_STEPS = 12 STEP_MINUTES = 5 HISTORY_STEPS = 2016 +LOGGER = logging.getLogger(__name__) + CLASSIFIER_FEATURES = [ "line_id_encoded", "sum_load_p", @@ -242,6 +247,9 @@ def __init__(self, artifact_dir: str | Path): } self.threshold = float(self.metadata.get("decision_threshold", 0.5)) self._validate_bundle() + self.rule_explainer = RuleExplainer( + self.artifact_dir / "explainability" / "rules" + ) self.history: Deque[ObservationHistoryEntry] = deque( maxlen=HISTORY_STEPS + 1 ) @@ -291,6 +299,35 @@ def _failure_probability(self, features: pd.DataFrame) -> float: raise ValueError("Failure classifier does not expose class 1") return float(probabilities[0, matches[0]]) + def _explain_prediction( + self, + line: str, + row: Dict[str, Any], + failure_prediction: int, + ) -> Dict[str, Any]: + """Best-effort explanation that never replaces the HGB prediction.""" + unavailable = { + "explanation": None, + "rule_prediction": None, + "rule_agrees_with_forecast": None, + } + explainer = getattr(self, "rule_explainer", None) + if explainer is None or line not in explainer.available_lines: + LOGGER.warning("No explanatory rule is available for line %s", line) + return unavailable + try: + explanation = explainer.explain(line, row) + except (KeyError, TypeError, ValueError, ArithmeticError) as exc: + LOGGER.warning("Explanation failed for line %s: %s", line, exc) + return unavailable + return { + "explanation": explanation.text, + "rule_prediction": explanation.prediction, + "rule_agrees_with_forecast": ( + explanation.prediction == failure_prediction + ), + } + def _forecast_context(self, env: Any, obs: Any) -> Dict[str, Any]: """Build the common t+12 operating state used by all contingencies.""" self.record_observation(obs) @@ -369,13 +406,17 @@ def _predict_line( {feature: row.get(feature, np.nan) for feature in self.features} ]) probability = self._failure_probability(classifier_input) + failure_prediction = int(probability >= self.threshold) return { - "failure_prediction": int(probability >= self.threshold), + "failure_prediction": failure_prediction, "failure_probability": round(probability, 3), "failure_probability_pct": round(probability * 100.0, 1), "forecast_horizon_steps": int(HORIZON_STEPS), "forecast_horizon_minutes": int(HORIZON_STEPS * STEP_MINUTES), "line": normalized_line, + **self._explain_prediction( + normalized_line, row, failure_prediction + ), } def predict(self, env: Any, obs: Any, line_name: str) -> Dict[str, Any]: @@ -407,6 +448,17 @@ def build_failure_forecast_event( "failure_probability": float(forecast["failure_probability"]), "failure_probability_pct": float(forecast["failure_probability_pct"]), "failure_prediction": int(forecast["failure_prediction"]), + "explanation": forecast.get("explanation"), + "rule_prediction": ( + None + if forecast.get("rule_prediction") is None + else int(forecast["rule_prediction"]) + ), + "rule_agrees_with_forecast": ( + None + if forecast.get("rule_agrees_with_forecast") is None + else bool(forecast["rule_agrees_with_forecast"]) + ), } for forecast in forecasts } From fa2c16d0cb5eb848cc14e28c0684b151a024fde4 Mon Sep 17 00:00:00 2001 From: Oklyagina Date: Fri, 30 Oct 2026 18:51:10 +0000 Subject: [PATCH 7/8] added explainability + 10 lines --- .../PowerGrid/app/models/explainability.py | 292 ++++++++++++++++++ 1 file changed, 292 insertions(+) create mode 100644 usecases_examples/PowerGrid/app/models/explainability.py diff --git a/usecases_examples/PowerGrid/app/models/explainability.py b/usecases_examples/PowerGrid/app/models/explainability.py new file mode 100644 index 00000000..d0951a80 --- /dev/null +++ b/usecases_examples/PowerGrid/app/models/explainability.py @@ -0,0 +1,292 @@ +"""Load, validate, evaluate, and describe failure-forecast rules. + +The rules are distilled offline in the RL failure-forecast project. This +module is intentionally self-contained so the PowerGrid runtime only needs the +deployed ``best_rule.py`` artifacts and never needs an LLM connection. +""" +from __future__ import annotations + +import ast +import logging +import textwrap +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable, Mapping + + +LOGGER = logging.getLogger(__name__) + +FEATURE_TEXT = { + "sum_load_p": "total active power load at the current time", + "sum_load_q": "total reactive power load at the current time", + "sum_gen_p": "total active power generation at the current time", + "var_line_rho": "variance of line loading at the current time", + "avg_line_rho": "average line loading at the current time", + "max_line_rho": "maximum line loading at the current time", + "nb_rho_ge_0.95": "number of highly loaded lines at the current time", + "load_gen_ratio": "load-to-generation ratio at the current time", + "aleatoric_load_p_mean": "forecast uncertainty for active load", + "aleatoric_load_q_mean": "forecast uncertainty for reactive load", + "aleatoric_gen_p_mean": "forecast uncertainty for generation", + "epistemic_before": "epistemic uncertainty at the current time", + "epistemic_after": "epistemic uncertainty in the forecast state", + "fcast_sum_load_p": "forecast total active power load one hour ahead", + "fcast_sum_load_q": "forecast total reactive power load one hour ahead", + "fcast_sum_gen_p": "forecast total generation one hour ahead", + "fcast_var_line_rho": "forecast variance of line loading one hour ahead", + "fcast_avg_line_rho": "forecast average line loading one hour ahead", + "fcast_max_line_rho": "forecast maximum line loading one hour ahead", + "fcast_nb_rho_ge_0.95": ( + "forecast number of highly loaded lines one hour ahead" + ), +} + +FEATURE_UNITS = { + "sum_load_p": "MW", + "sum_load_q": "MVAR", + "sum_gen_p": "MW", + "fcast_sum_load_p": "MW", + "fcast_sum_load_q": "MVAR", + "fcast_sum_gen_p": "MW", +} + +_OP_TEXT = { + ast.GtE: ">=", + ast.Gt: ">", + ast.LtE: "<=", + ast.Lt: "<", + ast.Eq: "=", + ast.NotEq: "!=", +} + +_INVERTED_OP_TEXT = { + ast.GtE: "<", + ast.Gt: "<=", + ast.LtE: ">", + ast.Lt: ">=", + ast.Eq: "!=", + ast.NotEq: "=", +} + +_ALLOWED_NODES = ( + ast.Module, + ast.FunctionDef, + ast.arguments, + ast.arg, + ast.If, + ast.Return, + ast.Compare, + ast.BoolOp, + ast.And, + ast.Or, + ast.Subscript, + ast.Name, + ast.Load, + ast.Constant, + ast.UnaryOp, + ast.USub, + ast.Gt, + ast.GtE, + ast.Lt, + ast.LtE, + ast.Eq, + ast.NotEq, +) + + +def _feature_name(node: ast.expr) -> str: + if not isinstance(node, ast.Subscript): + raise ValueError("A rule condition must read a feature from x[...].") + if not isinstance(node.value, ast.Name) or node.value.id != "x": + raise ValueError("A rule condition must read a feature from x[...].") + if isinstance(node.slice, ast.Constant) and isinstance(node.slice.value, str): + return node.slice.value + raise ValueError("A rule feature name must be a string literal.") + + +def _number(node: ast.expr) -> float: + if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)): + return float(node.value) + if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub): + return -_number(node.operand) + raise ValueError("A rule threshold must be numeric.") + + +def _format_threshold(value: float, unit: str | None) -> str: + rendered = f"{value:.0f}" if abs(value) >= 10 else f"{value:.4g}" + return f"{rendered} {unit}" if unit else rendered + + +def _condition_text(node: ast.expr, *, negated: bool = False) -> str: + if ( + isinstance(node, ast.Compare) + and len(node.ops) == len(node.comparators) == 1 + ): + feature = _feature_name(node.left) + value = _number(node.comparators[0]) + operators = _INVERTED_OP_TEXT if negated else _OP_TEXT + operator = operators.get(type(node.ops[0])) + if operator is None: + raise ValueError("Unsupported comparison operator in rule.") + description = FEATURE_TEXT.get(feature, feature.replace("_", " ")) + threshold = _format_threshold(value, FEATURE_UNITS.get(feature)) + return f"the {description} is {operator} {threshold}" + if isinstance(node, ast.BoolOp): + joiner = " or " if isinstance(node.op, ast.Or) else " and " + if negated: + joiner = " and " if joiner == " or " else " or " + return "(" + joiner.join( + _condition_text(value, negated=negated) for value in node.values + ) + ")" + raise ValueError("Unsupported rule condition.") + + +def validate_rule(rule_code: str) -> ast.FunctionDef: + """Validate the deliberately small, side-effect-free rule language.""" + try: + tree = ast.parse(textwrap.dedent(rule_code).strip()) + except (SyntaxError, TypeError) as exc: + raise ValueError(f"Invalid rule syntax: {exc}") from exc + + functions = [node for node in tree.body if isinstance(node, ast.FunctionDef)] + if len(tree.body) != 1 or len(functions) != 1 or functions[0].name != "rule": + raise ValueError("Rule source must contain exactly one function named 'rule'.") + + function = functions[0] + arguments = function.args + if ( + len(arguments.args) != 1 + or arguments.args[0].arg != "x" + or arguments.vararg is not None + or arguments.kwarg is not None + or arguments.defaults + or arguments.kw_defaults + ): + raise ValueError("Rule signature must be exactly def rule(x):") + + for node in ast.walk(tree): + if not isinstance(node, _ALLOWED_NODES): + raise ValueError(f"Unsupported syntax in rule: {type(node).__name__}") + if isinstance(node, ast.Name) and node.id not in {"x", "rule"}: + raise ValueError(f"Unsupported name in rule: {node.id}") + if isinstance(node, ast.Compare): + if len(node.ops) != 1 or len(node.comparators) != 1: + raise ValueError("Rule comparisons must use one numeric threshold.") + feature = _feature_name(node.left) + if feature not in FEATURE_TEXT: + raise ValueError(f"Unsupported feature in rule: {feature}") + _number(node.comparators[0]) + if isinstance(node, ast.Return): + try: + value = _number(node.value) if node.value is not None else None + except ValueError as exc: + raise ValueError( + "Rule return values must be exactly 0 or 1." + ) from exc + if value not in {0.0, 1.0}: + raise ValueError("Rule return values must be exactly 0 or 1.") + return function + + +def compile_rule(rule_code: str) -> Callable[[Mapping[str, float]], int]: + """Compile validated rule source without exposing Python built-ins.""" + validate_rule(rule_code) + namespace: dict[str, Any] = {} + exec( + compile(textwrap.dedent(rule_code), "", "exec"), + {"__builtins__": {}}, + namespace, + ) + return namespace["rule"] + + +def _failure_paths( + statements: list[ast.stmt], conditions: list[str] +) -> list[list[str]]: + if not statements: + return [] + first, rest = statements[0], statements[1:] + if isinstance(first, ast.Return): + return [conditions] if _number(first.value) == 1 else [] + if not isinstance(first, ast.If): + return _failure_paths(rest, conditions) + return ( + _failure_paths( + list(first.body) + rest, + conditions + [_condition_text(first.test)], + ) + + _failure_paths( + list(first.orelse) + rest, + conditions + [_condition_text(first.test, negated=True)], + ) + ) + + +def translate_rule_to_sentence(rule_code: str, line_name: str) -> str: + """Describe every branch that produces a rule-based failure prediction.""" + function = validate_rule(rule_code) + paths = _failure_paths(list(function.body), []) + line_name = str(line_name).removeprefix("line_") + if not paths: + return f"The explanatory rule for contingency {line_name} never predicts failure." + clauses = [" and ".join(path) if path else "all conditions" for path in paths] + return ( + f"For a contingency on line {line_name}, the simplified explanatory rule " + f"predicts that the RL agent may fail when " + + "; or when ".join(clauses) + + "." + ) + + +@dataclass(frozen=True) +class RuleExplanation: + line: str + prediction: int + text: str + + +class RuleExplainer: + """Load valid ``line_*/best_rule.py`` files from deployed artifacts.""" + + def __init__(self, rules_dir: str | Path): + self.rules_dir = Path(rules_dir) + self._codes: dict[str, str] = {} + self._rules: dict[str, Callable[[Mapping[str, float]], int]] = {} + if not self.rules_dir.exists(): + LOGGER.warning( + "Explanatory rules directory does not exist: %s", self.rules_dir + ) + return + for path in sorted(self.rules_dir.glob("line_*/best_rule.py")): + line = path.parent.name.removeprefix("line_") + try: + code = path.read_text(encoding="utf-8") + rule = compile_rule(code) + except (OSError, UnicodeError, ValueError, SyntaxError) as exc: + LOGGER.warning("Skipping invalid explanatory rule %s: %s", path, exc) + continue + self._rules[line] = rule + self._codes[line] = code + + @property + def available_lines(self) -> list[str]: + return sorted(self._rules) + + def describe(self, line: str) -> str: + name = str(line).removeprefix("line_") + if name not in self._codes: + raise KeyError(f"No explanatory rule is available for line {name!r}.") + return translate_rule_to_sentence(self._codes[name], name) + + def explain( + self, line: str, features: Mapping[str, float] + ) -> RuleExplanation: + name = str(line).removeprefix("line_") + if name not in self._rules: + raise KeyError(f"No explanatory rule is available for line {name!r}.") + prediction = int(self._rules[name](features)) + if prediction not in {0, 1}: + raise ValueError( + f"Rule for line {name!r} returned {prediction}, expected 0 or 1" + ) + return RuleExplanation(name, prediction, self.describe(name)) From 089badc0812eb2f890acede11a6c9b389832f6b6 Mon Sep 17 00:00:00 2001 From: Oklyagina Date: Fri, 30 Oct 2026 19:11:16 +0000 Subject: [PATCH 8/8] rephrasing --- usecases_examples/PowerGrid/app/models/explainability.py | 2 +- usecases_examples/PowerGrid/app/models/failure_forecast.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/usecases_examples/PowerGrid/app/models/explainability.py b/usecases_examples/PowerGrid/app/models/explainability.py index d0951a80..e6726f1b 100644 --- a/usecases_examples/PowerGrid/app/models/explainability.py +++ b/usecases_examples/PowerGrid/app/models/explainability.py @@ -223,7 +223,7 @@ def _failure_paths( def translate_rule_to_sentence(rule_code: str, line_name: str) -> str: - """Describe every branch that produces a rule-based failure prediction.""" + """Describe every branch that produces a rule-based predicted probability of failure.""" function = validate_rule(rule_code) paths = _failure_paths(list(function.body), []) line_name = str(line_name).removeprefix("line_") diff --git a/usecases_examples/PowerGrid/app/models/failure_forecast.py b/usecases_examples/PowerGrid/app/models/failure_forecast.py index 3f7ccb98..d8c308ad 100644 --- a/usecases_examples/PowerGrid/app/models/failure_forecast.py +++ b/usecases_examples/PowerGrid/app/models/failure_forecast.py @@ -480,9 +480,9 @@ def build_failure_forecast_event( return { "criticality": "HIGH", - "title": "Predicted failure probabilities by line", + "title": "Predicted probability of failure by line", "description": ( - f"One-hour-ahead failure predictions triggered by overload on " + f"One-hour-ahead predicted probabilities of failure triggered by overload on " f"line {line_name}." ), "start_date": f"{context_date}",