LNG_DIAGRAM_SUBPATHWAYS: read sub-pathway diagrams for drawn hand-offs (default off)

src/diagram_connectivity.py

@@ -71,30 +71,25 @@ def _pathway_reaction_stids(pathway_id: str) -> Set[str]:
     return set(rows or [])
 
 
-def diagram_shared_product_pairs(pathway_id: str) -> Set[Tuple[str, str]]:
-    """(producer_stId, consumer_stId) reaction pairs drawn as connected in the diagram.
+def _subpathway_diagram_stids(pathway_id: str) -> List[str]:
+    """stIds of the pathways under this one that have a diagram of their own.
 
-    A pair is emitted when a producer reaction's OUTPUT glyph is the same glyph a
-    consumer reaction takes as INPUT or CATALYST. Reaction identity is the
-    Reactome stId (mapped from the diagram's numeric reactomeId via the
-    companion ``.graph.json``). The diagram used is the pathway's own, else the
-    nearest diagrammed ancestor; pairs are then **restricted to reactions that
-    belong to this pathway** so an ancestor diagram only contributes the target
-    pathway's internal connectivity. Returns an empty set if no diagram covers it.
+    A top-level diagram draws such a sub-pathway as one box and its reactions
+    only on the sub-pathway's own diagram, so its drawn hand-offs are read
+    from there (deltasignal specs/070).
     """
-    ddir = _diagram_dir()
-    diagram_stid = _covering_diagram_stid(pathway_id)
-    if not diagram_stid:
-        logger.info(f"No diagram (own or ancestor) covers {pathway_id}; skipping diagram connectivity")
-        return set()
-    if diagram_stid != pathway_id:
-        logger.info(f"{pathway_id} has no own diagram; using ancestor diagram {diagram_stid}")
+    from src.neo4j_connector import get_graph
+    rows = get_graph().run(
+        "MATCH (:Pathway {stId:$pid})-[:hasEvent*]->(s:Pathway) "
+        "RETURN collect(DISTINCT s.stId) AS s",
+        pid=pathway_id,
+    ).evaluate()
+    return sorted(s for s in (rows or []) if _has_diagram(s))
 
-    layout = json.loads((ddir / f"{diagram_stid}.json").read_text())
-    graph = json.loads((ddir / f"{diagram_stid}.graph.json").read_text())
-    # Restrict to this pathway's reactions (isolate it within the ancestor diagram).
-    own_reactions = _pathway_reaction_stids(pathway_id)
 
+def _drawn_pairs(layout: dict, graph: dict) -> Set[Tuple[str, str]]:
+    """(producer_stId, consumer_stId) for every glyph one reaction outputs and
+    another takes as input or catalyst, on one diagram."""
     # reaction dbId -> stId (graph.json edges carry both)
     dbid_to_stid = {e["dbId"]: e["stId"] for e in graph.get("edges", []) if e.get("stId")}
 
@@ -120,13 +115,49 @@ def diagram_shared_product_pairs(pathway_id: str) -> Set[Tuple[str, str]]:
                     continue
                 p_st = dbid_to_stid.get(producer)
                 c_st = dbid_to_stid.get(consumer)
-                if not p_st or not c_st:
-                    continue
-                # Keep only pairs whose BOTH reactions belong to this pathway
-                # (so an ancestor diagram contributes only this pathway's flow).
-                if own_reactions and (p_st not in own_reactions or c_st not in own_reactions):
-                    continue
-                pairs.add((p_st, c_st))
+                if p_st and c_st:
+                    pairs.add((p_st, c_st))
+    return pairs
+
+
+def diagram_shared_product_pairs(pathway_id: str) -> Set[Tuple[str, str]]:
+    """(producer_stId, consumer_stId) reaction pairs drawn as connected in the diagram.
+
+    A pair is emitted when a producer reaction's OUTPUT glyph is the same glyph a
+    consumer reaction takes as INPUT or CATALYST. Reaction identity is the
+    Reactome stId (mapped from the diagram's numeric reactomeId via the
+    companion ``.graph.json``). The diagram used is the pathway's own, else the
+    nearest diagrammed ancestor; pairs are then **restricted to reactions that
+    belong to this pathway** so an ancestor diagram only contributes the target
+    pathway's internal connectivity. Returns an empty set if no diagram covers it.
+    """
+    ddir = _diagram_dir()
+    diagram_stid = _covering_diagram_stid(pathway_id)
+    if not diagram_stid:
+        logger.info(f"No diagram (own or ancestor) covers {pathway_id}; skipping diagram connectivity")
+        return set()
+    if diagram_stid != pathway_id:
+        logger.info(f"{pathway_id} has no own diagram; using ancestor diagram {diagram_stid}")
+
+    # Restrict to this pathway's reactions (isolate it within the ancestor diagram).
+    own_reactions = _pathway_reaction_stids(pathway_id)
+
+    # LNG_DIAGRAM_SUBPATHWAYS: also every sub-pathway's own diagram, where the
+    # covering diagram draws that sub-pathway as a box (deltasignal specs/070).
+    diagrams = [diagram_stid]
+    if env_flag("LNG_DIAGRAM_SUBPATHWAYS"):
+        diagrams += [s for s in _subpathway_diagram_stids(pathway_id) if s != diagram_stid]
+
+    pairs: Set[Tuple[str, str]] = set()
+    for stid in diagrams:
+        layout = json.loads((ddir / f"{stid}.json").read_text())
+        graph = json.loads((ddir / f"{stid}.graph.json").read_text())
+        for p_st, c_st in _drawn_pairs(layout, graph):
+            # Keep only pairs whose BOTH reactions belong to this pathway
+            # (so an ancestor diagram contributes only this pathway's flow).
+            if own_reactions and (p_st not in own_reactions or c_st not in own_reactions):
+                continue
+            pairs.add((p_st, c_st))
     return pairs
 
 

src/env_flags.py

@@ -23,6 +23,7 @@
     "LNG_DIAGRAM_BRIDGE": False,
     "LNG_DIAGRAM_CONNECTIVITY": True,
     "LNG_DIAGRAM_SET_MEMBER": False,
+    "LNG_DIAGRAM_SUBPATHWAYS": False,  # deltasignal specs/070
     "LNG_EMIT_ONE_SIDED": True,
     "LNG_HANDOFF_EDGES": False,
     "LNG_PATHOGEN_PROTEIN": True,   # default since deltasignal specs/048 amendment 2 (arm pathp: exp +7, held-out +1, 0 broken)

src/pathway_generator.py

@@ -53,6 +53,7 @@
     "LNG_PATHOGEN_PROTEIN",
     "LNG_VARIANT_NODES",
     "LNG_DIAGRAM_SET_MEMBER",
+    "LNG_DIAGRAM_SUBPATHWAYS",
     "LNG_SPECIALIZATION_BRIDGE",
     "LNG_SPECIALIZATION_CLOSED",
     # Of these three, only LNG_EMIT_ONE_SIDED is a genuine cache gap: it is

tests/test_diagram_connectivity.py

@@ -75,3 +75,25 @@ def test_augment_disabled_by_env(diagram_dir, monkeypatch):
                        "following_reaction_id": ["R-HSA-2"],
                        "event_status": ["x"]})
     assert dc.augment_reaction_connections("R-HSA-TEST", rc) is rc
+
+
+def test_subpathway_diagram_read_only_under_flag(diagram_dir, monkeypatch):
+    """A sub-pathway drawn as a box on the covering diagram has its hand-offs on
+    its own diagram; LNG_DIAGRAM_SUBPATHWAYS reads them (deltasignal specs/070)."""
+    (diagram_dir / "R-HSA-TOP.json").write_text(json.dumps({"nodes": [], "edges": []}))
+    (diagram_dir / "R-HSA-TOP.graph.json").write_text(json.dumps({"edges": []}))
+    _write_diagram(diagram_dir, "R-HSA-SUB")
+    monkeypatch.setattr(dc, "_subpathway_diagram_stids", lambda pid: ["R-HSA-SUB"])
+    assert dc.diagram_shared_product_pairs("R-HSA-TOP") == set()
+    monkeypatch.setenv("LNG_DIAGRAM_SUBPATHWAYS", "1")
+    assert dc.diagram_shared_product_pairs("R-HSA-TOP") == {("R-HSA-100", "R-HSA-200")}
+
+
+def test_subpathway_pairs_still_restricted_to_own_reactions(diagram_dir, monkeypatch):
+    (diagram_dir / "R-HSA-TOP.json").write_text(json.dumps({"nodes": [], "edges": []}))
+    (diagram_dir / "R-HSA-TOP.graph.json").write_text(json.dumps({"edges": []}))
+    _write_diagram(diagram_dir, "R-HSA-SUB")
+    monkeypatch.setattr(dc, "_subpathway_diagram_stids", lambda pid: ["R-HSA-SUB"])
+    monkeypatch.setattr(dc, "_pathway_reaction_stids", lambda pid: {"R-HSA-100"})
+    monkeypatch.setenv("LNG_DIAGRAM_SUBPATHWAYS", "1")
+    assert dc.diagram_shared_product_pairs("R-HSA-TOP") == set()