From 7570ad96f83e7bc76bdf42f99bc8538ca0211b7f Mon Sep 17 00:00:00 2001 From: = Date: Fri, 27 Sep 2024 19:47:21 -0700 Subject: [PATCH 1/8] pruned redundant paths in control succession search --- biobalm/control.py | 16 ++++++++++++++++ tests/control_test.py | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+) diff --git a/biobalm/control.py b/biobalm/control.py index 1f836ea..ccaa8c7 100644 --- a/biobalm/control.py +++ b/biobalm/control.py @@ -339,11 +339,24 @@ def successions_to_target( target=target, ) + found_valid_target_node = False + for s in succession_diagram.node_ids(): + # don't try to get to s if it's not in the target fixed_vars = succession_diagram.node_data(s)["space"] if not is_subspace(fixed_vars, target): continue + found_valid_target_node = True + # no need to specifically control to s if getting to any of its parents is sufficient + parents = succession_diagram.dag.predecessors(s) # type: ignore + if all( + is_subspace(succession_diagram.node_data(c)["space"], target) # type: ignore + for p in parents # type: ignore + for c in succession_diagram.dag.successors(p) # type: ignore + ): + continue + for path in cast( list[list[int]], nx.all_simple_paths( # type: ignore @@ -358,6 +371,9 @@ def successions_to_target( ] successions.append(succession) + if found_valid_target_node and len(successions) == 0: + successions = [[]] + return successions diff --git a/tests/control_test.py b/tests/control_test.py index 3f2f95d..0f7808a 100644 --- a/tests/control_test.py +++ b/tests/control_test.py @@ -305,3 +305,36 @@ def test_size_restriction(): sd, target, max_drivers_per_succession_node=1, successful_only=True ) assert len(interventions) == 0 + + +def test_no_control_needed(): + rules = """ + a, b + b, a | c + c, !a + """ + target: BooleanSpace = {"a": 1, "b": 1, "c": 0} + + sd = SuccessionDiagram.from_rules(rules) + sd.build() + interventions = succession_control(sd, target) + + assert len(interventions) == 1 + intervention = interventions[0] + assert intervention.successful + assert intervention.control == [] + + +def test_no_control_possible(): + rules = """ + a, b + b, a | c + c, !a + """ + target: BooleanSpace = {"a": 0, "b": 1, "c": 0} + + sd = SuccessionDiagram.from_rules(rules) + sd.build() + interventions = succession_control(sd, target) + + assert len(interventions) == 0 From b4e1a93ebbeaec7f7dad9d008ae8a7cbff3056e3 Mon Sep 17 00:00:00 2001 From: Samuel Pastva Date: Sat, 28 Sep 2024 15:08:35 +0200 Subject: [PATCH 2/8] Skip redundant paths that can be replaced by a simpler path. --- biobalm/control.py | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/biobalm/control.py b/biobalm/control.py index ccaa8c7..77c80b1 100644 --- a/biobalm/control.py +++ b/biobalm/control.py @@ -6,6 +6,7 @@ from __future__ import annotations from itertools import combinations, product +from functools import reduce from typing import Literal, cast import networkx as nx # type: ignore @@ -332,6 +333,11 @@ def successions_to_target( nested trap spaces that specify the target. """ successions: list[SubspaceSuccession] = [] + # Tracks the combined perturbation that needs to be applied + # for the whole succession to take effect. We use this to detect + # which successions are redundant when they can be replaced by + # a succession with a subset signature. + succession_signatures: list[BooleanSpace] = [] # expand the succession_diagram toward the target if expand_diagram: @@ -369,7 +375,28 @@ def successions_to_target( succession_diagram.edge_stable_motif(x, y, reduced=True) for x, y in zip(path[:-1], path[1:]) ] + signature = reduce(lambda x, y: x | y, succession) + # First, check if any existing successions can be eliminated + # because they are redundant w.r.t. to this succession. + # (`reversed` is important here, because that way a delete + # only impacts indices that we already processed) + skip_completely = False + for i in reversed(range(len(succession_signatures))): + existing_signature = succession_signatures[i] + if is_subspace(signature, existing_signature): + # The current `path` is already superseded by a path in successions. + skip_completely = True + print("Skipping", succession, path) + break + if is_subspace(existing_signature, signature): + # A path in successions is made redundant by the current path. + del succession_signatures[i] + del successions[i] + if skip_completely: + continue + successions.append(succession) + succession_signatures.append(signature) if found_valid_target_node and len(successions) == 0: successions = [[]] From bfec0c0962bcf3fc95c9b4baf5e380e41be713de Mon Sep 17 00:00:00 2001 From: Samuel Pastva Date: Sat, 28 Sep 2024 16:30:42 +0200 Subject: [PATCH 3/8] Remove redundant print. --- biobalm/control.py | 1 - 1 file changed, 1 deletion(-) diff --git a/biobalm/control.py b/biobalm/control.py index 77c80b1..8dc301f 100644 --- a/biobalm/control.py +++ b/biobalm/control.py @@ -386,7 +386,6 @@ def successions_to_target( if is_subspace(signature, existing_signature): # The current `path` is already superseded by a path in successions. skip_completely = True - print("Skipping", succession, path) break if is_subspace(existing_signature, signature): # A path in successions is made redundant by the current path. From c1eb7b5274d9b6fda6f7bc1fcf106cd2f96358f6 Mon Sep 17 00:00:00 2001 From: Samuel Pastva Date: Sat, 28 Sep 2024 17:15:47 +0200 Subject: [PATCH 4/8] Add the option to generate all combinations of control overrides for one intervention. --- biobalm/control.py | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/biobalm/control.py b/biobalm/control.py index 8dc301f..5e8725a 100644 --- a/biobalm/control.py +++ b/biobalm/control.py @@ -7,7 +7,7 @@ from itertools import combinations, product from functools import reduce -from typing import Literal, cast +from typing import Literal, cast, Iterator import networkx as nx # type: ignore from biodivine_aeon import AsynchronousGraph, BooleanNetwork @@ -159,6 +159,19 @@ def __str__(self): else: return "unknown strategy: " + self.__repr__() + def all_control_strategies(self) -> Iterator[ControlOverrides]: + """ + Returns all possible combinations of `ControlOverrides` sequences that + can be used to execute this `Intervention`. + + Internally, an intervention consists of multiple control steps that + need to be taken sequentially. For each step in the sequence, an intervention + can have multiple options of how to execute it. With this method, + we can generate the actual sequences that arise by combining all the + available options for each step. + """ + return map(lambda x: list(x), product(*self._control)) + def succession_control( succession_diagram: SuccessionDiagram, From 40bec57b766043191bd9c65315de0f36089368a7 Mon Sep 17 00:00:00 2001 From: = Date: Sat, 28 Sep 2024 08:24:46 -0700 Subject: [PATCH 5/8] better redundancy check --- biobalm/control.py | 39 +++++++++++++++++++++++++-------------- 1 file changed, 25 insertions(+), 14 deletions(-) diff --git a/biobalm/control.py b/biobalm/control.py index 77c80b1..4a916ea 100644 --- a/biobalm/control.py +++ b/biobalm/control.py @@ -5,14 +5,14 @@ from __future__ import annotations -from itertools import combinations, product from functools import reduce +from itertools import combinations, product from typing import Literal, cast import networkx as nx # type: ignore from biodivine_aeon import AsynchronousGraph, BooleanNetwork -from biobalm.space_utils import is_subspace, percolate_space +from biobalm.space_utils import intersect, is_subspace, percolate_space from biobalm.succession_diagram import SuccessionDiagram from biobalm.types import BooleanSpace, ControlOverrides, SubspaceSuccession @@ -345,21 +345,32 @@ def successions_to_target( target=target, ) - found_valid_target_node = False + # these contradict the target or have a motif avoidant attractor that isn't + # in full agreement with the target + hot_lava_nodes: set[int] = set() + descendant_map: dict[int, set[int]] = {} for s in succession_diagram.node_ids(): - # don't try to get to s if it's not in the target - fixed_vars = succession_diagram.node_data(s)["space"] - if not is_subspace(fixed_vars, target): + is_consistent = intersect(succession_diagram.node_data(s)["space"], target) + is_goal = is_subspace(succession_diagram.node_data(s)["space"], target) + is_minimal = succession_diagram.node_is_minimal(s) + if not is_consistent or (not is_goal and is_minimal): + hot_lava_nodes.add(s) + + descendant_map[s] = set(nx.descendants(succession_diagram.dag, s)) # type: ignore + descendant_map[s].add(s) # for our purposes, s is its own descendant + found_valid_target_node = False + for s in succession_diagram.node_ids(): + # a node is a valid end point if all + # 1) its descendents (including itself) are not "hot lava" + # 2) it has a parent that is or can reach "hot lava" (otherwise, we can just control to the parent) + # note that all nodes are either cold lava or hot lava, but not both or neither + if descendant_map[s] & hot_lava_nodes: continue - found_valid_target_node = True - # no need to specifically control to s if getting to any of its parents is sufficient - parents = succession_diagram.dag.predecessors(s) # type: ignore - if all( - is_subspace(succession_diagram.node_data(c)["space"], target) # type: ignore - for p in parents # type: ignore - for c in succession_diagram.dag.successors(p) # type: ignore + if not any( + descendant_map[p] & hot_lava_nodes + for p in succession_diagram.dag.predecessors(s) # type: ignore ): continue @@ -393,7 +404,7 @@ def successions_to_target( del succession_signatures[i] del successions[i] if skip_completely: - continue + pass successions.append(succession) succession_signatures.append(signature) From d978daa1636af27c73c5180060f080a63ca8b14d Mon Sep 17 00:00:00 2001 From: = Date: Sat, 28 Sep 2024 08:35:50 -0700 Subject: [PATCH 6/8] made succession skipping optional --- biobalm/control.py | 69 ++++++++++++++++++++++++++-------------------- 1 file changed, 39 insertions(+), 30 deletions(-) diff --git a/biobalm/control.py b/biobalm/control.py index c4491fd..808a847 100644 --- a/biobalm/control.py +++ b/biobalm/control.py @@ -317,16 +317,17 @@ def successions_to_target( succession_diagram: SuccessionDiagram, target: BooleanSpace, expand_diagram: bool = True, + skip_feedforward_successions: bool = False, ) -> list[SubspaceSuccession]: """Find lists of nested trap spaces (successions) that lead to the specified target subspace. Generally, it is not necessary to call this function directly, as it is automatically invoked by - :func:`succession_control`. It is primarily - provided in the public API for testing and benchmarking purposes, or in the - case that the user wants to implement a custom strategy to identify - succession drivers rather than relying on + :func:`succession_control`. It is + primarily provided in the public API for testing and benchmarking purposes, + or in the case that the user wants to implement a custom strategy to + identify succession drivers rather than relying on :func:`drivers_of_succession`. Parameters @@ -338,6 +339,12 @@ def successions_to_target( expand_diagram: bool Whether to ensure that the succession diagram is expanded enough to capture all paths to the target (default: True). + skip_feedforward_successions: bool + Whether to skip redundant successions (default: False). Skipping these + can reduce the number of interventions to test, yielding performance + improvements, but can also cause the algorithm to miss some valid + interventions, particularly in cases when the order of intervention + application is important. Returns ------- @@ -346,10 +353,10 @@ def successions_to_target( nested trap spaces that specify the target. """ successions: list[SubspaceSuccession] = [] - # Tracks the combined perturbation that needs to be applied - # for the whole succession to take effect. We use this to detect - # which successions are redundant when they can be replaced by - # a succession with a subset signature. + # Tracks the combined perturbation that needs to be applied for the whole + # succession to take effect. We use this to detect which successions are + # redundant when they can be replaced by a succession with a subset + # signature. Used when skip_feedforward_successions is True succession_signatures: list[BooleanSpace] = [] # expand the succession_diagram toward the target @@ -376,8 +383,9 @@ def successions_to_target( for s in succession_diagram.node_ids(): # a node is a valid end point if all # 1) its descendents (including itself) are not "hot lava" - # 2) it has a parent that is or can reach "hot lava" (otherwise, we can just control to the parent) - # note that all nodes are either cold lava or hot lava, but not both or neither + # 2) it has a parent that is or can reach "hot lava" (otherwise, we can + # just control to the parent) note that all nodes are either cold lava + # or hot lava, but not both or neither if descendant_map[s] & hot_lava_nodes: continue found_valid_target_node = True @@ -399,27 +407,28 @@ def successions_to_target( succession_diagram.edge_stable_motif(x, y, reduced=True) for x, y in zip(path[:-1], path[1:]) ] - signature = reduce(lambda x, y: x | y, succession) - # First, check if any existing successions can be eliminated - # because they are redundant w.r.t. to this succession. - # (`reversed` is important here, because that way a delete - # only impacts indices that we already processed) - skip_completely = False - for i in reversed(range(len(succession_signatures))): - existing_signature = succession_signatures[i] - if is_subspace(signature, existing_signature): - # The current `path` is already superseded by a path in successions. - skip_completely = True - break - if is_subspace(existing_signature, signature): - # A path in successions is made redundant by the current path. - del succession_signatures[i] - del successions[i] - if skip_completely: - pass - + if skip_feedforward_successions: + signature = reduce(lambda x, y: x | y, succession) + # First, check if any existing successions can be eliminated + # because they are redundant w.r.t. to this succession. + # (`reversed` is important here, because that way a delete + # only impacts indices that we already processed) + skip_completely = False + for i in reversed(range(len(succession_signatures))): + existing_signature = succession_signatures[i] + if is_subspace(signature, existing_signature): + # The current `path` is already superseded by a path in successions. + skip_completely = True + break + if is_subspace(existing_signature, signature): + # A path in successions is made redundant by the current path. + del succession_signatures[i] + del successions[i] + if skip_completely: + continue + + succession_signatures.append(signature) successions.append(succession) - succession_signatures.append(signature) if found_valid_target_node and len(successions) == 0: successions = [[]] From cc85aa54f650cad5d02c731d18a77cb520be1be6 Mon Sep 17 00:00:00 2001 From: Samuel Pastva Date: Sat, 28 Sep 2024 19:39:28 +0200 Subject: [PATCH 7/8] Make the `skip` flag available in the main control method. --- biobalm/control.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/biobalm/control.py b/biobalm/control.py index 808a847..d6ff6df 100644 --- a/biobalm/control.py +++ b/biobalm/control.py @@ -180,6 +180,7 @@ def succession_control( max_drivers_per_succession_node: int | None = None, forbidden_drivers: set[str] | None = None, successful_only: bool = True, + skip_feedforward_successions: bool = False, ) -> list[Intervention]: """ Performs succession-diagram control to reach a target subspace. @@ -294,7 +295,10 @@ def succession_control( interventions: list[Intervention] = [] successions = successions_to_target( - succession_diagram, target=target, expand_diagram=True + succession_diagram, + target=target, + expand_diagram=True, + skip_feedforward_successions=skip_feedforward_successions, ) for succession in successions: From b83d9f7f13980861964d5cf8b3102c86522ad1af Mon Sep 17 00:00:00 2001 From: Samuel Pastva Date: Sat, 28 Sep 2024 20:03:25 +0200 Subject: [PATCH 8/8] Store all stable motifs for each edge. --- biobalm/control.py | 50 ++++++++++++++++++----------------- biobalm/succession_diagram.py | 20 +++++++++++++- 2 files changed, 45 insertions(+), 25 deletions(-) diff --git a/biobalm/control.py b/biobalm/control.py index d6ff6df..86ceb45 100644 --- a/biobalm/control.py +++ b/biobalm/control.py @@ -407,32 +407,34 @@ def successions_to_target( target=s, ), ): - succession = [ - succession_diagram.edge_stable_motif(x, y, reduced=True) + motif_list = [ + succession_diagram.edge_all_stable_motifs(x, y, reduced=True) for x, y in zip(path[:-1], path[1:]) ] - if skip_feedforward_successions: - signature = reduce(lambda x, y: x | y, succession) - # First, check if any existing successions can be eliminated - # because they are redundant w.r.t. to this succession. - # (`reversed` is important here, because that way a delete - # only impacts indices that we already processed) - skip_completely = False - for i in reversed(range(len(succession_signatures))): - existing_signature = succession_signatures[i] - if is_subspace(signature, existing_signature): - # The current `path` is already superseded by a path in successions. - skip_completely = True - break - if is_subspace(existing_signature, signature): - # A path in successions is made redundant by the current path. - del succession_signatures[i] - del successions[i] - if skip_completely: - continue - - succession_signatures.append(signature) - successions.append(succession) + for succession_tuple in product(*motif_list): + succession = list(succession_tuple) + if skip_feedforward_successions: + signature = reduce(lambda x, y: x | y, succession) + # First, check if any existing successions can be eliminated + # because they are redundant w.r.t. to this succession. + # (`reversed` is important here, because that way a delete + # only impacts indices that we already processed) + skip_completely = False + for i in reversed(range(len(succession_signatures))): + existing_signature = succession_signatures[i] + if is_subspace(signature, existing_signature): + # The current `path` is already superseded by a path in successions. + skip_completely = True + break + if is_subspace(existing_signature, signature): + # A path in successions is made redundant by the current path. + del succession_signatures[i] + del successions[i] + if skip_completely: + continue + + succession_signatures.append(signature) + successions.append(succession) if found_valid_target_node and len(successions) == 0: successions = [[]] diff --git a/biobalm/succession_diagram.py b/biobalm/succession_diagram.py index d74f6e8..282a6a1 100644 --- a/biobalm/succession_diagram.py +++ b/biobalm/succession_diagram.py @@ -1087,6 +1087,22 @@ def node_percolated_petri_net( return percolated_pn + def edge_all_stable_motifs( + self, parent_id: int, child_id: int, reduced: bool = False + ) -> list[BooleanSpace]: + """ + Similar to `edge_stable_motif`, but returns all motifs associated with an edge. + """ + all_motifs: list[BooleanSpace] = cast(list[BooleanSpace], self.dag.edges[parent_id, child_id]["all_motifs"]) # type: ignore + if reduced: + result: list[BooleanSpace] = [] + node_space = self.node_data(parent_id)["space"] + for m in all_motifs: + result.append({k: v for k, v in m.items() if k not in node_space}) + return result + else: + return all_motifs + def edge_stable_motif( self, parent_id: int, child_id: int, reduced: bool = False ) -> BooleanSpace: @@ -1641,5 +1657,7 @@ def _ensure_edge(self, parent_id: int, child_id: int, stable_motif: BooleanSpace # can be reached through multiple stable motifs. Not sure how to # approach these... but this is probably good enough for now. if not self.dag.has_edge(parent_id, child_id): # type: ignore - self.dag.add_edge(parent_id, child_id, motif=stable_motif) # type: ignore + self.dag.add_edge(parent_id, child_id, motif=stable_motif, all_motifs=[stable_motif]) # type: ignore + else: + self.dag.edges[parent_id, child_id]["all_motifs"].append(stable_motif) # type: ignore self._update_node_depth(child_id, parent_id)