feat(overflow_graph): support extra lines to cut without overload styling

alphaDeesp/core/graphs/overflow_graph.py

@@ -41,6 +41,7 @@ def __init__(
         df_overflow: pd.DataFrame,
         layout: Optional[List[Tuple[float, float]]] = None,
         float_precision: str = "%.2f",
+        extra_lines_to_cut: Optional[Iterable[int]] = None,
     ) -> None:
         if "line_name" not in df_overflow.columns:
             df_overflow["line_name"] = [
@@ -49,6 +50,16 @@ def __init__(
             ]
 
         self.df = df_overflow
+        # Subset of ``lines_to_cut`` that the caller wants the cut-analysis
+        # to treat like overloads (so they get the same black/constrained
+        # styling and feed the structured-overload graph the same way) but
+        # WITHOUT being classified as overloads in the viewer's
+        # ``Overloads`` layer / ``is_overload`` flag.  Used by callers who
+        # want the recommender to find actions that prevent flow increase
+        # on otherwise-healthy lines (ExpertAgent's ``additionalLinesToCut``
+        # semantic).  ``None`` / empty means "no extras" — every cut line
+        # is a true overload, preserving the legacy behaviour.
+        self.extra_lines_cut = set(extra_lines_to_cut or [])
         super().__init__(topo, lines_to_cut, layout, float_precision)
 
     def build_graph(self) -> None:
@@ -74,14 +85,23 @@ def build_edges_from_df(self, g: nx.MultiDiGraph, lines_to_cut: List[int]) -> No
         )
 
         cols = ("idx_or", "idx_ex", "delta_flows", "gray_edges", "line_name")
