diff --git a/PIPELINE_SETTINGS.md b/PIPELINE_SETTINGS.md index 25ef211..1236cc1 100644 --- a/PIPELINE_SETTINGS.md +++ b/PIPELINE_SETTINGS.md @@ -73,6 +73,7 @@ what NOT to touch. Overriding is for A/B benchmarking only. | `LNG_POOL_ACTIVE_VIA` | `made_from` | how a pool's active form is found (deltasignal specs/039) | | `LNG_CAP_POOLS` | `1` | past the variant cap, one OR pool per set instead of an all-required bundle (deltasignal specs/045) | | `LNG_RECYCLE_POOLS` | `1` | write a modified protein's return to the complex it starts in as a pool with several resting states, plus a structural `recycle` edge from the free copy to the starting copy in `boundary_edges.csv` (p130/p107 in Mitotic G1; deltasignal specs/081) | +| `LNG_RELEASE_POOLS` | `1` | a protein bound, modified and released by a complex, whose released copy carries on (beta-catenin and the destruction complex), as a free/bound pool whose released copy reads the free state; writes `released_twin` and `recycle` edges (deltasignal specs/086) | | `LNG_CATALYST_BUNDLE` | `0` | legacy; subsumed by `LNG_COMPLEX_AS_NODE` | | `LNG_HANDOFF_EDGES` | `0` | precedingEvent hand-off bridges (net-negative, off) | diff --git a/src/env_flags.py b/src/env_flags.py index 071d88c..cf0ea57 100644 --- a/src/env_flags.py +++ b/src/env_flags.py @@ -29,6 +29,7 @@ "LNG_PATHOGEN_PROTEIN": True, # default since deltasignal specs/048 amendment 2 (arm pathp: exp +7, held-out +1, 0 broken) "LNG_POOL_JOIN_MEMBERS": False, # deltasignal specs/048 lever B1 "LNG_POOL_SHARED": False, # deltasignal specs/049 + "LNG_RELEASE_POOLS": True, # default since deltasignal specs/086 (WNT tuning +73, experimental +8, held-out 0) "LNG_RECYCLE_POOLS": True, # default since deltasignal specs/081 (tuning +70/0, experimental +8, held-out 0) "LNG_SET_EXPAND": True, "LNG_SET_MEMBERS_OR": False, diff --git a/src/logic_network_generator.py b/src/logic_network_generator.py index 2764803..cc512f9 100755 --- a/src/logic_network_generator.py +++ b/src/logic_network_generator.py @@ -4736,6 +4736,174 @@ def copies(a: str, b: str, rx: str) -> List[str]: return forms, transitions, edges +RELEASE_MAX_REACTIONS = 10 # deltasignal specs/086: beta-catenin's route is 8 reactions + + +def find_release_pools(pathway_logic_network: pd.DataFrame, reaction_id_map: pd.DataFrame, + uuid_mapping: Dict[str, str], profiles: Dict[str, Dict[str, Any]], + claimed: Optional[Set[str]] = None, stats: Optional[Dict[str, int]] = None): + """Release pools (deltasignal specs/086): a protein bound, modified and released by a + complex, whose released copy carries on downstream. + + beta-catenin in Signaling by WNT: its starting copy S binds the destruction complex; + CK1 and GSK3 phosphorylate it there, PP2A dephosphorylates it, and the released copy P + goes on to the nucleus. P's only producer is the release, so the nuclear signal was + the traffic through the destruction complex, and an APC knockout read DOWN. A candidate: + + - S, a node of a single protein R with no in-edge that is an ``input`` or ``catalyst`` + of a reaction; + - P, another node of the same entity that a reaction produces and that has an out-edge; + - a route S -> ... -> P of at most ``RELEASE_MAX_REACTIONS`` reactions (input or + catalyst to output) through forms that carry R; + - on that route a form whose modification signature for R differs from S's (a partner + released unchanged is a carrier, deltasignal specs/077, and is excluded). + + The pool's states are every form M_v a release copy that makes P consumes (one per + variant of the complex; all bases: at rest the protein is mostly bound), and S; the + forms between are intermediates (amendment 1). Paths: S -> ... -> M_v along each + M_v's own route, and M_v -> S through its release. P is the pool's carrier, its + release steps every release copy, so it reads its other producers; the + edge S -> P (``released_twin``, OR) makes that S. A structural ``recycle`` edge + P -> S (specs/081) makes the cycle one component. A candidate any of whose nodes is in + ``claimed`` is skipped and counted. + + Returns (forms, transitions, carriers, edges) in :func:`find_pools`'s row formats, the + edges as ``(source_uuid, target_uuid, edge_type)``.""" + from src.variant_keys import variant_parts + stats = stats if stats is not None else {} + for k in ("release_pools", "release_paths", "release_claimed"): + stats[k] = 0 + claimed = claimed or set() + if pathway_logic_network.empty: + return [], [], [], [] + ent = _uuid_to_stable_id_map(pathway_logic_network, uuid_mapping) + vr = {str(u): str(r) for u, r in zip(reaction_id_map["uid"], reaction_id_map["reactome_id"])} + feeds: Dict[str, Set[str]] = {} # entity -> reaction nodes it is an input or catalyst of + rin: Dict[str, Set[str]] = {} # reaction node -> its input and catalyst entities + rout: Dict[str, Set[str]] = {} + catalysed: Set[str] = set() + produced: Set[str] = set(); has_out: Set[str] = set(); has_in: Set[str] = set() + for _, e in pathway_logic_network.iterrows(): + s, t, et = str(e["source_id"]), str(e["target_id"]), e.get("edge_type") + has_out.add(s); has_in.add(t) + if et in ("input", "catalyst") and t in vr: + feeds.setdefault(s, set()).add(t); rin.setdefault(t, set()).add(s) + if et == "catalyst": + catalysed.add(t) + elif et == "output" and s in vr: + rout.setdefault(s, set()).add(t); produced.add(t) + + def pe_of(u: str) -> str: + return variant_parts(ent.get(u, ""))[0] + + def proteins(u: str) -> frozenset: + parent, members = variant_parts(ent.get(u, "")) + if not members: + return frozenset(profiles.get(parent, {}).get("proteins", ())) + out = set(profiles.get(parent, {}).get("fixed", ())) + for mm in members: + out |= set(profiles.get(mm, {}).get("proteins", ())) + return frozenset(out) + + def sig(u: str, r): + return profiles.get(pe_of(u), {}).get("sig", {}).get(r, frozenset()) + + by_pe: Dict[str, List[str]] = {} + for u in sorted(ent): + by_pe.setdefault(pe_of(u), []).append(u) + + def route(src: str, targets: Set[str], r) -> Optional[List[Tuple[str, str, str]]]: + prev: Dict[str, Optional[Tuple[str, str]]] = {src: None} + frontier = [src] + for _ in range(RELEASE_MAX_REACTIONS): + nxt = [] + for x in frontier: + for rx in sorted(feeds.get(x, ())): + for y in sorted(rout.get(rx, ())): + if y in prev or r not in proteins(y): + continue + prev[y] = (x, rx); nxt.append(y) + if y in targets: + steps: List[Tuple[str, str, str]] = [] + link = prev[y] + while link is not None: + a, rr = link; steps.append((a, rr, y)); y = a + link = prev[y] + return steps[::-1] + frontier = nxt + return None + + def copies(a: str, b: str, rx: str) -> List[str]: + return sorted(u for u in vr if vr[u] == vr[rx] and a in rin.get(u, ()) and b in rout.get(u, ())) + + forms: List[Tuple] = []; transitions: List[Tuple] = []; carriers: List[Tuple] = [] + edges: List[Tuple[str, str, str]] = [] + n = 0 + for s_node in sorted(ent): + if s_node in has_in or s_node in vr or not feeds.get(s_node): + continue + rs = proteins(s_node) + if len(rs) != 1: + continue + r = next(iter(rs)) + ps = {u for u in by_pe.get(pe_of(s_node), []) if u != s_node and u in produced and u in has_out} + if not ps: + continue + st = route(s_node, ps, r) + if not st or len(st) < 2: + continue + if not any(sig(y, r) != sig(s_node, r) for _, _, y in st[:-1]): + continue + p_node = st[-1][2] + rel_stid = vr[st[-1][1]] + # amendment 1: every release copy that makes P, each with its own consumed form + # M_v and its own route from S (one chain per variant of the complex) + chains = [] + for rc in sorted(u for u in vr if vr[u] == rel_stid and p_node in rout.get(u, ())): + for m_node in sorted(x for x in rin.get(rc, ()) if r in proteins(x) and x != s_node): + cst = route(s_node, {m_node}, r) + if cst: + chains.append((m_node, rc, cst)) + if not chains: + continue + nodes = {s_node} + fwd_by_m: Dict[str, List[Tuple[str, str, List[str]]]] = {} + rel_by_m: Dict[str, List[str]] = {} + inter: Set[str] = set() + for m_node, rc, cst in chains: + fwd_by_m.setdefault(m_node, [(a, y, copies(a, y, rx)) for a, rx, y in cst]) + rel_by_m.setdefault(m_node, []).append(rc) + inter |= {y for _, _, y in cst[:-1]} + nodes |= {m_node} + inter -= set(fwd_by_m) | {s_node} + nodes |= inter | {u for f in fwd_by_m.values() for _, _, cs in f for u in cs} | \ + {u for rs in rel_by_m.values() for u in rs} + if nodes & claimed or any(not cs for f in fwd_by_m.values() for _, _, cs in f): + stats["release_claimed"] += 1 + continue + n += 1 + pid = f"release{n}" + for m_node in sorted(fwd_by_m): + forms.append((pid, m_node, ent.get(m_node, ""), "state", True)) + forms.append((pid, s_node, ent.get(s_node, ""), "state", False)) + for u in sorted(inter): + forms.append((pid, u, ent.get(u, ""), "intermediate", False)) + for i, m_node in enumerate(sorted(fwd_by_m), 1): + fwd = fwd_by_m[m_node] + enz = any(u in catalysed for _, _, cs in fwd for u in cs) + for si, (a, b, cs) in enumerate(fwd, 1): + for u in cs: + transitions.append((pid, f"{pid}_f{i}", si, a, b, u, vr[u], enz)) + for u in sorted(set(rel_by_m[m_node])): + transitions.append((pid, f"{pid}_r{i}", 1, m_node, s_node, u, vr[u], u in catalysed)) + carriers.append((pid, p_node, u)) + stats["release_paths"] += 2 + edges += [(s_node, p_node, "released_twin"), (p_node, s_node, "recycle")] + claimed |= nodes + stats["release_pools"] += 1 + return forms, transitions, carriers, edges + + def export_pools(pathway_id: str, pathway_logic_network: pd.DataFrame, reaction_id_map: pd.DataFrame, uuid_mapping: Dict[str, str], pools_file: str, transitions_file: str, carriers_file: str, carrier_loops_file: Optional[str] = None, @@ -4784,6 +4952,24 @@ def export_pools(pathway_id: str, pathway_logic_network: pd.DataFrame, reaction_ logger.info(f"Recycle pools (specs/081): {rstats['recycle_pools']} pools, {rstats['recycle_states']} states, " f"{rstats['recycle_paths']} paths, {rstats['recycle_edges']} recycle edges, " f"{rstats['recycle_claimed']} skipped (nodes claimed by a pool)") + if env_flag("LNG_RELEASE_POOLS"): + # deltasignal specs/086: after the ordinary and recycle pools, which keep their nodes + claimed = {f[1] for f in forms} | {t[5] for t in transitions} + lstats: Dict[str, int] = {} + lf, lt, lc, ledges = find_release_pools(pathway_logic_network, reaction_id_map, uuid_mapping, profiles, + claimed, lstats) + forms = list(forms) + lf + transitions = list(transitions) + lt + carriers = list(carriers) + lc + if ledges: + if boundary_file is None: + raise ValueError("LNG_RELEASE_POOLS found release pools but no boundary file to write their edges to") + cols = list(pd.read_csv(boundary_file, nrows=0).columns) + add = pd.DataFrame([{"source_id": a, "target_id": b, "pos_neg": "pos", "and_or": "or", + "edge_type": t, "stoichiometry": 1} for a, b, t in ledges]) + add.reindex(columns=cols).to_csv(boundary_file, mode="a", header=False, index=False) + logger.info(f"Release pools (specs/086): {lstats['release_pools']} pools, {lstats['release_paths']} paths, " + f"{lstats['release_claimed']} skipped (nodes claimed by a pool or no reaction copies)") pd.DataFrame(forms, columns=["pool_id", "node_uuid", "stable_id", "role", "is_base"]).to_csv(pools_file, index=False) pd.DataFrame(transitions, columns=["pool_id", "path_id", "step", "source_uuid", "target_uuid", "reaction_uuid", "reaction_stid", "enzyme_driven"]).to_csv(transitions_file, index=False) diff --git a/tests/test_release_pools.py b/tests/test_release_pools.py new file mode 100644 index 0000000..d6acdbd --- /dev/null +++ b/tests/test_release_pools.py @@ -0,0 +1,99 @@ +"""Release pools (deltasignal specs/086): a protein bound, modified and released by a +complex, whose released copy carries on. Shape: beta-catenin (s) binds the destruction +complex (dc) into c1; a kinase phosphorylates it (c2); a phosphatase dephosphorylates it +(c3); the release (v4) makes p, which goes on to the nucleus (vt -> n). The pool: c3 (base) +and s, intermediates c1 and c2; paths s -> c1 -> c2 -> c3 and c3 -> s; p is the carrier +with release v4; edges s -> p (released_twin) and p -> s (recycle).""" +import pandas as pd + +import src.logic_network_generator as m + +NAMES = ["s", "dc", "v1", "c1", "kin", "v2", "c2", "pp", "v3", "c3", "v4", "p", "vt", "n"] +U = {k: f"00000000-0000-0000-0000-{i:012d}" for i, k in enumerate(NAMES, 1)} +CTNNB1, DC, KIN, PP = 1, 2, 3, 4 + + +def e(s, t, et): + return {"source_id": U[s], "target_id": U[t], "edge_type": et, "stoichiometry": 1} + + +def prof(proteins, sig=()): + return {"proteins": set(proteins), "fixed": set(proteins), "sig": {r: frozenset(sig) for r in proteins}, + "mods": 0, "comps": 0, "slots": 1} + + +def bcat(modified=True, carries_on=True): + edges = [e("s", "v1", "input"), e("dc", "v1", "input"), e("v1", "c1", "output"), + e("c1", "v2", "input"), e("kin", "v2", "catalyst"), e("v2", "c2", "output"), + e("c2", "v3", "input"), e("pp", "v3", "catalyst"), e("v3", "c3", "output"), + e("c3", "v4", "input"), e("v4", "p", "output")] + if carries_on: + edges += [e("p", "vt", "input"), e("vt", "n", "output")] + net = pd.DataFrame(edges) + rx = ["v1", "v2", "v3", "v4", "vt"] + rmap = pd.DataFrame({"uid": [U[k] for k in rx], "reactome_id": [f"R-HSA-{k.upper()}" for k in rx]}) + umap = {U["s"]: "R-HSA-CTNNB1", U["p"]: "R-HSA-CTNNB1", U["dc"]: "R-HSA-DC", U["c1"]: "R-HSA-C1", + U["c2"]: "R-HSA-C2", U["c3"]: "R-HSA-C3", U["kin"]: "R-HSA-KIN", U["pp"]: "R-HSA-PP", U["n"]: "R-HSA-N"} + phos = (("res", "p-S45"),) if modified else () + profiles = {"R-HSA-CTNNB1": prof([CTNNB1]), "R-HSA-DC": prof([DC]), + "R-HSA-C1": prof([CTNNB1, DC]), "R-HSA-C2": prof([CTNNB1, DC], phos), + "R-HSA-C3": prof([CTNNB1, DC]), "R-HSA-KIN": prof([KIN]), "R-HSA-PP": prof([PP]), + "R-HSA-N": prof([CTNNB1])} + return net, rmap, umap, profiles + + +def test_beta_catenin_is_a_free_bound_pool_whose_released_copy_reads_the_free_state(): + st = {} + forms, trans, carriers, edges = m.find_release_pools(*bcat(), stats=st) + assert [(u, role, base) for _, u, _, role, base in forms] == [ + (U["c3"], "state", True), (U["s"], "state", False), + (U["c1"], "intermediate", False), (U["c2"], "intermediate", False)] + rows = sorted((t[1], t[2], t[3], t[4], t[5]) for t in trans) + assert rows == sorted([("release1_f1", 1, U["s"], U["c1"], U["v1"]), ("release1_f1", 2, U["c1"], U["c2"], U["v2"]), + ("release1_f1", 3, U["c2"], U["c3"], U["v3"]), ("release1_r1", 1, U["c3"], U["s"], U["v4"])]) + assert [(c[1], c[2]) for c in carriers] == [(U["p"], U["v4"])] + assert edges == [(U["s"], U["p"], "released_twin"), (U["p"], U["s"], "recycle")] + assert st["release_pools"] == 1 and st["release_paths"] == 2 and st["release_claimed"] == 0 + + +def test_a_partner_released_unchanged_is_not_a_release_pool(): + assert m.find_release_pools(*bcat(modified=False)) == ([], [], [], []) + + +def test_a_released_copy_that_leads_nowhere_is_left_to_recycle_pools(): + assert m.find_release_pools(*bcat(carries_on=False)) == ([], [], [], []) + + +def test_a_candidate_on_a_claimed_node_is_skipped_and_counted(): + st = {} + assert m.find_release_pools(*bcat(), claimed={U["c2"]}, stats=st) == ([], [], [], []) + assert st["release_claimed"] == 1 + + +def test_the_flag_is_on_by_default(): + # default since deltasignal specs/086 was adopted (Adam, 2026-10-21) + from src.env_flags import BOOL_FLAGS + assert BOOL_FLAGS["LNG_RELEASE_POOLS"] is True + + +def test_every_variant_chain_is_in_the_pool(): + # amendment 1: a second variant of the complex (dc2 -> c1b -> c2b -> c3b) also releases p + net, rmap, umap, profiles = bcat() + extra = ["dc2", "w1", "c1b", "w2", "c2b", "w3", "c3b", "w4"] + for i, k in enumerate(extra, 100): + U[k] = f"00000000-0000-0000-0000-{i:012d}" + rows = [e("s", "w1", "input"), e("dc2", "w1", "input"), e("w1", "c1b", "output"), + e("c1b", "w2", "input"), e("kin", "w2", "catalyst"), e("w2", "c2b", "output"), + e("c2b", "w3", "input"), e("pp", "w3", "catalyst"), e("w3", "c3b", "output"), + e("c3b", "w4", "input"), e("w4", "p", "output")] + net = pd.concat([net, pd.DataFrame(rows)], ignore_index=True) + rmap = pd.concat([rmap, pd.DataFrame({"uid": [U[k] for k in ("w1", "w2", "w3", "w4")], + "reactome_id": ["R-HSA-V1", "R-HSA-V2", "R-HSA-V3", "R-HSA-V4"]})], + ignore_index=True) + umap = dict(umap); umap.update({U["dc2"]: "R-HSA-DC", U["c1b"]: "R-HSA-C1", U["c2b"]: "R-HSA-C2", U["c3b"]: "R-HSA-C3"}) + st = {} + forms, trans, carriers, edges = m.find_release_pools(net, rmap, umap, profiles, stats=st) + assert sorted(u for _, u, _, role, base in forms if base) == sorted([U["c3"], U["c3b"]]) + assert sorted(c[2] for c in carriers) == sorted([U["v4"], U["w4"]]) + assert {t[1] for t in trans} == {"release1_f1", "release1_r1", "release1_f2", "release1_r2"} + assert st["release_pools"] == 1 and st["release_paths"] == 4