Successive shortest path algorithm in Python
Send flow along shortest-cost augmenting paths in the residual graph repeatedly.
- File
successive_shortest_path.py, 85 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/successive-shortest-path/. The authors are listed in prov/13QZ7V.jsonl.
import heapq
import math
def successive_shortest_path(n: int, edges: list[tuple[int, int, int, int]], s: int, t: int,
d: float = math.inf) -> tuple[int, int, list[int]]:
"""Cheapest s-t flow of value min(d, maximum flow value).
The network is directed, with vertices 0 .. n-1 and edges (u, v, cap, cost):
integer cap >= 0 and integer cost of any sign. Parallel and antiparallel
edges are allowed and kept apart; self-loops are ignored. The edges with
cap > 0 must not form a cycle of negative cost; otherwise ValueError.
Returns (value, cost, flow) with flow[i] the flow on edge i. With
d = inf it returns a minimum-cost maximum flow.
"""
if s == t:
raise ValueError("s and t must differ")
head: list[int] = []
cap: list[int] = []
cost: list[int] = []
adj: list[list[int]] = [[] for _ in range(n)]
for u, v, c, w in edges:
if c < 0:
raise ValueError("capacities must be nonnegative")
if u == v:
c = 0
for a, b, cc, ww in ((u, v, c, w), (v, u, 0, -w)):
adj[a].append(len(head))
head.append(b)
cap.append(cc)
cost.append(ww)
x_value = 0 # @line L1
# Bellman-Ford from a virtual root joined to every vertex by cost-0 edges.
pi = [0] * n # @line L2
for rnd in range(n + 1):
changed = False
for a in range(n):
for r in adj[a]:
if cap[r] > 0 and pi[a] + cost[r] < pi[head[r]]:
pi[head[r]] = pi[a] + cost[r]
changed = True
if not changed:
break
if rnd == n:
raise ValueError("negative-cost cycle")
while x_value < d: # @line L3
dist = [math.inf] * n # @line L4
via = [-1] * n
dist[s] = 0
heap = [(0, s)]
while heap:
du, a = heapq.heappop(heap)
if du > dist[a]:
continue
for r in adj[a]:
b = head[r]
if cap[r] > 0:
nd = du + cost[r] + pi[a] - pi[b]
if nd < dist[b]:
dist[b], via[b] = nd, r
heapq.heappush(heap, (nd, b))
if dist[t] == math.inf: # @line L5
break
for v in range(n): # @line L6
if dist[v] < math.inf:
pi[v] += dist[v]
delta = d - x_value # @line L7
v = t
while v != s:
r = via[v]
delta = min(delta, cap[r])
v = head[r ^ 1]
v = t # @line L8
while v != s:
r = via[v]
cap[r] -= delta
cap[r ^ 1] += delta
v = head[r ^ 1]
x_value += delta
flow = [cap[2 * i + 1] for i in range(len(edges))]
total = sum(f * e[3] for f, e in zip(flow, edges)) # @line L9
return x_value, total, flow
test_successive_shortest_path.py 112 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/successive-shortest-path/. The authors are listed in prov/13QZ7V.jsonl.
import itertools
import math
import random
import pytest
from successive_shortest_path import successive_shortest_path as ssp
# s, a, b, t = 0, 1, 2, 3; edges (u, v, capacity, cost)
EXAMPLE = [(0, 1, 3, 1), (0, 2, 2, 4), (1, 2, 2, 1), (1, 3, 2, 5), (2, 3, 3, 1)]
def brute(n, edges, s, t):
"""Independent oracle: the least cost of an integral s-t flow of each value.
Self-loops are ignored by the algorithm, so they carry 0 here as well.
"""
best = {}
for x in itertools.product(*[range(e[2] + 1 if e[0] != e[1] else 1) for e in edges]):
bal = [0] * n
for (u, v, _, _), f in zip(edges, x):
bal[u] -= f
bal[v] += f
if all(bal[v] == 0 for v in range(n) if v not in (s, t)):
c = sum(f * e[3] for e, f in zip(edges, x))
k = bal[t]
if k >= 0 and (k not in best or c < best[k]):
best[k] = c
return best
def check(n, edges, s, t, value, cost, flow):
assert all(0 <= f <= e[2] for f, e in zip(flow, edges))
bal = [0] * n
for (u, v, _, _), f in zip(edges, flow):
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
assert cost == sum(f * e[3] for f, e in zip(flow, edges))
def test_example_each_value() -> None:
expected = {0: 0, 1: 3, 2: 6, 3: 11, 4: 17, 5: 25}
for d, c in expected.items():
value, cost, flow = ssp(4, EXAMPLE, 0, 3, d)
assert (value, cost) == (d, c)
check(4, EXAMPLE, 0, 3, value, cost, flow)
assert ssp(4, EXAMPLE, 0, 3) == (5, 25, [3, 2, 1, 2, 3])
assert ssp(4, EXAMPLE, 0, 3, 9)[:2] == (5, 25)
def test_zero_capacity_and_no_path() -> None:
assert ssp(2, [(0, 1, 0, -100)], 0, 1) == (0, 0, [0])
assert ssp(4, [(0, 1, 3, 1), (2, 3, 3, 1)], 0, 3) == (0, 0, [0, 0])
assert ssp(2, [], 0, 1) == (0, 0, [])
def test_negative_costs_and_antiparallel() -> None:
edges = [(0, 1, 2, -3), (1, 0, 2, 5), (1, 2, 3, 1), (0, 2, 1, 0), (1, 1, 4, -9)]
value, cost, flow = ssp(3, edges, 0, 2)
assert (value, cost) == (3, -4)
check(3, edges, 0, 2, value, cost, flow)
assert flow[1] == 0 and flow[4] == 0
def test_negative_cycle_rejected() -> None:
with pytest.raises(ValueError):
ssp(3, [(1, 2, 1, -2), (2, 1, 1, 1), (0, 1, 1, 0)], 0, 1)
# a negative cycle of capacity 0 does not count
assert ssp(3, [(1, 2, 0, -2), (2, 1, 1, 1), (0, 1, 1, 0)], 0, 1)[:2] == (1, 0)
def test_errors() -> None:
with pytest.raises(ValueError):
ssp(2, [(0, 1, 1, 1)], 1, 1)
with pytest.raises(ValueError):
ssp(2, [(0, 1, -1, 1)], 0, 1)
def has_negative_cycle(n, edges):
dist = [0] * n
for _ in range(n):
for u, v, c, w in edges:
if c > 0 and u != v and dist[u] + w < dist[v]:
dist[v] = dist[u] + w
return any(c > 0 and u != v and dist[u] + w < dist[v] for u, v, c, w in edges)
def test_random_against_brute_force() -> None:
rng = random.Random(2024)
tested = 0
while tested < 600:
n = rng.randint(2, 5)
m = rng.randint(0, 6)
edges = [(rng.randrange(n), rng.randrange(n), rng.randint(0, 3), rng.randint(-3, 6))
for _ in range(m)]
if has_negative_cycle(n, edges):
continue
s, t = rng.sample(range(n), 2)
best = brute(n, edges, s, t)
top = max(best)
for d in range(top + 1):
value, cost, flow = ssp(n, edges, s, t, d)
assert value == d and cost == best[d]
check(n, edges, s, t, value, cost, flow)
value, cost, _ = ssp(n, edges, s, t)
assert (value, cost) == (top, best[top])
tested += 1
What the next level needs
- the tests did not run in this build