diff --git a/expert_backend/services/network_service.py b/expert_backend/services/network_service.py index 90cb1905..5b533fc1 100644 --- a/expert_backend/services/network_service.py +++ b/expert_backend/services/network_service.py @@ -143,8 +143,8 @@ def _decode_network_gz_b64(self, b64_path: str) -> str: with open(b64_path, 'rb') as f: raw = gzip.decompress(base64.b64decode(f.read())) out_path = b64_path[:-len('.gz.b64')] # network.xiidm.gz.b64 -> network.xiidm - if os.path.isfile(out_path): - return out_path # already decoded — reuse + if os.path.isfile(out_path) and self._looks_like_network_xml(out_path): + return out_path # already decoded and valid — reuse try: with open(out_path, 'wb') as f: f.write(raw) @@ -156,18 +156,45 @@ def _decode_network_gz_b64(self, b64_path: str) -> str: logger.info("Decoded %s -> %s", b64_path, out_path) return out_path + @staticmethod + def _looks_like_network_xml(path: str) -> bool: + """True if the file starts like an XIIDM/XML network (`` 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. + decompressing a zip when the path is (or only exists as) a ``.zip``, + or decoding a ``.gz.b64`` (the France THT game grids ship this way). + + Handles: an explicit ``*.zip`` / ``*.gz.b64`` path; a missing + ``foo.xiidm`` whose sibling ``foo.xiidm.zip`` / ``foo.xiidm.gz.b64`` + exists; a directory that holds only a ``.zip`` archive; and a present- + but-unloadable ``foo.xiidm`` (truncated / LFS pointer) that has a + sibling ``.gz.b64`` to re-decode from. """ if network_path.lower().endswith('.zip') and os.path.isfile(network_path): return self._extract_network_zip(network_path) + # An explicit ``.gz.b64`` path (e.g. config pointed straight at the + # committed transport) — decode it to raw XML. + if network_path.endswith('.gz.b64') and os.path.isfile(network_path): + return self._decode_network_gz_b64(network_path) + if os.path.isfile(network_path): + # Present but unparseable (truncated decode / un-smudged LFS + # pointer) — re-decode from a sibling ``.gz.b64`` if one is there, + # rather than handing pypowsybl a file it will reject. + if (not self._looks_like_network_xml(network_path) + and os.path.isfile(network_path + '.gz.b64')): + return self._decode_network_gz_b64(network_path + '.gz.b64') return network_path if os.path.isdir(network_path): @@ -178,6 +205,10 @@ def _resolve_network_file(self, network_path: str) -> str: if zips: return self._extract_network_zip( os.path.join(network_path, zips[0])) + b64s = [f for f in os.listdir(network_path) if f.endswith('.gz.b64')] + if b64s: + return self._decode_network_gz_b64( + os.path.join(network_path, b64s[0])) return network_path # Missing path: try a sibling/companion .zip (e.g. the shipped diff --git a/expert_backend/tests/test_network_service.py b/expert_backend/tests/test_network_service.py index 5ef7b19f..e1102562 100644 --- a/expert_backend/tests/test_network_service.py +++ b/expert_backend/tests/test_network_service.py @@ -195,6 +195,32 @@ def test_resolve_decodes_and_is_cached(self, tmp_path): assert first == second and os.path.isfile(first) assert first.endswith("network.xiidm") + @patch("expert_backend.services.network_service.pn") + def test_redecodes_when_present_xiidm_is_invalid(self, mock_pn, tmp_path): + # A stale / truncated network.xiidm (e.g. an un-smudged LFS pointer) + # sits next to a valid .gz.b64 — the resolver must re-decode rather + # than hand pypowsybl the unparseable file. + self._make_gz_b64(tmp_path) + bad = tmp_path / "network.xiidm" + bad.write_text("version https://git-lfs.github.com/spec/v1\noid sha256:deadbeef\n") + mock_pn.load.return_value = MagicMock(id="g") + + NetworkService().load_network(str(bad)) + + loaded = mock_pn.load.call_args[0][0] + assert open(loaded, "rb").read() == b"" # the decoded XML, not the pointer + + @patch("expert_backend.services.network_service.pn") + def test_loads_when_given_the_gz_b64_path_directly(self, mock_pn, tmp_path): + b64_path = self._make_gz_b64(tmp_path) + mock_pn.load.return_value = MagicMock(id="g") + + NetworkService().load_network(b64_path) + + loaded = mock_pn.load.call_args[0][0] + assert loaded.endswith("network.xiidm") + assert open(loaded, "rb").read() == b"" + def test_decode_falls_back_to_tempdir_when_grid_dir_readonly(self, tmp_path): # Force the in-place write to fail (a read-only grid dir can't be # simulated as root, so raise OSError on the target open instead). diff --git a/frontend/public/game/preview-high.svg b/frontend/public/game/preview-high.svg index fca83eed..953c69b0 100644 --- a/frontend/public/game/preview-high.svg +++ b/frontend/public/game/preview-high.svg @@ -1 +1 @@ - \ No newline at end of file + \ No newline at end of file diff --git a/frontend/public/game/preview-tht.svg b/frontend/public/game/preview-tht.svg index 6775a004..00ee2746 100644 --- a/frontend/public/game/preview-tht.svg +++ b/frontend/public/game/preview-tht.svg @@ -1 +1 @@ - \ No newline at end of file + \ No newline at end of file diff --git a/scripts/game_mode/gen_network_previews.py b/scripts/game_mode/gen_network_previews.py index 85479295..084d0763 100644 --- a/scripts/game_mode/gen_network_previews.py +++ b/scripts/game_mode/gen_network_previews.py @@ -47,29 +47,53 @@ ] # Voltage colouring: the >= 350 kV backbone (380 / 400 kV) is red, everything -# below (220 / 225 kV) is green. The two shades are the Okabe-Ito "vermillion" -# and "bluish green" — a warm/cool pair chosen so red-green colour-blind -# viewers can still tell them apart (they separate on the blue channel and -# lightness, not only the red-green axis), and both sit at a medium luminance -# that reads on the light AND dark config-screen card (the SVG is a themeless -# ). +# below (220 / 225 kV) is green. Each voltage level's nominal kV is read from +# the network's ```` — the substation IDs (RTE codes +# like ``1ARGIP7``) do NOT carry a 3-digit kV, so a regex over the id would put +# the whole map on one colour. +# +# The two shades are chosen for red-green colour-blind viewers: a warm +# vermillion vs a cool blue-leaning teal that separate on the BLUE channel and +# on LUMINANCE (the teal is markedly darker), not only on the red-green axis +# that deuteranopes/protanopes cannot use. As a belt-and-braces redundant cue — +# so the backbone is legible even in total colour-blindness / greyscale — the +# HV backbone is also drawn THICKER and fully opaque, the LV layer thinner and +# slightly translucent. Both read on the light AND dark config-screen card +# (the SVG is a themeless ). _HV_THRESHOLD_KV = 350 -_HV_COLOR = "#d55e00" # >= 350 kV — "red" -_LV_COLOR = "#009e73" # < 350 kV — "green" +_HV_COLOR = "#d55e00" # >= 350 kV — warm vermillion "red" +_LV_COLOR = "#166a5a" # < 350 kV — dark teal "green" (darker + bluer than the +# old #009e73, for a wider luminance + blue-channel gap) _WIDTH = 900 _PADDING = 24 _NODE_RADIUS = 1.6 -_EDGE_WIDTH = 1.6 +_HV_EDGE_WIDTH = 2.3 # backbone drawn heavier — a non-colour cue on top of hue +_LV_EDGE_WIDTH = 1.2 _MAX_NODES = 2600 # stride-sample nodes in the fallback (no-edge) scatter _KV_RE = re.compile(r"-(\d{3})(?:\D|$)") _BRANCH_TAG_RE = re.compile(r"<(?:\w+:)?(?:line|twoWindingsTransformer)\b([^>]*)>") _V1_RE = re.compile(r'voltageLevelId1="([^"]+)"') _V2_RE = re.compile(r'voltageLevelId2="([^"]+)"') +_VL_TAG_RE = re.compile( + r'<(?:\w+:)?voltageLevel\b[^>]*\bid="([^"]+)"[^>]*\bnominalV="([^"]+)"') -def _kv_of(node_id: str) -> int: +def _nominal_kv_map(xml: str) -> dict[str, int]: + """{voltageLevelId: nominal kV} from the network's tags.""" + out: dict[str, int] = {} + for vid, v in _VL_TAG_RE.findall(xml): + try: + out[vid] = round(float(v)) + except ValueError: + continue + return out + + +def _kv_of(node_id: str, vmap: dict[str, int] | None = None) -> int: + if vmap is not None and node_id in vmap: + return vmap[node_id] m = _KV_RE.search(node_id) return int(m.group(1)) if m else 0 @@ -154,7 +178,8 @@ def _svg_header(height: float) -> str: ) -def _build_edge_map(layout: dict, edges: list[tuple[str, str]]) -> str: +def _build_edge_map(layout: dict, edges: list[tuple[str, str]], + vmap: dict[str, int] | None = None) -> str: project, height = _projector(layout) # Group edges by colour (max kV of the two endpoints → HV backbone on top). @@ -169,7 +194,7 @@ def _build_edge_map(layout: dict, edges: list[tuple[str, str]]) -> str: continue connected.add(vl1) connected.add(vl2) - color = _color_of(max(_kv_of(vl1), _kv_of(vl2))) + color = _color_of(max(_kv_of(vl1, vmap), _kv_of(vl2, vmap))) edge_paths.setdefault(color, []).append(f"M{p1[0]} {p1[1]}L{p2[0]} {p2[1]}") # Only VLs with no line at all get a dot (so islanded substations don't @@ -181,13 +206,18 @@ def _build_edge_map(layout: dict, edges: list[tuple[str, str]]) -> str: p = project(node_id) if p is None: continue - orphan_pts.setdefault(_color_of(_kv_of(node_id)), []).append(p) + orphan_pts.setdefault(_color_of(_kv_of(node_id, vmap)), []).append(p) parts = [_svg_header(height)] for color, segs in sorted(edge_paths.items(), key=lambda kc: _draw_order(kc[0])): + # Redundant (non-colour) encoding: the HV backbone is drawn thicker and + # fully opaque so it stands out by weight too, not by hue alone. + is_hv = color == _HV_COLOR + width = _HV_EDGE_WIDTH if is_hv else _LV_EDGE_WIDTH + opacity = 0.95 if is_hv else 0.7 parts.append( f'' ) for color, pts in sorted(orphan_pts.items(), key=lambda kc: _draw_order(kc[0])): @@ -198,7 +228,7 @@ def _build_edge_map(layout: dict, edges: list[tuple[str, str]]) -> str: return "".join(parts) -def _build_node_scatter(layout: dict) -> str: +def _build_node_scatter(layout: dict, vmap: dict[str, int] | None = None) -> str: """Fallback when the network topology isn't available: nodes only.""" project, height = _projector(layout) ids = list(layout) @@ -208,7 +238,7 @@ def _build_node_scatter(layout: dict) -> str: p = project(node_id) if p is None: continue - by_color.setdefault(_color_of(_kv_of(node_id)), []).append(p) + by_color.setdefault(_color_of(_kv_of(node_id, vmap)), []).append(p) parts = [_svg_header(height)] for color, pts in sorted(by_color.items(), key=lambda kc: _draw_order(kc[0])): @@ -231,10 +261,13 @@ def main() -> int: out_path = _OUT_DIR / out_name xml = _load_network_xml(grid_dir) if xml: + vmap = _nominal_kv_map(xml) edges = _edges(xml) - svg = _build_edge_map(layout, edges) + svg = _build_edge_map(layout, edges, vmap) + hv = sum(1 for v in vmap.values() if v >= _HV_THRESHOLD_KV) out_path.write_text(svg, encoding="utf-8") - print(f"{tier}: {len(layout)} nodes → {out_path.relative_to(_REPO_ROOT)} " + print(f"{tier}: {len(layout)} nodes ({hv} HV ≥{_HV_THRESHOLD_KV}kV) → " + f"{out_path.relative_to(_REPO_ROOT)} " f"[edge map, {len(edges)} lines, {len(svg) // 1024} KB]") elif out_path.is_file(): # Network file is an un-smudged LFS pointer here. Never downgrade a