Eppstein-Galil-Italiano-Nissenzweig algorithm in Python
Dynamic minimum spanning forest using sparsification.
- File
eppstein_galil_italiano_nissenzweig.py, 102 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/eppstein-galil-italiano-nissenzweig/. The authors are listed in prov/7VG548.jsonl.
"""Fully dynamic minimum spanning forest by sparsification.
The edges are split into groups of at most n edges, at the leaves of a
complete binary tree. Each node keeps the minimum spanning forest of the union
of the forests of its children, so it never holds more than 2(n - 1) edges.
An update changes at most two groups and recomputes the forests on their paths
to the root. Here each node is recomputed by Kruskal's algorithm, which gives
O(n log n log(m/n)) time per update; the paper puts Frederickson's dynamic
structure at each node to reach O(n^(1/2)). Ties between equal weights are
broken by the edge id, so the forest is unique. Deterministic; insertions and
deletions (fully dynamic).
"""
def _kruskal(edge_ids: list[int], edges: dict[int, tuple[int, int, float]]) -> list[int]:
order = sorted(edge_ids, key=lambda e: (edges[e][2], e))
parent: dict[int, int] = {}
def find(x: int) -> int:
root = x
while parent.get(root, root) != root:
root = parent[root]
while parent.get(x, x) != root:
parent[x], x = root, parent[x]
return root
out = []
for e in order:
u, v, _ = edges[e]
a, b = find(u), find(v)
if a != b:
parent[a] = b
out.append(e)
return out
class SparsifiedMSF:
def __init__(self, n: int) -> None:
self.n = n
self.cap = max(1, n) # group size
self.edges: dict[int, tuple[int, int, float]] = {}
self.where: dict[int, int] = {} # edge id -> group index
self.groups: list[list[int]] = []
self.leaves = 1
self.cert: list[list[int]] = [[] for _ in range(2)]
self.next_id = 0
def _recompute_path(self, g: int) -> None:
x = self.leaves + g
self.cert[x] = _kruskal(self.groups[g] if g < len(self.groups) else [], self.edges)
x //= 2
while x >= 1: # @line L6
self.cert[x] = _kruskal(self.cert[2 * x] + self.cert[2 * x + 1], self.edges) # @line L7
x //= 2
def _rebuild(self) -> None:
while self.leaves < len(self.groups):
self.leaves *= 2
self.cert = [[] for _ in range(2 * self.leaves)]
for g, grp in enumerate(self.groups):
self.cert[self.leaves + g] = _kruskal(grp, self.edges)
for x in range(self.leaves - 1, 0, -1):
self.cert[x] = _kruskal(self.cert[2 * x] + self.cert[2 * x + 1], self.edges)
def insert(self, u: int, v: int, w: float) -> int:
e = self.next_id
self.next_id += 1
self.edges[e] = (u, v, w)
if not self.groups or len(self.groups[-1]) >= self.cap:
self.groups.append([]) # @line L1
g = len(self.groups) - 1
self.groups[g].append(e)
self.where[e] = g
if len(self.groups) > self.leaves:
self._rebuild()
else:
self._recompute_path(g) # @line L2
return e
def delete(self, e: int) -> None:
g = self.where.pop(e)
self.groups[g].remove(e) # @line L3
last = len(self.groups) - 1
if g != last:
moved = self.groups[last].pop() # @line L4
self.groups[g].append(moved)
self.where[moved] = g
del self.edges[e]
if not self.groups[last]:
self.groups.pop()
self._recompute_path(g) # @line L5
if last != g:
self._recompute_path(last)
def forest(self) -> list[int]:
return list(self.cert[1]) # @line L8
def weight(self) -> float:
return sum(self.edges[e][2] for e in self.cert[1])
test_eppstein_galil_italiano_nissenzweig.py 67 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/eppstein-galil-italiano-nissenzweig/. The authors are listed in prov/7VG548.jsonl.
import random
from eppstein_galil_italiano_nissenzweig import SparsifiedMSF
def _msf(n: int, edges: dict[int, tuple[int, int, float]]) -> set[int]:
"""Static Kruskal on the whole current graph, ties broken by id."""
parent = list(range(n))
def find(x: int) -> int:
while parent[x] != x:
parent[x] = parent[parent[x]]
x = parent[x]
return x
out = set()
for e in sorted(edges, key=lambda e: (edges[e][2], e)):
u, v, _ = edges[e]
a, b = find(u), find(v)
if a != b:
parent[a] = b
out.add(e)
return out
def test_example() -> None:
s = SparsifiedMSF(4)
ids = {}
for name, (u, v, w) in {"01": (0, 1, 1), "02": (0, 2, 5), "03": (0, 3, 4),
"12": (1, 2, 2), "13": (1, 3, 6), "23": (2, 3, 3)}.items():
ids[name] = s.insert(u, v, w)
assert s.weight() == 6
s.delete(ids["12"])
assert s.weight() == 8
assert {s.edges[e][:2] for e in s.forest()} == {(0, 1), (2, 3), (0, 3)}
def test_random_against_kruskal() -> None:
rng = random.Random(31)
for _ in range(500):
n = rng.randint(1, 7)
s = SparsifiedMSF(n)
alive: list[int] = []
for _ in range(rng.randint(0, 40)):
if alive and rng.random() < 0.4:
e = alive.pop(rng.randrange(len(alive)))
s.delete(e)
else:
u, v = rng.randrange(n), rng.randrange(n)
alive.append(s.insert(u, v, rng.randint(1, 6)))
assert set(s.forest()) == _msf(n, s.edges)
# every node holds a forest, so at most n - 1 edges
assert all(len(c) <= max(0, n - 1) for c in s.cert)
def test_dense_graph() -> None:
rng = random.Random(5)
n = 30
s = SparsifiedMSF(n)
ids = [s.insert(u, v, rng.random()) for u in range(n) for v in range(u + 1, n)]
rng.shuffle(ids)
for e in ids[:200]:
s.delete(e)
assert set(s.forest()) == _msf(n, s.edges)
What the next level needs
- the tests did not run in this build
Read the full page about Eppstein-Galil-Italiano-Nissenzweig algorithm