Capacity scaling max-flow algorithm in Python
Augment only along paths with residual capacity above a decreasing threshold.
- File
capacity_scaling.py, 86 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/capacity-scaling/. The authors are listed in prov/H1FND5.jsonl.
from collections import deque
from typing import Callable, Optional
def capacity_scaling(
n: int,
edges: list[tuple[int, int, int]],
s: int,
t: int,
on_augment: Optional[Callable[[int, list[int], int], None]] = None,
) -> tuple[int, list[int], list[int]]:
"""Maximum s-t flow by capacity scaling (augment only along fat paths, then halve).
The network has the vertices 0 .. n-1 and the directed edges (u, v, cap) of the
list, with integer cap >= 0. Parallel, antiparallel and zero-capacity edges and
self-loops are allowed; each edge keeps its own flow. Requires s != t. The
optional callback gets (delta threshold, path as a vertex list, amount) after
each augmentation.
Returns (value, flow, source_side): flow[i] is the flow on edges[i], and
source_side lists, in increasing order, the vertices reachable from s in the final
residual graph, the source side of a minimum cut.
At most 2m augmentations per phase and floor(log2 U) + 1 phases: O(m^2 log U).
"""
if s == t:
raise ValueError("source and sink must differ")
head: list[int] = [] # arc 2i is edge i forward, arc 2i + 1 is edge i backward
res: list[int] = []
adj: list[list[int]] = [[] for _ in range(n)]
for u, v, cap in edges: # @line L1
if not isinstance(cap, int):
raise TypeError("capacity scaling needs integer capacities")
if cap < 0:
raise ValueError("capacities must be nonnegative")
adj[u].append(len(head))
head.append(v)
res.append(cap)
adj[v].append(len(head))
head.append(u)
res.append(0)
big = max(res, default=0)
delta = 1 << (big.bit_length() - 1) if big > 0 else 0 # @line L2
value = 0
while delta >= 1: # @line L3
while True: # @line L4
parent = _search(n, adj, head, res, s, delta)
if parent[t] == -1:
break
amount = None # @line L5
v = t
while v != s:
a = parent[v]
amount = res[a] if amount is None else min(amount, res[a])
v = head[a ^ 1]
path = [t]
v = t
while v != s: # @line L6
a = parent[v]
res[a] -= amount
res[a ^ 1] += amount
v = head[a ^ 1]
path.append(v)
value += amount
if on_augment is not None:
on_augment(delta, path[::-1], amount)
delta //= 2 # @line L7
parent = _search(n, adj, head, res, s, 1) # @line L8
flow = [res[2 * i + 1] for i in range(len(edges))]
return value, flow, [v for v in range(n) if parent[v] != -1] # @line L9
def _search(n: int, adj: list[list[int]], head: list[int], res: list[int], s: int, delta: int) -> list[int]:
"""BFS from s over arcs with residual capacity >= delta; parent arc, -2 at s, -1 if unreached."""
parent = [-1] * n
parent[s] = -2
queue = deque([s])
while queue:
u = queue.popleft()
for a in adj[u]:
v = head[a]
if res[a] >= delta and parent[v] == -1:
parent[v] = a
queue.append(v)
return parent
test_capacity_scaling.py 148 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/capacity-scaling/. The authors are listed in prov/H1FND5.jsonl.
import random
from fractions import Fraction
from itertools import combinations
import pytest
from capacity_scaling import capacity_scaling
S, A, B, C, D, T = range(6)
RUNNING = [(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)]
def min_cut_brute(n, edges, s, t):
"""Oracle: the least capacity over all 2^(n-2) s-t cuts."""
others = [v for v in range(n) if v not in (s, t)]
best = None
for k in range(len(others) + 1):
for extra in combinations(others, k):
side = {s, *extra}
cap = sum(c for u, v, c in edges if u in side and v not in side)
best = cap if best is None else min(best, cap)
return best
def check(n, edges, s, t, value, flow, side):
"""The flow is feasible, has the value, and the side is a cut of that capacity."""
exc = [0] * n
for (u, v, c), f in zip(edges, flow):
assert 0 <= f <= c
exc[u] -= f
exc[v] += f
for v in range(n):
if v not in (s, t):
assert exc[v] == 0
assert -exc[s] == value
assert s in side and t not in side
cut = set(side)
assert sum(c for u, v, c in edges if u in cut and v not in cut) == value
def test_running_example() -> None:
value, flow, side = capacity_scaling(6, RUNNING, S, T)
assert value == 19
assert side == [S, C]
check(6, RUNNING, S, T, value, flow, side)
def test_backward_edge_needed() -> None:
edges = [(0, 1, 1), (0, 2, 1), (1, 2, 1), (1, 3, 1), (2, 3, 1)]
value, flow, side = capacity_scaling(4, edges, 0, 3)
assert value == 2
check(4, edges, 0, 3, value, flow, side)
def test_no_path_and_zero_capacity() -> None:
assert capacity_scaling(3, [(0, 1, 5), (2, 1, 5)], 0, 2) == (0, [0, 0], [0, 1])
assert capacity_scaling(3, [(0, 1, 5), (1, 2, 0)], 0, 2) == (0, [0, 0], [0, 1])
assert capacity_scaling(2, [], 0, 1) == (0, [], [0])
def test_antiparallel_parallel_and_loops() -> None:
edges = [(0, 1, 3), (1, 2, 2), (2, 1, 1), (2, 3, 2), (0, 2, 1), (1, 1, 7), (0, 1, 2)]
value, flow, side = capacity_scaling(4, edges, 0, 3)
assert value == 2
check(4, edges, 0, 3, value, flow, side)
edges = [(0, 1, 4), (1, 0, 4), (1, 2, 3), (2, 1, 5)]
value, flow, side = capacity_scaling(3, edges, 0, 2)
assert value == 3
check(3, edges, 0, 2, value, flow, side)
def test_bad_input() -> None:
with pytest.raises(ValueError):
capacity_scaling(2, [(0, 1, 1)], 0, 0)
with pytest.raises(ValueError):
capacity_scaling(2, [(0, 1, -1)], 0, 1)
def test_large_capacities() -> None:
big = 2**62
edges = [(0, 1, big), (1, 2, big), (0, 2, 1)]
assert capacity_scaling(3, edges, 0, 2)[0] == big + 1
def test_random_against_brute_force_cut() -> None:
rng = random.Random(4417)
for _ in range(600):
n = rng.randint(2, 6)
edges = [(rng.randrange(n), rng.randrange(n), rng.randint(0, 6)) for _ in range(rng.randint(0, 12))]
s, t = rng.sample(range(n), 2)
value, flow, side = capacity_scaling(n, edges, s, t)
assert value == min_cut_brute(n, edges, s, t)
check(n, edges, s, t, value, flow, side)
def test_bad_case_for_plain_ford_fulkerson() -> None:
x = 10**15
edges = [(0, 1, x), (0, 2, x), (1, 2, 1), (1, 3, x), (2, 3, x)]
value, flow, side = capacity_scaling(4, edges, 0, 3)
assert value == 2 * x
assert flow[2] == 0 # the shortest paths never use the middle edge
def test_long_path_no_recursion_limit() -> None:
n = 5000
edges = [(i, i + 1, 3) for i in range(n - 1)]
assert capacity_scaling(n, edges, 0, n - 1)[0] == 3
def test_non_integer_capacities_rejected() -> None:
with pytest.raises(TypeError):
capacity_scaling(2, [(0, 1, 1.5)], 0, 1)
with pytest.raises(TypeError):
capacity_scaling(2, [(0, 1, Fraction(1, 2))], 0, 1)
def test_phases_of_running_example() -> None:
log = []
value, flow, side = capacity_scaling(6, RUNNING, S, T, lambda d, p, a: log.append((d, p, a)))
assert value == 19
assert log == [
(8, [S, A, D, T], 8),
(4, [S, C, D, B, T], 6),
(2, [S, A, B, T], 2),
(2, [S, C, D, T], 2),
(1, [S, C, D, A, B, T], 1),
]
def test_each_amount_at_least_delta_and_at_most_2m_per_phase() -> None:
rng = random.Random(5150)
for _ in range(500):
n = rng.randint(2, 7)
edges = [(rng.randrange(n), rng.randrange(n), rng.randint(0, 40)) for _ in range(rng.randint(0, 14))]
s, t = rng.sample(range(n), 2)
log = []
value, flow, side = capacity_scaling(n, edges, s, t, lambda d, p, a: log.append((d, a)))
assert value == min_cut_brute(n, edges, s, t)
check(n, edges, s, t, value, flow, side)
per_phase = {}
for d, a in log:
assert a >= d
per_phase[d] = per_phase.get(d, 0) + 1
for d, k in per_phase.items():
assert k <= 2 * max(1, len(edges))
What the next level needs
- the tests did not run in this build
Read the full page about Capacity scaling max-flow algorithm