Hub labeling in Python
Stores for each vertex a set of hubs and distances so that a query takes a merge of two label lists.
- File
hub_labeling.py, 90 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/hub-labeling/. The authors are listed in prov/AFFE5D.jsonl.
import heapq
import math
Label = list[tuple[int, float]]
def query(lout: list[Label], lin: list[Label], u: int, v: int) -> float:
"""Return the distance from u to v given by the labels, or math.inf.
Both labels are sorted by hub rank, so one merge finds the common hubs.
Time O(|lout[u]| + |lin[v]|).
"""
# @line L9
a, b = lout[u], lin[v]
i = j = 0
best = math.inf
while i < len(a) and j < len(b):
# @line L10
if a[i][0] == b[j][0]:
best = min(best, a[i][1] + b[j][1])
i += 1
j += 1
elif a[i][0] < b[j][0]:
i += 1
else:
j += 1
# @line L11
return best
def _pruned_search(graph, h: int, rank: int, lout, lin, forward: bool) -> None:
n = len(graph)
dist = [math.inf] * n
done = [False] * n
dist[h] = 0
heap = [(0, h)]
while heap:
# @line L6
d, x = heapq.heappop(heap)
if done[x]:
continue
done[x] = True
# @line L7
q = query(lout, lin, h, x) if forward else query(lout, lin, x, h)
if q <= d:
continue
# @line L8
(lin if forward else lout)[x].append((rank, d))
for y, w in graph[x]:
if d + w < dist[y]:
dist[y] = d + w
heapq.heappush(heap, (dist[y], y))
def hub_labeling(adj: list[list[tuple[int, float]]], order: list[int] | None = None) -> tuple[list[Label], list[Label]]:
"""Pruned landmark labeling for a directed graph with nonnegative weights.
adj[u] lists the arcs (v, w) out of u, for vertices 0 .. n-1. order lists the
vertices from the most to the least important; by default, vertices of larger
total degree come first. Returns (lout, lin): lout[u] holds pairs (rank, d) with
d = dist(u, hub) and lin[v] holds pairs (rank, d) with d = dist(hub, v), where
hub = order[rank]; both lists are sorted by rank. query(lout, lin, u, v) then
gives dist(u, v), or math.inf when v is not reachable from u. Raises ValueError
on a negative weight, so negative cycles are rejected too.
"""
n = len(adj)
radj: list[list[tuple[int, float]]] = [[] for _ in range(n)]
for u in range(n):
for v, w in adj[u]:
if w < 0:
raise ValueError(f"negative arc weight {w} on ({u}, {v})")
radj[v].append((u, w))
if order is None:
order = sorted(range(n), key=lambda v: (-(len(adj[v]) + len(radj[v])), v))
if sorted(order) != list(range(n)):
raise ValueError("order must list every vertex once")
# @line L1
lout: list[Label] = [[] for _ in range(n)]
lin: list[Label] = [[] for _ in range(n)]
# @line L2
for rank, h in enumerate(order):
# @line L3
_pruned_search(adj, h, rank, lout, lin, forward=True)
# @line L4
_pruned_search(radj, h, rank, lout, lin, forward=False)
# @line L5
return lout, lin
test_hub_labeling.py 106 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/hub-labeling/. The authors are listed in prov/AFFE5D.jsonl.
import heapq
import math
import random
import pytest
from hub_labeling import hub_labeling, query
C, D, E, F, G, H = range(6)
EXAMPLE = [(C, D, 3), (C, E, 2), (D, F, 4), (E, D, 1), (E, F, 2), (E, G, 3), (F, G, 2), (F, H, 1), (G, H, 2)]
def build(n, arcs):
adj = [[] for _ in range(n)]
for u, v, w in arcs:
adj[u].append((v, w))
return adj
def dijkstra(adj, s):
"""Independent oracle: plain Dijkstra with a binary heap."""
dist = [math.inf] * len(adj)
dist[s] = 0
heap = [(0, s)]
while heap:
d, u = heapq.heappop(heap)
if d > dist[u]:
continue
for v, w in adj[u]:
if d + w < dist[v]:
dist[v] = d + w
heapq.heappush(heap, (dist[v], v))
return dist
def check_all_pairs(adj, order=None):
lout, lin = hub_labeling(adj, order)
n = len(adj)
for u in range(n):
want = dijkstra(adj, u)
for v in range(n):
assert query(lout, lin, u, v) == want[v], (u, v)
for lab in lout + lin:
ranks = [r for r, _ in lab]
assert ranks == sorted(ranks) and len(set(ranks)) == len(ranks)
return lout, lin
def test_example_labels_and_queries():
adj = build(6, EXAMPLE)
order = [F, E, D, G, C, H]
lout, lin = check_all_pairs(adj, order)
assert lout[C] == [(0, 4), (1, 2), (4, 0)]
assert lin[H] == [(0, 1), (3, 2), (5, 0)]
assert sum(map(len, lout)) + sum(map(len, lin)) == 21
assert query(lout, lin, C, H) == 5
assert query(lout, lin, D, G) == 6
def test_unreachable_is_inf():
adj = build(6, EXAMPLE)
lout, lin = hub_labeling(adj)
assert query(lout, lin, H, C) == math.inf
adj = build(3, [(0, 1, 1)])
lout, lin = check_all_pairs(adj)
assert query(lout, lin, 0, 2) == math.inf
assert query(lout, lin, 2, 2) == 0
def test_negative_weight_and_negative_cycle_rejected():
with pytest.raises(ValueError):
hub_labeling(build(2, [(0, 1, -1)]))
with pytest.raises(ValueError):
hub_labeling(build(3, [(0, 1, 2), (1, 2, -3), (2, 1, 1)]))
def test_bad_order_rejected():
with pytest.raises(ValueError):
hub_labeling(build(3, [(0, 1, 1)]), [0, 0, 1])
def test_zero_weights_and_cycles():
adj = build(4, [(0, 1, 0), (1, 0, 0), (1, 2, 0), (2, 3, 5), (3, 1, 0)])
for order in ([0, 1, 2, 3], [3, 2, 1, 0], [2, 0, 3, 1]):
check_all_pairs(adj, order)
def test_empty_and_single_vertex():
assert hub_labeling([]) == ([], [])
lout, lin = hub_labeling([[]])
assert query(lout, lin, 0, 0) == 0
def test_random_against_dijkstra():
rng = random.Random(20261105)
for case in range(600):
n = rng.randint(1, 9)
m = rng.randint(0, 3 * n)
arcs = [(rng.randrange(n), rng.randrange(n), rng.choice([0, 1, 2, 3, 5, 8, 2.5])) for _ in range(m)]
adj = build(n, arcs)
order = list(range(n))
rng.shuffle(order)
check_all_pairs(adj, order if case % 2 else None)
What the next level needs
- the tests did not run in this build