Algorithm Search⌘K

Johnson's algorithm in Python

Reweights with Bellman-Ford potentials, then runs Dijkstra from every vertex; handles sparse graphs with negative weights.

File
johnson.py, 71 lines
Level
I0 not checked yet
Toolchain
Python 3.14, pytest 9
Test framework
pytest
Run the tests
python3 -m pytest
johnson.pyRawGitHub
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/johnson/. The authors are listed in prov/GE9795.jsonl.

import heapq

Edge = tuple[int, int, float]


def bellman_ford_from_new_source(n: int, edges: list[Edge]) -> list[float] | None:
    """Distances from an added vertex q with an arc of weight 0 to each vertex.

    Returns None when the graph has a negative cycle. Time O(nm).
    """
    h = [0.0] * n  # round 1 relaxes the arcs q -> v of weight 0 and sets every label to 0
    if n == 0:
        return h
    # With n + 1 vertices, rounds 2 .. n make the labels final when there is no negative
    # cycle, so round n + 1 (the last pass of this loop) changes nothing.
    for _ in range(n):
        changed = False
        for u, v, w in edges:
            if h[u] + w < h[v]:
                h[v] = h[u] + w
                changed = True
        if not changed:
            return h
    return None  # round n + 1 would still change a label


def dijkstra(n: int, adj: list[list[tuple[int, float]]], s: int) -> list[float | None]:
    """Dijkstra with a binary heap; all weights in adj must be nonnegative."""
    dist: list[float | None] = [None] * n
    best = [float("inf")] * n
    best[s] = 0.0
    pq = [(0.0, s)]
    while pq:
        d, u = heapq.heappop(pq)
        if dist[u] is not None:
            continue
        dist[u] = d
        for v, w in adj[u]:
            if dist[v] is None and d + w < best[v]:
                best[v] = d + w
                heapq.heappush(pq, (d + w, v))
    return dist


def johnson(n: int, edges: list[Edge]) -> list[list[float | None]] | None:
    """All-pairs shortest paths on vertices 0 .. n-1 with arcs (u, v, w), w any real.

    Returns the matrix D with D[u][v] = delta(u, v), or None when v is not reachable
    from u. Returns None (no matrix) when the graph has a negative cycle.
    Time O(nm log n) with the binary heap used here, extra space O(n^2 + m).
    """
    # @line L1: the added vertex q is implicit: its arcs of weight 0 start every label at 0
    h = bellman_ford_from_new_source(n, edges)  # @line L2
    if h is None:  # @line L3
        return None
    # @line L4: h[v] = delta(q, v)
    adj: list[list[tuple[int, float]]] = [[] for _ in range(n)]
    for u, v, w in edges:  # @line L5
        adj[u].append((v, w + h[u] - h[v]))
    dist: list[list[float | None]] = []
    for u in range(n):  # @line L6
        du = dijkstra(n, adj, u)  # @line L7
        row: list[float | None] = []
        for v in range(n):  # @line L8
            x = du[v]
            row.append(None if x is None else x - h[u] + h[v])
        dist.append(row)
    return dist  # @line L9
test_johnson.py 58 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/johnson/. The authors are listed in prov/GE9795.jsonl.

import random

from johnson import johnson

# CLRS Figure 23.6, vertices 1 .. 5 renamed 0 .. 4
CLRS = [(0, 1, 3), (0, 2, 8), (0, 4, -4), (1, 3, 1), (1, 4, 7), (2, 1, 4), (3, 0, 2), (3, 2, -5), (4, 3, 6)]


def floyd_warshall(n: int, edges: list[tuple[int, int, int]]) -> list[list[float | None]] | None:
    inf = float("inf")
    d = [[0.0 if i == j else inf for j in range(n)] for i in range(n)]
    for u, v, w in edges:
        d[u][v] = min(d[u][v], w)
    for k in range(n):
        for i in range(n):
            for j in range(n):
                if d[i][k] + d[k][j] < d[i][j]:
                    d[i][j] = d[i][k] + d[k][j]
    if any(d[i][i] < 0 for i in range(n)):
        return None
    return [[None if x == inf else x for x in row] for row in d]


def test_clrs_example() -> None:
    assert johnson(5, CLRS) == [
        [0, 1, -3, 2, -4],
        [3, 0, -4, 1, -1],
        [7, 4, 0, 5, 3],
        [2, -1, -5, 0, -2],
        [8, 5, 1, 6, 0],
    ]


def test_negative_cycle() -> None:
    assert johnson(3, [(0, 1, 1), (1, 2, -2), (2, 1, 1)]) is None
    assert johnson(1, [(0, 0, -1)]) is None  # a negative self-loop


def test_unreachable_and_empty() -> None:
    assert johnson(0, []) == []
    assert johnson(1, []) == [[0]]
    assert johnson(3, [(0, 1, -2)]) == [[0, -2, None], [None, 0, None], [None, None, 0]]


def test_zero_cycle_and_parallel_arcs() -> None:
    d = johnson(3, [(0, 1, 2), (1, 0, -2), (1, 2, 5), (1, 2, 1)])
    assert d == [[0, 2, 3], [-2, 0, 1], [None, None, 0]]


def test_random_against_floyd_warshall() -> None:
    rng = random.Random(1977)
    for _ in range(600):
        n = rng.randint(1, 7)
        edges = [(rng.randrange(n), rng.randrange(n), rng.randint(-4, 9)) for _ in range(rng.randint(0, 3 * n))]
        assert johnson(n, edges) == floyd_warshall(n, edges)

What the next level needs

  • the tests did not run in this build

Read the full page about Johnson's algorithm