Algorithm Search⌘K

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
fredman_tarjan_mst.pyRawGitHub
# 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

Read the full page about Fredman-Tarjan algorithm