+        # Operator-selected extras must NOT be coloured black: black is the
+        # visual signal for "overload contingency line" used by both the
+        # ``Overloads`` layer and the structured-overload analyser.  We
+        # therefore strip extras from the cut list passed to ``_edge_color``
+        # so they keep their natural flow polarity colour (coral / blue).
+        # They stay marked ``is_constrained`` and ``is_extra_cut`` so the
+        # downstream layers can still find them by flag.
+        cut_for_colour = [idx for idx in lines_to_cut if idx not in self.extra_lines_cut]
         for i, (origin, extremity, reported_flow, gray_edge, line_name) in enumerate(
                 zip(*(self.df[c] for c in cols))):
             self._add_overflow_edge(
                 g, origin, extremity, reported_flow, line_name,
-                color=self._edge_color(i, reported_flow, gray_edge, lines_to_cut),
+                color=self._edge_color(i, reported_flow, gray_edge, cut_for_colour),
                 scaling_factor=scaling_factor,
                 min_penwidth=min_penwidth,
-                is_constrained=(i in lines_to_cut))
+                is_constrained=(i in lines_to_cut),
+                is_extra_cut=(i in self.extra_lines_cut))
 
     @staticmethod
     def _edge_color(
@@ -105,6 +125,7 @@ def _add_overflow_edge(
         scaling_factor: float,
         min_penwidth: float,
         is_constrained: bool,
+        is_extra_cut: bool = False,
     ) -> None:
         """Add a single styled overflow edge to g."""
         fp = self.float_precision
@@ -119,6 +140,12 @@ def _add_overflow_edge(
         }
         if is_constrained:
             attrs["constrained"] = True
+        if is_extra_cut:
+            # Operator-supplied extra cut — gets the black/constrained
+            # styling like a real overload but the viewer's "Overloads"
+            # layer must skip it.  ``highlight_significant_line_loading``
+            # respects this flag when stamping ``is_overload``.
+            attrs["is_extra_cut"] = True
         g.add_edge(origin, extremity, **attrs)
 
     def keep_overloads_components(self) -> None:
@@ -191,6 +218,7 @@ def highlight_significant_line_loading(self, dict_line_loading: Dict[Any, Any])
         edge_names = nx.get_edge_attributes(self.g, "name")
         edge_colors = nx.get_edge_attributes(self.g, "color")
         edge_x_labels = nx.get_edge_attributes(self.g, "label")
+        edge_extra_cut = nx.get_edge_attributes(self.g, "is_extra_cut")
         label_font_color = {edge: "black" for edge in edge_names.keys()}
         color_label_highlight = "darkred"
 
@@ -204,18 +232,29 @@ def highlight_significant_line_loading(self, dict_line_loading: Dict[Any, Any])
             current_edge_color = edge_colors[edge]
             before = dict_line_loading[edge_name]["before"]
             after = dict_line_loading[edge_name]["after"]
+            is_extra = bool(edge_extra_cut.get(edge, False))
 
             # Every entry in dict_line_loading is a monitored / low-
             # margin line; the black ones are additionally overloads.
-            is_monitored_attrs[edge] = True
-            if current_edge_color == "black":
-                edge_x_labels[edge] = f'< {current_x_label} <BR/>  <B>{before}%</B>  → {after}%>'
-                is_overload_attrs[edge] = True
+            # Operator-selected extras (``is_extra_cut``) are kept out
+            # of both flags so the viewer's ``Overloads`` and
+            # ``Low margin lines`` layers reflect the recommender's
+            # detected state, not user-supplied targets.
+            if not is_extra:
+                is_monitored_attrs[edge] = True
+                if current_edge_color == "black":
+                    edge_x_labels[edge] = f'< {current_x_label} <BR/>  <B>{before}%</B>  → {after}%>'
+                    is_overload_attrs[edge] = True
+                else:
+                    edge_x_labels[edge] = f'< {current_x_label} <BR/>  {before}% → <B>{after}%</B> >'
+                edge_colors[edge] = f'"{current_edge_color}:yellow:{current_edge_color}"'
             else:
+                # Extras keep their natural flow colour; only the
+                # ``before → 0%`` annotation surfaces the cut so the
+                # operator sees how their choice materialises.
                 edge_x_labels[edge] = f'< {current_x_label} <BR/>  {before}% → <B>{after}%</B> >'
 
             label_font_color[edge] = color_label_highlight
-            edge_colors[edge] = f'"{current_edge_color}:yellow:{current_edge_color}"'
 
         nx.set_edge_attributes(self.g, edge_x_labels, "label")
         nx.set_edge_attributes(self.g, label_font_color, "fontcolor")

alphaDeesp/core/interactive_html.py

@@ -73,6 +73,12 @@
     {"key": "in_red_loop", "label": "Red-loop paths", "swatch": "red-loop", "scope": "both"},
     {"key": "is_overload", "label": "Overloads", "swatch": "overload", "scope": "edge"},
     {"key": "is_monitored", "label": "Low margin lines", "swatch": "monitored", "scope": "edge"},
+    # Operator-supplied extras (ExpertAgent's `additionalLinesToCut`):
+    # cut in the analysis like overloads but rendered with their
+    # natural flow colour and excluded from the Overloads /
+    # Low margin lines layers.  Surfaced as a dedicated layer so the
+    # operator can still see how their choice materialised.
+    {"key": "is_extra_cut", "label": "Extra lines to prevent flow increase", "swatch": "extra-cut", "scope": "edge"},
     {"key": "is_hub", "label": "Hubs", "swatch": "diamond", "scope": "node"},
 ]
 
@@ -104,6 +110,7 @@
     # Individual entities properties — per-edge / per-node flags.
     "semantic:is_overload": _SECTION_PROPERTIES,
     "semantic:is_monitored": _SECTION_PROPERTIES,
+    "semantic:is_extra_cut": _SECTION_PROPERTIES,
     "semantic:is_hub": _SECTION_PROPERTIES,
     "style:dashed": _SECTION_PROPERTIES,
     "style:dotted": _SECTION_PROPERTIES,
@@ -693,6 +700,7 @@ def _edge_repl(match: re.Match) -> str:
     if (swatch === 'constrained-path') return '<svg viewBox="0 0 14 6"><line x1="0" y1="3" x2="14" y2="3" stroke="black" stroke-width="2"/></svg>';
     if (swatch === 'overload') return '<svg viewBox="0 0 14 6"><line x1="0" y1="3" x2="14" y2="3" stroke="black" stroke-width="2.5"/><line x1="0" y1="3" x2="14" y2="3" stroke="yellow" stroke-width="0.8"/></svg>';
     if (swatch === 'monitored') return '<svg viewBox="0 0 14 6"><line x1="0" y1="3" x2="14" y2="3" stroke="coral" stroke-width="2.5"/><line x1="0" y1="3" x2="14" y2="3" stroke="yellow" stroke-width="0.8"/></svg>';
