Emit curated input/output stoichiometry; fix .env loading and duplicate reaction rows

bin/create-pathways.py

@@ -17,7 +17,7 @@
 from typing import List, Tuple
 
 import pandas as pd
-from dotenv import dotenv_values
+from dotenv import dotenv_values, load_dotenv
 
 sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
 
@@ -28,6 +28,11 @@
 
 def main() -> None:
     dotenv_path = os.path.join(os.path.dirname(__file__), "..", ".env")
+    # load_dotenv populates os.environ (without clobbering already-set vars) so
+    # NEO4J_URL/USER/PASSWORD in .env actually reach neo4j_connector.get_graph(),
+    # which reads them via os.getenv. dotenv_values alone only builds a local
+    # dict and would leave the connector on its hardcoded defaults.
+    load_dotenv(dotenv_path)
     env_vars = dotenv_values(dotenv_path)
     args = parse_args()
     configure_logging(args.debug, args.verbose)

src/logic_network_generator.py

@@ -727,33 +727,55 @@ def _resolve_vr_entities(
         Dict mapping vr_uid -> (input_node_ids, output_node_ids,
                                 input_stoich_map, output_stoich_map)
     """
-    from src.neo4j_connector import get_reaction_input_output_ids
+    from src.neo4j_connector import (
+        get_reaction_input_output_ids,
+        get_reaction_io_stoichiometry,
+    )
 
     annotated_cache: Dict[tuple, Set[str]] = {}
+    stoich_cache: Dict[tuple, Dict[str, int]] = {}
 
     def _annotated(reaction_id: str, io: str) -> Set[str]:
         key = (reaction_id, io)
         if key not in annotated_cache:
             annotated_cache[key] = set(get_reaction_input_output_ids(reaction_id, io))
         return annotated_cache[key]
 
+    def _annotated_stoich(reaction_id: str, io: str) -> Dict[str, int]:
+        key = (reaction_id, io)
+        if key not in stoich_cache:
+            stoich_cache[key] = get_reaction_io_stoichiometry(reaction_id, io)
+        return stoich_cache[key]
+
+    def _resolve_io(reaction_id: str, io: str, members: Set[str]) -> tuple:
+        # Node identity comes from the reaction's annotated entities; the
+        # curated stoichiometry is carried per annotated entity and attached to
+        # every node that entity maps to (e.g. each expanded set member inherits
+        # the set's coefficient). If two annotated entities map to the same node
+        # their coefficients sum. Absent curation, the coefficient defaults to 1.
+        by_entity = _annotated_stoich(reaction_id, io)
+        node_ids: Set[str] = set()
+        node_stoich: Dict[str, int] = {}
+        for e in _annotated(reaction_id, io):
+            s = by_entity.get(str(e), 1)
+            for n in _map_annotated_entity_to_nodes(str(e), members):
+                node_ids.add(n)
+                node_stoich[n] = node_stoich.get(n, 0) + s
+        return list(node_ids), node_stoich
+
     vr_entities: Dict[str, tuple] = {}
     for _, row in reaction_id_map.iterrows():
         vr_uid = row["uid"]
         reaction_id = str(row["reactome_id"])
         input_members = set(_resolve_to_terminal_reactome_ids(uid_index, row["input_hash"]))
         output_members = set(_resolve_to_terminal_reactome_ids(uid_index, row["output_hash"]))
 
-        input_ids: Set[str] = set()
-        for e in _annotated(reaction_id, "input"):
-            input_ids |= _map_annotated_entity_to_nodes(str(e), input_members)
-        output_ids: Set[str] = set()
-        for e in _annotated(reaction_id, "output"):
-            output_ids |= _map_annotated_entity_to_nodes(str(e), output_members)
+        input_ids, input_stoich = _resolve_io(reaction_id, "input", input_members)
+        output_ids, output_stoich = _resolve_io(reaction_id, "output", output_members)
 
         vr_entities[vr_uid] = (
-            list(input_ids), list(output_ids),
-            {n: 1 for n in input_ids}, {n: 1 for n in output_ids},
+            input_ids, output_ids,
+            input_stoich, output_stoich,
         )
     return vr_entities
 

src/neo4j_connector.py

@@ -270,7 +270,7 @@ def get_reaction_connections(pathway_id: str) -> pd.DataFrame:
         WHERE pathway.stId = $pathway_id
         OPTIONAL MATCH (r1)<-[:precedingEvent]-(r2:ReactionLikeEvent)<-[:hasEvent*]-(pathway)
         WHERE pathway.stId = $pathway_id
-        RETURN r1.stId AS preceding_reaction_id,
+        RETURN DISTINCT r1.stId AS preceding_reaction_id,
                r2.stId AS following_reaction_id,
                CASE WHEN r2 IS NULL THEN 'No Preceding Event' ELSE 'Has Preceding Event' END AS event_status
         """
@@ -548,6 +548,44 @@ def get_reaction_input_output_ids(reaction_id: str, input_or_output: str) -> Set
         raise
 
 
+def get_reaction_io_stoichiometry(reaction_id: str, input_or_output: str) -> Dict[str, int]:
+    """Curated input/output stoichiometry for a reaction, keyed by entity stId.
+
+    Reactome stores the coefficient on the input/output relationship (the same
+    ``stoichiometry`` property that hasComponent uses). An entity linked with no
+    explicit coefficient defaults to 1; where the same entity is linked by more
+    than one relationship the coefficients are summed. Companion to
+    :func:`get_reaction_input_output_ids`, which returns only the entity ids.
+    """
+    # Same relationship-type allowlist as get_reaction_input_output_ids: the
+    # type can't be parameterized in Cypher, so restrict it before embedding.
+    if input_or_output not in {"input", "output"}:
+        raise ValueError(f"input_or_output must be 'input' or 'output', got {input_or_output!r}")
+
+    query: str = (
+        f"MATCH (reaction)-[rel:{input_or_output}]-(io) "
+        f"WHERE (reaction:Reaction OR reaction:ReactionLikeEvent) "
+        f"AND reaction.stId = $reaction_id "
+        f"RETURN io.stId AS io_id, rel.stoichiometry AS stoichiometry"
+    )
+
+    try:
+        data = get_graph().run(query, reaction_id=reaction_id).data()
+    except Exception:
+        logger.error("Error in get_reaction_io_stoichiometry", exc_info=True)
+        raise
+
+    result: Dict[str, int] = {}
+    for row in data:
+        io_id = row["io_id"]
+        if io_id is None:
+            continue
+        stoich_raw = row.get("stoichiometry")
+        stoich = 1 if stoich_raw is None else int(stoich_raw)
+        result[io_id] = result.get(io_id, 0) + stoich
+    return result
+
+
 def get_reference_entity_id(entity_id: str) -> Union[str, None]:
     if entity_id in _reference_entity_cache:
         return _reference_entity_cache[entity_id]

tests/test_logic_network_generator.py

@@ -297,8 +297,9 @@ def test_no_duplicate_edges(self, monkeypatch):
 
         Since set-variant emission, node identities come from the reaction's
         annotated input/output entities (mapped to nodes), and duplicates are
-        collapsed via a set. We mock the two Neo4j calls the resolver makes:
-        the reaction's annotated entities and their labels (simple proteins).
+        collapsed via a set. We mock the Neo4j calls the resolver makes: the
+        reaction's annotated entities, their curated I/O stoichiometry, and
+        their labels (simple proteins).
         """
         import src.logic_network_generator as m
         from src import neo4j_connector
@@ -308,6 +309,11 @@ def test_no_duplicate_edges(self, monkeypatch):
             neo4j_connector, "get_reaction_input_output_ids",
             lambda rid, io: {"9933417"} if io == "input" else {"12345"},
         )
+        # Curated stoichiometry: 2x the input, 1x the output.
+        monkeypatch.setattr(
+            neo4j_connector, "get_reaction_io_stoichiometry",
+            lambda rid, io: {"9933417": 2} if io == "input" else {"12345": 1},
+        )
         # Both entities are simple (not complexes/sets) -> mapped to themselves.
         monkeypatch.setattr(
             neo4j_connector, "get_labels",
@@ -329,10 +335,13 @@ def test_no_duplicate_edges(self, monkeypatch):
         })
 
         vr_entities = _resolve_vr_entities(reaction_id_map, uid_index)
-        input_ids, output_ids, _in_stoich, _out_stoich = vr_entities["vr1"]
+        input_ids, output_ids, in_stoich, out_stoich = vr_entities["vr1"]
 
         assert input_ids == ["9933417"], f"expected one deduped input, got {input_ids}"
         assert output_ids == ["12345"], f"expected one output, got {output_ids}"
+        # Curated stoichiometry is carried onto the nodes (not hardcoded to 1).
+        assert in_stoich == {"9933417": 2}, f"expected curated input stoich, got {in_stoich}"
+        assert out_stoich == {"12345": 1}, f"expected curated output stoich, got {out_stoich}"
 
     def test_root_input_same_entity_gets_one_uuid(self):
         """Root input entity appearing at multiple reactions should share one UUID."""