diff --git a/src/env_flags.py b/src/env_flags.py
index a6f3d5f..0239a85 100644
--- a/src/env_flags.py
+++ b/src/env_flags.py
@@ -27,6 +27,7 @@
"LNG_SET_EXPAND": True,
"LNG_SET_MEMBERS_OR": False,
"LNG_SET_POOL": True,
+ "LNG_VARIANT_NODES": True, # default since deltasignal specs/046 (vn7: held-out +130, experimental -10 ns)
}
# Read elsewhere with their own parsing and validation.
diff --git a/src/logic_network_generator.py b/src/logic_network_generator.py
index 826d338..9790f0f 100755
--- a/src/logic_network_generator.py
+++ b/src/logic_network_generator.py
@@ -2,7 +2,7 @@
from src.env_flags import env_flag, validate_env
import uuid
from collections import Counter
-from typing import Dict, List, Any, NamedTuple, Optional, Set, Tuple
+from typing import Dict, List, Any, NamedTuple, Optional, Sequence, Set, Tuple
import pandas as pd
from pandas import DataFrame
@@ -659,10 +659,8 @@ def _build_reactome_to_vr_map(reaction_id_map: pd.DataFrame) -> Dict[str, List[s
def _parse_variant_members(variant_id: str) -> Set[str]:
"""Terminal member stIds encoded in a ``{parent}::variant::{m1_m2}`` id."""
- if "::variant::" not in variant_id:
- return set()
- tail = variant_id.split("::variant::", 1)[1]
- return {m for m in tail.split("_") if m}
+ from src.variant_keys import variant_parts
+ return set(variant_parts(variant_id)[1])
_variant_leafsets_cache: Dict[str, List[frozenset]] = {}
@@ -999,6 +997,10 @@ def _decompose_regulator_entity(
"""
from src.neo4j_connector import get_labels, get_complex_components, get_set_members
+ # A specs/046 variant key is one node: the copy already chose its members.
+ # So is a pool reference from a capped reaction (wired by _wire_variant_pool_refs).
+ if ("::variant::" in entity_id and "=" in entity_id) or entity_id.endswith("::pool"):
+ return [(entity_id, 1)]
labels = get_labels(entity_id)
if "Complex" in labels:
@@ -1218,6 +1220,208 @@ def _bridge_excluded_stids() -> frozenset:
return _cofactor_stids() | _UBIQUITIN_STIDS
+def _emit_variant_pool(base: str, member_uuids, pathway_logic_network_data: List[Dict[str, Any]],
+ reactome_id_to_uuid: Dict[str, str], pool_cache: Dict[tuple, str]) -> str:
+ """One node standing for ``base`` as a whole, fed by the given nodes of its
+ variants (deltasignal specs/046 decision D5).
+
+ The members arrive as OR inputs (``variant_pool`` edges), so the solver
+ combines them by its OR rule, ``mean``: the same answer as reading each
+ variant in its own reaction copy and OR-ing the copies. The node id is
+ ``::pool``, which no readout or gene resolves to, so readout and
+ perturbation resolution are unchanged. One pool per (base, member set) per
+ pathway."""
+ key = (base, frozenset(member_uuids))
+ pool = pool_cache.get(key)
+ if pool is None:
+ pool = str(uuid.uuid4())
+ pool_cache[key] = pool
+ reactome_id_to_uuid[pool] = f"{base}::pool"
+ for m in sorted(member_uuids):
+ pathway_logic_network_data.append({
+ "source_id": m, "target_id": pool, "pos_neg": "pos",
+ "and_or": "or", "edge_type": "variant_pool", "stoichiometry": 1,
+ })
+ return pool
+
+
+def _wire_variant_pool_refs(pathway_logic_network_data: List[Dict[str, Any]],
+ reactome_id_to_uuid: Dict[str, str]) -> None:
+ """Feed every ``::pool`` node a capped reaction reads (specs/046 D6)
+ from the node each variant's own copy would have read, so the pool's OR
+ equals expanding (D5). The variants are enumerated from the entity, not
+ looked up by id prefix: a bare set's variant key is its chosen MEMBER's
+ key, which carries no ``::variant::`` prefix (vn5: RPL10/RPL22/RPL5
+ in ROBO resolved to no node at all, because the pooled set's members
+ appeared nowhere else). For each key: the nodes some reaction produces;
+ else the key's existing nodes; else a new root node for the key, which is
+ exactly what the expanded copy would have registered. An entity with more
+ variants than the cap falls back to the nodes that exist (counted)."""
+ from src import variant_keys as vk
+ produced = {e["target_id"] for e in pathway_logic_network_data if e.get("edge_type") == "output"}
+ fed = {e["target_id"] for e in pathway_logic_network_data if e.get("edge_type") == "variant_pool"}
+ by_key: Dict[str, List[str]] = {}
+ refs: List[Tuple[str, str]] = []
+ for node_uuid, sid in list(reactome_id_to_uuid.items()):
+ if sid.endswith("::pool"):
+ if node_uuid not in fed:
+ refs.append((node_uuid, sid[:-len("::pool")]))
+ continue
+ by_key.setdefault(sid, []).append(node_uuid)
+ cap = vk._max_variants()
+ n_edges = n_new = n_over = 0
+ for pool_uuid, base in sorted(refs):
+ keys = sorted({vk.vkey(base, sig) for sig in
+ vk.reaction_choices([base], limit=(cap + 1) if cap > 0 else None)})
+ if cap > 0 and len(keys) > cap:
+ # Too many variants to enumerate: read the variant nodes that exist.
+ # A bare set's variants are its MEMBERS' keys (a member's own stId
+ # or its variant key), never "::variant::..." (review of vn6:
+ # RAS GEFs, 645 variants, was a pool with no input at all).
+ n_over += 1
+ owners = {base} | (set(vk.flat_members(base)) if vk.is_set(base) else set())
+ keys = sorted(k for k in by_key
+ if k != base and not k.endswith("::pool")
+ and vk.variant_parts(k)[0] in owners)
+ if not keys and vk.is_set(base):
+ # none exists: one root per member, by its plain stId
+ keys = sorted(set(vk.flat_members(base)))
+ members: List[str] = []
+ for key in keys:
+ nodes = by_key.get(key, [])
+ made = sorted(u for u in nodes if u in produced)
+ if made or nodes:
+ members.extend(made or sorted(nodes))
+ else:
+ root = str(uuid.uuid4())
+ reactome_id_to_uuid[root] = key
+ by_key[key] = [root]
+ members.append(root)
+ n_new += 1
+ for m in members:
+ pathway_logic_network_data.append({
+ "source_id": m, "target_id": pool_uuid, "pos_neg": "pos",
+ "and_or": "or", "edge_type": "variant_pool", "stoichiometry": 1,
+ })
+ n_edges += 1
+ if refs:
+ logger.info(f"Variant nodes: {len(refs)} pool references wired by {n_edges} "
+ f"variant_pool edges; {n_new} variant root nodes created; "
+ f"{n_over} entities over the cap read existing nodes only")
+
+
+def _wire_capped_outputs(pathway_logic_network_data: List[Dict[str, Any]],
+ reactome_id_to_uuid: Dict[str, str],
+ following_by_reaction: Optional[Dict[str, Set[str]]] = None,
+ vr_to_reaction: Optional[Dict[str, str]] = None) -> None:
+ """A reaction past every D6 step is ONE copy writing its outputs by plain
+ stable id, while consumers read the outputs' variant keys, so nothing fed
+ them (vn5 RAF: R-HSA-5672980 -> R-HSA-5672701, 34 experimental cases cut).
+ Feed each existing node of the output's variant keys from the plain output
+ by a ``variant_split`` OR edge: every variant reads the one copy, which is
+ what the expanded copies would each have produced. A plain node is such an
+ output exactly when it is produced and the entity has variant slots.
+
+ Only variant nodes nothing else produces are fed, and only where a
+ consumer is a curated following reaction of the producer (identity across
+ connected pairs, decision D1). Feeding every node of every key diluted
+ already-produced nodes and reached unrelated occurrences of a member
+ protein (review of vn6: 14 such edges, one onto a 33-producer node)."""
+ from src import variant_keys as vk
+ produced = {e["target_id"] for e in pathway_logic_network_data if e.get("edge_type") == "output"}
+ producer_rx: Dict[str, Set[str]] = {}
+ consumer_rx: Dict[str, Set[str]] = {}
+ vr_to_reaction = vr_to_reaction or {}
+ for e in pathway_logic_network_data:
+ et = e.get("edge_type")
+ if et == "output":
+ rx = e.get("edge_reaction_id") or vr_to_reaction.get(str(e["source_id"]), "")
+ producer_rx.setdefault(e["target_id"], set()).add(str(rx))
+ elif et in ("input", "catalyst", "regulator"):
+ rx = e.get("edge_reaction_id") or vr_to_reaction.get(str(e["target_id"]), "")
+ consumer_rx.setdefault(e["source_id"], set()).add(str(rx))
+ following_by_reaction = following_by_reaction or {}
+ by_key: Dict[str, List[str]] = {}
+ for node_uuid, sid in reactome_id_to_uuid.items():
+ by_key.setdefault(sid, []).append(node_uuid)
+ cap = vk._max_variants()
+ n_out = n_edges = 0
+ for sid in sorted(by_key):
+ if "::" in sid:
+ continue
+ plain = [u for u in by_key[sid] if u in produced]
+ if not plain:
+ continue
+ try:
+ if not (vk.is_set(sid) or (vk.is_complex(sid) and not vk.is_capped(sid) and vk.own_slots(sid))):
+ continue
+ except Exception:
+ continue
+ keys = {vk.vkey(sid, sig) for sig in vk.reaction_choices([sid], limit=(cap + 1) if cap > 0 else None)}
+ candidates = sorted(u for k in keys if k != sid for u in by_key.get(k, []) if u not in produced)
+ made = False
+ for src in sorted(plain):
+ follows = set().union(*[following_by_reaction.get(r, set()) for r in producer_rx.get(src, ())]) \
+ if producer_rx.get(src) else set()
+ targets = [t for t in candidates if consumer_rx.get(t, set()) & follows]
+ made = made or bool(targets)
+ for t in targets:
+ pathway_logic_network_data.append({
+ "source_id": src, "target_id": t, "pos_neg": "pos",
+ "and_or": "or", "edge_type": "variant_split", "stoichiometry": 1,
+ })
+ n_edges += 1
+ n_out += made
+ if n_out:
+ logger.info(f"Variant nodes: {n_out} capped plain outputs feed their variants "
+ f"by {n_edges} variant_split edges")
+
+
+def _pool_variant_depleters(pathway_logic_network_data: List[Dict[str, Any]], start: int,
+ reactome_id_to_uuid: Dict[str, str]) -> None:
+ """Replace depletion edges that reach ONE target from several variants of
+ ONE depleter entity with a single edge from a pool of those variants.
+
+ Depletion runs from a reaction's catalyst to its substrate node, and that
+ node is shared by every copy of the reaction. Once a catalyst set is
+ expanded, each copy's catalyst variant depleted the shared substrate on its
+ own, and the solver multiplies depletions: an n-variant catalyst acted n
+ times. Traced in TP53 (specs/046 research): TP53 produced at 2.05x read
+ 2.05 / 1.87^2 = 0.59 under two variants of R-HSA-6804885, which turned 49
+ correct AKT1-knockout calls into wrong ones. The variants are alternative
+ forms of one catalyst, so its activity is their OR."""
+ tail = pathway_logic_network_data[start:]
+
+ def base_of(e: Dict[str, Any]) -> str:
+ # the curated catalyst it came from; else the id's parent stId
+ return e.get("_origin") or \
+ reactome_id_to_uuid.get(e["source_id"], e["source_id"]).split("::variant::")[0]
+
+ groups: Dict[tuple, List[str]] = {}
+ for e in tail:
+ if e.get("edge_type") == "depletion":
+ groups.setdefault((e["target_id"], base_of(e)), []).append(e["source_id"])
+ pooled = {k: v for k, v in groups.items() if len(set(v)) > 1}
+ kept = [e for e in tail if not (
+ e.get("edge_type") == "depletion" and (e["target_id"], base_of(e)) in pooled)]
+ for e in kept:
+ e.pop("_origin", None)
+ del pathway_logic_network_data[start:]
+ pathway_logic_network_data.extend(kept)
+ if not pooled:
+ return
+ cache: Dict[tuple, str] = {}
+ for (target, base), sources in sorted(pooled.items()):
+ pool = _emit_variant_pool(base, set(sources), pathway_logic_network_data,
+ reactome_id_to_uuid, cache)
+ pathway_logic_network_data.append({
+ "source_id": pool, "target_id": target, "pos_neg": "neg",
+ "and_or": "and", "edge_type": "depletion", "stoichiometry": 1.0,
+ })
+ logger.info(f"Variant nodes: {len(pooled)} depletion targets read a pool of "
+ f"{sum(len(set(v)) for v in pooled.values())} catalyst variants")
+
+
def _emit_substrate_depletion_edges(
pathway_logic_network_data: List[Dict[str, Any]],
reactome_id_to_uuid: Dict[str, str],
@@ -1400,6 +1604,19 @@ def _pool_contains(cat_uuid: str, uid: str, sid: str) -> bool:
seen_edges: set = set()
n_emitted_phos = 0
n_emitted_ub = 0
+ first_depletion = len(pathway_logic_network_data)
+ # specs/046: which curated catalyst each (reaction copy, catalyst node)
+ # came from, so the variants of ONE catalyst are pooled even when, as for
+ # a bare set, their ids share no prefix (review of vn6: 6 DUSP members of
+ # one set each depleted the MAPK3 dimer).
+ origin_of: Dict[Tuple[str, str], str] = {}
+ if env_flag("LNG_VARIANT_NODES") and "participant" in catalyst_map.columns:
+ for _, row in catalyst_map.iterrows():
+ origin_of[(str(row["reaction_uuid"]), str(row["entity_id"]))] = str(row["participant"])
+
+ def _origin(rxn_uuid: str, cat_uuid: str) -> Dict[str, str]:
+ o = origin_of.get((rxn_uuid, reactome_id_to_uuid.get(cat_uuid, "")))
+ return {"_origin": o} if o else {}
# Phosphatase pass — existing behavior.
for rxn_uuid in phosphatase_rxn_uuids:
catalyst_uuids = by_target_catalysts.get(rxn_uuid, [])
@@ -1435,6 +1652,7 @@ def _pool_contains(cat_uuid: str, uid: str, sid: str) -> bool:
"and_or": "and",
"edge_type": "depletion",
"stoichiometry": 1.0,
+ **_origin(rxn_uuid, cat_uuid),
})
n_emitted_phos += 1
@@ -1470,8 +1688,12 @@ def _pool_contains(cat_uuid: str, uid: str, sid: str) -> bool:
"and_or": "and",
"edge_type": "depletion",
"stoichiometry": 1.0,
+ **_origin(ru, cat_uuid),
})
n_emitted_ub += 1
+ if env_flag("LNG_VARIANT_NODES"):
+ _pool_variant_depleters(pathway_logic_network_data, first_depletion,
+ reactome_id_to_uuid)
logger.info(
f"Emitted {n_emitted_phos + n_emitted_ub} substrate-depletion edges "
f"({len(phosphatase_rxn_uuids)} phosphatase reactions, "
@@ -1492,8 +1714,15 @@ def _node_leaves(node_id: str) -> frozenset:
"""
if node_id in _handoff_leaf_cache:
return _handoff_leaf_cache[node_id]
+ if node_id.endswith("::pool"):
+ _handoff_leaf_cache[node_id] = _node_leaves(node_id[:-len("::pool")])
+ return _handoff_leaf_cache[node_id]
if "::variant::" in node_id:
- s = {m for m in node_id.split("::variant::")[-1].split("_") if m.startswith("R-")}
+ from src.variant_keys import variant_parts, variant_leaves
+ if "=" in node_id: # specs/046 key: members may be complexes
+ s = {m for m in variant_leaves(node_id) if m.startswith("R-")}
+ else:
+ s = {m for m in variant_parts(node_id)[1] if m.startswith("R-")}
else:
try:
from src.neo4j_connector import get_labels
@@ -1830,6 +2059,7 @@ def _emit_boundary_decomposition_edges_inner(
how affected each member is *at that location* (read the sink; aggregate
across a member's sinks for an overall figure) without any cross-talk.
"""
+ variant_nodes = env_flag("LNG_VARIANT_NODES")
from src.neo4j_connector import get_labels
# Positional roots / terminals from the current edge list, IGNORING
@@ -1926,13 +2156,44 @@ def _is_complex(entity_id: str) -> bool:
# Tolerate any other unknown stIds too (e.g., entities added by other
# synthetic emissions) — if the lookup fails, assume it's not a complex
# and skip decomposition rather than crashing.
+ if entity_id.endswith("::pool"):
+ return False # specs/046: a pool is fed by its variants, never decomposed
if "::variant::" in entity_id:
- return False
+ # specs/046 variant keys of complexes are decomposed through the
+ # members their copy chose; legacy variant ids are not.
+ from src.variant_keys import is_variant_key, variant_parts
+ if not is_variant_key(entity_id):
+ return False
+ entity_id = variant_parts(entity_id)[0]
+ if variant_nodes:
+ # Under variant nodes, catalyst and regulator members bring in
+ # entities the pathway's cache prefetch never covered, and the
+ # connector's primed caches report those as unlabelled leaves
+ # (PIP3: KIT complex R-HSA-205310 became a leaf, so KIT had no
+ # root). Read structure directly (specs/046).
+ from src.variant_keys import is_complex as _vk_is_complex
+ return _vk_is_complex(entity_id)
try:
return "Complex" in get_labels(entity_id)
except IndexError:
return False
+ def _components_of(entity_id: str):
+ from src.variant_keys import _components, is_variant_key, variant_components
+ if is_variant_key(entity_id):
+ return variant_components(entity_id)
+ if variant_nodes:
+ return _components(entity_id)
+ return get_complex_components(entity_id) or {}
+
+ def _leaves_of(entity_id: str):
+ from src.variant_keys import is_variant_key, terminal_components, variant_leaves
+ if is_variant_key(entity_id):
+ return variant_leaves(entity_id)
+ if variant_nodes:
+ return terminal_components(entity_id)
+ return get_terminal_components(entity_id)
+
# Boundary expansion emits SYNTHETIC causal edges from a structural fact:
# assembly (leaf -> root complex, so a subunit can be perturbed) and
# dissociation (terminal complex -> a freshly minted readout sink, so a
@@ -2005,7 +2266,7 @@ def _emit(src: str, dst: str, kind: str = "assembly") -> None:
def _decompose_hier(container_uuid: str, container_stid: str, root_uuid: str, depth: int) -> None:
nonlocal nested_built
- for comp in sorted(get_complex_components(container_stid) or {}):
+ for comp in sorted(_components_of(container_stid)):
existing = _existing_upstream(comp, root_uuid)
if existing is not None:
_emit(existing, container_uuid) # a species the network already has
@@ -2023,7 +2284,7 @@ def _decompose_hier(container_uuid: str, container_stid: str, root_uuid: str, de
elif _is_set(comp) and depth < 6:
_emit(_set_node(comp, root_uuid, depth), container_uuid)
else:
- for leaf in sorted(get_terminal_components(comp)):
+ for leaf in sorted(_leaves_of(comp)):
_emit(_existing_upstream(leaf, root_uuid) or _leaf_uuid(leaf, root_uuid), container_uuid)
# A SET component is "any one of its members" (LNG code review F7,
@@ -2042,7 +2303,12 @@ def _set_node(set_stid: str, root_uuid: str, depth: int) -> str:
set_uuid = nested_registry[key] = str(uuid.uuid4())
reactome_id_to_uuid[set_uuid] = set_stid
nested_built += 1
- for m in sorted(get_set_members(set_stid) or {}):
+ if variant_nodes:
+ from src.variant_keys import _members as _vk_members
+ _set_members = _vk_members(set_stid)
+ else:
+ _set_members = sorted(get_set_members(set_stid) or {})
+ for m in _set_members:
existing = _existing_upstream(m, root_uuid)
if existing is not None:
_emit(existing, set_uuid, "set_member")
@@ -2063,8 +2329,12 @@ def _set_node(set_stid: str, root_uuid: str, depth: int) -> str:
return set_uuid
def _is_set(entity_id: str) -> bool:
- if "::variant::" in entity_id or entity_id in modifier_isoform_set_ids():
+ if ("::variant::" in entity_id or entity_id.endswith("::pool")
+ or entity_id in modifier_isoform_set_ids()):
return False
+ if variant_nodes:
+ from src.variant_keys import is_set as _vk_is_set
+ return _vk_is_set(entity_id)
try:
labels = get_labels(entity_id)
except IndexError:
@@ -2088,7 +2358,7 @@ def _is_set(entity_id: str) -> bool:
if hierarchy:
_decompose_hier(complex_uuid, stid, complex_uuid, 0)
continue
- leaves = get_terminal_components(stid)
+ leaves = _leaves_of(stid)
if leaves == {str(stid)}: # nothing below the complex to expose
continue
for leaf in leaves:
@@ -2111,7 +2381,7 @@ def _is_set(entity_id: str) -> bool:
stid = reactome_id_to_uuid.get(complex_uuid) or ""
if not stid or not _is_complex(stid):
continue
- leaves = get_terminal_components(stid)
+ leaves = _leaves_of(stid)
if leaves == {str(stid)}:
continue
for leaf in leaves:
@@ -2151,6 +2421,8 @@ def append_regulators(
pathway_logic_network_data: List[Dict[str, Any]],
reactome_id_to_uuid: Dict[str, str],
entity_uuid_registry: Optional[Dict[tuple, str]] = None,
+ produced_by_eid: Optional[Dict[str, List[Tuple[str, str]]]] = None,
+ preceding_by_reaction: Optional[Dict[str, Set[str]]] = None,
) -> None:
"""Append regulatory relationships to the pathway network.
@@ -2177,6 +2449,23 @@ def append_regulators(
set_pool_uuids: Dict[str, str] = {} # set stId -> its pool node (LNG_SET_POOL)
+ def _existing(member_id: str, reaction_id: str) -> Optional[str]:
+ """The node a regulator reads. By default the first registry entry
+ for the entity, which is registration order and can be a copy nothing
+ produces. With ``produced_by_eid`` (LNG_VARIANT_NODES, specs/046): the
+ copy a curated preceding reaction produces, else any produced copy,
+ else the default. Traced on PTEN catalysing R-HSA-199456: the first
+ entry was an unfed copy that read 10x under a PTEN knockout."""
+ if produced_by_eid:
+ cands = produced_by_eid.get(member_id)
+ if cands:
+ pre = (preceding_by_reaction or {}).get(reaction_id, ())
+ for u, prx in cands:
+ if prx in pre:
+ return u
+ return cands[0][0]
+ return stid_to_existing_uuid.get(member_id)
+
regulator_configs = [
(catalyst_map, "pos", "catalyst"),
(negative_regulator_map, "neg", "regulator"),
@@ -2284,8 +2573,9 @@ def append_regulators(
set_pool_uuids[entity_id] = pool_uuid
reactome_id_to_uuid[pool_uuid] = entity_id
for member_id, member_stoich in terminal_members:
- if member_id in stid_to_existing_uuid:
- member_uuid = stid_to_existing_uuid[member_id]
+ found = _existing(member_id, str(row["reaction_id"]))
+ if found is not None:
+ member_uuid = found
else:
member_uuid = str(uuid.uuid4())
stid_to_existing_uuid[member_id] = member_uuid
@@ -2313,8 +2603,9 @@ def append_regulators(
continue
for member_id, member_stoich in terminal_members:
- if member_id in stid_to_existing_uuid:
- member_uuid = stid_to_existing_uuid[member_id]
+ found = _existing(member_id, str(row["reaction_id"]))
+ if found is not None:
+ member_uuid = found
else:
member_uuid = str(uuid.uuid4())
stid_to_existing_uuid[member_id] = member_uuid
@@ -2422,6 +2713,7 @@ def create_pathway_logic_network(
best_matches: Any,
diagram_bridge_pairs: Optional[Set[Tuple[str, str]]] = None,
diagram_set_member_pairs: Optional[Set[Tuple[str, str]]] = None,
+ pathway_reaction_ids: Optional[Sequence[str]] = None,
) -> PathwayResult:
"""Create a pathway logic network from decomposed UID mappings and reaction connections.
@@ -2513,11 +2805,39 @@ def create_pathway_logic_network(
_calculate_reaction_statistics(reaction_connections)
# Create mappings and connections
- reaction_id_map = create_reaction_id_map(decomposed_uid_mapping, best_matches)
+ # deltasignal specs/046: under LNG_VARIANT_NODES the reaction copies and
+ # their entities come from variant keys (src/variant_emission.py), with
+ # catalysts and regulators expanded per copy. Phases 1-3 below are shared.
+ variant_nodes = env_flag("LNG_VARIANT_NODES")
graph = get_graph()
- catalyst_map = get_catalysts_for_reaction(reaction_id_map, graph)
- negative_regulator_map = get_negative_regulators_for_reaction(reaction_id_map, graph)
- positive_regulator_map = get_positive_regulators_for_reaction(reaction_id_map, graph)
+ if variant_nodes:
+ from src.variant_emission import STATS as _variant_stats, build_variant_reactions
+ # The pathway's reactions are the ones the matching layer decomposed:
+ # the same set the canonical path builds copies from. The
+ # reaction_connections reaching this function is the reduced
+ # connectivity table, and building from it dropped every reaction with
+ # no preceding/following event (PIP3: 66 of 89 reactions, taking the
+ # RTK catalyst sets with them; specs/046 research, first catalog arm).
+ # Every reaction of the pathway: the full list the pathway generator
+ # fetched (pathway_reaction_ids). The connectivity table reaching this
+ # function omits reactions with no preceding/following event, and
+ # decomposed_uid_mapping's reactome_id holds entity ids; building from
+ # either dropped reactions (PIP3: 8 of 89, among them R-HSA-2316434
+ # with the RTK catalyst set). specs/046 research, first catalog arm.
+ if pathway_reaction_ids is None:
+ from src.reaction_generator import reaction_ids_from_connections
+ pathway_reaction_ids = reaction_ids_from_connections(reaction_connections)
+ _variant_rids = sorted({str(r) for r in pathway_reaction_ids if r and str(r) != "nan"})
+ (reaction_id_map, _variant_vr_entities, catalyst_map,
+ negative_regulator_map, positive_regulator_map) = build_variant_reactions(graph, _variant_rids)
+ logger.info(f"Variant nodes: {len(_variant_rids)} reactions given; "
+ f"{reaction_id_map['reactome_id'].nunique()} emitted; {len(reaction_id_map)} copies")
+ logger.info(f"Variant nodes: {dict(_variant_stats)}")
+ else:
+ reaction_id_map = create_reaction_id_map(decomposed_uid_mapping, best_matches)
+ catalyst_map = get_catalysts_for_reaction(reaction_id_map, graph)
+ negative_regulator_map = get_negative_regulators_for_reaction(reaction_id_map, graph)
+ positive_regulator_map = get_positive_regulators_for_reaction(reaction_id_map, graph)
# Print regulator statistics
_print_regulator_statistics(positive_regulator_map, negative_regulator_map, catalyst_map)
@@ -2531,7 +2851,8 @@ def create_pathway_logic_network(
logger.debug(f"Built UID index with {len(uid_index)} entries")
# Resolve VR entities and build reactome-to-VR map
- vr_entities = _resolve_vr_entities(reaction_id_map, uid_index)
+ vr_entities = (_variant_vr_entities if variant_nodes
+ else _resolve_vr_entities(reaction_id_map, uid_index))
reactome_to_vr = _build_reactome_to_vr_map(reaction_id_map)
logger.debug(f"Processing {len(vr_entities)} virtual reactions in 3 phases")
@@ -2766,6 +3087,25 @@ def create_pathway_logic_network(
from src.neo4j_connector import prefetch_entity_decomposition_data
prefetch_entity_decomposition_data(list(cat_reg_entity_ids))
+ # specs/046: under variant nodes a regulator reads the copy of its entity
+ # that a curated preceding reaction produces (see append_regulators).
+ produced_by_eid: Optional[Dict[str, List[Tuple[str, str]]]] = None
+ preceding_by_reaction: Optional[Dict[str, Set[str]]] = None
+ if variant_nodes:
+ produced_by_eid = {}
+ for (eid, vr_uid, role), node_uuid in entity_uuid_registry.items():
+ if role != "output":
+ continue
+ prx = vr_to_reaction.get(str(vr_uid), "")
+ lst = produced_by_eid.setdefault(str(eid), [])
+ if all(u != node_uuid or r != prx for u, r in lst):
+ lst.append((node_uuid, prx))
+ preceding_by_reaction = {}
+ for _, conn in reaction_connections.iterrows():
+ if pd.isna(conn["preceding_reaction_id"]) or pd.isna(conn["following_reaction_id"]):
+ continue
+ preceding_by_reaction.setdefault(str(conn["following_reaction_id"]), set()).add(
+ str(conn["preceding_reaction_id"]))
append_regulators(
catalyst_map,
negative_regulator_map,
@@ -2773,6 +3113,8 @@ def create_pathway_logic_network(
pathway_logic_network_data,
reactome_id_to_uuid,
entity_uuid_registry=entity_uuid_registry,
+ produced_by_eid=produced_by_eid,
+ preceding_by_reaction=preceding_by_reaction,
)
# Substrate-depletion edges: for each catalytic reaction, emit edges
@@ -2787,6 +3129,15 @@ def create_pathway_logic_network(
reactome_id_to_uuid=reactome_id_to_uuid,
catalyst_map=catalyst_map,
)
+ if variant_nodes:
+ _wire_variant_pool_refs(pathway_logic_network_data, reactome_id_to_uuid)
+ following_by_reaction: Dict[str, Set[str]] = {}
+ for _, conn in reaction_connections.iterrows():
+ if pd.notna(conn["preceding_reaction_id"]) and pd.notna(conn["following_reaction_id"]):
+ following_by_reaction.setdefault(str(conn["preceding_reaction_id"]), set()).add(
+ str(conn["following_reaction_id"]))
+ _wire_capped_outputs(pathway_logic_network_data, reactome_id_to_uuid,
+ following_by_reaction, vr_to_reaction)
# Boundary expansion: every root-input and terminal-output complex
# occurrence gets synthetic assembly / dissociation edges to its member
@@ -3179,12 +3530,23 @@ def export_nodes(pathway_logic_network: pd.DataFrame,
s = uuid_to_str.get(u)
if s is None:
pass
+ elif s.endswith("::pool"):
+ # specs/046 D5: the entity as a whole, fed by its variants
+ kind, diagram = "variant_pool", s[:-len("::pool")]
elif "::variant::" in s:
kind = "set_variant"
- diagram = s.split("::variant::")[0]
- members = [m for m in s.split("::variant::")[-1].split("_")
- if m.startswith("R-")]
- sets, chosen = _derive_sets_and_chosen(diagram, set(members))
+ from src.variant_keys import is_variant_key, parse_variant_key, variant_leaves, variant_parts
+ if is_variant_key(s):
+ # specs/046: member_leaves are the variant's terminal
+ # leaves (what a gene resolves through); the slots and the
+ # members chosen in them are read from the key itself.
+ diagram, choice = parse_variant_key(s)
+ members = sorted(m for m in variant_leaves(s) if m.startswith("R-"))
+ sets, chosen = sorted(choice), sorted(set(choice.values()))
+ else:
+ diagram, members = variant_parts(s)
+ members = [m for m in members if m.startswith("R-")]
+ sets, chosen = _derive_sets_and_chosen(diagram, set(members))
else:
diagram = s
inc = incoming_types.get(u, set())
@@ -3358,7 +3720,16 @@ def export_containment(reactome_id_to_uuid: Dict[str, str],
# A failed lookup raises: treating it as "contains only itself" put a
# wrong row in containment.csv, which the consumer's default model
# reads (code review 2026-10-02).
- leaves = get_terminal_components(stid)
+ if stid.endswith("::pool"):
+ # specs/046 D5: a pool stands for its whole entity
+ leaves = get_terminal_components(stid[:-len("::pool")])
+ elif "::variant::" in stid and "=" in stid:
+ # specs/046: a variant contains only what its copy CHOSE: its
+ # chosen members, their leaves, and nothing of the alternatives.
+ from src.variant_keys import variant_leaves, variant_parts
+ leaves = set(variant_parts(stid)[1]) | variant_leaves(stid)
+ else:
+ leaves = get_terminal_components(stid)
leaves_by_stid[stid] = {str(x) for x in leaves} | {stid}
# Keyed by STABLE ID, not uuid. Containment is a property of the entity, not
@@ -3409,7 +3780,13 @@ def export_containment_structure(reactome_id_to_uuid: Dict[str, str],
graph = get_graph()
rows = []
seen: Set[str] = set()
- frontier = sorted({str(v) for v in reactome_id_to_uuid.values() if v})
+ from src.variant_keys import variant_parts
+ roots: Set[str] = set()
+ for v in reactome_id_to_uuid.values():
+ if v:
+ parent, chosen = variant_parts(str(v))
+ roots |= {parent, *chosen}
+ frontier: List[str] = sorted(roots)
while frontier:
seen.update(frontier)
found = graph.run(
@@ -3536,11 +3913,10 @@ def export_drugs(pathway_logic_network: pd.DataFrame,
# (stid_to_uuid_mapping.csv), so it is what must be listed (review of PR
# #98). A variant is drug-derived if its parent is, or if a member it
# CHOSE is: a variant that picks the drug out of a mixed set is the drug.
+ from src.variant_keys import variant_parts
+
def parts(node_id: str):
- if "::variant::" not in node_id:
- return node_id, []
- parent, members = node_id.split("::variant::", 1)
- return parent, [m for m in members.split("_") if m]
+ return variant_parts(node_id)
wanted = set()
for n in node_ids:
@@ -3692,11 +4068,8 @@ def find_pools(pathway_logic_network: pd.DataFrame, reaction_id_map: pd.DataFram
copies.setdefault(r, []).append(u)
def parts(u: str) -> Tuple[str, List[str]]:
- s = ent.get(u, "")
- if "::variant::" not in s:
- return s, []
- parent, members = s.split("::variant::", 1)
- return parent, [m for m in members.split("_") if m]
+ from src.variant_keys import variant_parts
+ return variant_parts(ent.get(u, ""))
def pe_of(u: str) -> str:
return parts(u)[0]
@@ -4045,10 +4418,11 @@ def export_pools(pathway_id: str, pathway_logic_network: pd.DataFrame, reaction_
ent = _uuid_to_stable_id_map(pathway_logic_network, uuid_mapping)
wanted = {pe for _, _, pe, _ in participants}
for s in ent.values():
- if "::variant::" in s:
- parent, members = s.split("::variant::", 1)
+ if "::variant::" in s or s.endswith("::pool"):
+ from src.variant_keys import variant_parts
+ parent, members = variant_parts(s)
wanted.add(parent)
- wanted.update(m for m in members.split("_") if m)
+ wanted.update(members)
else:
wanted.add(s)
profiles = get_form_profiles(wanted)
@@ -4160,11 +4534,16 @@ def add(stable_id: str, node_uuid: str, relation: str, depth: int,
# 2. Entity nodes: what each node directly stands for.
stid_to_uuids: Dict[str, Set[str]] = {}
for node_uuid, node_str in uuid_to_str.items():
- if "::variant::" in node_str:
+ if node_str.endswith("::pool"):
+ # specs/046 D5: recorded, under a relation no readout or gene
+ # resolution reads, so resolution is unchanged by pooling
+ add(node_str[:-len("::pool")], node_uuid, "variant_pool", 0)
+ elif "::variant::" in node_str:
parent = node_str.split("::variant::")[0]
add(parent, node_uuid, "variant", 0)
stid_to_uuids.setdefault(parent, set()).add(node_uuid)
- for member in node_str.split("::variant::")[-1].split("_"):
+ from src.variant_keys import variant_parts
+ for member in variant_parts(node_str)[1]:
if member.startswith("R-"):
add(member, node_uuid, "set_member", 1)
stid_to_uuids.setdefault(member, set()).add(node_uuid)
@@ -4195,7 +4574,7 @@ def add(stable_id: str, node_uuid: str, relation: str, depth: int,
continue
edge_rid = vr_to_reaction.get(rxn_uuid)
edge_node_str = uuid_to_str.get(node_uuid)
- if not edge_rid or not edge_node_str:
+ if not edge_rid or not edge_node_str or edge_node_str.endswith("::pool"):
continue
base = edge_node_str.split("::variant::")[0]
add(base, node_uuid,
diff --git a/src/pathway_generator.py b/src/pathway_generator.py
index c0021f4..fea0c84 100755
--- a/src/pathway_generator.py
+++ b/src/pathway_generator.py
@@ -42,6 +42,7 @@
"LNG_SET_POOL",
"LNG_CAP_POOLS",
"LNG_POOL_ACTIVE_VIA",
+ "LNG_VARIANT_NODES",
"LNG_DIAGRAM_SET_MEMBER",
# Of these three, only LNG_EMIT_ONE_SIDED is a genuine cache gap: it is
# read in reaction_generator.decompose_by_reactions, so it changes
@@ -433,10 +434,13 @@ def generate_pathway_file(
logger.warning("Could not read diagram set-member links",
exc_info=True)
+ from src.reaction_generator import reaction_ids_from_connections
result = create_pathway_logic_network(
decomposed_uid_mapping, connectivity, best_matches,
diagram_bridge_pairs=diagram_bridge_pairs,
diagram_set_member_pairs=set_member_pairs,
+ # every reaction of the pathway, from the full fetched table
+ pathway_reaction_ids=reaction_ids_from_connections(reaction_connections),
)
# Save logic network (main output file users need)
diff --git a/src/variant_emission.py b/src/variant_emission.py
new file mode 100644
index 0000000..9e2081e
--- /dev/null
+++ b/src/variant_emission.py
@@ -0,0 +1,309 @@
+"""Virtual reactions from variant keys (deltasignal specs/046, LNG_VARIANT_NODES).
+
+Replaces, under the flag, the two structures the emitter is built from:
+
+- ``reaction_id_map``: one row per reaction copy (``uid``, ``reactome_id``);
+- ``vr_entities``: ``uid -> (input_ids, output_ids, input_stoich, output_stoich)``,
+ where the ids are now variant keys (src/variant_keys.py);
+
+and the catalyst / regulator maps, whose ``entity_id`` per copy is the
+participant's variant key under that copy's choice. Everything after that
+(Phase 1 sharing per (key, reaction, role), Phase 2 merging across connected
+reaction pairs, Phase 3 edges, boundary layer) is unchanged, which is how
+decision D1 (identity across connected pairs only) is kept.
+
+Copies (decision D2): one choice per set per reaction, over inputs, outputs,
+catalysts and regulators (D4: negative regulators too). An output slot no
+input-side participant holds is BOUND to an input-side slot whose members
+correspond one to one by isoform-level reference identity; a tie is broken by
+sorted rank and counted, never by list position (replaces finding F9). An
+output slot that binds to nothing fans out.
+
+Past the cap (more than LNG_MAX_VARIANTS copies) decision D6 applies, in order:
+1. a participant none of whose slots reaches an output is read as ONE pool of
+ its variants (``::pool``, wired by the generator, combined by OR:
+ decision D5), and the reaction is enumerated over the rest;
+2. if still over, every input-side participant with a slot outside the output
+ variants is pooled, giving one copy per output variant;
+3. if still over, ONE copy reads every slot-bearing input-side participant as
+ a pool and writes its outputs by plain stable id (counted).
+The first catalog arm used only a plain-id single copy; nothing produced those
+plain ids, so a capped reaction was cut from everything upstream (IFN alpha/beta
+"Expression of IFN-induced genes": 8 perturbations x 25 readouts).
+"""
+import uuid
+from collections import Counter, defaultdict
+from typing import Dict, List, Optional, Sequence, Set, Tuple
+
+import pandas as pd
+
+from src import variant_keys as vk
+
+# `participant` is the curated catalyst/regulator the row came from (a bare
+# set's copies name its MEMBERS in entity_id), read by depleter pooling.
+_CAT_REG_COLUMNS = ["reaction_id", "entity_id", "edge_type", "uuid", "reaction_uuid", "participant"]
+
+STATS: Counter = Counter()
+
+_PARTICIPANT_CYPHER = """
+UNWIND $rids AS rid
+MATCH (r:ReactionLikeEvent {stId: rid})
+OPTIONAL MATCH (r)-[i:input]->(ie:PhysicalEntity)
+WITH r, collect(DISTINCT [ie.stId, coalesce(i.stoichiometry, 1)]) AS ins
+OPTIONAL MATCH (r)-[o:output]->(oe:PhysicalEntity)
+WITH r, ins, collect(DISTINCT [oe.stId, coalesce(o.stoichiometry, 1)]) AS outs
+OPTIONAL MATCH (r)-[:catalystActivity]->(:CatalystActivity)-[:physicalEntity]->(c:PhysicalEntity)
+WITH r, ins, outs, collect(DISTINCT c.stId) AS cats
+OPTIONAL MATCH (r)-[:regulatedBy]->(:PositiveRegulation)-[:regulator]->(p:PhysicalEntity)
+WITH r, ins, outs, cats, collect(DISTINCT p.stId) AS pos
+OPTIONAL MATCH (r)-[:regulatedBy]->(:NegativeRegulation)-[:regulator]->(n:PhysicalEntity)
+RETURN r.stId AS rid, ins, outs, cats, pos, collect(DISTINCT n.stId) AS neg
+"""
+
+
+def fetch_participants(graph, reaction_ids: Sequence[str]) -> Dict[str, dict]:
+ """reaction stId -> {"in": {stId: stoich}, "out": {...}, "cat": [...],
+ "pos": [...], "neg": [...]}, everything sorted for determinism."""
+ out: Dict[str, dict] = {}
+ rids = sorted(set(str(r) for r in reaction_ids))
+ for k in range(0, len(rids), 500):
+ for row in graph.run(_PARTICIPANT_CYPHER, rids=rids[k:k + 500]).data():
+ def stoich(pairs):
+ d: Dict[str, int] = {}
+ for s, n in pairs:
+ if s:
+ d[str(s)] = d.get(str(s), 0) + int(n or 1)
+ return dict(sorted(d.items()))
+ out[row["rid"]] = {
+ "in": stoich(row["ins"]), "out": stoich(row["outs"]),
+ "cat": sorted(x for x in row["cats"] if x),
+ "pos": sorted(x for x in row["pos"] if x),
+ "neg": sorted(x for x in row["neg"] if x),
+ }
+ return out
+
+
+def _slots_of(participants: Sequence[str]) -> Set[str]:
+ out: Set[str] = set()
+ for p in participants:
+ if vk.is_set(p):
+ out.add(p)
+ elif vk.is_complex(p) and not vk.is_capped(p):
+ out |= set(vk.own_slots(p))
+ return out
+
+
+_sig_cache: Dict[str, Tuple[str, ...]] = {}
+
+
+def _reference_signature(member: str) -> Tuple[str, ...]:
+ """Isoform-level identity of a set member: the sorted multiset of its
+ leaves' reference entities (a member complex keeps its leaf multiset)."""
+ if member in _sig_cache:
+ return _sig_cache[member]
+ from src.neo4j_connector import get_reference_entity_id
+ from src.reaction_generator import get_terminal_components
+ leaves = sorted(get_terminal_components(member)) if vk.is_complex(member) else [member]
+ sig = tuple(sorted(str(get_reference_entity_id(x) or x) for x in leaves))
+ _sig_cache[member] = sig
+ return sig
+
+
+def bind_output_slots(input_side: Set[str], output_slots: Set[str]) -> Dict[str, Tuple[str, Dict[str, str]]]:
+ """Output-only slot -> (input-side slot, {input member: output member}).
+
+ A binding exists when every member of the output slot has a member of the
+ input slot with the same reference signature, one to one. A signature
+ shared by several members is paired by sorted rank (counted). The first
+ input-side slot, in sorted order, that binds is used."""
+ bound: Dict[str, Tuple[str, Dict[str, str]]] = {}
+ free_in = sorted(s for s in input_side if s not in output_slots)
+ for w in sorted(output_slots - input_side):
+ wm = vk.flat_members(w)
+ for u in free_in:
+ um = vk.flat_members(u)
+ if len(um) != len(wm):
+ continue
+ by_sig_u, by_sig_w = defaultdict(list), defaultdict(list)
+ for m in um:
+ by_sig_u[_reference_signature(m)].append(m)
+ for m in wm:
+ by_sig_w[_reference_signature(m)].append(m)
+ if sorted((k, len(v)) for k, v in by_sig_u.items()) != \
+ sorted((k, len(v)) for k, v in by_sig_w.items()):
+ continue
+ mapping: Dict[str, str] = {}
+ for k, ms in by_sig_u.items():
+ if len(ms) > 1:
+ STATS["slot_binding_ties"] += 1
+ for a, b in zip(sorted(ms), sorted(by_sig_w[k])):
+ mapping[a] = b
+ bound[w] = (u, mapping)
+ STATS["output_slots_bound"] += 1
+ break
+ else:
+ STATS["output_slots_free"] += 1
+ return bound
+
+
+def _choices(parts: dict, limit: int) -> Tuple[List[Dict[str, str]], bool]:
+ """All copies of one reaction (choices), and whether the cap was hit."""
+ input_side = list(parts["in"]) + parts["cat"] + parts["pos"] + parts["neg"]
+ outputs = list(parts["out"])
+ in_slots, out_slots = _slots_of(input_side), _slots_of(outputs)
+ bound = bind_output_slots(in_slots, out_slots)
+ # outputs whose slots are all bound contribute no free variable
+ free_outputs = [o for o in outputs if not (_slots_of([o]) and _slots_of([o]) <= set(bound))]
+ base = list(vk.reaction_choices(input_side + free_outputs, limit=limit + 1))
+ result: List[Dict[str, str]] = []
+ for sig in base:
+ sig = dict(sig)
+ for w, (u, mapping) in bound.items():
+ if u in sig and sig[u] in mapping:
+ sig[w] = mapping[sig[u]]
+ # a bound member complex may open slots of its own: fan them out
+ pending = [sig]
+ while pending:
+ cur = pending.pop()
+ extra = _open_slots(outputs, cur)
+ if not extra:
+ result.append(cur)
+ continue
+ for more in vk.reaction_choices(extra, limit=limit + 1):
+ pending.append({**cur, **more})
+ if len(result) + len(pending) > limit:
+ return result + pending, True
+ if len(result) > limit:
+ return result, True
+ return result, len(result) > limit
+
+
+def _missing_slots(e: str, sig: Dict[str, str], depth: int = 0) -> Set[str]:
+ """Slots `e` needs under `sig` that `sig` does not assign, followed
+ through chosen members recursively (a bound set's chosen member complex
+ can open slots of its own)."""
+ if depth > 12:
+ return set()
+ if vk.is_set(e):
+ if e not in sig:
+ return {e}
+ return _missing_slots(sig[e], sig, depth + 1)
+ if vk.is_complex(e) and not vk.is_capped(e):
+ return {x for x in vk._complex_slots(e, sig) if x not in sig}
+ return set()
+
+
+def _open_slots(entities: Sequence[str], sig: Dict[str, str]) -> List[str]:
+ """Sets still unassigned in `entities` under `sig` (to be fanned out)."""
+ out: Set[str] = set()
+ for e in entities:
+ out |= _missing_slots(e, sig)
+ return sorted(out)
+
+
+POOL_SUFFIX = vk.POOL_SUFFIX
+
+
+def deep_slots(p: str) -> Set[str]:
+ """Every slot a participant can open, through chosen member complexes."""
+ out: Set[str] = set()
+ todo = list(_slots_of([p]))
+ while todo:
+ x = todo.pop()
+ if x in out:
+ continue
+ out.add(x)
+ for m in vk.flat_members(x):
+ if vk.is_complex(m) and not vk.is_capped(m):
+ todo += vk.own_slots(m)
+ return out
+
+
+def _without(parts: dict, pooled: Set[str]) -> dict:
+ return {"in": {e: n for e, n in parts["in"].items() if e not in pooled},
+ "out": parts["out"],
+ "cat": [e for e in parts["cat"] if e not in pooled],
+ "pos": [e for e in parts["pos"] if e not in pooled],
+ "neg": [e for e in parts["neg"] if e not in pooled]}
+
+
+def capped_fallback(parts: dict, limit: int) -> Tuple[List[Optional[Dict[str, str]]], Set[str], int]:
+ """(choices, pooled participants, D6 step) for a reaction over the cap."""
+ inside = list(parts["in"]) + parts["cat"] + parts["pos"] + parts["neg"]
+ outputs = list(parts["out"])
+ out_slots: Set[str] = set()
+ for o in outputs:
+ out_slots |= deep_slots(o)
+ bound = bind_output_slots(_slots_of(inside), _slots_of(outputs))
+ reach = out_slots | {u for (u, _) in bound.values()}
+ slots = {x: deep_slots(x) for x in inside}
+ # Step 1 pools only participants that share no slot with a kept one:
+ # pooling A while a kept B fixes A's slot S would let each copy read every
+ # S variant of A (decision D2, one choice per set per reaction; review of
+ # vn6). Shrink to a fixed point.
+ step1 = {x for x in inside if slots[x] and not (slots[x] & reach)}
+ while True:
+ kept_slots = set().union(*[slots[x] for x in inside if x not in step1]) if inside else set()
+ shrink = {x for x in step1 if slots[x] & kept_slots}
+ if not shrink:
+ break
+ step1 -= shrink
+ step2 = {x for x in inside if slots[x] and not (slots[x] <= reach)}
+ for step, pooled in ((1, step1), (2, step2)):
+ choices, over = _choices(_without(parts, pooled), limit)
+ if not over and choices:
+ return list(choices), set(pooled), step
+ return [None], {x for x in inside if slots[x]}, 3
+
+
+def build_variant_reactions(graph, reaction_ids: Sequence[str]
+ ) -> Tuple[pd.DataFrame, Dict[str, tuple], pd.DataFrame, pd.DataFrame, pd.DataFrame]:
+ """(reaction_id_map, vr_entities, catalyst_map, negative_regulator_map,
+ positive_regulator_map) for LNG_VARIANT_NODES=1."""
+ STATS.clear()
+ cap = vk._max_variants()
+ parts_by_rx = fetch_participants(graph, reaction_ids)
+ rows: List[dict] = []
+ vr_entities: Dict[str, tuple] = {}
+ cat_rows: List[dict] = []; neg_rows: List[dict] = []; pos_rows: List[dict] = []
+ for rx in sorted(parts_by_rx):
+ parts = parts_by_rx[rx]
+ found, over = _choices(parts, cap if cap > 0 else 10 ** 9)
+ choices: List[Optional[Dict[str, str]]] = list(found)
+ if not found and not over:
+ # Never drop a reaction silently: zero copies means a set with no
+ # members reached the enumeration, i.e. the structure is wrong.
+ raise ValueError(f"{rx}: the variant enumeration produced no copy")
+ pooled: Set[str] = set()
+ if over:
+ STATS["reactions_over_cap"] += 1
+ choices, pooled, step = capped_fallback(parts, cap)
+ STATS[f"capped_step{step}"] += 1
+ STATS["pooled_participants"] += len(pooled)
+ STATS["copies"] += len(choices)
+ for sig in choices:
+ uid = str(uuid.uuid4())
+
+ def name(e, s=sig, pooled=pooled, output=False):
+ # pooling applies to the input side only; an output is written
+ # by its key (or plain stId in step 3)
+ if e in pooled and not output:
+ return e + POOL_SUFFIX
+ return e if s is None else vk.vkey(e, s)
+ ins: Counter = Counter()
+ outs: Counter = Counter()
+ for e, n in parts["in"].items():
+ ins[name(e)] += n
+ for e, n in parts["out"].items():
+ outs[name(e, output=True)] += n
+ rows.append({"uid": uid, "reactome_id": rx, "input_hash": None, "output_hash": None})
+ vr_entities[uid] = (sorted(ins), sorted(outs), dict(ins), dict(outs))
+ for lst, target in ((parts["cat"], cat_rows), (parts["neg"], neg_rows), (parts["pos"], pos_rows)):
+ for e in lst:
+ target.append({"reaction_id": rx, "entity_id": name(e),
+ "edge_type": "catalyst" if target is cat_rows else "regulator",
+ "uuid": str(uuid.uuid4()), "reaction_uuid": uid,
+ "participant": e})
+ rid_map = pd.DataFrame(rows, columns=["uid", "reactome_id", "input_hash", "output_hash"])
+ mk = lambda r: pd.DataFrame(r, columns=_CAT_REG_COLUMNS) # noqa: E731
+ return rid_map, vr_entities, mk(cat_rows), mk(neg_rows), mk(pos_rows)
diff --git a/src/variant_keys.py b/src/variant_keys.py
new file mode 100644
index 0000000..b03b6b7
--- /dev/null
+++ b/src/variant_keys.py
@@ -0,0 +1,383 @@
+"""Variant keys: one canonical name per variant of an entity (deltasignal specs/046).
+
+A *choice* sigma maps slots to chosen members. A slot is a set occurrence: the
+OUTERMOST set on a hasComponent path from a participant, named by its stId.
+Its chosen member is always a non-set entity (nested sets are flattened, as
+break_apart_entity does). A chosen member that is itself a complex containing
+sets contributes its own slots, so choices are made recursively.
+
+ VK(e, sigma)
+ leaf, modifier-isoform set, set-free complex, capped complex -> e
+ bare set S -> VK(sigma[S], sigma)
+ complex C otherwise -> C + "::variant::" + "_".join(sorted(
+ f"{slot}={VK(sigma[slot], sigma)}" ...))
+
+Member complexes keep their identity (their stId, not their leaves), which is
+what lets a producer that outputs an INNER set and a consumer that inputs an
+OUTER set name the same node (the F10 fix). Keys are built from sorted
+strings only: no uuid, dict or hash-seed order reaches them.
+
+Design: specs/046 spec.md "Merged design" (Adam's decisions D1-D6) and
+derivation-fable.md section 1; decision D2 is "one choice per set per
+reaction", which `reaction_choices` implements by sharing slots across all of
+a reaction's participants.
+"""
+from itertools import product as _product # noqa: F401 (kept for callers)
+from typing import Any, Dict, Iterator, List, Optional, Sequence, Set, Tuple
+
+SET_LABELS = ("EntitySet", "DefinedSet", "CandidateSet", "OpenSet")
+VARIANT_SEP = "::variant::"
+
+# Injected lookups, so the module is testable without a database. The
+# defaults are the generator's cached Neo4j accessors.
+_lookups: Dict[str, Any] = {}
+# Per-process memo of structure lookups (one Neo4j round trip per entity).
+_memo: Dict[Tuple[str, str], List[str]] = {}
+
+
+def reset_caches() -> None:
+ """Forget memoised structure (tests, or a new database)."""
+ _memo.clear()
+ _variants_cache.clear()
+ _struct.clear()
+
+
+def _labels(s: str) -> List[str]:
+ if ("labels", s) in _memo:
+ return _memo[("labels", s)]
+ _memo[("labels", s)] = r = _labels_uncached(s)
+ return r
+
+
+# Structure is read from Neo4j DIRECTLY, not through the connector's entity
+# caches: once primed for a pathway, those answer "no members" for an entity
+# outside the prefetch, which gave a reaction zero copies and dropped it
+# silently (PIP3: 8 of 89 reactions; specs/046 first catalog arm).
+_STRUCT_CYPHER = """
+MATCH (e:DatabaseObject {stId: $s})
+OPTIONAL MATCH (e)-[:hasComponent]->(c)
+WITH e, collect(DISTINCT c.stId) AS comps
+OPTIONAL MATCH (e)-[:hasMember|hasCandidate]->(m)
+RETURN labels(e) AS labels, comps, collect(DISTINCT m.stId) AS members
+"""
+_struct: Dict[str, Tuple[List[str], List[str], List[str]]] = {}
+
+
+def _structure(s: str) -> Tuple[List[str], List[str], List[str]]:
+ if s not in _struct:
+ from src.neo4j_connector import get_graph
+ rows = get_graph().run(_STRUCT_CYPHER, s=s).data()
+ if not rows:
+ _struct[s] = ([], [], [])
+ else:
+ r = rows[0]
+ _struct[s] = (list(r["labels"] or []), sorted(x for x in r["comps"] if x),
+ sorted(x for x in r["members"] if x))
+ return _struct[s]
+
+
+def _labels_uncached(s: str) -> List[str]:
+ f = _lookups.get("labels")
+ if f is not None:
+ try:
+ return list(f(s) or [])
+ except IndexError:
+ return []
+ return _structure(s)[0]
+
+
+def _components(s: str) -> List[str]:
+ if ("components", s) in _memo:
+ return _memo[("components", s)]
+ _memo[("components", s)] = r = _components_uncached(s)
+ return r
+
+
+def _components_uncached(s: str) -> List[str]:
+ f = _lookups.get("components")
+ if f is not None:
+ return sorted(str(x) for x in (f(s) or {}))
+ return _structure(s)[1]
+
+
+def _members(s: str) -> List[str]:
+ if ("members", s) in _memo:
+ return _memo[("members", s)]
+ _memo[("members", s)] = r = _members_uncached(s)
+ return r
+
+
+def _members_uncached(s: str) -> List[str]:
+ f = _lookups.get("members")
+ if f is not None:
+ return sorted(str(x) for x in (f(s) or {}))
+ return _structure(s)[2]
+
+
+def _atomic_sets() -> Set[str]:
+ f = _lookups.get("atomic_sets")
+ if f is not None:
+ return set(f())
+ from src.reaction_generator import modifier_isoform_set_ids
+ return set(modifier_isoform_set_ids())
+
+
+def _max_variants() -> int:
+ v: Any = _lookups.get("max_variants")
+ if v is None:
+ from src.reaction_generator import MAX_VARIANTS
+ v = MAX_VARIANTS
+ return int(v() if callable(v) else v)
+
+
+def is_set(s: str) -> bool:
+ """A set with members (an empty curated set is a leaf, not a slot)."""
+ return (any(t in _labels(s) for t in SET_LABELS) and s not in _atomic_sets()
+ and bool(_members(s)))
+
+
+def is_complex(s: str) -> bool:
+ return "Complex" in _labels(s)
+
+
+def flat_members(s: str, _depth: int = 0) -> List[str]:
+ """A set's alternatives with nested sets flattened: non-set entities only."""
+ out: Set[str] = set()
+ if _depth > 12:
+ return []
+ for m in _members(s):
+ if is_set(m):
+ out |= set(flat_members(m, _depth + 1))
+ else:
+ out.add(m)
+ return sorted(out)
+
+
+def own_slots(e: str, _depth: int = 0) -> List[str]:
+ """The set occurrences directly reachable from `e` (a complex or a bare
+ set) without passing through another set: the slots `e` itself opens.
+ A bare set is its own (single) slot. Sorted."""
+ if _depth > 12:
+ return []
+ if is_set(e):
+ return [e]
+ if not is_complex(e):
+ return []
+ out: Set[str] = set()
+ for c in _components(e):
+ out |= set(own_slots(c, _depth + 1))
+ return sorted(out)
+
+
+_variants_cache: Dict[str, int] = {}
+
+
+def variant_count(e: str, _depth: int = 0) -> int:
+ """How many variants `e` has (product over its slots of the summed
+ variants of each slot's members). Used to decide whether `e` is capped,
+ identically wherever it appears."""
+ if e in _variants_cache:
+ return _variants_cache[e]
+ n = 1
+ if _depth <= 12:
+ if is_set(e):
+ n = sum(variant_count(m, _depth + 1) for m in flat_members(e)) or 1
+ elif is_complex(e):
+ for s in own_slots(e):
+ n *= variant_count(s, _depth + 1)
+ _variants_cache[e] = n
+ return n
+
+
+def is_capped(e: str) -> bool:
+ mv = _max_variants()
+ return mv > 0 and is_complex(e) and variant_count(e) > mv
+
+
+def _complex_slots(e: str, sigma: Dict[str, str], _depth: int = 0) -> List[str]:
+ """Every slot of complex `e` under choice `sigma`, the chosen member
+ complexes' own slots hoisted in. Sorted, unique."""
+ out: Set[str] = set()
+ if _depth > 12:
+ return []
+ for s in own_slots(e):
+ out.add(s)
+ m = sigma.get(s)
+ if m is not None and is_complex(m) and not is_capped(m):
+ out |= set(_complex_slots(m, sigma, _depth + 1))
+ return sorted(out)
+
+
+def vkey(e: str, sigma: Dict[str, str]) -> str:
+ """The variant key of `e` under choice `sigma` (every slot `e` opens,
+ recursively, must be assigned)."""
+ e = str(e)
+ if e in _atomic_sets():
+ return e
+ if is_set(e):
+ if e not in sigma:
+ raise KeyError(f"slot {e} has no chosen member")
+ return vkey(sigma[e], sigma)
+ if not is_complex(e) or is_capped(e):
+ return e
+ slots = _complex_slots(e, sigma)
+ if not slots:
+ return e
+ missing = [s for s in slots if s not in sigma]
+ if missing:
+ raise KeyError(f"{e}: slots {missing} have no chosen member")
+ tokens = []
+ for s in slots:
+ m = sigma[s]
+ # The member complex keeps its identity; its own slots are already
+ # tokens of this key (hoisted), so name it by its plain stId here.
+ tokens.append(f"{s}={m}")
+ return e + VARIANT_SEP + "_".join(sorted(tokens))
+
+
+def parse_variant_key(key: str) -> Tuple[str, Dict[str, str]]:
+ """(parent stId, {slot: chosen member}) — the one parser every consumer
+ of variant ids should use."""
+ if VARIANT_SEP not in key:
+ return key, {}
+ parent, tail = key.split(VARIANT_SEP, 1)
+ choice: Dict[str, str] = {}
+ for tok in tail.split("_"):
+ if not tok:
+ continue
+ if "=" not in tok:
+ raise ValueError(f"malformed variant token {tok!r} in {key!r}")
+ s, m = tok.split("=", 1)
+ choice[s] = m
+ return parent, choice
+
+
+def variant_leaves(key: str, _depth: int = 0) -> Set[str]:
+ """The leaf entities a variant is made of: the parent's structure with
+ every slot replaced by its chosen member."""
+ parent, sigma = parse_variant_key(key)
+ return _leaves(parent, sigma, _depth)
+
+
+def _leaves(e: str, sigma: Dict[str, str], _depth: int) -> Set[str]:
+ if _depth > 12:
+ return {e}
+ if e in _atomic_sets():
+ return {e}
+ if is_set(e):
+ m = sigma.get(e)
+ return _leaves(m, sigma, _depth + 1) if m else {e}
+ if is_complex(e) and not is_capped(e):
+ out: Set[str] = set()
+ for c in _components(e):
+ out |= _leaves(c, sigma, _depth + 1)
+ return out or {e}
+ return {e}
+
+
+def reaction_choices(participants: Sequence[str],
+ limit: Optional[int] = None) -> Iterator[Dict[str, str]]:
+ """Every choice for a reaction whose (non-negative-regulator) participants
+ are `participants`, with ONE variable per slot shared across all of them
+ (decision D2). A chosen member complex opens its own slots. Yields
+ choices in a deterministic order; stops after `limit` (the caller treats
+ reaching it as "over the cap"). Capped complexes open no slots."""
+ roots: Set[str] = set()
+ for p in participants:
+ if is_set(p):
+ roots.add(p)
+ elif is_complex(p) and not is_capped(p):
+ roots |= set(own_slots(p))
+ count = [0]
+
+ def rec(sigma: Dict[str, str], open_slots: List[str]) -> Iterator[Dict[str, str]]:
+ pending = sorted(s for s in open_slots if s not in sigma)
+ if not pending:
+ count[0] += 1
+ yield dict(sigma)
+ return
+ s = pending[0]
+ for m in flat_members(s):
+ if limit is not None and count[0] >= limit:
+ return
+ sigma[s] = m
+ new = set(open_slots)
+ if is_complex(m) and not is_capped(m):
+ new |= set(own_slots(m))
+ yield from rec(sigma, sorted(new))
+ del sigma[s]
+
+ yield from rec({}, sorted(roots))
+
+
+POOL_SUFFIX = "::pool"
+
+
+def is_pool_id(node_id: str) -> bool:
+ """A pool of an entity's variants (``::pool``, decision D5)."""
+ return node_id.endswith(POOL_SUFFIX)
+
+
+def pool_base(node_id: str) -> str:
+ """The entity a pool id stands for; any other id unchanged."""
+ return node_id[:-len(POOL_SUFFIX)] if is_pool_id(node_id) else node_id
+
+
+def variant_parts(node_id: str) -> Tuple[str, List[str]]:
+ """(parent stId, chosen members) of a node id in EITHER format: the
+ specs/046 key (``parent::variant::slot=member_...``) or the legacy
+ ``parent::variant::m1_m2`` id. A plain stId gives (stId, []). Every
+ consumer that needs a variant's members goes through this, so the two
+ formats cannot be misread as each other. A pool id gives (its entity, []):
+ it stands for the entity as a whole, with no chosen member."""
+ node_id = pool_base(node_id)
+ if VARIANT_SEP not in node_id:
+ return node_id, []
+ parent, tail = node_id.split(VARIANT_SEP, 1)
+ toks = [t for t in tail.split("_") if t]
+ if any("=" in t for t in toks):
+ return parent, sorted({t.split("=", 1)[1] for t in toks})
+ return parent, toks
+
+
+def is_variant_key(node_id: str) -> bool:
+ """True for a specs/046 key (``parent::variant::slot=member...``)."""
+ return VARIANT_SEP in node_id and "=" in node_id.split(VARIANT_SEP, 1)[1]
+
+
+def variant_components(key: str) -> List[str]:
+ """The components of a variant, one level down: the parent's components
+ with each set slot replaced by the key of its chosen member, and each
+ component complex that holds sets by its own variant key. Used by the
+ boundary layer to decompose a root variant through what its copy chose."""
+ parent, sigma = parse_variant_key(key)
+ out: List[str] = []
+ for c in _components(parent):
+ if is_set(c):
+ m = sigma.get(c)
+ out.append(vkey(m, sigma) if m else c)
+ elif is_complex(c) and not is_capped(c) and own_slots(c):
+ out.append(vkey(c, sigma))
+ else:
+ out.append(c)
+ return out
+
+
+def terminal_components(e: str, _depth: int = 0) -> Set[str]:
+ """Every leaf below `e` through components AND all set members (the
+ generator's get_terminal_components semantics: modifier-isoform sets are
+ atomic), read through this module's direct structure lookups, so it is
+ correct for entities outside a pathway's cache prefetch."""
+ if _depth > 12 or e in _atomic_sets():
+ return {e}
+ if is_set(e):
+ out: Set[str] = set()
+ for m in _members(e):
+ out |= terminal_components(m, _depth + 1)
+ return out or {e}
+ if is_complex(e):
+ out = set()
+ for c in _components(e):
+ out |= terminal_components(c, _depth + 1)
+ return out or {e}
+ return {e}
diff --git a/tests/test_boundary_hierarchy.py b/tests/test_boundary_hierarchy.py
index b2bb9f3..2ef8a38 100644
--- a/tests/test_boundary_hierarchy.py
+++ b/tests/test_boundary_hierarchy.py
@@ -11,6 +11,16 @@
import src.logic_network_generator as lng
from src.logic_network_generator import _emit_boundary_decomposition_edges
+
+@pytest.fixture(autouse=True)
+def _flag_off_boundary_layer(monkeypatch):
+ # These tests pin the FLAG-OFF boundary layer, whose structure comes from
+ # the mocked connector lookups. Under LNG_VARIANT_NODES (the default since
+ # deltasignal specs/046) the layer reads structure from Neo4j through
+ # src/variant_keys.py instead; that path has its own tests
+ # (tests/test_variant_*.py).
+ monkeypatch.setenv("LNG_VARIANT_NODES", "0")
+
K, N, I = "R-HSA-K", "R-HSA-N", "R-HSA-I" # K = ISGF3:KPNA1:KPNB1, N = ISGF3:KPNA1, I = ISGF3
A1, B1, S1, S2, IR = "R-HSA-KPNA1", "R-HSA-KPNB1", "R-HSA-STAT1", "R-HSA-STAT2", "R-HSA-IRF9"
LABELS = {K: ["Complex"], N: ["Complex"], I: ["Complex"],
diff --git a/tests/test_boundary_leaf_reuse.py b/tests/test_boundary_leaf_reuse.py
index 8e3cd37..ed6681a 100644
--- a/tests/test_boundary_leaf_reuse.py
+++ b/tests/test_boundary_leaf_reuse.py
@@ -10,6 +10,16 @@
import src.logic_network_generator as lng
from src.logic_network_generator import _emit_boundary_decomposition_edges
+
+@pytest.fixture(autouse=True)
+def _flag_off_boundary_layer(monkeypatch):
+ # These tests pin the FLAG-OFF boundary layer, whose structure comes from
+ # the mocked connector lookups. Under LNG_VARIANT_NODES (the default since
+ # deltasignal specs/046) the layer reads structure from Neo4j through
+ # src/variant_keys.py instead; that path has its own tests
+ # (tests/test_variant_*.py).
+ monkeypatch.setenv("LNG_VARIANT_NODES", "0")
+
C, P, Q = "R-HSA-100", "R-HSA-200", "R-HSA-300" # C: root complex of P and Q
LABELS = {C: ["Complex"], P: ["EntityWithAccessionedSequence"], Q: ["EntityWithAccessionedSequence"]}
diff --git a/tests/test_boundary_set_nodes.py b/tests/test_boundary_set_nodes.py
index 7bf3195..b00fb30 100644
--- a/tests/test_boundary_set_nodes.py
+++ b/tests/test_boundary_set_nodes.py
@@ -8,6 +8,16 @@
import src.neo4j_connector as nc
import src.logic_network_generator as lng
import src.reaction_generator as rg
+
+
+@pytest.fixture(autouse=True)
+def _flag_off_boundary_layer(monkeypatch):
+ # These tests pin the FLAG-OFF boundary layer, whose structure comes from
+ # the mocked connector lookups. Under LNG_VARIANT_NODES (the default since
+ # deltasignal specs/046) the layer reads structure from Neo4j through
+ # src/variant_keys.py instead; that path has its own tests
+ # (tests/test_variant_*.py).
+ monkeypatch.setenv("LNG_VARIANT_NODES", "0")
from src.logic_network_generator import _emit_boundary_decomposition_edges
CX, GDP, SET, RHOA, RAC1, CDC42 = "R-HSA-CX", "R-ALL-GDP", "R-HSA-SET", "R-HSA-RHOA", "R-HSA-RAC1", "R-HSA-CDC42"
diff --git a/tests/test_logic_network_generator.py b/tests/test_logic_network_generator.py
index aeb91e7..d3166be 100644
--- a/tests/test_logic_network_generator.py
+++ b/tests/test_logic_network_generator.py
@@ -2,6 +2,8 @@
from typing import Dict, List, Any
import os
+
+import pytest
import sys
from pathlib import Path
from unittest.mock import patch
@@ -408,6 +410,13 @@ def test_non_boundary_entity_gets_separate_uuids(self):
# tests/test_boundary_hierarchy.py and tests/test_boundary_leaf_reuse.py.
@patch.dict(os.environ, {"LNG_BOUNDARY_HIERARCHY": "0"})
class TestBoundaryLeavesReuseExistingUUIDs:
+
+ @pytest.fixture(autouse=True)
+ def _flag_off_boundary_layer(self, monkeypatch):
+ # Pins the FLAG-OFF boundary layer (mocked connector lookups); under
+ # LNG_VARIANT_NODES (default since deltasignal specs/046) the layer reads
+ # Neo4j through src/variant_keys.py, tested in tests/test_variant_*.py.
+ monkeypatch.setenv("LNG_VARIANT_NODES", "0")
"""Boundary expansion must NOT mint a fresh UUID for a leaf if that
leaf's stId already has a UUID elsewhere in the network. Otherwise
perturbing 'MDM2 the regulator' wouldn't propagate to 'MDM2 the
diff --git a/tests/test_variant_emission.py b/tests/test_variant_emission.py
new file mode 100644
index 0000000..d300766
--- /dev/null
+++ b/tests/test_variant_emission.py
@@ -0,0 +1,155 @@
+"""specs/046 variant emitter: copies and output binding (no database)."""
+import pytest
+
+import src.variant_emission as ve
+import src.variant_keys as vk
+
+LABELS = {"CX": ["Complex"], "SET": ["EntitySet"], "OSET": ["EntitySet"], "GDP": ["SimpleEntity"],
+ "DIMER": ["Complex"], "CAT": ["Complex"], "TIE": ["EntitySet"], "OTIE": ["EntitySet"]}
+COMP = {"CX": ["GDP", "SET"], "DIMER": ["SET", "SET"], "CAT": ["SET", "GDP"]}
+MEM = {"SET": ["RHOA", "RAC1"], "OSET": ["pRAC1", "pRHOA"], "TIE": ["A1", "A2"], "OTIE": ["B1", "B2"]}
+SIG = {"RHOA": ("RHOA",), "pRHOA": ("RHOA",), "RAC1": ("RAC1",), "pRAC1": ("RAC1",),
+ "A1": ("A",), "A2": ("A",), "B1": ("A",), "B2": ("A",)}
+
+
+@pytest.fixture(autouse=True)
+def stub(monkeypatch):
+ vk._lookups.update(labels=lambda s: LABELS.get(s, ["EWAS"]), components=lambda s: COMP.get(s, []),
+ members=lambda s: MEM.get(s, []), atomic_sets=lambda: set(), max_variants=lambda: 512)
+ vk.reset_caches(); ve.STATS.clear(); ve._sig_cache.clear()
+ monkeypatch.setattr(ve, "_reference_signature", lambda m: SIG.get(m, (m,)))
+ yield
+ vk._lookups.clear(); vk.reset_caches()
+
+
+def parts(**kw):
+ p = {"in": {}, "out": {}, "cat": [], "pos": [], "neg": []}
+ p.update(kw); return p
+
+
+def test_catalyst_equal_to_input_is_one_choice_per_copy():
+ choices, over = ve._choices(parts(**{"in": {"CX": 1}, "cat": ["CX"]}), 512)
+ assert not over and len(choices) == 2
+ for sig in choices:
+ assert vk.vkey("CX", sig) == f"CX::variant::SET={sig['SET']}"
+
+
+def test_dimer_of_one_set_gives_homo_copies_only():
+ choices, _ = ve._choices(parts(**{"in": {"DIMER": 1}}), 512)
+ assert sorted(c["SET"] for c in choices) == ["RAC1", "RHOA"]
+
+
+def test_output_only_set_is_bound_by_identity_not_multiplied():
+ choices, _ = ve._choices(parts(**{"in": {"SET": 1}, "out": {"OSET": 1}}), 512)
+ assert len(choices) == 2
+ assert {(c["SET"], c["OSET"]) for c in choices} == {("RHOA", "pRHOA"), ("RAC1", "pRAC1")}
+ assert ve.STATS["output_slots_bound"] == 1
+
+
+def test_a_signature_tie_is_paired_by_rank_and_counted():
+ choices, _ = ve._choices(parts(**{"in": {"TIE": 1}, "out": {"OTIE": 1}}), 512)
+ assert {(c["TIE"], c["OTIE"]) for c in choices} == {("A1", "B1"), ("A2", "B2")}
+ assert ve.STATS["slot_binding_ties"] == 1
+
+
+def test_negative_regulator_shares_the_copy_choice():
+ choices, _ = ve._choices(parts(**{"in": {"CX": 1}, "neg": ["CAT"]}), 512)
+ assert len(choices) == 2 # D4: expanded, but the shared set is one variable
+
+
+def test_over_the_cap_is_reported():
+ choices, over = ve._choices(parts(**{"in": {"CX": 1}, "out": {"OSET": 1}}), 1)
+ assert over
+
+
+def test_a_bound_member_complex_opens_its_own_slots():
+ # output set bound to an input set; the mapped member is a complex that
+ # holds a set of its own, which must be fanned out (IL-3 R-HSA-879909)
+ LABELS.update({"ISET": ["EntitySet"], "OSET2": ["EntitySet"], "MCX": ["Complex"], "INNER": ["EntitySet"]})
+ MEM.update({"ISET": ["RHOA"], "OSET2": ["MCX"], "INNER": ["X1", "X2"]})
+ COMP["MCX"] = ["INNER"]
+ SIG["MCX"] = ("RHOA",)
+ choices, over = ve._choices(parts(**{"in": {"ISET": 1}, "out": {"OSET2": 1}}), 512)
+ assert not over and sorted(c["INNER"] for c in choices) == ["X1", "X2"]
+ for c in choices:
+ assert vk.vkey("OSET2", c).startswith("MCX::variant::INNER=")
+
+
+# --- D6 cap fallback (specs/046 research: the IFN alpha/beta cap seam) ---
+
+CAP_LABELS = {"REG": ["Complex"], "RS1": ["EntitySet"], "RS2": ["EntitySet"], "OUT": ["EntitySet"],
+ "BIG": ["EntitySet"], "IC": ["Complex"]}
+CAP_COMP = {"REG": ["RS1", "RS2"], "IC": ["GENE", "OUT"]}
+CAP_MEM = {"RS1": ["r1", "r2", "r3"], "RS2": ["s1", "s2", "s3"], "OUT": ["o1", "o2", "o3", "o4"],
+ "BIG": [f"b{i}" for i in range(12)]}
+
+
+@pytest.fixture
+def cap_world():
+ vk._lookups.update(labels=lambda s: CAP_LABELS.get(s, ["EWAS"]),
+ components=lambda s: CAP_COMP.get(s, []),
+ members=lambda s: CAP_MEM.get(s, []))
+ vk.reset_caches()
+
+
+def test_step1_pools_a_regulator_whose_choice_reaches_no_output(cap_world):
+ # 9 regulator variants x 4 free output members = 36 copies, over a cap of 10.
+ p = parts(**{"in": {"GENE": 1}, "out": {"OUT": 1}, "pos": ["REG"]})
+ assert ve._choices(p, 10)[1]
+ choices, pooled, step = ve.capped_fallback(p, 10)
+ assert step == 1 and pooled == {"REG"}
+ assert sorted(c["OUT"] for c in choices) == ["o1", "o2", "o3", "o4"]
+
+
+def test_step1_keeps_a_participant_holding_an_output_slot(cap_world):
+ # IC holds OUT's slot (reaches the output) and is kept; REG is pooled.
+ p = parts(**{"in": {"IC": 1}, "out": {"OUT": 1}, "neg": ["REG"]})
+ choices, pooled, step = ve.capped_fallback(p, 5)
+ assert step == 1 and pooled == {"REG"} and len(choices) == 4
+ assert all(vk.vkey("IC", c) == f"IC::variant::OUT={c['OUT']}" for c in choices)
+
+
+def test_step2_gives_one_copy_per_output_variant(cap_world):
+ # MIX holds OUT's slot AND RS1: kept by step 1 (4 x 3 = 12 > 5), pooled by step 2.
+ CAP_LABELS["MIX"] = ["Complex"]; CAP_COMP["MIX"] = ["OUT", "RS1"]
+ try:
+ vk.reset_caches()
+ p = parts(**{"in": {"MIX": 1}, "out": {"OUT": 1}})
+ choices, pooled, step = ve.capped_fallback(p, 5)
+ assert step == 2 and pooled == {"MIX"} and len(choices) == 4
+ finally:
+ del CAP_LABELS["MIX"], CAP_COMP["MIX"]
+
+
+def test_step3_single_copy_when_outputs_alone_exceed_the_cap(cap_world):
+ p = parts(**{"in": {"GENE": 1}, "out": {"BIG": 1}, "cat": ["REG"]})
+ choices, pooled, step = ve.capped_fallback(p, 10)
+ assert step == 3 and choices == [None] and pooled == {"REG"}
+
+
+def test_emitted_copy_reads_the_pool_reference(cap_world, monkeypatch):
+ monkeypatch.setattr(vk, "_max_variants", lambda: 10)
+ monkeypatch.setattr(ve, "fetch_participants", lambda g, r: {
+ "R1": parts(**{"in": {"GENE": 1}, "out": {"OUT": 1}, "pos": ["REG"]})})
+ rid_map, vr, cat, neg, pos = ve.build_variant_reactions(None, ["R1"])
+ assert len(rid_map) == 4 and set(pos["entity_id"]) == {"REG::pool"}
+ assert {o for (_, outs, _, _) in vr.values() for o in outs} == {
+ f"o{i}" for i in range(1, 5)} # a bare set's key is its member's key
+ assert ve.STATS["capped_step1"] == 1
+
+
+def test_step1_does_not_pool_a_participant_sharing_a_slot_with_a_kept_one(cap_world):
+ # review of vn6: catalyst A = {S, T} shares S with input B = {S, OUT}.
+ # Pooling A would let each copy (fixed S for B) read every S variant of A.
+ CAP_LABELS.update(A=["Complex"], B=["Complex"], S=["EntitySet"], T=["EntitySet"])
+ CAP_COMP.update(A=["S", "T"], B=["S", "OUT"])
+ CAP_MEM.update(S=["s1", "s2", "s3"], T=["t1", "t2", "t3"])
+ try:
+ vk.reset_caches()
+ p = parts(**{"in": {"B": 1}, "out": {"OUT": 1}, "cat": ["A"]})
+ choices, pooled, step = ve.capped_fallback(p, 20)
+ assert "A" not in pooled or step != 1
+ finally:
+ for d, ks in ((CAP_LABELS, "ABST"), (CAP_COMP, "AB"), (CAP_MEM, "ST")):
+ for k in ks:
+ d.pop(k, None)
diff --git a/tests/test_variant_keys.py b/tests/test_variant_keys.py
new file mode 100644
index 0000000..781e002
--- /dev/null
+++ b/tests/test_variant_keys.py
@@ -0,0 +1,100 @@
+"""specs/046 variant keys (no database: lookups injected)."""
+import pytest
+
+import src.variant_keys as vk
+
+# GPVI-like: CX = GDP + SET{RHOA, RAC1}; FGFR2-like: OUTER{INNER{C1, C2}} with
+# C1 = HS + FGFSET{F7, F10} (a member complex that itself holds a set)
+LABELS = {"CX": ["Complex"], "SET": ["EntitySet", "CandidateSet"], "GDP": ["SimpleEntity"],
+ "RHOA": ["EWAS"], "RAC1": ["EWAS"], "OUTER": ["EntitySet", "DefinedSet"],
+ "INNER": ["EntitySet", "DefinedSet"], "C1": ["Complex"], "C2": ["Complex"],
+ "HS": ["SimpleEntity"], "FGFSET": ["EntitySet", "DefinedSet"], "F7": ["EWAS"],
+ "F10": ["EWAS"], "R2": ["EWAS"], "UB": ["EntitySet"], "DIMER": ["Complex"],
+ "CAT": ["Complex"]}
+COMP = {"CX": ["GDP", "SET"], "C1": ["HS", "FGFSET"], "C2": ["R2"], "DIMER": ["SET", "SET"],
+ "CAT": ["SET", "GDP"]}
+MEM = {"SET": ["RHOA", "RAC1"], "OUTER": ["INNER"], "INNER": ["C1", "C2"], "FGFSET": ["F7", "F10"],
+ "UB": ["x", "y"]}
+
+
+@pytest.fixture(autouse=True)
+def stub():
+ vk._lookups.update(labels=lambda s: LABELS.get(s, []), components=lambda s: COMP.get(s, []),
+ members=lambda s: MEM.get(s, []), atomic_sets=lambda: {"UB"},
+ max_variants=lambda: 512)
+ vk.reset_caches()
+ yield
+ vk._lookups.clear(); vk.reset_caches()
+
+
+def test_complex_key_names_slot_and_member():
+ assert vk.vkey("CX", {"SET": "RAC1"}) == "CX::variant::SET=RAC1"
+ assert vk.parse_variant_key("CX::variant::SET=RAC1") == ("CX", {"SET": "RAC1"})
+ assert vk.variant_leaves("CX::variant::SET=RAC1") == {"GDP", "RAC1"}
+
+
+def test_bare_set_dissolves_to_member_key_and_nested_sets_flatten():
+ assert vk.flat_members("OUTER") == ["C1", "C2"]
+ # outer set, member complex with its own set: hoisted slot
+ s = {"OUTER": "C1", "FGFSET": "F7"}
+ assert vk.vkey("OUTER", s) == vk.vkey("C1", s) == "C1::variant::FGFSET=F7"
+
+
+def test_producer_of_inner_set_and_consumer_of_outer_set_name_the_same_node():
+ # F10: R-HSA-190408 outputs INNER, R-HSA-5654404 inputs OUTER
+ assert vk.vkey("INNER", {"INNER": "C1", "FGFSET": "F10"}) == \
+ vk.vkey("OUTER", {"OUTER": "C1", "FGFSET": "F10"})
+
+
+def test_modifier_sets_and_set_free_entities_are_plain():
+ assert vk.vkey("UB", {}) == "UB"
+ assert vk.vkey("GDP", {}) == "GDP"
+ assert vk.vkey("C2", {}) == "C2"
+
+
+def test_capped_complex_is_its_plain_stid(monkeypatch):
+ vk._lookups["max_variants"] = lambda: 1
+ vk.reset_caches()
+ assert vk.is_capped("CX")
+ assert vk.vkey("CX", {}) == "CX"
+
+
+def test_reaction_choices_share_one_variable_per_set():
+ # D2: SET appears as input (CX) and as catalyst (CAT): one choice each copy
+ choices = list(vk.reaction_choices(["CX", "CAT"]))
+ assert choices == [{"SET": "RAC1"}, {"SET": "RHOA"}]
+ # a stoichiometry-2 set gives homo variants only (D3)
+ assert list(vk.reaction_choices(["DIMER"])) == [{"SET": "RAC1"}, {"SET": "RHOA"}]
+
+
+def test_member_complex_opens_its_slots_recursively():
+ got = list(vk.reaction_choices(["OUTER"]))
+ assert {"OUTER": "C1", "FGFSET": "F10"} in got and {"OUTER": "C1", "FGFSET": "F7"} in got
+ assert {"OUTER": "C2"} in got and len(got) == 3
+
+
+def test_choices_respect_the_limit():
+ assert len(list(vk.reaction_choices(["OUTER"], limit=2))) == 2
+
+
+def test_keys_do_not_depend_on_lookup_order():
+ a = vk.vkey("CX", {"SET": "RHOA"})
+ MEM["SET"] = ["RAC1", "RHOA"][::-1]
+ vk.reset_caches()
+ assert vk.vkey("CX", {"SET": "RHOA"}) == a
+
+
+def test_malformed_key_is_an_error():
+ with pytest.raises(ValueError):
+ vk.parse_variant_key("CX::variant::SETRAC1")
+
+
+def test_variant_parts_reads_both_formats():
+ assert vk.variant_parts("CX::variant::SET=RAC1") == ("CX", ["RAC1"])
+ assert vk.variant_parts("CX::variant::R-HSA-1_R-HSA-2") == ("CX", ["R-HSA-1", "R-HSA-2"])
+ assert vk.variant_parts("R-HSA-9") == ("R-HSA-9", [])
+
+
+def test_variant_components_replace_slots_by_chosen_members():
+ assert vk.variant_components("CX::variant::SET=RAC1") == ["GDP", "RAC1"]
+ assert vk.is_variant_key("CX::variant::SET=RAC1") and not vk.is_variant_key("CX::variant::A_B")
diff --git a/tests/test_variant_pools.py b/tests/test_variant_pools.py
new file mode 100644
index 0000000..07cc074
--- /dev/null
+++ b/tests/test_variant_pools.py
@@ -0,0 +1,139 @@
+"""specs/046: pool nodes for variant depleters and for capped-reaction pool
+references, and the regulator's produced-copy preference (no database)."""
+from src import logic_network_generator as lng
+
+
+def edge(s, t, et, pn="pos", ao="and"):
+ return {"source_id": s, "target_id": t, "pos_neg": pn, "and_or": ao, "edge_type": et, "stoichiometry": 1}
+
+
+def test_variants_of_one_depleter_become_one_pool_edge():
+ ids = {"c1": "C::variant::S=a", "c2": "C::variant::S=b", "d": "D", "tp53": "TP53"}
+ data = [edge("x", "y", "input"),
+ edge("c1", "tp53", "depletion", "neg"), edge("c2", "tp53", "depletion", "neg"),
+ edge("d", "tp53", "depletion", "neg")]
+ lng._pool_variant_depleters(data, 1, ids)
+ dep = [e for e in data if e["edge_type"] == "depletion"]
+ assert len(dep) == 2 # D kept, C's two variants -> one pool
+ pool = next(e["source_id"] for e in dep if e["source_id"] != "d")
+ assert ids[pool] == "C::pool"
+ assert sorted(e["source_id"] for e in data if e["edge_type"] == "variant_pool" and e["target_id"] == pool) == ["c1", "c2"]
+ assert all(e["and_or"] == "or" for e in data if e["edge_type"] == "variant_pool")
+ assert data[0] == edge("x", "y", "input") # edges before the pass untouched
+
+
+def test_a_single_depleter_is_left_alone():
+ ids = {"c1": "C::variant::S=a", "t": "T"}
+ data = [edge("c1", "t", "depletion", "neg")]
+ lng._pool_variant_depleters(data, 0, ids)
+ assert data == [edge("c1", "t", "depletion", "neg")]
+
+
+import pytest
+from src import variant_keys as vk
+
+
+@pytest.fixture
+def world():
+ vk._lookups.update(labels=lambda s: {"R": ["Complex"], "S": ["EntitySet"], "SET": ["EntitySet"]}.get(s, ["EWAS"]),
+ components=lambda s: {"R": ["S", "G"]}.get(s, []),
+ members=lambda s: {"S": ["a", "b", "c"], "SET": ["m1", "m2"]}.get(s, []),
+ atomic_sets=lambda: set(), max_variants=lambda: 512)
+ vk.reset_caches()
+ yield
+ vk._lookups.clear(); vk.reset_caches()
+
+
+def test_pool_reference_reads_produced_variants_else_roots(world):
+ ids = {"p": "R::pool", "v1": "R::variant::S=a", "v1b": "R::variant::S=a",
+ "v2": "R::variant::S=b", "plain": "R", "rx": "RX"}
+ data = [edge("rx", "v1", "output"), # v1 produced, v1b an unfed copy of the same key
+ edge("p", "rx2", "regulator")]
+ lng._wire_variant_pool_refs(data, ids)
+ members = sorted(e["source_id"] for e in data if e["edge_type"] == "variant_pool")
+ assert "v1" in members and "v2" in members and "v1b" not in members and "plain" not in members
+ new = [m for m in members if m not in ("v1", "v2")]
+ assert [ids[m] for m in new] == ["R::variant::S=c"] # S=c existed nowhere: a new root
+
+
+def test_a_pooled_bare_set_reads_its_members(world):
+ # vn5 RPL10: a bare set's variant keys are its MEMBERS' keys
+ ids = {"p": "SET::pool", "x": "m1"}
+ data = [edge("p", "rx", "regulator")]
+ lng._wire_variant_pool_refs(data, ids)
+ members = sorted(ids[e["source_id"]] for e in data if e["edge_type"] == "variant_pool")
+ assert members == ["m1", "m2"] # m1's existing node, m2 a new root
+
+
+def test_regulator_prefers_the_copy_a_preceding_reaction_produces(monkeypatch):
+ import pandas as pd
+ monkeypatch.setenv("LNG_COMPLEX_AS_NODE", "1")
+ key = "PTEN::variant::S=x"
+ cat = pd.DataFrame([{"reaction_id": "R-DEPHOS", "entity_id": key, "edge_type": "catalyst",
+ "uuid": "u", "reaction_uuid": "rxcopy"}])
+ empty = pd.DataFrame(columns=cat.columns)
+ registry = {(key, "vr_in", "input"): "unfed", (key, "vr_a", "output"): "fed_other",
+ (key, "vr_b", "output"): "fed_preceding"}
+ data, ids = [], {}
+ lng.append_regulators(cat, empty, empty, data, ids, entity_uuid_registry=registry,
+ produced_by_eid={key: [("fed_other", "R-OTHER"), ("fed_preceding", "R-TRANSLATE")]},
+ preceding_by_reaction={"R-DEPHOS": {"R-TRANSLATE"}})
+ assert [e["source_id"] for e in data] == ["fed_preceding"]
+ data = []
+ lng.append_regulators(cat, empty, empty, data, ids, entity_uuid_registry=registry)
+ assert [e["source_id"] for e in data] == ["unfed"] # default: first registry entry (unchanged)
+
+
+def redge(s, t, et, rx, pn="pos"):
+ e = edge(s, t, et, pn); e["edge_reaction_id"] = rx; return e
+
+
+def test_a_capped_plain_output_feeds_its_variants(world):
+ # vn5 RAF: one copy writes plain R; consumers read R's variant keys
+ ids = {"out": "R", "va": "R::variant::S=a", "vb": "R::variant::S=b", "vc": "R::variant::S=c"}
+ data = [redge("rx", "out", "output", "P"),
+ redge("va", "c1", "input", "F"), redge("vb", "c2", "input", "F"),
+ redge("vc", "c3", "input", "UNRELATED")] # not a curated follower
+ lng._wire_capped_outputs(data, ids, {"P": {"F"}})
+ split = sorted((e["source_id"], e["target_id"]) for e in data if e["edge_type"] == "variant_split")
+ assert split == [("out", "va"), ("out", "vb")]
+
+
+def test_split_skips_a_variant_another_reaction_produces(world):
+ ids = {"out": "R", "va": "R::variant::S=a", "vb": "R::variant::S=b"}
+ data = [redge("rx", "out", "output", "P"), redge("rx9", "va", "output", "Q"),
+ redge("va", "c1", "input", "F"), redge("vb", "c2", "input", "F")]
+ lng._wire_capped_outputs(data, ids, {"P": {"F"}})
+ split = sorted(e["target_id"] for e in data if e["edge_type"] == "variant_split")
+ assert split == ["vb"] # va already has a producer
+
+
+def test_an_unproduced_plain_node_is_not_split(world):
+ ids = {"out": "R", "va": "R::variant::S=a"}
+ data = [redge("va", "rx2", "input", "F")]
+ lng._wire_capped_outputs(data, ids, {"P": {"F"}})
+ assert not [e for e in data if e["edge_type"] == "variant_split"]
+
+
+def test_members_of_one_set_catalyst_deplete_once(world):
+ # review of vn6: six DUSP members of ONE set each depleted the MAPK3 dimer.
+ # Their ids share no prefix; the curated participant (_origin) groups them.
+ ids = {"d1": "DUSP1", "d2": "DUSP6", "o": "OTHER", "t": "MAPK3"}
+ data = [dict(edge("d1", "t", "depletion", "neg"), _origin="DUSPSET"),
+ dict(edge("d2", "t", "depletion", "neg"), _origin="DUSPSET"),
+ dict(edge("o", "t", "depletion", "neg"), _origin="OTHER")]
+ lng._pool_variant_depleters(data, 0, ids)
+ dep = [e for e in data if e["edge_type"] == "depletion"]
+ assert len(dep) == 2 and all("_origin" not in e for e in data)
+ pool = next(e["source_id"] for e in dep if e["source_id"] != "o")
+ assert ids[pool] == "DUSPSET::pool"
+
+
+def test_an_over_cap_set_pool_reads_its_members(world, monkeypatch):
+ # review of vn6: RAS GEFs (645 variants) was a pool with no input
+ monkeypatch.setattr(vk, "_max_variants", lambda: 1)
+ ids = {"p": "SET::pool", "x": "m1", "y": "m2::variant::Q=z", "other": "zz"}
+ data = []
+ lng._wire_variant_pool_refs(data, ids)
+ members = sorted(ids[e["source_id"]] for e in data if e["edge_type"] == "variant_pool")
+ assert members == ["m1", "m2::variant::Q=z"]