Coverage for graphqler / chains / strategies / dfs_strategy.py: 37%

1"""DFS-based chain generation strategy.""" 

2 

3import networkx 

4 

5from graphqler.chains.chain import Chain 

6from graphqler.chains.strategies.base_strategy import BaseChainStrategy 

7from graphqler.graph.node import Node 

8 

9 

10class DFSChainStrategy(BaseChainStrategy): 

11 """Generates chains via depth-first search from the starter nodes. 

12 

13 For every node visited during DFS, the current prefix path is recorded as a :class:`Chain`. 

14 This means a path A -> B -> C produces three chains: [A], [A, B], [A, B, C]. 

15 

16 Cycles are avoided by tracking the nodes already present in the current path. 

17 

18 Nodes whose ``mutation_type`` is in *filter_mutation_type* are excluded from chains; 

19 DFS stops at those nodes and does not recurse into their subtrees. 

20 """ 

21 

22 def generate(self, graph: networkx.DiGraph, starter_nodes: list[Node], 

23 filter_mutation_type: list[str] | None = None) -> list[Chain]: 

24 """Run DFS from each starter node and collect all prefix chains. 

25 

26 Args: 

27 graph (networkx.DiGraph): The dependency graph. 

28 starter_nodes (list[Node]): Nodes to begin DFS from. 

29 filter_mutation_type (list[str] | None): Mutation types to exclude. 

30 Nodes whose ``mutation_type`` is in this list (and their entire subtrees) 

31 will be skipped. Pass ``None`` or ``[]`` to include all nodes. 

32 

33 Returns: 

34 list[Chain]: All prefix chains discovered during DFS. 

35 """ 

36 excluded = set(filter_mutation_type) if filter_mutation_type else set() 

37 chains: list[Chain] = [] 

38 for start_node in starter_nodes: 

39 self._dfs(graph, start_node, [], chains, excluded) 

40 return chains 

41 

42 def _dfs(self, graph: networkx.DiGraph, node: Node, current_path: list[Node], 

43 chains: list[Chain], excluded: set[str]) -> None: 

44 """Recursively performs DFS, appending a Chain for every prefix path. 

45 

46 If a node's ``mutation_type`` is in *excluded*, it and its entire subtree are skipped. 

47 

48 Args: 

49 graph (networkx.DiGraph): The dependency graph. 

50 node (Node): The current node being visited. 

51 current_path (list[Node]): Nodes visited so far on this path (not including *node*). 

52 chains (list[Chain]): Accumulator for discovered chains. 

53 excluded (set[str]): Mutation types to skip. 

54 """ 

55 if node.mutation_type in excluded: 

56 return # stop recursion — this node and its subtree are excluded 

57 

58 new_path = current_path + [node] 

59 chains.append(Chain(nodes=new_path)) 

60 

61 for neighbor in graph.successors(node): 

62 if neighbor not in new_path: # avoid cycles 

63 self._dfs(graph, neighbor, new_path, chains, excluded)