+    if (swatch === 'extra-cut') return '<svg viewBox="0 0 14 6"><line x1="0" y1="3" x2="14" y2="3" stroke="blue" stroke-width="2" stroke-dasharray="3 2"/></svg>';
     // Match the upstream node fillcolors set in build_nodes:
     //   prod (prod_minus_load > 0)  → coral
     //   load (prod_minus_load < 0)  → lightblue

alphaDeesp/tests/test_interactive_html.py

@@ -167,6 +167,35 @@ def test_layer_index_emits_semantic_layers_from_source_flags():
     assert set(by_key["semantic:is_monitored"]["nodes"]) == {"A", "B"}
 
 
+def test_layer_index_emits_extra_cut_layer_with_endpoints():
+    """``is_extra_cut`` is an edge-only semantic flag; like the other
+    edge-only layers it must include the endpoint nodes so the
+    substations stay visible when the operator ticks it on alone, and
+    the layer must be assigned to the "Properties" section."""
+    edges = [
+        {"id": "edge1", "source": "A", "target": "B",
+         "attrs": {"color": "blue", "is_extra_cut": "True"}},
+        {"id": "edge2", "source": "B", "target": "C",
+         "attrs": {"color": "coral"}},
+    ]
+    nodes = [
+        {"name": "A", "attrs": {}},
+        {"name": "B", "attrs": {}},
+        {"name": "C", "attrs": {}},
+    ]
+    layers = _build_layer_index(edges, nodes)
+    by_key = {l["key"]: l for l in layers}
+
+    assert "semantic:is_extra_cut" in by_key
+    layer = by_key["semantic:is_extra_cut"]
+    assert layer["edges"] == ["edge1"]
+    assert set(layer["nodes"]) == {"A", "B"}
+    assert layer["swatch"] == "extra-cut"
+    assert layer["label"] == "Extra lines to prevent flow increase"
+    # Section assignment matches the other per-entity property layers.
+    assert layer["section"] == "Individual entities properties"
+
+
 def test_layer_index_skips_semantic_layer_when_no_match():
     """No noise: empty semantic buckets do NOT produce a layer entry."""
     edges = [
@@ -177,6 +206,7 @@ def test_layer_index_skips_semantic_layer_when_no_match():
     keys = {l["key"] for l in _build_layer_index(edges, nodes)}
     assert "semantic:is_hub" not in keys
     assert "semantic:in_red_loop" not in keys
+    assert "semantic:is_extra_cut" not in keys
     assert "color:coral" in keys
 
 

alphaDeesp/tests/test_overflow_graph.py

@@ -352,6 +352,124 @@ def test_min_penwidth_clamping(self):
         assert penwidth == 1.5
 
 
+# ──────────────────────────────────────────────────────────────────────
+# extra_lines_to_cut: operator-supplied extras keep their natural flow
+# colour, are flagged ``is_extra_cut`` (alongside ``constrained``), and
+# stay out of the ``is_overload`` / ``is_monitored`` semantic layers.
+# ──────────────────────────────────────────────────────────────────────
+
+
+def _three_line_df():
+    """L1 positive overload, L2 negative extra-cut, L3 healthy positive."""
+    return pd.DataFrame({
+        "idx_or": [0, 1, 2],
+        "idx_ex": [1, 2, 0],
+        "delta_flows": [1000.0, -100.0, 50.0],
+        "gray_edges": [False, False, False],
+        "line_name": ["L1", "L2", "L3"],
+    })
+
+
+def _edge_by_name(g, name):
+    for u, v, k, data in g.edges(keys=True, data=True):
+        if data.get("name") == name:
+            return (u, v, k), data
+    raise AssertionError(f"edge {name!r} not found")
+
+
+class TestExtraLinesToCut:
+
+    def test_default_extras_is_empty(self):
+        ofg = OverFlowGraph(_basic_topo(3), [0, 1], _three_line_df())
+        assert ofg.extra_lines_cut == set()
+
+    def test_extras_stored_as_set(self):
+        ofg = OverFlowGraph(
+            _basic_topo(3), [0, 1], _three_line_df(),
+            extra_lines_to_cut=[1, 1],
+        )
+        assert ofg.extra_lines_cut == {1}
+
+    def test_extras_keep_natural_flow_colour(self):
+        """An extra-cut line never gets the black overload colour — it
+        keeps coral / blue based on its delta-flow polarity."""
+        ofg = OverFlowGraph(
+            _basic_topo(3), [0, 1], _three_line_df(),
+            extra_lines_to_cut=[1],
+        )
+        _, l1 = _edge_by_name(ofg.g, "L1")  # in lines_to_cut, NOT extra
+        _, l2 = _edge_by_name(ofg.g, "L2")  # in lines_to_cut AND extra
+        _, l3 = _edge_by_name(ofg.g, "L3")  # not cut at all
+        assert l1["color"] == "black"
+        assert l2["color"] == "blue"   # natural — delta_flows = -100
+        assert l3["color"] == "coral"  # natural — delta_flows = +50
+
+    def test_extras_are_constrained_and_flagged(self):
+        ofg = OverFlowGraph(
+            _basic_topo(3), [0, 1], _three_line_df(),
+            extra_lines_to_cut=[1],
+        )
+        _, l1 = _edge_by_name(ofg.g, "L1")
+        _, l2 = _edge_by_name(ofg.g, "L2")
+        _, l3 = _edge_by_name(ofg.g, "L3")
+        # Real overload: constrained, not extra.
+        assert l1.get("constrained") is True
+        assert "is_extra_cut" not in l1
+        # Extra cut: both flags set so downstream layers can find it.
+        assert l2.get("constrained") is True
+        assert l2.get("is_extra_cut") is True
+        # Untouched line carries neither flag.
+        assert "constrained" not in l3
+        assert "is_extra_cut" not in l3
+
+    def test_extras_skipped_in_overload_and_monitored(self):
+        """``highlight_significant_line_loading`` must not stamp
+        ``is_overload`` / ``is_monitored`` on extras, must not yellow-tint
+        their colour, but should still annotate the edge label."""
+        ofg = OverFlowGraph(
+            _basic_topo(3), [0, 1], _three_line_df(),
+            extra_lines_to_cut=[1],
+        )
+        ofg.highlight_significant_line_loading({
+            "L1": {"before": 110, "after": 80},
+            "L2": {"before": 90, "after": 0},
+            "L3": {"before": 75, "after": 60},
+        })
+        _, l1 = _edge_by_name(ofg.g, "L1")
+        _, l2 = _edge_by_name(ofg.g, "L2")
+        _, l3 = _edge_by_name(ofg.g, "L3")
+
+        # L1 is a real overload: yellow-tinted, both flags set.
+        assert l1["color"] == '"black:yellow:black"'
+        assert l1.get("is_overload") is True
+        assert l1.get("is_monitored") is True
+
+        # L2 is the extra cut: keeps natural blue (no yellow tint), no
+        # is_overload / is_monitored, but the loading annotation still
+        # fires so the operator sees how their choice materialises.
+        assert l2["color"] == "blue"
+        assert l2.get("is_overload") is None
+        assert l2.get("is_monitored") is None
+        assert "90% → <B>0%</B>" in l2["label"]
+
+        # L3 is a low-margin line (not overload, not extra).
+        assert l3["color"] == '"coral:yellow:coral"'
+        assert l3.get("is_overload") is None
+        assert l3.get("is_monitored") is True
+
+    def test_legacy_behaviour_when_no_extras(self):
+        """Without ``extra_lines_to_cut`` the contingency lines render
+        black and get tagged as overloads — the legacy contract."""
+        ofg = OverFlowGraph(_basic_topo(3), [0], _three_line_df())
+        ofg.highlight_significant_line_loading({
+            "L1": {"before": 110, "after": 80},
+        })
+        _, l1 = _edge_by_name(ofg.g, "L1")
+        assert l1["color"] == '"black:yellow:black"'
+        assert l1.get("is_overload") is True
+        assert l1.get("is_extra_cut") is None
+
+
 # ──────────────────────────────────────────────────────────────────────
 # detect_edges_to_keep (full method)
 # ──────────────────────────────────────────────────────────────────────