Fibonacci heap in Python
Heap with O(1) amortized insert, merge and decrease-key, giving O(m + n log n) Dijkstra.
- File
fibonacci_heap.py, 255 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/fibonacci-heap/. The authors are listed in prov/1EX9SR.jsonl.
from collections.abc import Iterator
from typing import Any, Generic, Protocol, TypeVar
class Ordered(Protocol):
def __lt__(self, other: Any, /) -> bool: ...
K = TypeVar("K", bound=Ordered)
V = TypeVar("V")
class Node(Generic[K, V]):
"""A node of a Fibonacci heap. The caller keeps it as the handle of its item."""
__slots__ = ("key", "item", "p", "child", "left", "right", "degree", "mark", "in_heap")
def __init__(self, key: K, item: V) -> None:
self.key = key
self.item = item
self.p: "Node[K, V] | None" = None
self.child: "Node[K, V] | None" = None
self.left: "Node[K, V]" = self
self.right: "Node[K, V]" = self
self.degree = 0
self.mark = False
self.in_heap = False
def _siblings(x: "Node[K, V]") -> "list[Node[K, V]]":
"""The nodes of the circular list of x, starting at x."""
out = [x]
y = x.right
while y is not x:
out.append(y)
y = y.right
return out
def _splice_before(a: "Node[K, V]", x: "Node[K, V]") -> None:
"""Put the single node x just before a in the circular list of a."""
x.right = a
x.left = a.left
a.left.right = x
a.left = x
def _unlink(x: "Node[K, V]") -> None:
"""Take x out of its circular list and make it a list of one node."""
x.left.right = x.right
x.right.left = x.left
x.left = x.right = x
class FibonacciHeap(Generic[K, V]):
"""A Fibonacci min-heap (Fredman and Tarjan, 1987).
insert, find_min, union and decrease_key take O(1) amortized time;
extract_min and delete take O(log n) amortized time. Node objects are the
handles for decrease_key and delete. Keys need only "<". All loops are
iterative, so a long chain of cascading cuts does not use the Python stack.
"""
def __init__(self) -> None:
self.min: Node[K, V] | None = None
self.n = 0
def __len__(self) -> int:
return self.n
def _add_root(self, x: Node[K, V]) -> None:
x.p = None
if self.min is None:
x.left = x.right = x
self.min = x
else:
_splice_before(self.min, x)
def insert(self, key: K, item: V = None) -> Node[K, V]: # type: ignore[assignment]
"""Add an item with the given key and return its node. O(1) time."""
x: Node[K, V] = Node(key, item)
x.in_heap = True # @line L1
self._add_root(x) # @line L2
assert self.min is not None
if x.key < self.min.key: # @line L3
self.min = x
self.n += 1 # @line L4
return x
def find_min(self) -> Node[K, V]:
"""Return the node of a minimum key. Raise IndexError on an empty heap."""
if self.min is None:
raise IndexError("heap underflow")
return self.min
def union(self, other: "FibonacciHeap[K, V]") -> None:
"""Move all items of other into this heap and empty other. O(1) time."""
if other.min is not None: # @line L5
if self.min is None:
self.min = other.min
else:
a, b = self.min, other.min
a_last, b_last = a.left, b.left
a_last.right, b.left = b, a_last
b_last.right, a.left = a, b_last
if b.key < a.key: # @line L6
self.min = b
self.n += other.n # @line L7
other.min, other.n = None, 0 # @line L8
def extract_min(self) -> Node[K, V]:
"""Remove and return the node of a minimum key. O(log n) amortized time."""
z = self.min # @line L9
if z is None: # @line L10
raise IndexError("heap underflow")
if z.child is not None:
for x in _siblings(z.child): # @line L11
_unlink(x)
self._add_root(x)
z.child = None
if z.right is z: # @line L12
self.min = None
else:
self.min = z.right
_unlink(z)
if self.n == 1: # @line L13
self.min = None
else:
self._consolidate() # @line L14
self.n -= 1 # @line L15
z.in_heap = False
z.degree, z.mark = 0, False
return z # @line L16
def _consolidate(self) -> None:
assert self.min is not None
table: dict[int, Node[K, V]] = {} # @line L17
for w in _siblings(self.min): # @line L18
x, d = w, w.degree # @line L19
while d in table: # @line L20
y = table[d] # @line L21
if y.key < x.key: # @line L22
x, y = y, x
self._link(y, x) # @line L23
del table[d] # @line L24
d += 1
table[d] = x # @line L25
self.min = None # @line L26
for d in sorted(table): # @line L27
x = table[d] # @line L28
_unlink(x)
self._add_root(x) # @line L29
assert self.min is not None
if x.key < self.min.key: # @line L30
self.min = x
def _link(self, y: Node[K, V], x: Node[K, V]) -> None:
_unlink(y) # @line L31
y.p = x # @line L32
if x.child is None:
x.child = y
else:
_splice_before(x.child, y)
x.degree += 1
y.mark = False # @line L33
def decrease_key(self, x: Node[K, V], key: K) -> None:
"""Lower the key of node x to key. O(1) amortized time."""
if not x.in_heap:
raise ValueError("the node is not in the heap")
if x.key < key: # @line L34
raise ValueError("new key is larger than the current key")
x.key = key # @line L35
y = x.p
if y is not None and x.key < y.key: # @line L36
self._cut(x, y) # @line L37
self._cascading_cut(y) # @line L38
assert self.min is not None
if x.key < self.min.key: # @line L39
self.min = x
def _cut(self, x: Node[K, V], y: Node[K, V]) -> None:
if y.child is x: # @line L40
y.child = None if x.right is x else x.right
_unlink(x)
y.degree -= 1
self._add_root(x) # @line L41
x.mark = False
def _cascading_cut(self, y: Node[K, V]) -> None:
while True:
z = y.p # @line L42
if z is None: # @line L43
return
if not y.mark: # @line L44
y.mark = True
return
self._cut(y, z) # @line L45
y = z
def delete(self, x: Node[K, V]) -> None:
"""Remove node x from the heap. O(log n) amortized time."""
if not x.in_heap:
raise ValueError("the node is not in the heap")
y = x.p # @line L46: decrease the key of x to minus infinity
if y is not None:
self._cut(x, y)
self._cascading_cut(y)
self.min = x
self.extract_min() # @line L47
def roots(self) -> list[Node[K, V]]:
"""The roots, in the order of the root list from the minimum."""
return [] if self.min is None else _siblings(self.min)
def __iter__(self) -> Iterator[Node[K, V]]:
stack = self.roots()
while stack:
x = stack.pop()
yield x
if x.child is not None:
stack.extend(_siblings(x.child))
def check_invariant(self) -> None:
"""Check the heap order, the links, the degrees, the count, the minimum and the size bound."""
count = 0
fib = [1, 2]
while len(fib) < 100:
fib.append(fib[-1] + fib[-2])
roots = self.roots()
for r in roots:
assert r.p is None and r.in_heap and r.right.left is r
assert self.min is not None and not r.key < self.min.key
sizes: dict[int, int] = {}
order: list[Node[K, V]] = []
stack = list(roots)
while stack:
x = stack.pop()
order.append(x)
count += 1
kids = [] if x.child is None else _siblings(x.child)
assert len(kids) == x.degree
for c in kids:
assert c.p is x and c.right.left is c and c.in_heap
assert not c.key < x.key
stack.extend(kids)
for x in reversed(order):
kids = [] if x.child is None else _siblings(x.child)
sizes[id(x)] = 1 + sum(sizes[id(c)] for c in kids)
assert sizes[id(x)] >= fib[x.degree] # size >= F(degree + 2)
assert count == self.n
assert (self.min is None) == (self.n == 0)
test_fibonacci_heap.py 135 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/fibonacci-heap/. The authors are listed in prov/1EX9SR.jsonl.
import random
import pytest
from fibonacci_heap import FibonacciHeap, Node, _siblings
def shape(h: FibonacciHeap[int, None]) -> list[str]:
def t(x: Node[int, None]) -> str:
kids = [] if x.child is None else _siblings(x.child)
s = str(x.key) + ("*" if x.mark else "")
return s + ("(" + " ".join(t(c) for c in kids) + ")" if kids else "")
return sorted(t(r) for r in h.roots())
def test_example() -> None:
h: FibonacciHeap[int, None] = FibonacciHeap()
nodes = {k: h.insert(k) for k in [7, 18, 38, 24, 17, 23, 21, 39, 41]}
assert len(h.roots()) == 9 and h.find_min().key == 7
assert h.extract_min().key == 7
assert shape(h) == ["17(24 18(38) 21(23 39(41)))"]
h.decrease_key(nodes[41], 15)
assert shape(h) == ["15", "17(24 18(38) 21(23 39*))"]
h.decrease_key(nodes[39], 16)
assert shape(h) == ["15", "16", "17(24 18(38) 21*(23))"]
h.decrease_key(nodes[23], 5)
assert shape(h) == ["15", "16", "17(24 18(38))", "21", "5"]
assert h.find_min().key == 5
assert h.extract_min().key == 5
assert shape(h) == ["15(21)", "16", "17(24 18(38))"]
h.delete(nodes[38])
assert shape(h) == ["15(21)", "16", "17(24 18*)"]
h.check_invariant()
assert [h.extract_min().key for _ in range(len(h))] == [15, 16, 17, 18, 21, 24]
def test_edges() -> None:
h: FibonacciHeap[int, str] = FibonacciHeap()
with pytest.raises(IndexError):
h.extract_min()
with pytest.raises(IndexError):
h.find_min()
x = h.insert(3, "a")
with pytest.raises(ValueError):
h.decrease_key(x, 4)
h.decrease_key(x, 3)
assert h.extract_min().item == "a"
assert len(h) == 0
with pytest.raises(ValueError):
h.decrease_key(x, 1)
for _ in range(5):
h.insert(2, "same")
assert [h.extract_min().key for _ in range(5)] == [2] * 5
a: FibonacciHeap[int, str] = FibonacciHeap()
b: FibonacciHeap[int, str] = FibonacciHeap()
a.union(b)
b.insert(1, "b")
a.union(b)
assert len(a) == 1 and len(b) == 0 and a.find_min().item == "b"
def test_long_cascade() -> None:
h: FibonacciHeap[int, None] = FibonacciHeap()
nodes = [h.insert(k) for k in range(1, 2050)]
h.extract_min()
h.check_invariant()
rng = random.Random(5)
for x in rng.sample(nodes[1:], 1500):
h.decrease_key(x, x.key - 10000)
h.check_invariant() if rng.random() < 0.01 else None
h.check_invariant()
keys = [h.extract_min().key for _ in range(len(h))]
assert keys == sorted(keys)
def test_random_against_model() -> None:
rng = random.Random(1987)
for _ in range(500):
h: FibonacciHeap[int, int] = FibonacciHeap()
other: FibonacciHeap[int, int] = FibonacciHeap()
live: dict[int, Node[int, int]] = {}
model: dict[int, int] = {}
nxt = 0
for _ in range(rng.randint(0, 60)):
op = rng.random()
if op < 0.35:
k = rng.randint(-50, 50)
live[nxt] = h.insert(k, nxt)
model[nxt] = k
nxt += 1
elif op < 0.45:
k = rng.randint(-50, 50)
live[nxt] = other.insert(k, nxt)
model[nxt] = k
nxt += 1
if rng.random() < 0.3:
h.union(other)
elif op < 0.65:
if len(h):
x = h.extract_min()
in_h = [i for i in model if live[i].in_heap or i == x.item]
in_h = [i for i in in_h if i == x.item or _in(h, live[i])]
assert x.key == min(model[i] for i in in_h)
del model[x.item], live[x.item]
else:
with pytest.raises(IndexError):
h.extract_min()
elif op < 0.9:
ids = [i for i in model if _in(h, live[i])]
if ids:
i = rng.choice(ids)
k = model[i] - rng.randint(0, 30)
h.decrease_key(live[i], k)
model[i] = k
else:
ids = [i for i in model if _in(h, live[i])]
if ids:
i = rng.choice(ids)
h.delete(live[i])
del model[i], live[i]
h.check_invariant()
other.check_invariant()
got = sorted(x.key for x in h)
assert got == sorted(model[i] for i in model if _in(h, live[i]))
assert len(h) + len(other) == len(model)
def _in(h: FibonacciHeap[int, int], x: Node[int, int]) -> bool:
while x.p is not None:
x = x.p
return x in h.roots()
What the next level needs
- the tests did not run in this build