leaf_refs.csv: leaf -> reference entity (deltasignal specs/107)

src/logic_network_generator.py

@@ -4037,6 +4037,24 @@ def export_rnas(pathway_logic_network: pd.DataFrame,
                     get_rna_entities, "RNA", by_member=False)
 
 
+def export_leaf_refs(nodes_file: str, output_file: str) -> None:
+    """Write leaf_refs.csv (deltasignal specs/107): every leaf named in the
+    ``member_leaves`` of ``nodes_file`` (this bundle's nodes.csv) that has a
+    reference entity, with that reference entity's stable id, so a consumer can tell
+    that GRIP2 at the plasma membrane and GRIP2 in the cytosol are one protein.
+    Always written (a header alone when no leaf has a reference entity), so a
+    missing file means an older bundle, not "no references"."""
+    from src.neo4j_connector import get_leaf_references
+    leaves: set[str] = set()
+    nodes = pd.read_csv(nodes_file, dtype=str, keep_default_na=False)
+    for v in nodes.get("member_leaves", pd.Series(dtype=str)):
+        leaves.update(x for x in str(v).split("|") if x)
+    refs = get_leaf_references(leaves)
+    rows = [{"leaf_stid": leaf, "reference_stid": refs[leaf]} for leaf in sorted(refs)]
+    pd.DataFrame(rows, columns=["leaf_stid", "reference_stid"]).to_csv(output_file, index=False)
+    logger.info(f"Exported {len(rows)} leaf references to {output_file}")
+
+
 def _export_derived(pathway_logic_network: pd.DataFrame,
                     reactome_id_to_uuid: Dict[str, str],
                     output_file: str, finder, label: str, by_member: bool = True) -> None:

src/neo4j_connector.py

@@ -956,6 +956,20 @@ def get_rna_entities(stable_ids) -> Dict[str, Dict[str, str]]:
     return {r["s"]: {"schema_class": r["c"], "name": r["n"]} for r in rows}
 
 
+def get_leaf_references(stable_ids) -> Dict[str, str]:
+    """Each of ``stable_ids`` that has a reference entity, mapped to that reference
+    entity's stable id (``uniprot:Q9C0E4``, ``chebi:57613``): deltasignal specs/107.
+    Copies of one protein in different compartments, or modified, are different
+    entities with one reference entity. Entities without one are left out."""
+    ids = sorted({s for s in stable_ids if s})
+    if not ids:
+        return {}
+    rows = get_graph().run(
+        "UNWIND $ids AS s MATCH (e:PhysicalEntity {stId: s})-[:referenceEntity]->(r:ReferenceEntity) "
+        "RETURN s, r.stId AS ref", ids=ids).data()
+    return {r["s"]: r["ref"] for r in rows if r["ref"]}
+
+
 def get_pathogen_entities(stable_ids) -> Dict[str, Dict[str, str]]:
     """The entities among ``stable_ids`` that are PATHOGEN-DERIVED, with their
     schema class and display name (deltasignal specs/048).

src/pathway_generator.py

@@ -16,6 +16,7 @@
     export_pathogens,
     export_pools,
     export_rnas,
+    export_leaf_refs,
     export_containment,
     export_containment_structure,
     export_entity_reaction_proxy_mapping,
@@ -237,7 +238,7 @@ def _write_cache_fingerprint(
     "logic_network.csv", "stid_to_uuid_mapping.csv", "entity_reaction_proxy_mapping.csv",
     "nodes.csv", "node_reaction_context.csv", "node_resolution.csv", "node_exclusions.csv",
     "boundary_edges.csv", "cofactors.csv", "containment.csv", "containment_structure.csv",
-    "drugs.csv", "pathogens.csv", "rnas.csv",
+    "drugs.csv", "pathogens.csv", "rnas.csv", "leaf_refs.csv",
     "pools.csv", "pool_transitions.csv", "pool_carriers.csv", "carrier_loops.csv",
 )
 
@@ -629,6 +630,16 @@ def generate_pathway_file(
             logger.error(f"Failed to write rnas.csv: {e}", exc_info=True)
             raise
 
+        # Leaf reference entities (deltasignal specs/107), likewise on their own.
+        try:
+            export_leaf_refs(
+                str(pathway_output_dir / "nodes.csv"),
+                str(pathway_output_dir / "leaf_refs.csv"),
+            )
+        except Exception as e:
+            logger.error(f"Failed to write leaf_refs.csv: {e}", exc_info=True)
+            raise
+
         # Interconversion pools (deltasignal specs/039), in their own block.
         try:
             export_pools(

tests/test_export_failfast.py

@@ -24,7 +24,7 @@
 EXPORTS = ["export_uuid_to_reactome_mapping", "export_entity_reaction_proxy_mapping",
            "export_nodes", "export_node_reaction_context", "export_node_resolution",
            "export_cofactors", "export_containment", "export_containment_structure",
-           "export_drugs", "export_pathogens", "export_rnas", "export_pools"]
+           "export_drugs", "export_pathogens", "export_rnas", "export_leaf_refs", "export_pools"]
 
 
 @pytest.fixture

tests/test_provenance_exports.py

@@ -512,6 +512,31 @@ def test_pathogen_polymer_counts_through_its_repeated_unit(monkeypatch):
 
 
 
+# --- leaf_refs.csv (deltasignal specs/107) -------------------------------------
+
+def test_export_leaf_refs_maps_every_leaf_with_a_reference(tmp_path, monkeypatch):
+    seen = {}
+    def refs(stids):
+        seen["ids"] = set(stids)
+        return {"R-HSA-GM": "uniprot:Q9C0E4", "R-HSA-GC": "uniprot:Q9C0E4", "R-ALL-PE": "chebi:57613"}
+    monkeypatch.setattr(neo4j_connector, "get_leaf_references", refs)
+    nodes = tmp_path / "nodes.csv"
+    pd.DataFrame([{"uuid": "u1", "node_kind": "simple_entity", "member_leaves": "R-HSA-GM"},
+                  {"uuid": "u2", "node_kind": "simple_complex", "member_leaves": "R-HSA-GC|R-ALL-PE|R-HSA-X"},
+                  {"uuid": "u3", "node_kind": "reaction", "member_leaves": ""}]).to_csv(nodes, index=False)
+    out = tmp_path / "leaf_refs.csv"
+    m.export_leaf_refs(str(nodes), str(out))
+    assert seen["ids"] == {"R-HSA-GM", "R-HSA-GC", "R-ALL-PE", "R-HSA-X"}
+    df = pd.read_csv(out)
+    assert list(df.columns) == ["leaf_stid", "reference_stid"]
+    assert list(df.itertuples(index=False, name=None)) == [
+        ("R-ALL-PE", "chebi:57613"), ("R-HSA-GC", "uniprot:Q9C0E4"), ("R-HSA-GM", "uniprot:Q9C0E4")]
+    # nothing with a reference: the header alone, so the file still exists
+    monkeypatch.setattr(neo4j_connector, "get_leaf_references", lambda s: {})
+    m.export_leaf_refs(str(nodes), str(out))
+    assert list(pd.read_csv(out).columns) == ["leaf_stid", "reference_stid"] and len(pd.read_csv(out)) == 0
+
+
 # --- rnas.csv (deltasignal specs/100) ------------------------------------------
 
 def test_export_rnas_lists_rna_species(tmp_path, monkeypatch):