john/daglib
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

subgraph isolation

Browse Source
This commit is contained in:
John Lancaster
2026-10-10 11:04:53 -05:00
parent 308c06c31d
commit 1d69539b9b
1 changed files with 5 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/daglib/dag.py
+5 -2
View File
@@ -23,7 +23,9 @@ class DAG[T](MutableMapping[T, DAGSetView[T]]):
""" """
_succ: dictset[T] = field(default_factory=lambda: defaultdict(set)) _succ: dictset[T] = field(default_factory=lambda: defaultdict(set))
"""Mapping of nodes to the set of nodes that succeed them in the DAG."""
_pred: dictset[T] = field(default_factory=lambda: defaultdict(set)) _pred: dictset[T] = field(default_factory=lambda: defaultdict(set))
"""Mapping of nodes to the set of nodes that precede them in the DAG."""
on_add: Callable[[T, T], None] | None = None on_add: Callable[[T, T], None] | None = None
on_remove: Callable[[T, T], None] | None = None on_remove: Callable[[T, T], None] | None = None
@@ -119,10 +121,11 @@ class DAG[T](MutableMapping[T, DAGSetView[T]]):
self._pred.pop(u, None) self._pred.pop(u, None)
def subgraph(self, sub: T | Iterable[T]) -> Self: def subgraph(self, sub: T | Iterable[T]) -> Self:
"""Return an independent graph induced by the successor closure of sub."""
closure = transitive_closure(self, sub) closure = transitive_closure(self, sub)
subgraph = type(self)( subgraph = type(self)(
_succ=defaultdict(set, {k: self._succ[k] for k in set(self._succ.keys()) & closure}), _succ=defaultdict(set, {node: self._succ[node] & closure for node in self._succ.keys() & closure}),
_pred=defaultdict(set, {k: self._pred[k] for k in set(self._pred.keys()) & closure}), _pred=defaultdict(set, {node: self._pred[node] & closure for node in self._pred.keys() & closure}),
) )
subgraph.on_add = self.on_add subgraph.on_add = self.on_add
subgraph.on_remove = self.on_remove subgraph.on_remove = self.on_remove