Indexed priority queue in Python
Heap keyed by item indices, allowing O(log n) decrease-key and delete of a given item.
- File
indexed_priority_queue.py, 138 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/indexed-priority-queue/. The authors are listed in prov/QQ610A.jsonl.
from typing import Any, Generic, Optional, Protocol, TypeVar
class Ordered(Protocol):
def __lt__(self, other: Any, /) -> bool: ...
T = TypeVar("T", bound=Ordered)
class IndexedMinPQ(Generic[T]):
"""A binary min-heap of item indices 0 .. m - 1, ordered by their keys.
pq is the heap, a 0-based list of indices; qp[i] is the position of the
index i in pq, or -1 if i is not in the queue; key[i] is the key of i.
So pq[qp[i]] == i for each i in the queue. insert, extract_min,
decrease_key, change_key and delete take O(log n) time; contains and
key_of take O(1) time. Keys need only "<".
"""
def __init__(self, m: int) -> None:
if m < 0:
raise ValueError("capacity must be at least 0")
self.m = m
self.pq: list[int] = []
self.qp: list[int] = [-1] * m
self.key: list[Optional[T]] = [None] * m
def __len__(self) -> int:
return len(self.pq)
def _check_index(self, i: int) -> None:
if not 0 <= i < self.m:
raise IndexError("index out of range")
def contains(self, i: int) -> bool:
self._check_index(i)
return self.qp[i] != -1
def key_of(self, i: int) -> T:
if not self.contains(i):
raise KeyError("index not in the queue")
k = self.key[i]
assert k is not None
return k
def _less(self, a: int, b: int) -> bool:
ka, kb = self.key[self.pq[a]], self.key[self.pq[b]]
assert ka is not None and kb is not None
return ka < kb
def _exch(self, a: int, b: int) -> None:
pq = self.pq
pq[a], pq[b] = pq[b], pq[a] # @line L13
self.qp[pq[a]] = a
self.qp[pq[b]] = b
def _swim(self, j: int) -> None:
while j > 0 and self._less(j, (j - 1) // 2): # @line L14
self._exch(j, (j - 1) // 2) # @line L15
j = (j - 1) // 2
def _sink(self, j: int) -> None:
n = len(self.pq)
while 2 * j + 1 < n: # @line L16
c = 2 * j + 1 # @line L17
if c + 1 < n and self._less(c + 1, c):
c += 1
if not self._less(c, j): # @line L18
return
self._exch(j, c) # @line L19
j = c
def insert(self, i: int, k: T) -> None:
if self.contains(i): # @line L1
raise KeyError("index already in the queue")
self.qp[i] = len(self.pq) # @line L2
self.pq.append(i)
self.key[i] = k
self._swim(self.qp[i]) # @line L3
def min_index(self) -> int:
if not self.pq:
raise IndexError("min_index from an empty queue")
return self.pq[0]
def extract_min(self) -> tuple[int, T]:
if not self.pq: # @line L4
raise IndexError("extract_min from an empty queue")
i = self.pq[0] # @line L5
k = self.key_of(i)
self._exch(0, len(self.pq) - 1)
self.pq.pop()
self._sink(0) # @line L6
self.qp[i] = -1
self.key[i] = None
return i, k # @line L7
def decrease_key(self, i: int, k: T) -> None:
if self.key_of(i) < k: # @line L8
raise ValueError("new key is larger than the current key")
self.key[i] = k # @line L9
self._swim(self.qp[i])
def change_key(self, i: int, k: T) -> None:
old = self.key_of(i)
self.key[i] = k
if k < old:
self._swim(self.qp[i])
else:
self._sink(self.qp[i])
def delete(self, i: int) -> T:
k = self.key_of(i)
j = self.qp[i] # @line L10
self._exch(j, len(self.pq) - 1)
self.pq.pop()
if j < len(self.pq): # @line L11
self._swim(j)
self._sink(j)
self.qp[i] = -1 # @line L12
self.key[i] = None
return k
def check_invariant(self) -> None:
n = len(self.pq)
for j, i in enumerate(self.pq):
assert self.qp[i] == j
assert self.key[i] is not None
if j > 0:
assert not self._less(j, (j - 1) // 2)
assert sum(1 for p in self.qp if p != -1) == n
for i in range(self.m):
if self.qp[i] == -1:
assert self.key[i] is None
test_indexed_priority_queue.py 130 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/indexed-priority-queue/. The authors are listed in prov/QQ610A.jsonl.
import heapq
import random
import pytest
from indexed_priority_queue import IndexedMinPQ
def test_example() -> None:
q: IndexedMinPQ[int] = IndexedMinPQ(6)
for i, k in [(0, 50), (1, 30), (2, 40), (3, 10), (4, 60)]:
q.insert(i, k)
assert q.pq == [3, 1, 2, 0, 4]
assert q.qp == [3, 1, 2, 0, 4, -1]
q.decrease_key(4, 5)
assert q.pq == [4, 3, 2, 0, 1]
assert q.qp == [3, 4, 2, 1, 0, -1]
assert q.delete(3) == 10
assert q.pq == [4, 1, 2, 0]
assert q.qp == [3, 1, 2, -1, 0, -1]
assert q.extract_min() == (4, 5)
assert q.pq == [1, 0, 2]
q.check_invariant()
def test_edges() -> None:
with pytest.raises(ValueError):
IndexedMinPQ(-1)
q: IndexedMinPQ[int] = IndexedMinPQ(2)
with pytest.raises(IndexError):
q.extract_min()
with pytest.raises(IndexError):
q.min_index()
with pytest.raises(IndexError):
q.insert(2, 0)
q.insert(0, 5)
with pytest.raises(KeyError):
q.insert(0, 1)
with pytest.raises(ValueError):
q.decrease_key(0, 6)
with pytest.raises(KeyError):
q.delete(1)
assert q.min_index() == 0
assert q.delete(0) == 5
assert len(q) == 0
q.insert(0, 3)
assert q.key_of(0) == 3
q.check_invariant()
def test_delete_last_cell() -> None:
q: IndexedMinPQ[int] = IndexedMinPQ(3)
q.insert(0, 1)
q.insert(1, 2)
assert q.delete(1) == 2
q.check_invariant()
assert q.extract_min() == (0, 1)
def test_dijkstra() -> None:
rng = random.Random(5)
for _ in range(60):
n = rng.randrange(1, 15)
adj = [[(rng.randrange(n), rng.randrange(10)) for _ in range(rng.randrange(4))] for _ in range(n)]
dist = [None] * n
q: IndexedMinPQ[int] = IndexedMinPQ(n)
q.insert(0, 0)
best = {0: 0}
while len(q):
u, d = q.extract_min()
dist[u] = d
for v, w in adj[u]:
if dist[v] is None and (v not in best or d + w < best[v]):
if q.contains(v):
q.decrease_key(v, d + w)
else:
q.insert(v, d + w)
best[v] = d + w
ref = [None] * n
h = [(0, 0)]
while h:
d, u = heapq.heappop(h)
if ref[u] is not None:
continue
ref[u] = d
for v, w in adj[u]:
if ref[v] is None:
heapq.heappush(h, (d + w, v))
assert dist == ref
def test_random_against_model() -> None:
rng = random.Random(99)
for _ in range(500):
m = rng.randrange(1, 15)
q: IndexedMinPQ[int] = IndexedMinPQ(m)
model: dict[int, int] = {}
for _ in range(rng.randrange(1, 50)):
op = rng.randrange(5)
free = [i for i in range(m) if i not in model]
if (op == 0 or not model) and free:
i, k = rng.choice(free), rng.randrange(30)
q.insert(i, k)
model[i] = k
elif not model:
continue
elif op == 1:
i, k = q.extract_min()
assert k == min(model.values())
assert model.pop(i) == k
elif op == 2:
i = rng.choice(sorted(model))
k = rng.randrange(model[i] + 1)
q.decrease_key(i, k)
model[i] = k
elif op == 3:
i = rng.choice(sorted(model))
k = rng.randrange(30)
q.change_key(i, k)
model[i] = k
else:
i = rng.choice(sorted(model))
assert q.delete(i) == model.pop(i)
q.check_invariant()
assert len(q) == len(model)
for i in range(m):
assert q.contains(i) == (i in model)
What the next level needs
- the tests did not run in this build