diff --git a/wizard-work/beam/bridges.py b/wizard-work/beam/bridges.py new file mode 100644 index 0000000..ffeafc0 --- /dev/null +++ b/wizard-work/beam/bridges.py @@ -0,0 +1,108 @@ +"""Undirected graph: bridges (cut edges) and articulation points (iterative Tarjan).""" +from __future__ import annotations + + +def _normalize(graph): + from collections import Counter + counts = Counter() + for u, nbrs in (graph or {}).items(): + for v in (nbrs or ()): + counts[frozenset((u, v))] += 1 + edge_list = [] + for e, c in counts.items(): + if c <= 1: + mult = 1 + elif c % 2 == 0: + mult = c // 2 + else: + mult = c + edge_list.extend([e] * mult) + # nodes + nodes = set((graph or {}).keys()) + for e in edge_list: + for n in e: + nodes.add(n) + # build adj with unique edge ids by index + adj = {n: [] for n in nodes} + for i, e in enumerate(edge_list): + els = tuple(e) + if len(els) == 1: + a = els[0] + adj[a].append((a, i)) + else: + a, b = els + adj[a].append((b, i)) + adj[b].append((a, i)) + return nodes, adj, edge_list + + +def analyze(graph): + nodes, adj, edge_list = _normalize(graph) + if not nodes: + return set(), set() + + disc = {} + low = {} + timer = [0] + bridges = set() + art = set() + children_root = {} + + for start in sorted(nodes, key=lambda x: (type(x).__name__, repr(x))): + if start in disc: + continue + root = start + children_root[root] = 0 + disc[root] = low[root] = timer[0] + timer[0] += 1 + stack = [root] + parent_node = {root: None} + parent_edge = {root: None} + frames = {root: 0} # next child index + cur = root + while True: + if frames[cur] < len(adj[cur]): + nbr, eid_ = adj[cur][frames[cur]] + frames[cur] += 1 + if eid_ == parent_edge[cur]: + continue + if nbr not in disc: + parent_node[nbr] = cur + parent_edge[nbr] = eid_ + disc[nbr] = low[nbr] = timer[0] + timer[0] += 1 + stack.append(nbr) + frames[nbr] = 0 + if cur == root: + children_root[root] += 1 + cur = nbr + else: + if nbr in stack: + low[cur] = min(low[cur], disc[nbr]) + else: + stack.pop() + if stack: + pnode = stack[-1] + low[pnode] = min(low[pnode], low[cur]) + if low[cur] > disc[pnode]: + bridges.add(edge_list[parent_edge[cur]]) + if pnode != root and low[cur] >= disc[pnode]: + art.add(pnode) + del frames[cur] + if not stack: + break + cur = stack[-1] + if children_root.get(root, 0) > 1: + art.add(root) + + return bridges, art + + +def bridges(graph): + b, _ = analyze(graph) + return b + + +def articulation_points(graph): + _, a = analyze(graph) + return a diff --git a/wizard-work/beam/test_bridges.py b/wizard-work/beam/test_bridges.py new file mode 100644 index 0000000..65d92b1 --- /dev/null +++ b/wizard-work/beam/test_bridges.py @@ -0,0 +1,137 @@ +import random +from collections import deque +from bridges import analyze + + +def _components(nodes, edges): + adj = {n: [] for n in nodes} + for e in edges: + els = tuple(e) + if len(els) < 2: + continue + a, b = els + adj[a].append(b); adj[b].append(a) + seen = set(); comp = 0 + for n in nodes: + if n in seen: continue + comp += 1 + dq = deque([n]); seen.add(n) + while dq: + u = dq.popleft() + for w in adj[u]: + if w not in seen: + seen.add(w); dq.append(w) + return comp + + +def brute_bridges(graph): + from bridges import _normalize + nodes, adj_ids, edge_list = _normalize(graph) + adj = {n: [] for n in nodes} + for e in edge_list: + els = tuple(e) + if len(els) < 2: continue + a, b = els + adj[a].append(b); adj[b].append(a) + res = set() + for e in edge_list: + els = tuple(e) + if len(els) < 2: continue + a, b = els + if edge_list.count(e) > 1: + continue + remaining = [x for x in edge_list if x != e] + if _components(nodes, remaining) > _components(nodes, edge_list): + res.add(e) + return res + + +def brute_art(graph): + from bridges import _normalize + nodes, adj_ids, edge_list = _normalize(graph) + adj = {n: [] for n in nodes} + for e in edge_list: + els = tuple(e) + if len(els) < 2: continue + a, b = els + adj[a].append(b); adj[b].append(a) + res = set() + for node in nodes: + rest = [x for x in nodes if x != node] + if not rest: continue + def connected_original(x, y): + seen = set([x]); dq = deque([x]) + while dq: + u = dq.popleft() + if u == y: return True + for w in adj[u]: + if w not in seen: + seen.add(w); dq.append(w) + return False + is_art = False + for x in rest: + for y in rest: + if x >= y: continue + if not connected_original(x, y): continue + seen2 = set([node, x]); dq = deque([x]); ok = False + while dq: + u = dq.popleft() + if u == y: ok = True; break + for w in adj[u]: + if w not in seen2: + seen2.add(w); dq.append(w) + if not ok: + is_art = True; break + if is_art: break + if is_art: + res.add(node) + return res + + +def test_random_small(): + rng = random.Random(42) + for _ in range(3000): + n = rng.randint(1, 7) + nodes = list(range(n)) + adj = {i: [] for i in nodes} + m = rng.randint(0, n + 3) + for _ in range(m): + u = rng.randrange(n); v = rng.randrange(n) + adj[u].append(v); adj[v].append(u) + b, a = analyze(adj) + assert b == brute_bridges(adj), (adj, b, brute_bridges(adj)) + assert a == brute_art(adj), (adj, a, brute_art(adj)) + + +def test_path(): + g = {i: [i-1, i+1] for i in range(5)} + for i in g: g[i] = [x for x in g[i] if 0 <= x < 5] + b, a = analyze(g) + assert len(b) == 4 and a == {1,2,3} + + +def test_cycle(): + g = {0:[1,2],1:[0,2],2:[0,1]} + b,a = analyze(g) + assert b == set() and a == set() + + +def test_empty(): + assert analyze({}) == (set(), set()) + + +def test_single(): + assert analyze({0: []}) == (set(), set()) + + +def test_parallel_edges(): + g = {0: [1, 1], 1: [0, 0]} + b, a = analyze(g) + assert b == set() and a == set() + + +if __name__ == '__main__': + for name in ['test_random_small','test_path','test_cycle','test_empty','test_single','test_parallel_edges']: + globals()[name]() + print('PASS', name) + print('ALL OK') \ No newline at end of file