Algorithm Search⌘K

Interval heap in Python

Double-ended heap in which each node holds an interval of two keys.

File
interval_heap.py, 134 lines
Level
I0 not checked yet
Toolchain
Python 3.14, pytest 9
Test framework
pytest
Run the tests
python3 -m pytest
interval_heap.pyRawGitHub
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/interval-heap/. The authors are listed in prov/XN8WFN.jsonl.

from collections.abc import Iterable
from typing import Any, Generic, Protocol, TypeVar


class Ordered(Protocol):
    def __lt__(self, other: Any, /) -> bool: ...


T = TypeVar("T", bound=Ordered)


class IntervalHeap(Generic[T]):
    """A double-ended priority queue in one 0-based list.

    The node j holds the keys a[2j] (low) and a[2j + 1] (high); the last
    node may hold only a low key. The parent of the node j is (j - 1) // 2.
    Each node has low <= high, and the keys of a node lie in the interval of
    its parent. insert, extract_min and extract_max take O(log n) time;
    find_min and find_max take O(1) time. Keys need only "<".
    """

    def __init__(self, keys: Iterable[T] = ()) -> None:
        self.a: list[T] = []
        for k in keys:
            self.insert(k)

    def __len__(self) -> int:
        return len(self.a)

    def _hi(self, j: int) -> int:
        return 2 * j + 1 if 2 * j + 1 < len(self.a) else 2 * j

    def _sift_up_min(self, j: int) -> None:
        a = self.a
        while j > 0 and a[2 * j] < a[2 * ((j - 1) // 2)]:  # @line L4
            p = (j - 1) // 2
            a[2 * j], a[2 * p] = a[2 * p], a[2 * j]  # @line L5
            j = p

    def _sift_up_max(self, j: int) -> None:
        a = self.a
        while j > 0 and a[2 * ((j - 1) // 2) + 1] < a[self._hi(j)]:  # @line L6
            p = (j - 1) // 2
            h = self._hi(j)
            a[h], a[2 * p + 1] = a[2 * p + 1], a[h]  # @line L7
            j = p

    def insert(self, x: T) -> None:
        a = self.a
        a.append(x)  # @line L1
        k = len(a) - 1
        j = k // 2
        if k % 2 == 1 and a[k] < a[k - 1]:  # @line L2
            a[k], a[k - 1] = a[k - 1], a[k]
        self._sift_up_min(j)  # @line L3
        self._sift_up_max(j)

    def find_min(self) -> T:
        if not self.a:
            raise IndexError("find_min from an empty heap")
        return self.a[0]

    def find_max(self) -> T:
        if not self.a:
            raise IndexError("find_max from an empty heap")
        return self.a[1] if len(self.a) > 1 else self.a[0]

    def _children(self, j: int) -> list[int]:
        return [c for c in (2 * j + 1, 2 * j + 2) if 2 * c < len(self.a)]

    def extract_min(self) -> T:
        a = self.a
        if not a:  # @line L8
            raise IndexError("extract_min from an empty heap")
        r = a[0]  # @line L9
        x = a.pop()
        if not a:
            return r
        a[0] = x  # @line L10
        j = 0
        while True:
            cs = self._children(j)  # @line L11
            if not cs:
                return r
            c = cs[0]
            if len(cs) == 2 and a[2 * cs[1]] < a[2 * cs[0]]:
                c = cs[1]
            if not a[2 * c] < a[2 * j]:  # @line L12
                return r
            a[2 * j], a[2 * c] = a[2 * c], a[2 * j]  # @line L13
            h = self._hi(c)
            if a[h] < a[2 * c]:
                a[h], a[2 * c] = a[2 * c], a[h]
            j = c  # @line L14

    def extract_max(self) -> T:
        a = self.a
        if not a:  # @line L15
            raise IndexError("extract_max from an empty heap")
        if len(a) <= 2:  # @line L16
            return a.pop()
        r = a[1]  # @line L17
        a[1] = a.pop()
        j = 0
        while True:
            cs = self._children(j)  # @line L18
            if not cs:
                return r
            c = cs[0]
            if len(cs) == 2 and a[self._hi(cs[0])] < a[self._hi(cs[1])]:
                c = cs[1]
            if not a[2 * j + 1] < a[self._hi(c)]:  # @line L19
                return r
            h = self._hi(c)
            a[2 * j + 1], a[h] = a[h], a[2 * j + 1]  # @line L20
            if a[h] < a[2 * c]:
                a[h], a[2 * c] = a[2 * c], a[h]
            j = c  # @line L21

    def nodes(self) -> list[tuple[T, ...]]:
        return [tuple(self.a[2 * j : 2 * j + 2]) for j in range((len(self.a) + 1) // 2)]

    def check_invariant(self) -> None:
        a = self.a
        for j in range((len(a) + 1) // 2):
            lo, hi = a[2 * j], a[self._hi(j)]
            assert not hi < lo
            if j > 0:
                p = (j - 1) // 2
                assert not lo < a[2 * p]
                assert not a[2 * p + 1] < hi
test_interval_heap.py 73 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/interval-heap/. The authors are listed in prov/XN8WFN.jsonl.

import random

import pytest

from interval_heap import IntervalHeap


def test_example() -> None:
    h: IntervalHeap[int] = IntervalHeap()
    for x in [5, 2, 8, 6, 1, 9, 3]:
        h.insert(x)
        h.check_invariant()
    assert h.nodes() == [(1, 9), (3, 6), (2, 8), (5,)]
    h.insert(7)
    assert h.nodes() == [(1, 9), (3, 7), (2, 8), (5, 6)]
    assert h.extract_min() == 1
    assert h.nodes() == [(2, 9), (3, 7), (6, 8), (5,)]
    assert h.extract_max() == 9
    assert h.nodes() == [(2, 8), (3, 7), (5, 6)]
    assert (h.find_min(), h.find_max()) == (2, 8)
    h.check_invariant()


def test_edges() -> None:
    h: IntervalHeap[int] = IntervalHeap()
    for f in (h.find_min, h.find_max, h.extract_min, h.extract_max):
        with pytest.raises(IndexError):
            f()
    h.insert(4)
    assert (h.find_min(), h.find_max()) == (4, 4)
    assert h.extract_max() == 4
    h.insert(4)
    assert h.extract_min() == 4
    h = IntervalHeap([3, 1])
    assert h.extract_min() == 1
    assert h.extract_max() == 3
    assert len(h) == 0


def test_duplicates() -> None:
    h = IntervalHeap([2, 2, 2, 1, 1, 3, 3])
    h.check_invariant()
    out = [h.extract_min() for _ in range(7)]
    assert out == [1, 1, 2, 2, 2, 3, 3]


def test_random_against_model() -> None:
    rng = random.Random(31)
    for _ in range(500):
        h: IntervalHeap[int] = IntervalHeap()
        model: list[int] = []
        for _ in range(rng.randrange(1, 60)):
            op = rng.randrange(3)
            if op == 0 or not model:
                x = rng.randrange(20)
                h.insert(x)
                model.append(x)
            elif op == 1:
                m = min(model)
                model.remove(m)
                assert h.extract_min() == m
            else:
                m = max(model)
                model.remove(m)
                assert h.extract_max() == m
            h.check_invariant()
            assert sorted(h.a) == sorted(model)
            if model:
                assert h.find_min() == min(model)
                assert h.find_max() == max(model)

What the next level needs

  • the tests did not run in this build

Read the full page about Interval heap