Gomory-Hu tree algorithm in Python
Build a Gomory-Hu tree using n-1 max-flow computations.
- File
gomory_hu_algorithm.py, 137 lines- Level
- I0 not checked yet
- Toolchain
- Python 3.14, pytest 9
- Test framework
- pytest
- Run the tests
python3 -m pytest
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/gomory-hu-algorithm/. The authors are listed in prov/PSCBG5.jsonl.
from collections import deque
from fractions import Fraction
from typing import Callable, Optional, Union
Number = Union[int, Fraction, float]
def _min_cut(k: int, arcs: list[tuple[int, int, Number]], s: int, t: int) -> tuple[Number, set[int]]:
"""Minimum s-t cut of a directed network on 0 .. k-1 by shortest augmenting paths.
Returns (value, S) with S the vertices reachable from s in the final residual graph.
"""
head: list[int] = []
res: list[Number] = []
adj: list[list[int]] = [[] for _ in range(k)]
for u, v, cap in arcs:
adj[u].append(len(head))
head.append(v)
res.append(cap)
adj[v].append(len(head))
head.append(u)
res.append(0)
value: Number = 0
while True:
parent = [-1] * k
parent[s] = -2
queue = deque([s])
while queue:
u = queue.popleft()
for a in adj[u]:
v = head[a]
if res[a] > 0 and parent[v] == -1:
parent[v] = a
queue.append(v)
if parent[t] == -1:
return value, {v for v in range(k) if parent[v] != -1}
delta = None
v = t
while v != s:
a = parent[v]
delta = res[a] if delta is None else min(delta, res[a])
v = head[a ^ 1]
v = t
while v != s:
a = parent[v]
res[a] -= delta
res[a ^ 1] += delta
v = head[a ^ 1]
value += delta
def gomory_hu_algorithm(
n: int,
edges: list[tuple[int, int, Number]],
on_split: Optional[Callable[[int, int, Number, list[int], list[int]], None]] = None,
) -> list[tuple[int, int, Number]]:
"""Gomory-Hu cut tree of an undirected graph by the original contraction method.
The graph has the vertices 0 .. n-1 and the undirected edges (u, v, w), w >= 0
(int, Fraction or float). Parallel and zero-weight edges are allowed; self-loops
are ignored. Each undirected edge is sent to the flow routine as two arcs u->v
and v->u of capacity w.
Returns the n - 1 tree edges (a, b, lam) with a < b, sorted. For each pair u, v the
least lam on the tree path is the minimum u-v cut value, and deleting that tree
edge splits the vertices into a minimum u-v cut.
on_split(s, t, lam, xs, xt) is called after each split, with xs and xt the two
halves of the split tree node.
n - 1 maximum flow computations on contracted graphs, O(n + m) space besides them.
"""
for u, v, w in edges:
if w < 0:
raise ValueError("weights must be nonnegative")
if not (0 <= u < n and 0 <= v < n):
raise ValueError("vertex out of range")
if n <= 1:
return []
nodes: list[list[int]] = [list(range(n))] # @line L1
tree: list[list[tuple[int, Number]]] = [[]] # tree[i]: (neighbour node, weight)
while True:
x = next((i for i, X in enumerate(nodes) if len(X) >= 2), -1) # @line L2
if x == -1:
break
s, t = nodes[x][0], nodes[x][1] # @line L3
label = [-1] * n # vertex of the contracted graph G_X for each vertex of G
k = 0
for v in nodes[x]:
label[v] = k
k += 1
comp_of_nbr: dict[int, int] = {}
for y, _ in tree[x]: # @line L4
comp = k
k += 1
comp_of_nbr[y] = comp
seen = {x, y}
stack = [y]
while stack:
z = stack.pop()
for v in nodes[z]:
label[v] = comp
for z2, _ in tree[z]:
if z2 not in seen:
seen.add(z2)
stack.append(z2)
arcs: list[tuple[int, int, Number]] = []
for u, v, w in edges:
a, b = label[u], label[v]
if a != b:
arcs.append((a, b, w))
arcs.append((b, a, w))
lam, side = _min_cut(k, arcs, label[s], label[t]) # @line L5
xs = [v for v in nodes[x] if label[v] in side] # @line L6
xt = [v for v in nodes[x] if label[v] not in side]
nodes[x] = xs # @line L7
nodes.append(xt)
y_new = len(nodes) - 1
old = tree[x]
tree[x] = [(y_new, lam)]
tree.append([(x, lam)])
for y, w in old: # @line L8
if comp_of_nbr[y] in side:
tree[x].append((y, w))
else:
tree[y_new].append((y, w))
tree[y] = [(y_new if z == x else z, w2) for z, w2 in tree[y]]
if on_split is not None:
on_split(s, t, lam, sorted(xs), sorted(xt))
out = [] # @line L9
for i, X in enumerate(nodes):
for j, w in tree[i]:
if i < j:
a, b = X[0], nodes[j][0]
out.append((min(a, b), max(a, b), w))
out.sort(key=lambda e: (e[0], e[1]))
return out
test_gomory_hu_algorithm.py 159 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/gomory-hu-algorithm/. The authors are listed in prov/PSCBG5.jsonl.
import random
from fractions import Fraction
from itertools import combinations
import pytest
from gomory_hu_algorithm import gomory_hu_algorithm
P, Q, R, S, U, V = range(6)
SIX = [(P, Q, 1), (P, R, 7), (Q, R, 1), (Q, S, 3), (Q, U, 2), (R, U, 4), (S, U, 1), (S, V, 6), (U, V, 2)]
def cut_value(edges, side):
return sum(w for u, v, w in edges if (u in side) != (v in side))
def lam_brute(n, edges, a, b):
"""Oracle: the least cut over all vertex sets holding a and not b."""
others = [v for v in range(n) if v not in (a, b)]
best = None
for k in range(len(others) + 1):
for extra in combinations(others, k):
c = cut_value(edges, {a, *extra})
best = c if best is None else min(best, c)
return best
def path_min(n, tree, a, b):
"""Least weight on the tree path from a to b, and the tree edge achieving it."""
adj = [[] for _ in range(n)]
for i, (x, y, w) in enumerate(tree):
adj[x].append((y, w, i))
adj[y].append((x, w, i))
stack = [(a, -1, None, None)]
while stack:
v, par, best, arg = stack.pop()
if v == b:
return best, arg
for y, w, i in adj[v]:
if y != par:
if best is None or w < best:
stack.append((y, v, w, i))
else:
stack.append((y, v, best, arg))
raise AssertionError("tree is not connected")
def side_without(n, tree, i, a):
"""Vertices on the side of a after deleting tree edge i."""
adj = [[] for _ in range(n)]
for j, (x, y, _) in enumerate(tree):
if j != i:
adj[x].append(y)
adj[y].append(x)
seen = {a}
stack = [a]
while stack:
v = stack.pop()
for y in adj[v]:
if y not in seen:
seen.add(y)
stack.append(y)
return seen
def check(n, edges, tree):
assert len(tree) == n - 1
for a, b in combinations(range(n), 2):
lam, i = path_min(n, tree, a, b)
assert lam == lam_brute(n, edges, a, b)
side = side_without(n, tree, i, a)
assert b not in side
assert cut_value(edges, side) == lam
def test_six_vertex_example() -> None:
log = []
tree = gomory_hu_algorithm(6, SIX, lambda s, t, lam, xs, xt: log.append((s, t, lam, xs, xt)))
assert log == [
(P, Q, 6, [P, R], [Q, S, U, V]),
(P, R, 8, [P], [R]),
(Q, S, 6, [Q, U], [S, V]),
(Q, U, 7, [Q], [U]),
(S, V, 8, [S], [V]),
]
assert tree == [(P, R, 8), (Q, U, 7), (R, U, 6), (S, U, 6), (S, V, 8)]
check(6, SIX, tree)
def test_tiny() -> None:
assert gomory_hu_algorithm(0, []) == []
assert gomory_hu_algorithm(1, [(0, 0, 5)]) == []
assert gomory_hu_algorithm(2, [(0, 1, 3), (1, 0, 2)]) == [(0, 1, 5)]
def test_disconnected_and_zero_weight() -> None:
edges = [(0, 1, 4), (2, 3, 0), (3, 4, 2)]
tree = gomory_hu_algorithm(5, edges)
check(5, edges, tree)
assert path_min(5, tree, 0, 3)[0] == 0
assert path_min(5, tree, 2, 3)[0] == 0
def test_no_edges() -> None:
tree = gomory_hu_algorithm(4, [])
assert [w for _, _, w in tree] == [0, 0, 0]
def test_running_network_undirected() -> None:
s, a, b, c, d, t = range(6)
edges = [(s, a, 10), (s, c, 10), (a, b, 4), (a, c, 2), (a, d, 8), (c, d, 9), (d, b, 6), (b, t, 10), (d, t, 10)]
tree = gomory_hu_algorithm(6, edges)
check(6, edges, tree)
assert sorted(w for _, _, w in tree) == [20, 20, 20, 21, 21]
def test_fractions_and_floats() -> None:
edges = [(0, 1, Fraction(1, 3)), (1, 2, Fraction(1, 2)), (0, 2, Fraction(1, 6))]
tree = gomory_hu_algorithm(3, edges)
check(3, edges, tree)
tree = gomory_hu_algorithm(3, [(0, 1, 0.5), (1, 2, 0.25), (0, 2, 0.25)])
assert sorted(w for _, _, w in tree) == [0.5, 0.75]
def test_parallel_edges() -> None:
edges = [(0, 1, 1), (0, 1, 2), (1, 2, 1)]
tree = gomory_hu_algorithm(3, edges)
check(3, edges, tree)
def test_errors() -> None:
with pytest.raises(ValueError):
gomory_hu_algorithm(2, [(0, 1, -1)])
with pytest.raises(ValueError):
gomory_hu_algorithm(2, [(0, 2, 1)])
def test_random_against_brute_force() -> None:
rng = random.Random(4242)
for _ in range(600):
n = rng.randint(2, 7)
edges = []
for _ in range(rng.randint(0, 14)):
u, v = rng.randrange(n), rng.randrange(n)
edges.append((u, v, rng.choice([0, 1, 2, 3, 5, 8])))
tree = gomory_hu_algorithm(n, edges)
check(n, edges, tree)
def test_at_most_n_minus_1_distinct_values() -> None:
rng = random.Random(77)
for _ in range(50):
n = rng.randint(2, 7)
edges = [(rng.randrange(n), rng.randrange(n), rng.randint(0, 9)) for _ in range(12)]
values = {lam_brute(n, edges, a, b) for a, b in combinations(range(n), 2)}
assert len(values) <= n - 1
What the next level needs
- the tests did not run in this build