Algorithm Search⌘K

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
maximum_capacity_augmenting_path.pyRawGitHub
# 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