# Copyright (c) 2025-2026, RTE (https://www.rte-france.com) # This Source Code Form is subject to the terms of the Mozilla Public License, version 2.0. # If a copy of the Mozilla Public License, version 2.0 was not distributed with this file, # you can obtain one at http://mozilla.org/MPL/2.0/. # SPDX-License-Identifier: MPL-2.0 # This file is part of Co-Study4Grid a Power Grid Study tool Assistant Interface to help solve contigencies for a grid state under study. """Stateless helpers for simulation_mixin. These are the small, self-contained pieces that used to live as inline blocks inside `simulate_manual_action` and `compute_superposition`. Extracting them: - Shrinks the orchestrator methods (they were 599 / 285 lines), - Lets each concern be unit-tested in isolation, and - Makes the data flow explicit (arguments vs. closures). """ from __future__ import annotations import logging import math import re from typing import Any import numpy as np from expert_backend.services.sanitize import sanitize_for_json logger = logging.getLogger(__name__) # Topology keys recognised by `_build_action_entry_from_topology` — also # used to distinguish "single topology dict" from "dict of per-action # topologies" in the caller-provided `action_content`. TOPO_KEYS: frozenset[str] = frozenset({ "lines_ex_bus", "lines_or_bus", "gens_bus", "loads_bus", "substations", "switches", "loads_p", "gens_p", "pst_tap", }) # Minimum disconnected load (MW) required to flag an islanding event. # Below this, an extra connected component is treated as an AC load-flow # numerical artefact (typically fast-mode local non-convergence on # extreme PST taps) rather than a real outage. ISLANDING_MW_THRESHOLD: float = 1.0 def canonicalize_action_id(action_id: str) -> str: """Return a canonical "+"-joined ID (components sorted alphabetically).""" if not action_id or "+" not in action_id: return action_id return "+".join(sorted(p.strip() for p in action_id.split("+"))) def is_switch_only_content(content: Any) -> bool: """True if ``content`` carries only a non-empty ``switches`` dict. Used by ``simulate_manual_action`` to decide whether to auto-build a human-readable description for user-built SLD-edit actions, instead of falling back to the raw ``action_id`` (which is just a generated placeholder like ``user_topo__``). """ if not isinstance(content, dict): return False switches = content.get("switches") if not isinstance(switches, dict) or not switches: return False other_keys = [k for k in content.keys() if k != "switches"] return len(other_keys) == 0 def build_switch_action_description( switches: dict[str, bool], voltage_level_id: str | None = None, ) -> str: """Return ``"Manoeuvre manuelle sur : A ouvert, B fermé"`` .""" if not switches: return "Manoeuvre manuelle (aucun switch)" parts: list[str] = [] for sw_id, is_open in switches.items(): verb = "ouvert" if is_open else "fermé" parts.append(f"{sw_id} {verb}") body = ", ".join(parts) if voltage_level_id: return f"Manoeuvre manuelle sur {voltage_level_id}: {body}" return f"Manoeuvre manuelle: {body}" def _fmt_setpoint_mw(value: Any) -> str: """Format an active-power setpoint for an action description (1 decimal).""" try: return f"{float(value):.1f}" except (TypeError, ValueError): return str(value) def build_manual_action_description( content: dict[str, Any], voltage_level_id: str | None = None, ) -> str: """Describe a user-built SLD-edit action (switch toggles AND/OR injection setpoint changes) in one human-readable French clause. Generalises :func:`build_switch_action_description` to the combined interactive-SLD gesture: the operator can, from the same diagram, flip breakers / disconnectors **and** retune the active power of loads / generators before simulating one manual action. For a switch-only ``content`` the output is byte-identical to :func:`build_switch_action_description`, so the existing card / filter contract is preserved. ``content`` is the post-``_build_action_entry_from_topology`` form, i.e. ``switches`` plus ``set_gen_p`` / ``set_load_p`` (absolute MW setpoints). """ parts: list[str] = [] for sw_id, is_open in (content.get("switches") or {}).items(): parts.append(f"{sw_id} {'ouvert' if is_open else 'fermé'}") for gen_id, p in (content.get("set_gen_p") or {}).items(): parts.append(f"{gen_id} P={_fmt_setpoint_mw(p)} MW") for load_id, p in (content.get("set_load_p") or {}).items(): parts.append(f"{load_id} P={_fmt_setpoint_mw(p)} MW") if not parts: return "Manoeuvre manuelle (vide)" body = ", ".join(parts) if voltage_level_id: return f"Manoeuvre manuelle sur {voltage_level_id}: {body}" return f"Manoeuvre manuelle: {body}" def compute_reduction_setpoint( element_name: str, element_type: str, target_mw: float | None, obs_n1: Any, ) -> float: """Compute the remaining MW setpoint after a target reduction. ``element_type`` is 'load' or 'gen'. Returns ``max(0, |current| - target)``. Falls back to 0.0 if ``target_mw`` is ``None``, ``obs_n1`` is missing, or the element cannot be found on the observation — i.e. the function degrades to "full reduction" rather than raising. """ if target_mw is None or obs_n1 is None: return 0.0 try: if element_type == "load": idx = list(obs_n1.name_load).index(element_name) current_mw = float(obs_n1.load_p[idx]) else: idx = list(obs_n1.name_gen).index(element_name) current_mw = float(obs_n1.gen_p[idx]) remaining = max(0.0, abs(current_mw) - float(target_mw)) return round(remaining, 2) except Exception as e: logger.warning( "[compute_reduction_setpoint] could not compute setpoint for %s: %s — falling back to 0.0", element_name, e, ) return 0.0 def compute_redispatch_setpoint( gen_name: str, delta_mw: float | None, obs_n1: Any, default_delta_mw: float = 10.0, ) -> float: """Compute the new generator setpoint for a redispatch (current + signed delta). Unlike :func:`compute_reduction_setpoint` (which *subtracts* a reduction from the current output), redispatch *adds* a signed delta: a positive ``delta_mw`` raises production, a negative one lowers it. Works in production-positive magnitude (pypowsybl ``target_p`` convention) and is floored at 0. Falls back to ``default_delta_mw`` (raise) when no delta is supplied, and to ``default_delta_mw`` itself when the generator cannot be located on the observation. """ delta = float(delta_mw) if delta_mw is not None else float(default_delta_mw) if obs_n1 is None: return round(max(0.0, delta), 2) try: idx = list(obs_n1.name_gen).index(gen_name) current_mw = abs(float(obs_n1.gen_p[idx])) return round(max(0.0, current_mw + delta), 2) except Exception as e: logger.warning( "[compute_redispatch_setpoint] could not compute setpoint for %s: %s — using delta only", gen_name, e, ) return round(max(0.0, delta), 2) _PST_TAP_PATTERN = re.compile(r"(pst(?:_tap)?_(.+))_(inc|dec)(\d+)$") def parse_pst_tap_id(action_id: str) -> tuple[str, int] | None: """Parse a dynamic PST action ID like ``pst_tap__inc2``. Returns ``(pst_id, signed_variation)`` or ``None`` if the ID doesn't match the expected shape. ``signed_variation`` is positive for ``inc`` and negative for ``dec``. """ match = _PST_TAP_PATTERN.match(action_id) if not match: return None _, pst_id, direction, val_str = match.groups() val = int(val_str) return pst_id, val if direction == "inc" else -val def clamp_tap(target_tap: int, pst_info: dict[str, int] | None) -> int: """Clamp a requested tap position to the PST's ``[low_tap, high_tap]`` bounds. ``pst_info`` can be ``None`` (unknown bounds) — in that case the input is returned as-is. """ if not pst_info: return int(target_tap) return max(int(pst_info["low_tap"]), min(int(pst_info["high_tap"]), int(target_tap))) def classify_action_content(action_content: Any, action_ids: list[str]) -> dict[str, Any]: """Normalise a caller-provided ``action_content`` into ``{aid: topology}``. ``action_content`` may be: - A single topology dict (any of the keys in ``TOPO_KEYS``) — broadcast to every ``aid`` in ``action_ids``. - A dict mapping ``aid -> topology`` — returned as-is. """ if not action_content: return {} if any(k in action_content for k in TOPO_KEYS): return {aid: action_content for aid in action_ids} return action_content def is_pst_action(action_id: str, dict_action: dict | None, classifier: Any) -> bool: """Detect PST actions — mirrors the logic used by the library's ``compute_all_pairs_superposition``. """ desc = (dict_action or {}).get(action_id, {}) action_type = classifier.identify_action_type(desc, by_description=True) return ( action_type in {"pst", "pst_tap"} or "pst_tap" in action_id or "pst_" in action_id ) # Injection-action detection — kept in sync with the recommender library's # ``superposition.is_injection_action`` (id prefix + classifier action type). # Replicated here (rather than imported) so it stays correct when the library # module is stubbed by the test mock layer, exactly like ``is_pst_action``. _INJECTION_ID_PREFIXES: tuple[str, ...] = ("load_shedding_", "curtail_", "redispatch_") _INJECTION_ACTION_TYPES: frozenset[str] = frozenset({ "load_power_reduction", "gen_power_reduction", "gen_redispatch", "open_load", "open_gen", }) def is_injection_action(action_id: str, dict_action: dict | None, classifier: Any) -> bool: """Detect injection actions (load shedding / curtailment / redispatch). These change only nodal injections (``set_load_p`` / ``set_gen_p``), not the topology, and are combined with topology actions through the Generalized Superposition Theorem (GST). Detection is by action-id prefix and, when the classifier resolves a type, by the injection action types — mirroring the library's ``superposition.is_injection_action``. """ if action_id and action_id.startswith(_INJECTION_ID_PREFIXES): return True desc = (dict_action or {}).get(action_id, {}) action_type = classifier.identify_action_type(desc, by_description=True) return action_type in _INJECTION_ACTION_TYPES def pst_fallback_line_idxs( action_id: str, dict_action: dict | None, all_actions: dict | None, name_line: list[str], ) -> list[int]: """Locate the PST transformer line index from an action's ``pst_tap`` content. Used when ``_identify_action_elements`` returns empty for a PST action (PST tap changes are not topology switches — they don't appear in ``lines_ex_bus`` / ``lines_or_bus``). """ entry = (dict_action or {}).get(action_id) or (all_actions or {}).get(action_id, {}) pst_tap = entry.get("content", {}).get("pst_tap", {}) if not pst_tap: pst_tap = entry.get("action_topology", {}).get("pst_tap", {}) if not pst_tap: return [] idxs: list[int] = [] for pst_name in pst_tap: if pst_name in name_line: idxs.append(name_line.index(pst_name)) return idxs def _to_1d(arr: Any) -> np.ndarray: """Coerce an array-like (including MagicMock-friendly lists) to 1D numpy.""" return np.atleast_1d(arr) def build_half_open_reactive(network: Any) -> dict[str, float]: """Return ``{branch_id_or_name: live_end_reactive_mvar}`` for lines + 2-winding transformers that are open at EXACTLY ONE terminal in the current variant. A branch open at one end is out of service for active-power transfer, but its line capacitance stays energised from the live end, so pypowsybl reports a real REACTIVE charging current there (e.g. a 225 kV line opened at one end shows ~16 MVAr at the connected terminal while p ~ 0). The current-based loading ``rho`` then reads a small non-zero value — physically correct, but easy to misread as a residual overload when the operator opened the line to relieve one. Returning the live-end reactive power lets the UI explain that an "after" loading on such a branch is capacitive charging current, not flow. The value is ``abs(q)`` at the still-connected terminal. Keys cover BOTH the IIDM id and the friendly ``name`` so a caller holding a grid2op / operator name (``MARSIL61PRAGN``) can look the branch up. Returns ``{}`` on any pypowsybl failure — the annotation is additive and must not break a run. """ out: dict[str, float] = {} for getter in ("get_lines", "get_2_windings_transformers"): try: df = getattr(network, getter)( attributes=["name", "connected1", "connected2", "q1", "q2"] ) except Exception as e: logger.debug("build_half_open_reactive: %s(attrs) failed: %s", getter, e) try: df = getattr(network, getter)() except Exception as e2: logger.debug("build_half_open_reactive: %s fallback failed: %s", getter, e2) continue try: cols = list(getattr(df, "columns", [])) if "connected1" not in cols or "connected2" not in cols: continue has_name = "name" in cols for eid, row in df.iterrows(): c1, c2 = bool(row["connected1"]), bool(row["connected2"]) if c1 == c2: continue # both connected, or both open — not "half open" q_live = row.get("q1") if c1 else row.get("q2") try: reactive = abs(float(q_live)) except (TypeError, ValueError): reactive = 0.0 if not math.isfinite(reactive): reactive = 0.0 out[str(eid)] = reactive if has_name: nm = row.get("name") if nm is not None and str(nm) != "nan": out[str(nm)] = reactive except Exception as e: logger.debug("build_half_open_reactive: scan failed for %s: %s", getter, e) continue return out def build_care_mask( action_names: np.ndarray, action_rho: np.ndarray, base_rho: np.ndarray, lines_we_care_about: Any, branches_with_limits: Any, lines_overloaded_ids: list[int], monitoring_factor: float, worsening_threshold: float, ) -> np.ndarray: """Build a boolean mask over ``action_names`` selecting "monitored" lines. Rules: 1. Must be in ``lines_we_care_about`` AND ``branches_with_limits``. 2. Exclude pre-existing N-state overloads UNLESS the action moved the line outside the symmetric ``±worsening_threshold`` band around its N value (``not_impacted``). Lines whose rho barely changes belong to "other issues" and should not pollute the action's max_rho; lines whose rho moved significantly — in either direction — belong to the action's sensitive area and must remain monitored, even when they were already overloaded. 3. Always force-include lines at ``lines_overloaded_ids`` (active monitoring). NOTE: ``action_rho`` and ``base_rho`` come from grid2op observations, where ``obs.rho`` is already pre-scaled by the monitoring factor: ``obs.rho = current / (permanent_limit * monitoring_factor)``. So ``obs.rho >= 1.0`` already means "at or above the monitoring threshold", and the comparison MUST use 1.0 — not ``monitoring_factor`` — otherwise we double-apply the factor and flag lines as overloaded at ``mf**2 ≈ 90.25 %`` of the permanent limit instead of 95 %. The ``monitoring_factor`` parameter is kept only because it sets the symmetric-impact band scale via ``worsening_threshold`` and for parity with the diagram-side ``get_overloaded_lines``, which compares pypowsybl currents (NOT pre-scaled) against ``limit * monitoring_factor``. Falls back to an all-False mask if numpy comparisons fail (legacy tests pass MagicMocks for observations). """ mask = np.isin(action_names, list(lines_we_care_about)) mask &= np.isin(action_names, list(branches_with_limits)) wt = float(worsening_threshold) OBS_RHO_OVERLOAD_THRESHOLD = 1.0 # see docstring — obs.rho is mf-scaled try: pre_existing = base_rho >= OBS_RHO_OVERLOAD_THRESHOLD not_impacted = (action_rho >= base_rho * (1 - wt)) & (action_rho <= base_rho * (1 + wt)) mask &= ~(pre_existing & not_impacted) except Exception as e: logger.warning("build_care_mask: vectorised comparison failed (mock context?): %s", e) mask = np.zeros(len(action_names), dtype=bool) for idx in lines_overloaded_ids: if idx < len(mask): mask[idx] = True return mask def resolve_lines_overloaded( obs_simu_defaut: Any, obs_n: Any, analysis_context_overloaded: list[str] | None, caller_overloaded: list[str] | None, lines_we_care_about: Any, branches_with_limits: Any, monitoring_factor: float, worsening_threshold: float, ) -> tuple[list[int], list[str]]: """Determine which lines are treated as "overloaded" for reporting. Priority: analysis context > caller-provided list > vectorised recomputation. Returns ``(ids, names)`` matching ``obs_simu_defaut.name_line``. """ name_line = obs_simu_defaut.name_line name_to_idx = {l: i for i, l in enumerate(name_line)} if analysis_context_overloaded: ids = [name_to_idx[l] for l in analysis_context_overloaded if l in name_to_idx] return ids, [name_line[i] for i in ids] if caller_overloaded: ids = [name_to_idx[l] for l in caller_overloaded if l in name_to_idx] return ids, [name_line[i] for i in ids] action_names = _to_1d(obs_simu_defaut.name_line) action_rho = _to_1d(obs_simu_defaut.rho) base_rho = _to_1d(obs_n.rho) wt = float(worsening_threshold) # obs.rho is already pre-scaled by monitoring_factor — see # ``build_care_mask`` docstring. Threshold is 1.0, not mf. OBS_RHO_OVERLOAD_THRESHOLD = 1.0 mask = np.isin(action_names, list(lines_we_care_about)) mask &= np.isin(action_names, list(branches_with_limits)) try: rho_mask = action_rho >= OBS_RHO_OVERLOAD_THRESHOLD pre_existing = base_rho >= OBS_RHO_OVERLOAD_THRESHOLD # Symmetric impact rule — see ``build_care_mask`` for rationale. not_impacted = (action_rho >= base_rho * (1 - wt)) & (action_rho <= base_rho * (1 + wt)) mask &= rho_mask & ~(pre_existing & not_impacted) except Exception as e: logger.warning("resolve_lines_overloaded: vectorised comparison failed: %s", e) mask = np.zeros(len(action_names), dtype=bool) ids = np.where(mask)[0].tolist() names = action_names[mask].tolist() return ids, names def half_open_overload_notes( obs: Any, lines_overloaded_names: list[str], rho_after: list[float] ) -> dict[str, float]: """Return ``{line_name: live_end_reactive_mvar}`` for still-"overloaded" lines the action leaves open at ONE end with a loading above ~1 %. Such a line carries no real flow (the diagrams show p = 0) but its capacitance draws reactive charging current from the live end, so its current-based loading stays non-zero (the reported ~33 %). Surfacing the live-end reactive power lets the ActionCard annotate the value as capacitive charging current rather than it reading as a residual overload. Reads the post-action variant via :func:`half_open_branch_reactive_from_obs`. """ if not lines_overloaded_names: return {} half_open = half_open_branch_reactive_from_obs(obs) if not half_open: return {} notes: dict[str, float] = {} for i, name in enumerate(lines_overloaded_names): try: rho = float(rho_after[i]) if i < len(rho_after) else 0.0 except (TypeError, ValueError): rho = 0.0 if name in half_open and rho > 0.01: notes[name] = half_open[name] return notes def half_open_branch_reactive_from_obs(obs: Any) -> dict[str, float]: """Return ``{branch_id_or_name: live_end_reactive_mvar}`` for branches open at exactly one terminal in ``obs``'s post-action pypowsybl variant (see :func:`build_half_open_reactive`). Reads the SAME variant the SLD / NAD diagrams render (``obs._network_manager`` on ``obs._variant_id``). Best-effort — returns ``{}`` on any failure and always restores the network manager's working variant so the shared network is never left mutated. """ nm = getattr(obs, "_network_manager", None) variant_id = getattr(obs, "_variant_id", None) network = getattr(nm, "network", None) if nm is not None else None if nm is None or network is None or variant_id is None: return {} try: original = network.get_working_variant_id() except Exception as e: logger.debug("half_open_branch_reactive_from_obs: cannot read working variant: %s", e) return {} try: nm.set_working_variant(variant_id) return build_half_open_reactive(network) except Exception as e: logger.debug("half_open_branch_reactive_from_obs: read failed: %s", e) return {} finally: try: nm.set_working_variant(original) except Exception as e: logger.debug("half_open_branch_reactive_from_obs: variant restore failed: %s", e) def compute_action_metrics( obs: Any, obs_simu_defaut: Any, obs_simu_action: Any, info_action: dict, lines_overloaded_ids: list[int], lines_we_care_about: Any, branches_with_limits: Any, monitoring_factor: float, worsening_threshold: float, ) -> dict[str, Any]: """Post-process a single-action simulation result into a scalar summary. Returns a dict with: ``rho_before``, ``rho_after``, ``max_rho``, ``max_rho_line``, ``is_rho_reduction``, ``is_islanded``, ``n_components_after``, ``disconnected_mw``, ``lines_overloaded_after``. Handles the non-convergence case by zeroing action-side fields. """ mf = float(monitoring_factor) rho_before = ( (_to_1d(obs_simu_defaut.rho)[lines_overloaded_ids] * mf).tolist() if lines_overloaded_ids else [] ) result = { "rho_before": rho_before, "rho_after": None, "max_rho": 0.0, "max_rho_line": "N/A", "is_rho_reduction": False, "is_islanded": False, "n_components_after": 1, "disconnected_mw": 0.0, "lines_overloaded_after": [], } if info_action.get("exception"): return result n_components_after = obs_simu_action.n_components result["n_components_after"] = n_components_after if ( n_components_after > obs.n_components or n_components_after > obs_simu_defaut.n_components ): # Topology shows more components, but only flag a real islanding # when actual load left the main component. Sub-threshold deltas # are almost always AC-LF numerical artefacts (extreme PST taps, # fast-mode local non-convergence) rather than physical outages. disconnected_mw = float( max(0.0, obs_simu_defaut.main_component_load_mw - obs_simu_action.main_component_load_mw) ) if disconnected_mw >= ISLANDING_MW_THRESHOLD: result["is_islanded"] = True result["disconnected_mw"] = disconnected_mw rho_after = (_to_1d(obs_simu_action.rho)[lines_overloaded_ids] * mf).tolist() result["rho_after"] = rho_after if rho_before: try: result["is_rho_reduction"] = bool( np.all(np.array(rho_after) + 0.01 < np.array(rho_before)) ) except Exception as e: logger.debug("compute_action_metrics: rho reduction check failed: %s", e) action_names = _to_1d(obs_simu_action.name_line) action_rho = _to_1d(obs_simu_action.rho) base_rho = _to_1d(obs.rho) care_mask = build_care_mask( action_names, action_rho, base_rho, lines_we_care_about, branches_with_limits, lines_overloaded_ids, monitoring_factor, worsening_threshold, ) try: monitored_rho = action_rho[care_mask] monitored_names = action_names[care_mask] # obs.rho is mf-scaled — see ``build_care_mask`` docstring — so # the "overloaded" boundary on ``monitored_rho`` is 1.0, not mf. overload_mask = monitored_rho >= 1.0 result["lines_overloaded_after"] = monitored_names[overload_mask].tolist() if len(monitored_rho) > 0: result["max_rho"] = float(np.max(monitored_rho)) * mf result["max_rho_line"] = monitored_names[int(np.argmax(monitored_rho))] except Exception as e: logger.warning("compute_action_metrics: max_rho / overload calc failed: %s", e) return result def extract_action_topology(action: Any, action_id: str, dict_action: dict | None) -> dict: """Read topology fields off a Grid2Op action and sanitise them for JSON. Supplements switches + heuristic (curtail / load_shedding) power setpoints that don't appear as public attributes on standard actions. """ topo: dict[str, Any] = {} for field in ( "lines_ex_bus", "lines_or_bus", "gens_bus", "loads_bus", "pst_tap", "substations", "switches", "loads_p", "gens_p", ): val = getattr(action, field, None) if val: topo[field] = sanitize_for_json(val) if not topo.get("switches") and dict_action: entry = dict_action.get(action_id) if entry: sw = entry.get("switches") if not sw: content_in_dict = entry.get("content") if isinstance(content_in_dict, dict): sw = content_in_dict.get("switches") if sw: topo["switches"] = sanitize_for_json(sw) # Back-fill injection setpoints from the action content so a user-built # combined injection action (``set_gen_p`` AND ``set_load_p``) reports # BOTH the generator and the load in its topology — the SLD / NAD # highlight reads ``gens_p`` / ``loads_p`` to mark the affected feeders, # and the grid2op action object doesn't always expose them as public # attributes. Covers every action, not just the ``curtail_`` / # ``load_shedding_`` prefixes handled below. if dict_action: content = (dict_action.get(action_id) or {}).get("content") if isinstance(content, dict): if not topo.get("gens_p") and content.get("set_gen_p"): topo["gens_p"] = sanitize_for_json(content["set_gen_p"]) if not topo.get("loads_p") and content.get("set_load_p"): topo["loads_p"] = sanitize_for_json(content["set_load_p"]) if action_id.startswith("curtail_") and not topo.get("gens_p"): gen_name = action_id.replace("curtail_", "") reg = (dict_action or {}).get(action_id, {}).get("content", {}).get("set_gen_p", {}) topo["gens_p"] = {gen_name: reg.get(gen_name, 0.0)} elif action_id.startswith("load_shedding_") and not topo.get("loads_p"): load_name = action_id.replace("load_shedding_", "") reg = (dict_action or {}).get(action_id, {}).get("content", {}).get("set_load_p", {}) topo["loads_p"] = {load_name: reg.get(load_name, 0.0)} return topo def serialize_action_result(action_id: str, action_data: dict) -> dict: """Build the JSON-serialisable payload returned by ``simulate_manual_action``.""" return { "action_id": action_id, "description_unitaire": action_data.get("description_unitaire") or "No description available", "rho_before": sanitize_for_json(action_data.get("rho_before")), "rho_after": sanitize_for_json(action_data.get("rho_after")), "max_rho": sanitize_for_json(action_data.get("max_rho")), "max_rho_line": action_data.get("max_rho_line", ""), "is_rho_reduction": bool(action_data.get("is_rho_reduction", False)), "is_islanded": bool(action_data.get("is_islanded", False)), "disconnected_mw": sanitize_for_json(action_data.get("disconnected_mw", 0.0)), "n_components": int(action_data.get("n_components", 1)), "non_convergence": action_data.get("non_convergence"), "lines_overloaded": sanitize_for_json(action_data.get("lines_overloaded_after", [])), "lines_overloaded_after": sanitize_for_json(action_data.get("lines_overloaded_after", [])), "half_open_overloads": sanitize_for_json(action_data.get("half_open_overloads", {})), "is_estimated": False, "action_topology": action_data.get("action_topology"), "curtailment_details": action_data.get("curtailment_details"), "load_shedding_details": action_data.get("load_shedding_details"), "redispatch_details": action_data.get("redispatch_details"), "pst_details": action_data.get("pst_details"), "content": action_data.get("content"), } def normalise_non_convergence(exception: Any) -> str | None: """Convert a list or scalar simulation exception into a user-facing message.""" if not exception: return None if isinstance(exception, list): return "; ".join(str(e) for e in exception) return str(exception) def build_combined_description( action_ids: list[str], dict_action: dict | None, recent_actions: dict | None, ) -> str: """Build a ``"[COMBINED] desc1 + desc2"`` string for multi-action IDs. For single IDs, returns the single description (falling back through the same resolution chain the caller would use). """ def _get_desc(aid: str) -> str: entry = (dict_action or {}).get(aid) if entry: return ( entry.get("description_unitaire") or entry.get("description") or aid ) recent = (recent_actions or {}).get(aid, {}) return ( recent.get("description_unitaire") or recent.get("description") or aid ) if len(action_ids) == 1: return str(_get_desc(action_ids[0])) return "[COMBINED] " + " + ".join(str(_get_desc(aid)) for aid in action_ids) def compute_combined_rho( obs_start: Any, obs_act1: Any, obs_act2: Any, betas: list[float], ) -> np.ndarray: """Compute the superposed rho vector: (1 - Σβ)·ρ_start + β₁·ρ₁ + β₂·ρ₂. Absolute value — betas can be negative and the library convention is magnitude-only downstream. """ return np.abs( (1.0 - sum(betas)) * obs_start.rho + betas[0] * obs_act1.rho + betas[1] * obs_act2.rho ) def compute_target_max_rho( rho_combined: np.ndarray, name_line_list: Any, lines_overloaded_ids: list[int], ) -> tuple[float, str]: """Pick max rho / line over the user-selected overloaded lines only. Rationale: the global ``max_rho`` scan across every monitored line has to stay broad to catch NEW overloads that the action pair may introduce (see ``test_superposition_max_rho_filtering_regression`` which pins that behaviour). But on lines far from either action, linearisation error can put an arbitrary high-loaded line at the top of the scan — a line with no relation to the contingency the user is resolving. The "target" max reports the effect ON THE LINES THE USER CARES ABOUT — the contingency's actual overloads — so the UI can surface it alongside the global max and give a direct estimated-vs-simulated comparison on the same line set. Returns ``(0.0, "N/A")`` when no overload ids are available or all are out of range (caller should treat that as "no target info"). """ if not lines_overloaded_ids: return 0.0, "N/A" n_lines = len(rho_combined) focus_ids = [int(i) for i in lines_overloaded_ids if 0 <= int(i) < n_lines] if not focus_ids: return 0.0, "N/A" focus_rho = rho_combined[focus_ids] argmax = int(np.argmax(focus_rho)) names = list(name_line_list) return float(focus_rho[argmax]), str(names[focus_ids[argmax]])