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
# 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