93 lines
2.8 KiB
Python
93 lines
2.8 KiB
Python
"""Deterministic topological sort with cycle detection.
|
|
|
|
Public API:
|
|
topo_sort(nodes, edges) -> (order, cycle)
|
|
|
|
- nodes: iterable of hashable node labels.
|
|
- edges: iterable of (u, v) meaning u must come before v.
|
|
- Returns (order, None) for a DAG: `order` contains every node exactly once
|
|
in a topologically valid order (ties broken by node ordering for determinism).
|
|
- Returns (order, cycle) when a cycle exists: `order` contains the nodes that
|
|
could be ordered, `cycle` is a list of node labels forming a directed cycle
|
|
(last -> first edge closes the cycle).
|
|
- Raises ValueError on unknown nodes in edges or duplicate node labels.
|
|
"""
|
|
|
|
import heapq
|
|
import sys
|
|
|
|
|
|
def topo_sort(nodes, edges):
|
|
node_list = list(nodes)
|
|
if len(node_list) != len(set(node_list)):
|
|
raise ValueError("duplicate node labels")
|
|
adj = {n: [] for n in node_list}
|
|
indeg = {n: 0 for n in node_list}
|
|
seen = set()
|
|
for u, v in edges:
|
|
if u not in adj or v not in adj:
|
|
raise ValueError("unknown node in edge: %r -> %r" % (u, v))
|
|
if (u, v) in seen:
|
|
continue
|
|
seen.add((u, v))
|
|
adj[u].append(v)
|
|
indeg[v] += 1
|
|
|
|
# Kahn's algorithm with a min-heap for deterministic tie-breaking.
|
|
q = [n for n in node_list if indeg[n] == 0]
|
|
heapq.heapify(q)
|
|
order = []
|
|
work = dict(indeg)
|
|
while q:
|
|
u = heapq.heappop(q)
|
|
order.append(u)
|
|
for v in sorted(adj[u]):
|
|
work[v] -= 1
|
|
if work[v] == 0:
|
|
heapq.heappush(q, v)
|
|
|
|
if len(order) == len(node_list):
|
|
return order, None
|
|
|
|
remaining = set(n for n in node_list if n not in set(order))
|
|
# subgraph induced by remaining nodes
|
|
radj = {n: [] for n in remaining}
|
|
for u in remaining:
|
|
for v in adj[u]:
|
|
if v in remaining:
|
|
radj[u].append(v)
|
|
|
|
sys.setrecursionlimit(max(sys.getrecursionlimit(), len(remaining) + 100))
|
|
color = {n: 0 for n in remaining} # 0=unvisited,1=visiting,2=done
|
|
parent = {n: None for n in remaining}
|
|
cycle = [None]
|
|
|
|
def dfs(u):
|
|
color[u] = 1
|
|
for v in radj[u]:
|
|
if color[v] == 0:
|
|
parent[v] = u
|
|
if dfs(v):
|
|
return True
|
|
elif color[v] == 1:
|
|
# back edge u -> v: cycle is v .. u -> v
|
|
chain = []
|
|
cur = u
|
|
while cur != v and cur is not None:
|
|
chain.append(cur)
|
|
cur = parent[cur]
|
|
chain.append(v)
|
|
chain.reverse()
|
|
cycle[0] = chain
|
|
return True
|
|
color[u] = 2
|
|
return False
|
|
|
|
for s in sorted(remaining):
|
|
if color[s] == 0:
|
|
parent[s] = None
|
|
if dfs(s):
|
|
break
|
|
|
|
return order, cycle[0]
|