Colour the THT preview backbone by real kV + harden network loader

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 (``<?xml`` or a
+        ``<...network`` root). Guards against a truncated decode, a Git-LFS
+        pointer smudged in as text, or a ``.gz.b64`` mistaken for raw XML —
+        any of which make pypowsybl raise "Unsupported file format"."""
+        try:
+            with open(path, 'rb') as f:
+                head = f.read(512).lstrip()
+        except OSError:
+            return False
+        return head[:5] == b'<?xml' or (head[:1] == b'<' and b'network' in head[:400].lower())
+
     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.
+        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

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"<network/>"  # 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"<network/>"
+
     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).

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
-# <img>).
+# below (220 / 225 kV) is green. Each voltage level's nominal kV is read from
+# the network's ``<voltageLevel nominalV=...>`` — 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 <img>).
 _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 <voltageLevel> 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'<path d="{"".join(segs)}" fill="none" stroke="{color}" '
-            f'stroke-width="{_EDGE_WIDTH}" stroke-opacity="0.85" '
+            f'stroke-width="{width}" stroke-opacity="{opacity}" '
             f'stroke-linecap="round" stroke-linejoin="round"/>'
         )
     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