Source code for wholecell.utils.dependency_graph

"""Tools for working with dependency graphs"""

from typing import Iterable


[docs] class InvalidDependencyGraphError(Exception): """Exception for invalid dependency graphs""" pass
[docs] class ExplorationStatus(object): #: Node has not yet been visited. UNEXPLORED = 1 #: Node has been visited but not added to list. This means that we #: have not yet finished with the node. EXPLORING = 2 #: Node has been visited and added to list. EXPLORED = 3
[docs] class DependencyGraph(object): """Represents a dependency graph Attributes: dependencies: Mapping from node to that node's dependents """ def __init__(self): # type: () -> None """Initialize dependencies to empty dictionary""" self.dependencies: dict[str, list[str]] = {}
[docs] def add_nodes(self, nodes: Iterable[str]): """Add nodes with no dependencies Arguments: nodes: Nodes to add """ for node in nodes: self.dependencies[node] = []
[docs] def add_dep_relation(self, a: str, b: str): """Add an edge such that a depends on b If a or b does not exist yet as a node, it will be created. Arguments: a: The name of the node that depends on b b: The name of the node that is depended-upon by a """ self.dependencies.setdefault(a, []).append(b) if b not in self.dependencies: self.dependencies[b] = []
[docs] def get_topological_ordering(self) -> list[str]: """Get a topological ordering of the nodes Returns: List of dependency names such that the dependencies can be loaded in the order in which they appear in the list without violating dependency relationships. Raises: InvalidDependencyGraphError: If the graph contains a cycle """ explored = {name: ExplorationStatus.UNEXPLORED for name in self.dependencies} reverse_ordering: list[str] = [] for node in self.dependencies: if explored[node] != ExplorationStatus.EXPLORED: self._topo_sort_dfs(node, explored, reverse_ordering) return reverse_ordering
[docs] def _topo_sort_dfs( self, node: str, explored: dict[str, int], reverse_ordering: list[str] ): explored[node] = ExplorationStatus.EXPLORING for dependency in self.dependencies[node]: if explored[dependency] == ExplorationStatus.UNEXPLORED: self._topo_sort_dfs(dependency, explored, reverse_ordering) elif explored[dependency] == ExplorationStatus.EXPLORING: # We have reached a node that we have already # visited, but that we have not added to the # ordering. Thus a call to dependency is still on # the stack. This means there exists a path from # dependency to node, so there is a cycle. raise InvalidDependencyGraphError( "Dependency graphs must not contain cycles" ) reverse_ordering.append(node) explored[node] = ExplorationStatus.EXPLORED