# 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. import pypowsybl.network as pn import logging import os import tempfile import zipfile logger = logging.getLogger(__name__) class NetworkService: def __init__(self): self.network = None # Cached equipment tables + derived column lookups. The Network is # read-only after load, so fetching the generators / loads DataFrame # once (one pypowsybl/Java boundary crossing each) and serving every # subsequent metadata query from in-process dicts avoids re-fetching # the entire table per generator/load. Action enrichment calls these # accessors once per generator per prioritized action — on the French # grid (~3k generators) the un-cached path dominated the enrichment # phase. All cleared in ``load_network`` when the Network changes. self._generators_df = None self._gen_vl_map = None # gen_id -> voltage_level_id self._gen_source_map = None # gen_id -> energy_source self._gen_limits_map = None # gen_id -> (min_p, max_p) self._loads_df = None self._load_vl_map = None # load_id -> voltage_level_id def _invalidate_equipment_caches(self) -> None: """Drop the cached generator / load tables (called when the Network changes).""" self._generators_df = None self._gen_vl_map = None self._gen_source_map = None self._gen_limits_map = None self._loads_df = None self._load_vl_map = None def _get_generators_df(self): """Return the generators DataFrame, fetched once and memoized. The default ``get_generators()`` column set already includes ``energy_source``, ``min_p``, ``max_p`` and ``voltage_level_id``, so a single cached table backs every generator metadata accessor below. """ if not self.network: raise ValueError("Network not loaded") if self._generators_df is None: self._generators_df = self.network.get_generators() return self._generators_df def _get_gen_vl_map(self) -> dict: if self._gen_vl_map is None: df = self._get_generators_df() self._gen_vl_map = ( df['voltage_level_id'].to_dict() if df is not None and 'voltage_level_id' in df.columns else {} ) return self._gen_vl_map def _get_gen_source_map(self) -> dict: if self._gen_source_map is None: df = self._get_generators_df() self._gen_source_map = ( df['energy_source'].to_dict() if df is not None and 'energy_source' in df.columns else {} ) return self._gen_source_map def _get_gen_limits_map(self) -> dict: if self._gen_limits_map is None: df = self._get_generators_df() if df is not None and 'min_p' in df.columns and 'max_p' in df.columns: self._gen_limits_map = { gid: (mn, mx) for gid, mn, mx in zip(df.index, df['min_p'], df['max_p']) } else: self._gen_limits_map = {} return self._gen_limits_map def _get_loads_df(self): """Return the loads DataFrame, fetched once and memoized.""" if not self.network: raise ValueError("Network not loaded") if self._loads_df is None: self._loads_df = self.network.get_loads() return self._loads_df def _get_load_vl_map(self) -> dict: if self._load_vl_map is None: df = self._get_loads_df() self._load_vl_map = ( df['voltage_level_id'].to_dict() if df is not None and 'voltage_level_id' in df.columns else {} ) return self._load_vl_map def _extract_network_zip(self, zip_path: str) -> str: """Extract the first .xiidm/.xml inside ``zip_path`` and return its path. Extraction targets the zip's own directory so the result is cached for subsequent loads; if that directory is read-only, fall back to a temp dir. """ with zipfile.ZipFile(zip_path) as zf: members = [n for n in zf.namelist() if n.lower().endswith(('.xiidm', '.xml'))] if not members: raise FileNotFoundError( f"No .xiidm or .xml file found inside {zip_path}") member = members[0] out_name = os.path.basename(member) out_dir = os.path.dirname(os.path.abspath(zip_path)) out_path = os.path.join(out_dir, out_name) if os.path.isfile(out_path): return out_path # already decompressed — reuse data = zf.read(member) try: with open(out_path, 'wb') as f: f.write(data) except OSError: tmp_dir = tempfile.mkdtemp(prefix='cs4g_net_') out_path = os.path.join(tmp_dir, out_name) with open(out_path, 'wb') as f: f.write(data) logger.info("Decompressed %s -> %s", zip_path, out_path) return out_path def _resolve_network_file(self, network_path: str) -> str: """Resolve a network path to a loadable file, transparently decompressing a zip when the path is (or only exists as) a ``.zip``. Handles: an explicit ``*.zip`` path; a missing ``foo.xiidm`` whose sibling ``foo.xiidm.zip`` exists; and a directory that holds only a ``.zip`` archive. """ if network_path.lower().endswith('.zip') and os.path.isfile(network_path): return self._extract_network_zip(network_path) if os.path.isfile(network_path): return network_path if os.path.isdir(network_path): has_net = any(f.endswith(('.xiidm', '.xml')) for f in os.listdir(network_path)) if not has_net: zips = [f for f in os.listdir(network_path) if f.endswith('.zip')] if zips: return self._extract_network_zip( os.path.join(network_path, zips[0])) return network_path # Missing path: try a sibling/companion .zip (e.g. the shipped # ``network.xiidm.zip`` for a ``network.xiidm`` request). for candidate in (network_path + '.zip', os.path.splitext(network_path)[0] + '.zip'): if os.path.isfile(candidate): return self._extract_network_zip(candidate) return network_path def load_network(self, network_path: str) -> dict: network_path = self._resolve_network_file(network_path) if not os.path.exists(network_path): raise FileNotFoundError(f"Network file/directory not found: {network_path}") # Determine if it's a file or directory and load accordingly # Assuming bare_env is a directory of xiidm files or a single xiidm file # pypowsybl can load from file. # If it's a directory, we might need to pick the xiidm file inside. if os.path.isdir(network_path): files = [f for f in os.listdir(network_path) if f.endswith('.xiidm') or f.endswith('.xml')] if not files: raise FileNotFoundError(f"No .xiidm or .xml file found in {network_path}") file_path = os.path.join(network_path, files[0]) else: file_path = network_path # NOTE: pypowsybl exposes `allow_variant_multi_thread_access=True` # on `pn.load()` which looks like a silver bullet for the # `/api/config` contention between the NAD prefetch worker and # grid2op env setup. It is NOT safe to enable here, see # docs/performance/history/concurrent-variants.md: when ON, every thread that # touches the Network must FIRST call `n.set_working_variant(...)`, # otherwise pypowsybl raises "Variant index not set for current # thread". FastAPI serves each request on a thread-pool worker # whose identity is unstable — the read-only endpoints # (`/api/branches`, `/api/voltage-levels`, `/api/nominal-voltages`) # would need a per-endpoint variant-set guard, which we currently # do NOT have. Keeping the default (False) preserves correctness; # the contention (~2-3 s) is an accepted residual cost. self.network = pn.load(file_path) self._invalidate_equipment_caches() return {"message": "Network loaded successfully", "id": self.network.id} def get_disconnectable_elements(self) -> list: if not self.network: raise ValueError("Network not loaded") # get lines and two winding transformers lines = self.network.get_lines() transformers = self.network.get_2_windings_transformers() elements = [] if lines is not None and not lines.empty: elements.extend(lines.index.tolist()) if transformers is not None and not transformers.empty: elements.extend(transformers.index.tolist()) return sorted(elements) def get_element_names(self) -> dict | None: """Return {element_id: display_name} for all lines and transformers. The display name is the pypowsybl ``name`` field when it is set and differs from the element ID; otherwise the ID itself. For lines/transformers whose name is still a raw OSM identifier (e.g. ``way/426020732-400``), a composite name is built from the voltage-level names at each endpoint (e.g. ``CHARPENAY — ST-VULBAS-EST``). """ if not self.network: raise ValueError("Network not loaded") import re _RAW_OSM_RE = re.compile(r'^(way|relation)[/_]') # Pre-load VL display names for fallback construction vl_names: dict[str, str] = {} voltage_levels = self.network.get_voltage_levels() if voltage_levels is not None and not voltage_levels.empty and 'name' in voltage_levels.columns: for vl_id, row in voltage_levels.iterrows(): n = row.get('name') if n and str(n) != 'nan': # Strip trailing " 400kV" etc. for a cleaner composite name clean = re.sub(r'\s+\d+\s*kV$', '', str(n)) vl_names[vl_id] = clean def _display_name(eid: str, row, name_col_exists: bool, vl1_col: str, vl2_col: str) -> str | None: """Return a human-readable name, or None to skip.""" n = row.get('name') if name_col_exists else None if n and str(n) != str(eid) and str(n) != 'nan' and not _RAW_OSM_RE.match(str(n)): return str(n) # Name is missing or is a raw OSM ID → build from VL endpoint names vl1 = row.get(vl1_col) if vl1_col in row.index else None vl2 = row.get(vl2_col) if vl2_col in row.index else None name1 = vl_names.get(str(vl1), '') if vl1 else '' name2 = vl_names.get(str(vl2), '') if vl2 else '' if name1 and name2 and name1 != name2: return f"{name1} \u2014 {name2}" if name1: return name1 if name2: return name2 # Fallback: use the raw name if it exists and differs from ID if n and str(n) != str(eid) and str(n) != 'nan': return str(n) return None name_map: dict[str, str] = {} lines = self.network.get_lines() if lines is not None and not lines.empty: has_name = 'name' in lines.columns for eid, row in lines.iterrows(): display = _display_name(eid, row, has_name, 'voltage_level1_id', 'voltage_level2_id') if display: name_map[eid] = display transformers = self.network.get_2_windings_transformers() if transformers is not None and not transformers.empty: has_name = 'name' in transformers.columns for eid, row in transformers.iterrows(): display = _display_name(eid, row, has_name, 'voltage_level1_id', 'voltage_level2_id') if display: name_map[eid] = display return name_map def get_monitored_elements(self) -> list: """Return the list of element IDs that have at least one permanent operational limit.""" if not self.network: raise ValueError("Network not loaded") # Narrow query — only (element_id, type, acceptable_duration) are # consumed below, and all three live in the pypowsybl MultiIndex. # `value`, `element_type`, `name`, `group_name` are fetched by the # default call but unused here. `attributes=[]` drops those # columns and saves ~90 ms on the 55 k-row limit table of the # PyPSA-EUR France grid (265 ms → 175 ms). A `6835 × 0` # DataFrame is reported as `.empty` by pandas, so we check # `len(index)` instead. limits = self.network.get_operational_limits(attributes=[]) if limits is None or len(limits.index) == 0: return [] limits = limits.reset_index() # Filter for limits of type 'CURRENT' with acceptable_duration == -1 (permanent) # Note: some networks might use 'THERMAL' or other types, but 'CURRENT' is standard for ampere limits. # Expert Assist uses 'CURRENT' (see recommender_service.py:601) permanent_limits = limits[(limits['type'] == 'CURRENT') & (limits['acceptable_duration'] == -1)] if permanent_limits.empty: return [] ids = sorted(permanent_limits['element_id'].unique().tolist()) return ids def get_voltage_levels(self) -> list: if not self.network: raise ValueError("Network not loaded") # Narrow query — only the index is consumed downstream. Requesting # `attributes=[]` skips pypowsybl's Java→Python serialisation of # `nominal_v`, `name`, `topology_kind`, etc. (~3-4 ms saved on the # 6 835-VL PyPSA-EUR France grid). We check `len(index)` rather # than `.empty`, because a DataFrame with rows but 0 columns is # still reported as empty by pandas. voltage_levels = self.network.get_voltage_levels(attributes=[]) if voltage_levels is not None and len(voltage_levels.index) > 0: return sorted(voltage_levels.index.tolist()) return [] def get_voltage_level_substations(self) -> dict: """Return ``{vl_id: substation_id}`` for every voltage level. Used by the frontend to anchor action-overview pins on the overflow graph: the overflow graph nodes are pypowsybl substation IDs, while action data references voltage-level IDs — this map closes that gap. Returns an empty dict if the ``substation_id`` column is missing (pure-VL networks without substations). """ if not self.network: raise ValueError("Network not loaded") # `substation_id` ships in the default attribute set so a narrow # query is enough; we don't pull `name` / `nominal_v` here. voltage_levels = self.network.get_voltage_levels(attributes=['substation_id']) if voltage_levels is None or voltage_levels.empty: return {} if 'substation_id' not in voltage_levels.columns: return {} sub_ids = voltage_levels['substation_id'].tolist() idx = voltage_levels.index.tolist() return { vl_id: str(sub_id) for vl_id, sub_id in zip(idx, sub_ids) if sub_id is not None and str(sub_id) != 'nan' } def get_voltage_level_names(self) -> dict: """Return {vl_id: display_name} for all voltage levels.""" if not self.network: raise ValueError("Network not loaded") name_map: dict[str, str] = {} voltage_levels = self.network.get_voltage_levels() if voltage_levels is not None and not voltage_levels.empty and 'name' in voltage_levels.columns: for vl_id, row in voltage_levels.iterrows(): n = row.get('name') if n and str(n) != str(vl_id) and str(n) != 'nan': name_map[vl_id] = str(n) return name_map def get_nominal_voltages(self) -> dict: """Return {vl_id: nominal_v_kv} mapping for all voltage levels, snapped to detected grid values. Optimised path — narrow pypowsybl query + vectorised final dict build (no pandas `iterrows`). Measured on the 6 835-VL PyPSA-EUR France grid: 144 ms → 6.6 ms (~22× speedup). Output strictly identical. """ if not self.network: raise ValueError("Network not loaded") # Narrow query — only `nominal_v` is needed. `get_voltage_levels()` # with `all_attributes=True` materialises `name`, `topology_kind`, # `substation_id`, ... adding ~4 ms of Java→Python serialisation. voltage_levels = self.network.get_voltage_levels(attributes=['nominal_v']) if voltage_levels is None or voltage_levels.empty: return {} # Pull the column as a plain numpy array once — avoids repeated # pandas column access in the final dict comprehension. nom_v_arr = voltage_levels['nominal_v'].values idx_list = voltage_levels.index.tolist() # 1. Collect all unique nominal voltages import numpy as np raw_voltages = sorted(np.unique(nom_v_arr).tolist()) if not raw_voltages: return {} # 2. Cluster voltages within 2% of each other clusters = [] current_cluster = [raw_voltages[0]] for v in raw_voltages[1:]: # If v is within 2% of the cluster average, add it avg = sum(current_cluster) / len(current_cluster) if abs(v - avg) / avg < 0.02: current_cluster.append(v) else: clusters.append(current_cluster) current_cluster = [v] clusters.append(current_cluster) # 3. Create representative cleaned values for each cluster # Map each raw voltage to its clean representative raw_to_clean = {} for cluster in clusters: avg = sum(cluster) / len(cluster) # Bucketing: anything < 25kV goes into the 25kV bucket if avg < 25: clean_v = 25.0 else: # Clean representative: round to int clean_v = round(avg, 0) for v in cluster: raw_to_clean[v] = clean_v # 4. Map each voltage level to its clean representative. # Vectorised over the numpy array (avoids iterrows which was the # dominant cost — ~130 ms for 6 835 rows). nom_v_list = nom_v_arr.tolist() return { idx_list[i]: raw_to_clean[float(nom_v_list[i])] for i in range(len(idx_list)) } def get_element_voltage_levels(self, element_id: str) -> list: """Resolve an equipment ID (line, transformer, or VL) to its voltage level IDs.""" if not self.network: raise ValueError("Network not loaded") # Check if it's already a voltage level voltage_levels = self.network.get_voltage_levels() if voltage_levels is not None and element_id in voltage_levels.index: return [element_id] # Check lines (have voltage_level1_id and voltage_level2_id columns) lines = self.network.get_lines() if lines is not None and element_id in lines.index: row = lines.loc[element_id] vls = set() if 'voltage_level1_id' in row.index: vls.add(row['voltage_level1_id']) if 'voltage_level2_id' in row.index: vls.add(row['voltage_level2_id']) return sorted(vls) # Check 2-winding transformers transformers = self.network.get_2_windings_transformers() if transformers is not None and element_id in transformers.index: row = transformers.loc[element_id] vls = set() if 'voltage_level1_id' in row.index: vls.add(row['voltage_level1_id']) if 'voltage_level2_id' in row.index: vls.add(row['voltage_level2_id']) return sorted(vls) return [] def get_load_voltage_level(self, load_id: str) -> str | None: """Return the voltage level ID that a given load belongs to.""" return self._get_load_vl_map().get(load_id) def get_load_voltage_levels_bulk(self, load_ids: list[str]) -> dict[str, str]: """Return {load_id: voltage_level_id} for a list of loads.""" vl_map = self._get_load_vl_map() return {lid: vl_map[lid] for lid in load_ids if lid in vl_map} def get_generator_voltage_level(self, gen_id: str) -> str | None: """Return the voltage level ID that a given generator belongs to.""" return self._get_gen_vl_map().get(gen_id) def get_generator_active_power_limits(self, gen_id: str) -> tuple[float, float] | None: """Return ``(min_p, max_p)`` active-power limits (MW) of a generator. Used to expose the maximum redispatch headroom on a remedial action card (raise: ``max_p - current``; lower: ``current - min_p``).""" limits = self._get_gen_limits_map().get(gen_id) if limits is None: return None try: return float(limits[0]), float(limits[1]) except (TypeError, ValueError): return None def get_generator_type(self, gen_id: str) -> str | None: """Return the energy source type of a given generator.""" return self._get_gen_source_map().get(gen_id) def get_generator_types_bulk(self, gen_ids: list[str]) -> dict[str, str]: """Return {gen_id: energy_source} for a list of generators.""" source_map = self._get_gen_source_map() return {gid: source_map[gid] for gid in gen_ids if gid in source_map} network_service = NetworkService()