Maximum-capacity augmenting path algorithm in Python
Augmenting path method for maximum flow that always chooses the path with the largest bottleneck capacity.
- File
maximum_capacity_augmenting_path.py, 87 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/maximum-capacity-augmenting-path/. The authors are listed in prov/QFY05A.jsonl.
import heapq
def widest_path(n: int, head: list[int], cap: list[int], adj: list[list[int]],
s: int, t: int) -> tuple[list[int], int]:
"""Residual edges of an s-t path whose least residual capacity is largest.
Residual edge 2i is edge i forward, 2i + 1 is edge i backward; head[r] is
the vertex that r enters and cap[r] its residual capacity. Returns the
path as a list of residual edge ids from s to t and its bottleneck, or
([], 0) when t is not reachable. This is Dijkstra with (max, min) in place
of (min, +); time O(m log n).
"""
best = [0] * n # @line L7
via = [-1] * n
best[s] = -1 # stands for infinity
heap = [(-float("inf"), s)]
done = [False] * n
while heap: # @line L8
_, u = heapq.heappop(heap) # @line L9
if done[u]:
continue
done[u] = True
bu = float("inf") if u == s else best[u]
for r in adj[u]: # @line L10
v = head[r]
w = min(bu, cap[r])
if cap[r] > 0 and not done[v] and v != s and w > best[v]: # @line L11
best[v], via[v] = w, r # @line L12
heapq.heappush(heap, (-w, v))
if t == s or best[t] <= 0: # @line L13
return [], 0
path, v = [], t
while v != s:
r = via[v]
path.append(r)
v = head[r ^ 1]
path.reverse()
return path, best[t]
def maximum_capacity_augmenting_path(n: int, edges: list[tuple[int, int, int]],
s: int, t: int) -> tuple[int, list[int], set[int]]:
"""Maximum s-t flow by augmenting along a widest residual path each time.
The network is directed, with vertices 0 .. n-1 and edges (u, v, c) with
integer c >= 0. Parallel and antiparallel edges are allowed and kept
apart; self-loops carry no flow. Returns (value, flow, S): flow[i] is the
flow on edge i, and S is the set of vertices that s reaches in the final
residual graph, the source side of a minimum cut. At most
m ln(f*) + 1 augmentations, each O(m log n).
"""
if s == t:
raise ValueError("s and t must differ")
head: list[int] = []
cap: list[int] = []
adj: list[list[int]] = [[] for _ in range(n)]
for u, v, c in edges:
if c < 0:
raise ValueError("capacities must be nonnegative")
adj[u].append(len(head))
head.append(v)
cap.append(c)
adj[v].append(len(head))
head.append(u)
cap.append(0)
value = 0 # @line L1
while True: # @line L2
path, delta = widest_path(n, head, cap, adj, s, t) # @line L3
if not path: # @line L4
break
for r in path: # @line L5
cap[r] -= delta
cap[r ^ 1] += delta
value += delta
flow = [cap[2 * i + 1] for i in range(len(edges))]
seen, stack = {s}, [s]
while stack:
u = stack.pop()
for r in adj[u]:
if cap[r] > 0 and head[r] not in seen:
seen.add(head[r])
stack.append(head[r])
return value, flow, seen # @line L6
test_maximum_capacity_augmenting_path.py 89 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/maximum-capacity-augmenting-path/. The authors are listed in prov/QFY05A.jsonl.
import itertools
import random
import pytest
from maximum_capacity_augmenting_path import maximum_capacity_augmenting_path as mcap
# s, a, b, c, d, t = 0 .. 5
EXAMPLE = [(0, 1, 10), (0, 2, 10), (1, 2, 2), (1, 3, 4), (1, 4, 8),
(2, 4, 9), (4, 3, 6), (3, 5, 10), (4, 5, 10)]
def min_cut_brute(n, edges, s, t):
"""Independent oracle: the least capacity of a set S with s in S, t not in S."""
others = [v for v in range(n) if v not in (s, t)]
best = None
for r in range(len(others) + 1):
for extra in itertools.combinations(others, r):
side = {s, *extra}
c = sum(w for u, v, w in edges if u in side and v not in side)
best = c if best is None else min(best, c)
return best
def check_flow(n, edges, s, t, value, flow, side):
assert all(0 <= f <= c for f, (_, _, c) in zip(flow, edges))
bal = [0] * n
for f, (u, v, _) in zip(flow, edges):
bal[u] -= f
bal[v] += f
assert all(bal[v] == 0 for v in range(n) if v not in (s, t))
assert bal[t] == value and -bal[s] == value
assert s in side and t not in side
assert value == sum(c for u, v, c in edges if u in side and v not in side)
def test_example() -> None:
value, flow, side = mcap(6, EXAMPLE, 0, 5)
assert value == 19 and side == {0, 2}
check_flow(6, EXAMPLE, 0, 5, value, flow, side)
def test_zero_capacity_and_no_path() -> None:
edges = [(0, 1, 5), (1, 2, 0), (0, 2, 0)]
assert mcap(3, edges, 0, 2) == (0, [0, 0, 0], {0, 1})
assert mcap(4, [(0, 1, 3), (2, 3, 3)], 0, 3) == (0, [0, 0], {0, 1})
assert mcap(2, [], 0, 1) == (0, [], {0})
def test_antiparallel_parallel_and_loops() -> None:
edges = [(0, 1, 3), (1, 0, 2), (1, 2, 4), (0, 1, 2), (1, 1, 7), (2, 1, 1)]
value, flow, side = mcap(3, edges, 0, 2)
assert value == 4
check_flow(3, edges, 0, 2, value, flow, side)
assert flow[4] == 0
def test_backward_edge_is_needed() -> None:
edges = [(0, 1, 1), (0, 2, 1), (1, 2, 1), (1, 3, 1), (2, 3, 1)]
assert mcap(4, edges, 0, 3)[0] == 2
def test_errors() -> None:
with pytest.raises(ValueError):
mcap(2, [(0, 1, 1)], 0, 0)
with pytest.raises(ValueError):
mcap(2, [(0, 1, -1)], 0, 1)
def test_large_values() -> None:
big = 2 ** 62
value, _, _ = mcap(3, [(0, 1, big), (1, 2, big), (0, 2, 5)], 0, 2)
assert value == big + 5
def test_random_against_brute_force() -> None:
rng = random.Random(4711)
for _ in range(600):
n = rng.randint(2, 7)
m = rng.randint(0, 16)
edges = [(rng.randrange(n), rng.randrange(n), rng.choice([0, 1, 2, 3, 5, 8, 20]))
for _ in range(m)]
s, t = rng.sample(range(n), 2)
value, flow, side = mcap(n, edges, s, t)
assert value == min_cut_brute(n, edges, s, t)
check_flow(n, edges, s, t, value, flow, side)
What the next level needs
- the tests did not run in this build
Read the full page about Maximum-capacity augmenting path algorithm