Algorithm Search⌘K

Deap in Python

Double-ended heap with a min side and a max side, both stored in one complete tree.

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

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


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


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


class Deap(Generic[T]):
    """A deap: a double-ended heap in one 1-based array with an empty root.

    The positions 2 .. n hold the keys. The subtree of 2 is a min-heap and
    the subtree of 3 is a max-heap. Each min node i has a partner j in the
    max-heap at the same level, or the parent of that place if it is past
    n, and A[i] <= A[j]. 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[Optional[T]] = [None, None]
        for k in keys:
            self.insert(k)

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

    @property
    def n(self) -> int:
        return len(self.a) - 1

    def _k(self, i: int) -> T:
        v = self.a[i]
        assert v is not None
        return v

    @staticmethod
    def is_max_node(i: int) -> bool:
        k = i.bit_length() - 1
        return (i >> (k - 1)) == 3

    @staticmethod
    def min_partner(j: int) -> int:
        return j - (1 << (j.bit_length() - 2))

    def max_partner(self, i: int) -> int:
        j = i + (1 << (i.bit_length() - 2))
        return j if j <= self.n else j // 2

    def _min_sift_up(self, i: int, x: T) -> None:
        a = self.a
        while i > 3 and x < self._k(i // 2):  # @line L6
            a[i] = a[i // 2]
            i //= 2
        a[i] = x  # @line L7

    def _max_sift_up(self, j: int, x: T) -> None:
        a = self.a
        while j > 3 and self._k(j // 2) < x:  # @line L8
            a[j] = a[j // 2]
            j //= 2
        a[j] = x  # @line L9

    def insert(self, x: T) -> None:
        a = self.a
        a.append(x)  # @line L1
        n = self.n
        if n == 2:
            return
        if self.is_max_node(n):  # @line L2
            i = self.min_partner(n)
            if x < self._k(i):  # @line L3
                a[n] = a[i]
                self._min_sift_up(i, x)
            else:
                self._max_sift_up(n, x)
        else:
            j = self.max_partner(n)  # @line L4
            if self._k(j) < x:  # @line L5
                a[n] = a[j]
                self._max_sift_up(j, x)
            else:
                self._min_sift_up(n, x)

    def find_min(self) -> T:
        if self.n < 2:
            raise IndexError("find_min from an empty deap")
        return self._k(2)

    def find_max(self) -> T:
        if self.n < 2:
            raise IndexError("find_max from an empty deap")
        return self._k(3) if self.n >= 3 else self._k(2)

    def extract_min(self) -> T:
        a = self.a
        if self.n < 2:  # @line L10
            raise IndexError("extract_min from an empty deap")
        r = self._k(2)
        x = self._k(self.n)
        a.pop()
        n = self.n
        if n < 2:  # @line L11
            return r
        if n == 2:
            a[2] = x
            return r
        i = 2
        while 2 * i <= n:  # @line L12
            c = 2 * i
            if c + 1 <= n and self._k(c + 1) < self._k(c):
                c += 1
            a[i] = a[c]
            i = c
        j = self.max_partner(i)  # @line L13
        if self._k(j) < x:  # @line L14
            a[i] = a[j]
            self._max_sift_up(j, x)
        else:
            self._min_sift_up(i, x)
        return r

    def extract_max(self) -> T:
        a = self.a
        if self.n < 2:  # @line L15
            raise IndexError("extract_max from an empty deap")
        if self.n == 2:
            v = a.pop()
            assert v is not None
            return v
        r = self._k(3)  # @line L16
        x = self._k(self.n)
        a.pop()
        n = self.n
        if n == 2:  # @line L17
            return r
        j = 3
        while 2 * j <= n:  # @line L18
            c = 2 * j
            if c + 1 <= n and self._k(c) < self._k(c + 1):
                c += 1
            a[j] = a[c]
            j = c
        i = self.min_partner(j)  # @line L19
        if 2 * i <= n:
            c = 2 * i
            if c + 1 <= n and self._k(c) < self._k(c + 1):
                c += 1
            i = c
        if x < self._k(i):  # @line L20
            a[j] = a[i]
            self._min_sift_up(i, x)
        else:
            self._max_sift_up(j, x)
        return r

    def keys(self) -> list[T]:
        return [self._k(i) for i in range(2, self.n + 1)]

    def check_invariant(self) -> None:
        n = self.n
        for i in range(4, n + 1):
            p = i // 2
            if self.is_max_node(i):
                assert not self._k(p) < self._k(i)
            else:
                assert not self._k(i) < self._k(p)
        for i in range(2, n + 1):
            if not self.is_max_node(i):
                j = self.max_partner(i)
                if j >= 3:
                    assert not self._k(j) < self._k(i)
test_deap.py 86 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/deap/. The authors are listed in prov/5CWRB5.jsonl.

import random

import pytest

from deap import Deap


START = [5, 45, 10, 8, 25, 40, 15, 19, 9, 30, 20]


def test_example() -> None:
    d = Deap(START)
    assert d.keys() == START
    d.check_invariant()
    d.insert(4)
    assert d.keys() == [4, 45, 5, 8, 25, 40, 15, 10, 9, 30, 20, 19]
    d.insert(50)
    assert d.keys() == [4, 50, 5, 8, 25, 45, 15, 10, 9, 30, 20, 19, 40]
    assert d.extract_min() == 4
    assert d.keys() == [5, 50, 10, 8, 40, 45, 15, 19, 9, 30, 20, 25]
    assert d.extract_max() == 50
    assert d.keys() == [5, 45, 10, 8, 40, 30, 15, 19, 9, 25, 20]
    d.check_invariant()


def test_partners() -> None:
    assert [Deap.is_max_node(i) for i in range(2, 16)] == [
        False, True, False, False, True, True,
        False, False, False, False, True, True, True, True,
    ]
    assert Deap.min_partner(13) == 9
    assert Deap.min_partner(7) == 5
    d = Deap(range(12))
    assert d.n == 13
    assert d.max_partner(9) == 13
    assert d.max_partner(10) == 7
    assert d.max_partner(4) == 6


def test_edges() -> None:
    d: Deap[int] = Deap()
    for f in (d.find_min, d.find_max, d.extract_min, d.extract_max):
        with pytest.raises(IndexError):
            f()
    d.insert(3)
    assert (d.find_min(), d.find_max()) == (3, 3)
    assert d.extract_max() == 3
    d.insert(3)
    assert d.extract_min() == 3
    d = Deap([2, 7])
    assert d.extract_min() == 2
    assert d.find_max() == 7
    d = Deap([2, 7, 5])
    assert d.extract_max() == 7
    assert d.extract_max() == 5
    assert d.extract_min() == 2
    assert len(d) == 0


def test_random_against_model() -> None:
    rng = random.Random(77)
    for _ in range(600):
        d: Deap[int] = Deap()
        model: list[int] = []
        for _ in range(rng.randrange(1, 50)):
            op = rng.randrange(3)
            if op == 0 or not model:
                x = rng.randrange(25)
                d.insert(x)
                model.append(x)
            elif op == 1:
                m = min(model)
                model.remove(m)
                assert d.extract_min() == m
            else:
                m = max(model)
                model.remove(m)
                assert d.extract_max() == m
            d.check_invariant()
            assert sorted(d.keys()) == sorted(model)
            if model:
                assert d.find_min() == min(model)
                assert d.find_max() == max(model)

What the next level needs

  • the tests did not run in this build

Read the full page about Deap