diff --git a/src/env_flags.py b/src/env_flags.py index dcaa72d..8bd66cf 100644 --- a/src/env_flags.py +++ b/src/env_flags.py @@ -26,6 +26,8 @@ "LNG_EMIT_ONE_SIDED": True, "LNG_HANDOFF_EDGES": False, "LNG_PATHOGEN_PROTEIN": True, # default since deltasignal specs/048 amendment 2 (arm pathp: exp +7, held-out +1, 0 broken) + "LNG_POOL_JOIN_MEMBERS": False, # deltasignal specs/048 lever B1 + "LNG_POOL_SHARED": False, # deltasignal specs/049 "LNG_SET_EXPAND": True, "LNG_SET_MEMBERS_OR": False, "LNG_SET_POOL": True, @@ -39,6 +41,7 @@ "LNG_HANDOFF_HUB_MAX", # _int_env "LNG_MAX_VARIANTS", # _int_env "LNG_POOL_ACTIVE_VIA", # pool_active_via() + "LNG_POOL_REGEN", # pool_regen() "LNG_PYTHON", # catalog.sh: the interpreter, not a switch }) diff --git a/src/logic_network_generator.py b/src/logic_network_generator.py index 48a21f6..455c24b 100755 --- a/src/logic_network_generator.py +++ b/src/logic_network_generator.py @@ -512,6 +512,7 @@ def _reject_removed_env() -> None: raise ValueError(f"{name} was removed: {why}") validate_env(removed=_REMOVED_ENV) pool_active_via() + pool_regen() _POOL_ACTIVE_VIA = ("direct", "made_from") @@ -533,6 +534,24 @@ def pool_active_via() -> str: return value +_POOL_REGEN = ("entity", "protein") + + +def pool_regen() -> str: + """LNG_POOL_REGEN (deltasignal specs/048 lever B2): how a multi-step pool + path proves it regenerates what it consumes (specs/039 amendment 5). + ``entity`` (default): every consumed non-R, non-small input is output again + by a step of the path, compared by stable id. ``protein``: every protein a + consumed input carries is output again, in any form, so a partner released + under another stable id (RAS:GTP:activated RAF dimers, R-HSA-5672718 in and + R-HSA-5672712 out) or modified (MAP2K in, p-2S MAP2K out) still counts. + Any other value is an error at startup.""" + value = os.environ.get("LNG_POOL_REGEN", "entity") + if value not in _POOL_REGEN: + raise ValueError(f"LNG_POOL_REGEN must be one of {_POOL_REGEN}, got {value!r}") + return value + + def _register_phase1( vr_entities: Dict[str, tuple], entity_uuid_registry: Dict[tuple, str], @@ -3972,14 +3991,16 @@ def parts(node_id: str): "shared_nodes_dropped", "ties", "carriers", "autocat_source", "autocat_product", "autocat_other", "carrier_loops", "no_path_pools", "off_path_intermediates", "off_path_states", "ambiguous_copies", "budget_dropped", "merged_proteins", "copy_rows", - "carrier_conflicts", "nonregenerating_paths", "oriented_by_activity", "activity_fallbacks") + "carrier_conflicts", "nonregenerating_paths", "oriented_by_activity", "activity_fallbacks", + "carrier_loop_sets", "carrier_loops_closed", "carrier_loop_forms", "carrier_loop_conflicts") def find_pools(pathway_logic_network: pd.DataFrame, reaction_id_map: pd.DataFrame, uuid_mapping: Dict[str, str], steps, profiles: Dict[str, Dict[str, Any]], donor_reactions: Optional[Set[str]] = None, set_members: Optional[Dict[str, Set[str]]] = None, - stats: Optional[Dict[str, int]] = None): + stats: Optional[Dict[str, int]] = None, + carrier_loops: Optional[List[Tuple]] = None): """Modification-cycle pools at NODE level (deltasignal specs/039 amendment 2). ``steps`` are the pathway's R-steps at stId level, ``(reaction, R, input @@ -4043,6 +4064,18 @@ def find_pools(pathway_logic_network: pd.DataFrame, reaction_id_map: pd.DataFram source, target and reaction_stid and differ in reaction_uuid. ``enzyme_driven`` is the STEP's flag; a path is enzyme-driven if any step is. carriers: [(pool_id, carrier_uuid, release_reaction_uuid)] + ``carrier_loops``, when given a list, receives the carrier loops (deltasignal + specs/049): each R-form set rejected above as a carrier loop on the first + pass, de-duplicated across R. One row per (loop, free form, loop source): + ``(loop_id, r, closed, free_uuid, free_stid, loop_source_uuid)``. A free form + is a loop node whose proteins are exactly the loop's core; a loop source is + a node inside the loop (a step reaction node, a form of the loop, or a plain + node a step writes that variant_split hands to a form) with a positive + in-edge into that free form (any type but input, catalyst and regulator). + The loop is closed when no form of it has such an in-edge from outside the + loop. Free forms and conflicts are counted once per node, and a (free form, + source) pair in several loops is closed if any of them is. A free form that a pool manages (state, intermediate, + step node or carrier) is not listed (``carrier_loop_conflicts``). ``stats`` receives the counts in ``POOL_STATS_KEYS``; the autocatalysis counts are distinct (reaction node, catalyst) pairs where a form of the pool catalyses a step of its own pool, split by whether the catalyst is the @@ -4065,6 +4098,11 @@ def find_pools(pathway_logic_network: pd.DataFrame, reaction_id_map: pd.DataFram acts: Dict[str, Set[str]] = {} # node -> reaction nodes it catalyses or positively regulates set_pools_of: Dict[str, Set[str]] = {} # member node -> the set-pool nodes it feeds (specs/033) feeds: Dict[str, Set[str]] = {} # node -> reaction nodes it is an input of + into: Dict[str, Set[str]] = {} # entity node -> sources of its entity-level in-edges (specs/049) + pool_members: Dict[str, Set[str]] = {} # pool node -> the member nodes feeding it (specs/048 B1) + split_to: Dict[str, Set[str]] = {} # plain output node -> the variants it is split to (specs/048 B1) + join_members = env_flag("LNG_POOL_JOIN_MEMBERS") + regen_by_protein = pool_regen() == "protein" active_via = pool_active_via() has_st = "stoichiometry" in pathway_logic_network.columns for _, e in pathway_logic_network.iterrows(): @@ -4088,8 +4126,17 @@ def find_pools(pathway_logic_network: pd.DataFrame, reaction_id_map: pd.DataFram acts.setdefault(s, set()).add(t) elif et == "set_member": set_pools_of.setdefault(s, set()).add(t) + if et in ("set_member", "variant_pool"): + pool_members.setdefault(t, set()).add(s) + elif et == "variant_split": + split_to.setdefault(s, set()).add(t) if et == "input" and t in vr: feeds.setdefault(s, set()).add(t) + if et not in ("input", "catalyst", "regulator", "depletion") and t not in vr: + # every positive in-edge into an entity counts as supply: output, + # variant_split/variant_pool/set_member, and the flagged derived + # classes (diagram_bridge, composition, handoff) when present + into.setdefault(t, set()).add(s) copies: Dict[str, List[str]] = {} for u, r in vr.items(): copies.setdefault(r, []).append(u) @@ -4130,7 +4177,24 @@ def maps(u: str, pe: str) -> bool: for rx, r, a, b, enz in steps: for rx_u in sorted(copies.get(rx, [])): cin = [u for u, st in ins.get(rx_u, {}).items() if st == 1 and maps(u, a) and r in proteins(u)] + if join_members: + # specs/048 B1: a step reading a pool node (variant_pool or + # set_member target) starts from the pool's members that carry + # R, so a cycle closed through the pool (MAPK3 released by + # DUSPs re-entering R-HSA-169291::pool) is visible here as it + # is to the solver. + cin = sorted({m for u in cin for m in ( + [m for m in sorted(pool_members[u]) if r in proteins(m)] + if u in pool_members else [u])}) cout = [u for u, st in outs.get(rx_u, {}).items() if st == 1 and maps(u, b) and r in proteins(u)] + if join_members: + # and a step writing a plain node that variant_split edges hand + # to its variants (a capped reaction's output) ends at those + # variants that carry R: R-HSA-5672980 releases R-HSA-169289, + # split to p-MAPK1 and p-MAPK3. + cout = sorted({m for u in cout for m in ( + [m for m in sorted(split_to[u]) if r in proteins(m)] + if u in split_to else [u])}) if len(cin) != 1 or len(cout) != 1: if len(cin) > 1 or len(cout) > 1: stats["ambiguous_copies"] += 1 @@ -4139,6 +4203,7 @@ def maps(u: str, pe: str) -> bool: ustep[(r, cin[0], cout[0], rx_u)] = (rx, bool(enz)) if not ustep: return [], [], [] + first_loops: List[Tuple[Any, frozenset, frozenset, Set[str]]] = [] # (R, forms, core, step reaction nodes) def build(removed: Set[str]): local = {k: 0 for k in POOL_STATS_KEYS} @@ -4154,9 +4219,16 @@ def regenerates(path) -> bool: produced: Set[str] = set() for a_u, _, _, _, cp in path: for rx_u in cp: - consumed |= {ent.get(j, "") for j in ins.get(rx_u, {}) - if j != a_u and not prof(j, "small", False) and pe_of(j) not in _UBIQUITIN_STIDS} - produced |= {ent.get(o, "") for o in outs.get(rx_u, {})} + eaten = [j for j in ins.get(rx_u, {}) + if j != a_u and not prof(j, "small", False) and pe_of(j) not in _UBIQUITIN_STIDS + and not (join_members and a_u in pool_members.get(j, ()))] + if regen_by_protein: + # specs/048 B2: by the proteins carried, in any form + consumed |= {p for j in eaten for p in proteins(j)} + produced |= {p for o in outs.get(rx_u, {}) for p in proteins(o)} + else: + consumed |= {ent.get(j, "") for j in eaten} + produced |= {ent.get(o, "") for o in outs.get(rx_u, {})} if consumed <= produced: return True local["nonregenerating_paths"] += 1 @@ -4184,6 +4256,9 @@ def regenerates(path) -> bool: by_sig.setdefault(sig(u, r), []).append(u) if len(by_sig) < 2: local["carrier_loops"] += 1 + if not removed: + first_loops.append((r, frozenset(scc), core, + {x for a_u, b_u, x, _, _ in by_r[r] if a_u in scc and b_u in scc})) continue states: Set[str] = set() for members in by_sig.values(): @@ -4259,6 +4334,7 @@ def walk(u, so_far, seen): removed: Set[str] = set() dropped_shared = 0 + keep_shared = env_flag("LNG_POOL_SHARED") while True: pools, local = build(removed) claimed: Dict[str, int] = {} @@ -4266,6 +4342,12 @@ def walk(u, so_far, seen): for n in p["nodes"]: claimed[n] = claimed.get(n, 0) + 1 shared = {n for n, c in claimed.items() if c > 1} + if keep_shared: + # deltasignal specs/049: an intermediate or step reaction node that + # two pools pass through (MAPK1 and MAPK3 through one capped + # RAF:MAP2K:MAPKs complex) stays in both; the consumer writes it from + # the flux of every pool through it. A shared STATE still goes. + shared = {n for n in shared if any(n in p["states"] for p in pools)} if not shared: break removed |= shared @@ -4397,6 +4479,45 @@ def acts_on(u: str) -> Set[str]: stats["copy_rows"] += sum(len(cp) for path in paths for *_, cp in path) stats["multi_step_pools"] += int(any(len(path) > 1 for path in paths)) stats["carriers"] += len({j for j, _ in found}) + if carrier_loops is not None: + # deltasignal specs/049: a carrier loop's free form reads its supply, not + # its own release; the consumer drops these in-edges (all of them for a + # closed loop, whose free form then reads baseline) + pool_carriers = {c for _, c, _ in carriers} + merged: Dict[frozenset, Tuple[Any, frozenset, Set[str]]] = {} + for r, scc, core, rx in first_loops: + if scc in merged: + merged[scc][2].update(rx) + else: + merged[scc] = (r, core, set(rx)) + # one row per (free form, loop source); a pair in several loops (one per + # R) keeps the first loop's id and is closed if ANY of its loops is + rows: Dict[Tuple[str, str], List[Any]] = {} + loop_conflicts: Set[str] = set() + order = sorted(merged, key=lambda f: min(ent.get(u, "") + u for u in f)) + for i, scc in enumerate(order, start=1): + r, core, rx = merged[scc] + # a step that writes a plain node handing on to the loop's forms by + # variant_split (a capped release, under LNG_POOL_JOIN_MEMBERS) is + # inside the loop, and so is that plain node + inside = set(scc) | rx | {pn for x in rx for pn in outs.get(x, {}) if split_to.get(pn, set()) & scc} + closed = not any(into.get(u, set()) - inside for u in scc) + stats["carrier_loop_sets"] += 1 + stats["carrier_loops_closed"] += int(closed) + for u in sorted(scc, key=lambda u: (ent.get(u, ""), u)): + if proteins(u) != core: + continue + if u in managed or u in pool_carriers: + loop_conflicts.add(u) + continue + for src in sorted(into.get(u, set()) & inside): + if (u, src) in rows: + rows[(u, src)][2] = rows[(u, src)][2] or closed + else: + rows[(u, src)] = [f"loop{i}", r, closed, u, ent.get(u, ""), src] + stats["carrier_loop_conflicts"] = len(loop_conflicts) + stats["carrier_loop_forms"] = len({u for u, _ in rows}) + carrier_loops.extend(tuple(v) for v in rows.values()) return forms, transitions, carriers @@ -4433,7 +4554,7 @@ def r_steps(participants, profiles: Dict[str, Dict[str, Any]], ubiquitin_stids=( 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) -> None: + carriers_file: str, carrier_loops_file: Optional[str] = None) -> None: """Write pools.csv, pool_transitions.csv and pool_carriers.csv (deltasignal specs/039 amendment 2): each protein's modification-cycle pool, for a consumer that solves it at steady state (pi = pi P) instead of multiplying @@ -4457,12 +4578,16 @@ def export_pools(pathway_id: str, pathway_logic_network: pd.DataFrame, reaction_ donors = get_donor_reactions({s[0] for s in steps}, _UBIQUITIN_STIDS) set_members = get_set_descendants({x for s in steps for x in (s[2], s[3])}) stats: Dict[str, int] = {} + loops: List[Tuple] = [] forms, transitions, carriers = find_pools(pathway_logic_network, reaction_id_map, uuid_mapping, steps, - profiles, donors, set_members, stats) + profiles, donors, set_members, stats, carrier_loops=loops) 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) pd.DataFrame(carriers, columns=["pool_id", "carrier_uuid", "release_reaction_uuid"]).to_csv(carriers_file, index=False) + if carrier_loops_file is not None: + pd.DataFrame(loops, columns=["loop_id", "r", "closed", "free_uuid", "free_stid", "loop_source_uuid"]).to_csv( + carrier_loops_file, index=False) logger.info( f"Exported {stats['pools']} pools ({stats['states']} states, {stats['intermediates']} intermediates, " f"{stats['paths']} paths in {stats['copy_rows']} step-copy rows, {stats['multi_step_pools']} multi-step; " @@ -4472,7 +4597,9 @@ def export_pools(pathway_id: str, pathway_logic_network: pd.DataFrame, reaction_ f"{stats['shared_nodes_dropped']} shared nodes dropped, {stats['oriented_by_activity']} oriented by the active " f"state, {stats['activity_fallbacks']} by residues/donor/components of which {stats['ties']} by tie-break, " f"{stats['carriers']} carriers ({stats['carrier_conflicts']} excluded as pool nodes), " - f"{stats['carrier_loops']} carrier loops, " + f"{stats['carrier_loops']} carrier loops ({stats['carrier_loop_sets']} distinct, " + f"{stats['carrier_loops_closed']} closed, {stats['carrier_loop_forms']} free forms listed, " + f"{stats['carrier_loop_conflicts']} excluded as pool nodes), " f"{stats['merged_proteins']} co-travelling proteins merged, {stats['no_path_pools']} candidates with no path, " f"{stats['off_path_states']}/{stats['off_path_intermediates']} states/intermediates off every path, " f"{stats['ambiguous_copies']} ambiguous reaction copies, {stats['budget_dropped']} over the path budget; " diff --git a/src/pathway_generator.py b/src/pathway_generator.py index d5260f7..d80164e 100755 --- a/src/pathway_generator.py +++ b/src/pathway_generator.py @@ -43,6 +43,9 @@ "LNG_SET_POOL", "LNG_CAP_POOLS", "LNG_POOL_ACTIVE_VIA", + "LNG_POOL_JOIN_MEMBERS", + "LNG_POOL_REGEN", + "LNG_POOL_SHARED", "LNG_BIND_STOICH", "LNG_PATHOGEN_PROTEIN", "LNG_VARIANT_NODES", @@ -223,7 +226,7 @@ def _write_cache_fingerprint( "nodes.csv", "node_reaction_context.csv", "node_resolution.csv", "node_exclusions.csv", "boundary_edges.csv", "cofactors.csv", "containment.csv", "containment_structure.csv", "drugs.csv", "pathogens.csv", - "pools.csv", "pool_transitions.csv", "pool_carriers.csv", + "pools.csv", "pool_transitions.csv", "pool_carriers.csv", "carrier_loops.csv", ) @@ -590,6 +593,7 @@ def generate_pathway_file( str(pathway_output_dir / "pools.csv"), str(pathway_output_dir / "pool_transitions.csv"), str(pathway_output_dir / "pool_carriers.csv"), + str(pathway_output_dir / "carrier_loops.csv"), ) except Exception as e: logger.error(f"Failed to write pools.csv: {e}", exc_info=True) diff --git a/tests/test_pools.py b/tests/test_pools.py index 96957e5..bfa0b82 100644 --- a/tests/test_pools.py +++ b/tests/test_pools.py @@ -662,3 +662,173 @@ def test_unknown_pool_active_via_is_a_startup_error(monkeypatch): m.find_pools(*two_state()[:3], two_state()[4], two_state()[3]) monkeypatch.setenv("LNG_POOL_ACTIVE_VIA", "made_from") m._reject_removed_env() + + +# --- carrier loops (deltasignal specs/049) ------------------------------------------------------------------ + +def enzyme_loop(synthesis=False): + edges = [e("e", "v1", "input"), e("v1", "se", "output"), e("se", "v3", "input"), e("v3", "e", "output")] + rxs = ["v1", "v3"] + if synthesis: # the enzyme is also made outside its loop + edges.append(e("v5", "e", "output")) + rxs.append("v5") + net, rmap, umap = network(edges, rxs, {"e": "R-HSA-E", "se": "R-HSA-SE"}) + profiles = {"R-HSA-E": prof([KIN]), "R-HSA-SE": prof([KIN, SUB], slots=2, comps=2)} + steps = [("R-HSA-v1", KIN, "R-HSA-E", "R-HSA-SE", True), ("R-HSA-v3", KIN, "R-HSA-SE", "R-HSA-E", False)] + return net, rmap, umap, profiles, steps + + +def test_a_closed_carrier_loop_lists_its_free_form_and_the_release_into_it(): + net, rmap, umap, profiles, steps = enzyme_loop() + st, loops = {}, [] + assert m.find_pools(net, rmap, umap, steps, profiles, set(), {}, st, carrier_loops=loops) == ([], [], []) + assert loops == [("loop1", KIN, True, U["e"], "R-HSA-E", U["v3"])] # E:S is bound, not free + assert st["carrier_loop_sets"] == 1 and st["carrier_loops_closed"] == 1 + assert st["carrier_loop_forms"] == 1 and st["carrier_loop_conflicts"] == 0 + + +def test_a_fed_carrier_loop_lists_only_the_in_edges_from_inside_the_loop(): + net, rmap, umap, profiles, steps = enzyme_loop(synthesis=True) + st, loops = {}, [] + m.find_pools(net, rmap, umap, steps, profiles, set(), {}, st, carrier_loops=loops) + assert loops == [("loop1", KIN, False, U["e"], "R-HSA-E", U["v3"])] # synthesis (v5) is supply, kept + assert st["carrier_loops_closed"] == 0 + + +def test_a_carrier_loop_free_form_that_a_pool_manages_is_not_listed(): + # the GAP's own loop (GAP -> RAS:GTP:GAP -> GAP); the GAP is the RAS pool's carrier + net, rmap, umap, profiles, participants = ras_network() + st, loops = {}, [] + _, _, carriers = m.find_pools(net, rmap, umap, m.r_steps(participants, profiles), profiles, + {"R-HSA-vgef"}, {}, st, carrier_loops=loops) + assert carriers == [("pool1", U["gap"], U["vrel"])] + assert loops == [] and st["carrier_loop_sets"] == 1 and st["carrier_loop_conflicts"] == 1 + + +def test_carrier_loops_are_not_collected_unless_asked_and_change_nothing_else(): + net, rmap, umap, profiles, participants = ring_network() + steps = m.r_steps(participants, profiles) + st0, st1, loops = {}, {}, [] + assert m.find_pools(net, rmap, umap, steps, profiles, set(), {}, st0) == \ + m.find_pools(net, rmap, umap, steps, profiles, set(), {}, st1, carrier_loops=loops) + assert {k: v for k, v in st0.items() if not k.startswith("carrier_loop")} == \ + {k: v for k, v in st1.items() if not k.startswith("carrier_loop")} + # the kinase and the phosphatase loops are the pool's carriers: conflicts, not listed + assert loops == [] and st1["carrier_loop_conflicts"] == 2 + + +# --- pools that share an intermediate (deltasignal specs/049, LNG_POOL_SHARED) ------------------------------ + +def two_pools_one_complex(): + # S1 + S2 -> S1:S2 -> pS1 + pS2; pS1 -> S1 and pS2 -> S2 by their phosphatases. + # Both pools pass through the one complex node (MAPK1 and MAPK3 through a + # capped RAF:MAP2K:MAPKs complex). + S2P = 20 + edges = [e("s", "va", "input"), e("y", "va", "input"), e("va", "xy", "output"), + e("xy", "vb", "input"), e("vb", "sx", "output"), e("vb", "ys", "output"), + e("sx", "vc", "input"), e("vc", "s", "output"), e("p", "vc", "catalyst"), + e("ys", "vd", "input"), e("vd", "y", "output"), e("p", "vd", "catalyst")] + net, rmap, umap = network(edges, ["va", "vb", "vc", "vd"], + {"s": "R-HSA-S1", "sx": "R-HSA-S1P", "y": "R-HSA-S2", "ys": "R-HSA-S2P", + "xy": "R-HSA-S1S2", "p": "R-HSA-P"}) + ph = [("res", "p")] + profiles = {"R-HSA-S1": prof([SUB]), "R-HSA-S1P": prof([SUB], ph, mods=1), + "R-HSA-S2": prof([S2P]), "R-HSA-S2P": prof([S2P], ph, mods=1), "R-HSA-P": prof([PHOS]), + "R-HSA-S1S2": {"proteins": {SUB, S2P}, "fixed": {SUB, S2P}, "mods": 0, "comps": 2, "slots": 2, + "sig": {SUB: frozenset(), S2P: frozenset()}}} + steps = [("R-HSA-va", SUB, "R-HSA-S1", "R-HSA-S1S2", True), ("R-HSA-va", S2P, "R-HSA-S2", "R-HSA-S1S2", True), + ("R-HSA-vb", SUB, "R-HSA-S1S2", "R-HSA-S1P", False), ("R-HSA-vb", S2P, "R-HSA-S1S2", "R-HSA-S2P", False), + ("R-HSA-vc", SUB, "R-HSA-S1P", "R-HSA-S1", True), ("R-HSA-vd", S2P, "R-HSA-S2P", "R-HSA-S2", True)] + return net, rmap, umap, steps, profiles + + +def test_pools_sharing_an_intermediate_are_dropped_by_default(monkeypatch): + monkeypatch.setenv("LNG_POOL_REGEN", "protein") # the partner comes back phosphorylated + monkeypatch.delenv("LNG_POOL_SHARED", raising=False) + st = {} + forms, _, _ = m.find_pools(*two_pools_one_complex(), set(), {}, st) + assert {u for _, u, *_ in forms} & {U["xy"]} == set() + assert st["shared_nodes_dropped"] > 0 + + +def test_pools_sharing_an_intermediate_both_keep_it_under_lng_pool_shared(monkeypatch): + monkeypatch.setenv("LNG_POOL_REGEN", "protein") + monkeypatch.setenv("LNG_POOL_SHARED", "1") + st = {} + forms, trans, _ = m.find_pools(*two_pools_one_complex(), set(), {}, st) + assert st["pools"] == 2 and st["shared_nodes_dropped"] == 0 + owners = {pid for pid, u, _, role, _ in forms if u == U["xy"] and role == "intermediate"} + assert len(owners) == 2 # one complex, an intermediate of both + assert {pid for pid, *_, rx, _, _ in trans if rx == U["va"]} == owners # and both pools step through va + + + +def test_a_carrier_loop_released_through_a_variant_split_is_closed(monkeypatch): + # under LNG_POOL_JOIN_MEMBERS the release writes a plain node x that hands the + # free form on by variant_split (a capped release); x is inside the loop, and + # the edge the solver must drop is x -> e + monkeypatch.setenv("LNG_POOL_JOIN_MEMBERS", "1") + edges = [e("e", "v1", "input"), e("v1", "se", "output"), e("se", "v3", "input"), e("v3", "x", "output"), + e("x", "e", "variant_split")] + net, rmap, umap = network(edges, ["v1", "v3"], {"e": "R-HSA-E", "se": "R-HSA-SE", "x": "R-HSA-E"}) + profiles = {"R-HSA-E": prof([KIN]), "R-HSA-SE": prof([KIN, SUB], slots=2, comps=2)} + steps = [("R-HSA-v1", KIN, "R-HSA-E", "R-HSA-SE", True), ("R-HSA-v3", KIN, "R-HSA-SE", "R-HSA-E", False)] + st, loops = {}, [] + m.find_pools(net, rmap, umap, steps, profiles, set(), {}, st, carrier_loops=loops) + assert st["carrier_loops_closed"] == 1 + assert loops == [("loop1", KIN, True, U["e"], "R-HSA-E", U["x"])] + + +def test_a_carrier_loop_whose_bound_form_is_split_is_closed(monkeypatch): + # the binding step writes the plain complex se, split to its variant sxe + monkeypatch.setenv("LNG_POOL_JOIN_MEMBERS", "1") + edges = [e("e", "v1", "input"), e("v1", "se", "output"), e("se", "sxe", "variant_split"), + e("sxe", "v3", "input"), e("v3", "e", "output")] + net, rmap, umap = network(edges, ["v1", "v3"], {"e": "R-HSA-E", "se": "R-HSA-SE", "sxe": "R-HSA-SE"}) + profiles = {"R-HSA-E": prof([KIN]), "R-HSA-SE": prof([KIN, SUB], slots=2, comps=2)} + steps = [("R-HSA-v1", KIN, "R-HSA-E", "R-HSA-SE", True), ("R-HSA-v3", KIN, "R-HSA-SE", "R-HSA-E", False)] + st, loops = {}, [] + m.find_pools(net, rmap, umap, steps, profiles, set(), {}, st, carrier_loops=loops) + assert st["carrier_loops_closed"] == 1 + assert loops == [("loop1", KIN, True, U["e"], "R-HSA-E", U["v3"])] + + +def test_a_fed_loop_with_two_releases_lists_both_and_counts_the_free_form_once(): + # E binds S and binds Q; both bound forms release E. One R, so one strongly + # connected set, fed because v5 makes E:Q from outside. (The cross-loop + # "closed if any loop is" merge needs one free form in two different sets, + # one per R; it is not reachable with a single-protein carrier.) + edges = [e("e", "v1", "input"), e("v1", "se", "output"), e("se", "v3", "input"), e("v3", "e", "output"), + e("e", "v2", "input"), e("v2", "xy", "output"), e("xy", "v4", "input"), e("v4", "e", "output"), + e("v5", "xy", "output")] + net, rmap, umap = network(edges, ["v1", "v2", "v3", "v4", "v5"], + {"e": "R-HSA-E", "se": "R-HSA-SE", "xy": "R-HSA-EQ"}) + profiles = {"R-HSA-E": prof([KIN]), "R-HSA-SE": prof([KIN, SUB], slots=2, comps=2), + "R-HSA-EQ": prof([KIN, PHOS], slots=2, comps=2)} + steps = [("R-HSA-v1", KIN, "R-HSA-E", "R-HSA-SE", True), ("R-HSA-v3", KIN, "R-HSA-SE", "R-HSA-E", False), + ("R-HSA-v2", KIN, "R-HSA-E", "R-HSA-EQ", True), ("R-HSA-v4", KIN, "R-HSA-EQ", "R-HSA-E", False)] + st, loops = {}, [] + m.find_pools(net, rmap, umap, steps, profiles, set(), {}, st, carrier_loops=loops) + # one R, so one strongly connected set {e, se, xy}: fed (v5 makes xy from outside) + assert st["carrier_loop_sets"] == 1 and st["carrier_loops_closed"] == 0 + assert sorted(src for *_, src in loops) == sorted([U["v3"], U["v4"]]) + assert st["carrier_loop_forms"] == 1 + + +def test_a_pool_member_input_is_exempt_from_regeneration_only_under_join_members(monkeypatch): + # a step reads its source s and also y, a pool node s feeds: before specs/048 + # B1 y was a consumed partner (the path does not regenerate), and at the + # defaults it still is + net, rmap, umap, profiles, participants = ring_network() + net = pd.concat([net, pd.DataFrame([e("y", "v1", "input"), e("s", "y", "variant_pool")])], ignore_index=True) + umap[U["y"]] = "R-HSA-Y" + profiles["R-HSA-Y"] = prof([20]) + steps = m.r_steps(participants, profiles) + monkeypatch.delenv("LNG_POOL_JOIN_MEMBERS", raising=False) + st = {} + m.find_pools(net, rmap, umap, steps, profiles, set(), {}, st) + assert st["pools"] == 0 and st["nonregenerating_paths"] >= 1 + monkeypatch.setenv("LNG_POOL_JOIN_MEMBERS", "1") + st = {} + m.find_pools(net, rmap, umap, steps, profiles, set(), {}, st) + assert st["pools"] == 1