Compressed sparse row in Python
Adjacency lists packed into one edge array with an offset array; cache-friendly and compact.
- File
compressed_sparse_row.py, 96 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/compressed-sparse-row/. The authors are listed in prov/J0260T.jsonl.
from collections.abc import Iterable
def _counting_pass(n: int, pairs: list[tuple[int, int]], key: int) -> tuple[list[int], list[tuple[int, int]]]:
"""Stable counting sort of pairs by pair[key]; returns the offsets and the sorted pairs."""
cnt = [0] * (n + 1) # @line L1
for p in pairs:
cnt[p[key] + 1] += 1 # @line L2
off = [0] * (n + 1)
for u in range(1, n + 1):
off[u] = off[u - 1] + cnt[u] # @line L3
pos = off[:n] # @line L4
out: list[tuple[int, int]] = [(0, 0)] * len(pairs)
for p in pairs: # @line L5
out[pos[p[key]]] = p # @line L6
pos[p[key]] += 1
return off, out
class CompressedSparseRow:
"""A static graph on 0, ..., n - 1 as an offset array and one neighbour array.
The neighbours of u are adj[off[u]:off[u + 1]], in increasing order.
Building takes Theta(n + m) time, neighbors takes Theta(d(u)), degree O(1)
and adjacent O(log d(u)). The space is n + 1 + M entries, with M = m for
a directed graph and M = 2m for an undirected one.
"""
def __init__(self, n: int, edges: Iterable[tuple[int, int]], directed: bool = False) -> None:
if n < 0:
raise ValueError("n must be at least 0")
self.n = n
self.directed = directed
pairs: list[tuple[int, int]] = []
for u, v in edges:
if not (0 <= u < n and 0 <= v < n):
raise IndexError("vertex out of range")
if not directed and u == v:
raise ValueError("an undirected graph has no self-loops")
pairs.append((u, v))
if not directed:
pairs.append((v, u))
# Line L7 in Theta(n + M): a stable pass by v, then a stable pass by u.
_, by_v = _counting_pass(n, pairs, 1) # @line L7
self.off, by_u = _counting_pass(n, by_v, 0)
self.adj: list[int] = [v for _, v in by_u]
for u in range(n):
row = self.adj[self.off[u] : self.off[u + 1]]
if any(row[i] == row[i + 1] for i in range(len(row) - 1)):
raise ValueError("repeated edge")
def neighbors(self, u: int) -> list[int]:
if not 0 <= u < self.n:
raise IndexError("vertex out of range")
return [self.adj[i] for i in range(self.off[u], self.off[u + 1])] # @line L8
def degree(self, u: int) -> int:
if not 0 <= u < self.n:
raise IndexError("vertex out of range")
return self.off[u + 1] - self.off[u] # @line L9
def adjacent(self, u: int, v: int) -> bool:
if not (0 <= u < self.n and 0 <= v < self.n):
raise IndexError("vertex out of range")
lo, hi = self.off[u], self.off[u + 1] # @line L10
while lo < hi: # @line L11
mid = (lo + hi) // 2 # @line L12
if self.adj[mid] < v: # @line L13
lo = mid + 1
else:
hi = mid
return lo < self.off[u + 1] and self.adj[lo] == v # @line L14
def edges(self) -> list[tuple[int, int]]:
return [
(u, self.adj[i])
for u in range(self.n)
for i in range(self.off[u], self.off[u + 1])
if self.directed or u < self.adj[i]
]
def check_invariant(self) -> None:
off, adj, n = self.off, self.adj, self.n
assert len(off) == n + 1 and off[0] == 0 and off[n] == len(adj)
for u in range(n):
assert off[u] <= off[u + 1]
row = adj[off[u] : off[u + 1]]
assert all(row[i] < row[i + 1] for i in range(len(row) - 1))
assert all(0 <= v < n for v in row)
if not self.directed:
assert u not in row
for v in row:
assert self.adjacent(v, u)
test_compressed_sparse_row.py 71 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/compressed-sparse-row/. The authors are listed in prov/J0260T.jsonl.
import random
import pytest
from compressed_sparse_row import CompressedSparseRow, _counting_pass
def test_example() -> None:
e = [(3, 4), (0, 2), (2, 3), (3, 1), (0, 1), (1, 2)]
off, by_u = _counting_pass(5, e, 0)
assert off == [0, 2, 3, 4, 6, 6]
assert [v for _, v in by_u] == [2, 1, 2, 3, 4, 1]
g = CompressedSparseRow(5, e, directed=True)
g.check_invariant()
assert g.off == [0, 2, 3, 4, 6, 6]
assert g.adj == [1, 2, 2, 3, 1, 4]
assert g.neighbors(3) == [1, 4]
assert g.degree(3) == 2 and g.degree(4) == 0
assert g.adjacent(3, 4)
assert not g.adjacent(1, 0)
def test_undirected_example() -> None:
e = [(0, 1), (0, 2), (1, 2), (1, 3), (2, 3), (3, 4), (4, 5)]
g = CompressedSparseRow(6, e)
g.check_invariant()
assert g.off == [0, 2, 5, 8, 11, 13, 14]
assert g.adj == [1, 2, 0, 2, 3, 0, 1, 3, 1, 2, 4, 3, 5, 4]
assert g.edges() == e
def test_edge_cases() -> None:
g = CompressedSparseRow(0, [])
assert g.off == [0] and g.adj == [] and g.edges() == []
h = CompressedSparseRow(3, [])
assert h.off == [0, 0, 0, 0] and not h.adjacent(0, 2)
d = CompressedSparseRow(1, [(0, 0)], directed=True)
assert d.adjacent(0, 0) and d.neighbors(0) == [0]
with pytest.raises(ValueError):
CompressedSparseRow(2, [(1, 1)])
with pytest.raises(ValueError):
CompressedSparseRow(2, [(0, 1), (1, 0)])
with pytest.raises(IndexError):
CompressedSparseRow(2, [(0, 2)])
with pytest.raises(IndexError):
h.degree(3)
@pytest.mark.parametrize("directed", [False, True])
def test_random_against_set_model(directed: bool) -> None:
rng = random.Random(260)
for _ in range(300):
n = rng.randint(0, 8)
pairs = [(u, v) for u in range(n) for v in range(n) if directed or u < v]
model = set(rng.sample(pairs, rng.randint(0, len(pairs))))
given = [(v, u) if not directed and rng.random() < 0.5 else (u, v) for u, v in model]
rng.shuffle(given)
g = CompressedSparseRow(n, given, directed)
g.check_invariant()
assert set(g.edges()) == model
for u in range(n):
nb = {b for a, b in model if a == u}
if not directed:
nb |= {a for a, b in model if b == u}
assert g.neighbors(u) == sorted(nb)
assert g.degree(u) == len(nb)
for v in range(n):
assert g.adjacent(u, v) == (v in nb)
What the next level needs
- the tests did not run in this build