Fredman-Tarjan algorithm in Python
Borůvka and Prim steps on Fibonacci heaps, O(m β(m,n)) time.
- File
fredman_tarjan_mst.py, 213 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/fredman-tarjan-mst/. The authors are listed in prov/37AZFW.jsonl.
def fredman_tarjan_mst(
n: int,
edges: list[tuple[int, int, float]],
heap_limit: int | None = None,
log: list[tuple[int, int, float, int]] | None = None,
) -> list[int]:
"""Return the indices of a minimum spanning forest, by Fredman and Tarjan.
The vertices are 0 .. n - 1 and edges[i] = (u, v, w) is an undirected
edge. Weights may tie (ties are broken by the edge index); the graph may
be disconnected, and parallel edges and self-loops are allowed.
Each phase grows trees one after another, Prim style, with a Fibonacci
heap of the vertices next to the tree. A tree adds at least one edge,
and stops when its heap holds more than k vertices, when it reaches a
tree of the same phase, or when its heap runs empty. Then every tree is
contracted to one vertex. The limit is k = 2^(2m/t), where t is the number of vertices of the phase
and m the number of edges of the input after loops and parallel edges
are removed, unless heap_limit fixes it. If log is a list, each phase
appends (t, edges of the phase, k, number of trees).
Cost: O(m) per phase and at most beta(m, n) + 1 phases, so
O(m beta(m, n)) time; O(n + m) space.
"""
forest: list[int] = []
cur, t = _clean(edges, list(range(n)))
m0 = len(cur)
while cur: # @line L1
m = len(cur)
k = heap_limit if heap_limit is not None else 2.0 ** min(2.0 * m0 / t, 60.0) # @line L2
adj: list[list[tuple[int, tuple[float, int]]]] = [[] for _ in range(t)]
for a, b, w, i in cur:
adj[a].append((b, (w, i)))
adj[b].append((a, (w, i)))
tree_of = [-1] * t
trees = 0
for v0 in range(t): # @line L3
if tree_of[v0] >= 0 or not adj[v0]:
continue
in_tree = {v0} # @line L4
heap = _FibHeap()
node: dict[int, _Node] = {}
_scan(v0, adj, in_tree, heap, node)
join = -1
while heap.size > 0: # @line L5
(_, i), u = heap.extract_min() # @line L6
del node[u]
forest.append(i)
if tree_of[u] >= 0: # @line L7
join = tree_of[u]
break
in_tree.add(u) # @line L8
_scan(u, adj, in_tree, heap, node)
if heap.size > k:
break
if join < 0: # @line L9
join = trees
trees += 1
for v in in_tree:
tree_of[v] = join
if log is not None:
log.append((t, m, k, trees))
cur, t = _clean([(a, b, w) for a, b, w, _ in cur], tree_of, [i for *_, i in cur]) # @line L10
return sorted(forest) # @line L11
def _scan(u, adj, in_tree, heap, node):
"""Offer each edge at u to the heap of the vertices next to the tree."""
for x, key in adj[u]:
if x in in_tree:
continue
if x not in node:
node[x] = heap.insert(key, x)
elif key < node[x].key:
heap.decrease_key(node[x], key)
def _clean(edges, label, index=None):
"""Relabel the ends, drop loops, keep the lightest of parallel edges.
The vertices that keep an edge are numbered 0 .. t - 1; return (edges, t).
"""
best: dict[tuple[int, int], tuple[float, int]] = {}
for j, (u, v, w) in enumerate(edges):
i = j if index is None else index[j]
a, b = label[u], label[v]
if a == b:
continue
pair = (a, b) if a < b else (b, a)
if pair not in best or (w, i) < best[pair]:
best[pair] = (w, i)
new: dict[int, int] = {}
for a, b in best:
new.setdefault(a, len(new))
new.setdefault(b, len(new))
return [(new[a], new[b], w, i) for (a, b), (w, i) in best.items()], len(new)
class _Node:
__slots__ = ("key", "item", "parent", "child", "left", "right", "degree", "mark")
def __init__(self, key, item):
self.key = key
self.item = item
self.parent = None
self.child = None
self.left = self
self.right = self
self.degree = 0
self.mark = False
class _FibHeap:
"""A Fibonacci heap with insert, decrease-key and extract-min."""
def __init__(self):
self.min = None
self.size = 0
def insert(self, key, item):
x = _Node(key, item)
self._add_root(x)
self.size += 1
return x
def _add_root(self, x):
x.parent = None
x.mark = False
if self.min is None:
x.left = x.right = x
self.min = x
else:
x.left = self.min
x.right = self.min.right
self.min.right.left = x
self.min.right = x
if x.key < self.min.key:
self.min = x
def extract_min(self):
z = self.min
children = []
c = z.child
if c is not None:
children.append(c)
c = c.right
while c is not z.child:
children.append(c)
c = c.right
roots = [z.right]
r = z.right.right
while r is not z:
roots.append(r)
r = r.right
if z.right is z:
roots = []
self.size -= 1
table: dict[int, _Node] = {}
for x in roots + children:
x.parent = None
while x.degree in table:
y = table.pop(x.degree)
if y.key < x.key:
x, y = y, x
self._link(y, x)
table[x.degree] = x
self.min = None
for x in table.values():
self._add_root(x)
return z.key, z.item
def _link(self, y, x):
"""Make the root y a child of the root x."""
y.parent = x
y.mark = False
if x.child is None:
x.child = y
y.left = y.right = y
else:
y.left = x.child
y.right = x.child.right
x.child.right.left = y
x.child.right = y
x.degree += 1
def decrease_key(self, x, key):
x.key = key
p = x.parent
if p is not None and key < p.key:
self._cut(x, p)
while p.parent is not None and p.mark:
q = p.parent
self._cut(p, q)
p = q
if p.parent is not None:
p.mark = True
if key < self.min.key:
self.min = x
def _cut(self, x, p):
if x.right is x:
p.child = None
else:
x.left.right = x.right
x.right.left = x.left
if p.child is x:
p.child = x.right
p.degree -= 1
self._add_root(x)
test_fredman_tarjan_mst.py 135 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/fredman-tarjan-mst/. The authors are listed in prov/37AZFW.jsonl.
import heapq
import random
from fredman_tarjan_mst import _FibHeap, fredman_tarjan_mst
EXAMPLE = [(0, 1, 4), (0, 2, 3), (1, 2, 1), (1, 3, 2), (2, 3, 4), (2, 4, 5), (3, 4, 7), (3, 5, 6), (4, 5, 2)]
def kruskal_weight(n, edges):
"""The weight and the number of edges of a minimum spanning forest."""
parent = list(range(n))
def find(x):
while parent[x] != x:
parent[x] = parent[parent[x]]
x = parent[x]
return x
total, count = 0, 0
for u, v, w in sorted(edges, key=lambda e: e[2]):
a, b = find(u), find(v)
if a != b:
parent[a] = b
total += w
count += 1
return total, count
def is_forest(n, edges, chosen):
parent = list(range(n))
def find(x):
while parent[x] != x:
x = parent[x]
return x
for i in chosen:
a, b = find(edges[i][0]), find(edges[i][1])
if a == b:
return False
parent[a] = b
return True
def check(n, edges, chosen):
assert len(set(chosen)) == len(chosen)
assert is_forest(n, edges, chosen)
assert (sum(edges[i][2] for i in chosen), len(chosen)) == kruskal_weight(n, edges)
def test_example():
assert fredman_tarjan_mst(6, EXAMPLE) == [1, 2, 3, 5, 8]
def test_example_two_phases():
log = []
assert fredman_tarjan_mst(6, EXAMPLE, heap_limit=1, log=log) == [1, 2, 3, 5, 8]
assert [(t, m, trees) for t, m, _, trees in log] == [(6, 9, 2), (2, 1, 1)]
def test_trivial_graphs():
assert fredman_tarjan_mst(0, []) == []
assert fredman_tarjan_mst(1, []) == []
assert fredman_tarjan_mst(1, [(0, 0, 3)]) == []
assert fredman_tarjan_mst(3, []) == []
def test_parallel_edges_and_loops():
edges = [(0, 1, 5), (0, 1, 2), (1, 1, -9), (1, 2, 2), (2, 1, 1)]
assert fredman_tarjan_mst(3, edges) == [1, 4]
def test_ties_take_smaller_index():
edges = [(0, 1, 1), (1, 2, 1), (0, 2, 1)]
assert fredman_tarjan_mst(3, edges) == [0, 1]
def test_disconnected():
edges = [(0, 1, 3), (2, 3, 1), (3, 4, 2), (2, 4, 0)]
got = fredman_tarjan_mst(6, edges)
check(6, edges, got)
assert got == [0, 1, 3]
def test_fibonacci_heap_sorts():
rng = random.Random(3)
for _ in range(200):
h = _FibHeap()
ref = []
nodes = []
for j in range(rng.randint(1, 40)):
key = rng.randint(0, 50)
nodes.append([h.insert((key, j), j), key, j, True])
for _ in range(rng.randint(0, 30)):
entry = rng.choice(nodes)
if entry[3]:
if rng.random() < 0.5:
key, item = h.extract_min()
best = min((e for e in nodes if e[3]), key=lambda e: (e[1], e[2]))
assert key == (best[1], best[2]) and item == best[2]
best[3] = False
else:
entry[1] = rng.randint(-20, entry[1])
h.decrease_key(entry[0], (entry[1], entry[2]))
ref = sorted((e[1], e[2]) for e in nodes if e[3])
out = [h.extract_min()[0] for _ in range(h.size)]
assert out == ref
assert heapq.nsmallest(len(ref), ref) == out
def test_random_against_kruskal():
rng = random.Random(20261105)
for trial in range(700):
n = rng.randint(1, 12)
m = rng.randint(0, 30)
edges = [(rng.randrange(n), rng.randrange(n), rng.randint(-3, 6)) for _ in range(m)]
limit = None if trial % 3 == 0 else rng.randint(1, 4)
check(n, edges, fredman_tarjan_mst(n, edges, heap_limit=limit))
def test_large_sparse_many_phases():
rng = random.Random(11)
for _ in range(20):
n = rng.randint(50, 200)
edges = [(v, rng.randrange(v), rng.randint(0, 10**6)) for v in range(1, n)]
edges += [(rng.randrange(n), rng.randrange(n), rng.randint(0, 10**6)) for _ in range(n // 2)]
log = []
got = fredman_tarjan_mst(n, edges, log=log)
check(n, edges, got)
assert len(got) == n - 1
for t, m, k, trees in log:
assert trees <= max(1, 2 * m / k)
What the next level needs
- the tests did not run in this build