Algorithm Search⌘K

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

Read the full page about Indexed priority queue