diff --git a/alphaDeesp/core/graphs/overflow_graph.py b/alphaDeesp/core/graphs/overflow_graph.py index 3681f61..d2931e2 100644 --- a/alphaDeesp/core/graphs/overflow_graph.py +++ b/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}
{before}% → {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}
{before}% → {after}%>' + is_overload_attrs[edge] = True + else: + edge_x_labels[edge] = f'< {current_x_label}
{before}% → {after}% >' + 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}
{before}% → {after}% >' 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") diff --git a/alphaDeesp/core/interactive_html.py b/alphaDeesp/core/interactive_html.py index 5af7693..4562669 100644 --- a/alphaDeesp/core/interactive_html.py +++ b/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 ''; if (swatch === 'overload') return ''; if (swatch === 'monitored') return ''; + if (swatch === 'extra-cut') return ''; // Match the upstream node fillcolors set in build_nodes: // prod (prod_minus_load > 0) → coral // load (prod_minus_load < 0) → lightblue diff --git a/alphaDeesp/tests/test_interactive_html.py b/alphaDeesp/tests/test_interactive_html.py index c70c099..282b2a2 100644 --- a/alphaDeesp/tests/test_interactive_html.py +++ b/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 diff --git a/alphaDeesp/tests/test_overflow_graph.py b/alphaDeesp/tests/test_overflow_graph.py index 4092848..66ac2f3 100644 --- a/alphaDeesp/tests/test_overflow_graph.py +++ b/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% → 0%" 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) # ──────────────────────────────────────────────────────────────────────