Dial's algorithm in Python
Dijkstra with a bucket queue for small integer weights, running in O(m + nC) time.
- File
dial_algorithm.py, 48 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/dial-algorithm/. The authors are listed in prov/T0RAP2.jsonl.
import math
INF = math.inf
def dial(
adj: list[list[tuple[int, int]]], s: int
) -> tuple[list[float], list[int | None]]:
"""Dijkstra's algorithm with a circular array of C + 1 buckets.
The graph has the vertices 0 .. n-1; adj[u] lists the edges (v, w) out of u.
Precondition: every weight w is an integer with 0 <= w (a ValueError is raised
otherwise); C is the largest weight. Returns (d, parent) with d[v] = math.inf and
parent[v] = None for vertices not reachable from s. Bucket i mod (C + 1) holds
the vertices with tentative distance i; an entry whose vertex already has a
smaller label is stale and skipped. Time O(m + nC), extra space O(n + m + C).
"""
n = len(adj)
C = 0
for u in range(n):
for _, w in adj[u]:
if not isinstance(w, int) or isinstance(w, bool) or w < 0:
raise ValueError("Dial's algorithm needs nonnegative integer weights")
C = max(C, w)
d: list[float] = [INF] * n # @line L1
parent: list[int | None] = [None] * n
buckets: list[list[int]] = [[] for _ in range(C + 1)]
d[s] = 0 # @line L2
buckets[0].append(s)
i = 0 # @line L3
left = 1
while left > 0: # @line L4
while not buckets[i % (C + 1)]: # @line L5
i += 1
u = buckets[i % (C + 1)].pop() # @line L6
left -= 1
if d[u] < i: # @line L7
continue
for v, w in adj[u]: # @line L8
if i + w < d[v]: # @line L9
d[v] = i + w # @line L10
parent[v] = u
buckets[(i + w) % (C + 1)].append(v)
left += 1
return d, parent # @line L11
test_dial_algorithm.py 77 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/dial-algorithm/. The authors are listed in prov/T0RAP2.jsonl.
import math
import random
import pytest
from dial_algorithm import dial
INF = math.inf
# s=0, a=1, b=2, c=3, t=4, z=5
EXAMPLE = [[(1, 2), (2, 3)], [(2, 0), (3, 3)], [(3, 3), (4, 3)], [(4, 1)], [], [(0, 1)]]
def bellman_ford(adj, s):
"""Independent oracle: distances from s by n - 1 rounds of relaxing every edge."""
n = len(adj)
d = [INF] * n
d[s] = 0
for _ in range(n - 1):
for u in range(n):
for v, w in adj[u]:
if d[u] + w < d[v]:
d[v] = d[u] + w
return d
def test_example() -> None:
d, parent = dial(EXAMPLE, 0)
assert d == [0, 2, 2, 5, 5, INF]
assert parent == [None, 0, 1, 1, 2, None]
def test_unreachable_and_single_vertex() -> None:
assert dial([[]], 0) == ([0], [None])
d, _ = dial([[], [(0, 4)]], 0)
assert d == [0, INF]
def test_zero_weights_only() -> None:
adj = [[(1, 0)], [(2, 0), (0, 0)], [(2, 0)]]
assert dial(adj, 0)[0] == [0, 0, 0]
def test_negative_or_fractional_weight_rejected() -> None:
# a negative edge (here on the negative cycle 0 -> 1 -> 0) is outside the model
with pytest.raises(ValueError):
dial([[(1, 1)], [(0, -2)]], 0)
with pytest.raises(ValueError):
dial([[(1, 1.5)], []], 0)
def test_wraparound_of_the_circular_array() -> None:
# a long path of weight-C edges sends the index around the array many times
n, C = 30, 7
adj = [[(i + 1, C)] if i + 1 < n else [] for i in range(n)]
adj[0].append((n - 1, C * n))
d, _ = dial(adj, 0)
assert d == [C * i for i in range(n)]
def test_random_against_bellman_ford() -> None:
rng = random.Random(1969)
for _ in range(800):
n = rng.randint(1, 9)
C = rng.choice([0, 1, 3, 10])
adj = [[] for _ in range(n)]
for _ in range(rng.randint(0, 3 * n)):
adj[rng.randrange(n)].append((rng.randrange(n), rng.randint(0, C)))
s = rng.randrange(n)
d, parent = dial(adj, s)
assert d == bellman_ford(adj, s)
for v in range(n):
if v != s and d[v] < INF:
u = parent[v]
assert d[v] == d[u] + min(w for x, w in adj[u] if x == v)
What the next level needs
- the tests did not run in this build