Algorithm Search⌘K

Catenable deque in Python

A persistent double-ended queue supporting concatenation in O(1) worst-case time.

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

"""A persistent catenable deque in the style of Okasaki's simple c-deque.

A c-deque is Shallow(d) with 0 to 4 elements, or Deep(f, m, r) where the
front f and the rear r hold 2 to 4 elements and the middle m is a c-deque
whose elements are tuples of 2 or 3 elements of the level above.  This
version is strict: the middle is built at once, not suspended.  Each
operation returns a new deque and never changes an old one.
"""

from collections.abc import Iterable, Iterator
from typing import Any, Generic, TypeVar, Union

T = TypeVar("T")


class _Shallow:
    __slots__ = ("d",)

    def __init__(self, d: tuple[Any, ...]) -> None:
        self.d = d


class _Deep:
    __slots__ = ("f", "m", "r")

    def __init__(self, f: tuple[Any, ...], m: "_CD", r: tuple[Any, ...]) -> None:
        self.f = f
        self.m = m
        self.r = r


_CD = Union[_Shallow, _Deep]
_EMPTY = _Shallow(())


def _is_empty(q: _CD) -> bool:
    return isinstance(q, _Shallow) and not q.d


def _from_seq(s: tuple[Any, ...]) -> _CD:
    """A c-deque of at most 8 elements."""
    if len(s) <= 4:
        return _Shallow(s)
    h = len(s) // 2
    return _Deep(s[:h], _EMPTY, s[h:])


def _cons(x: Any, q: _CD) -> _CD:
    if isinstance(q, _Shallow) and len(q.d) < 4:  # @line L1
        return _Shallow((x,) + q.d)
    if isinstance(q, _Shallow):  # @line L2
        a, b, c, d = q.d
        return _Deep((x, a), _EMPTY, (b, c, d))
    f, m, r = q.f, q.m, q.r  # @line L3
    if len(f) < 4:  # @line L4
        return _Deep((x,) + f, m, r)
    a, b, c, d = f  # @line L5
    return _Deep((x, a), _cons((b, c, d), m), r)  # @line L6


def _snoc(q: _CD, x: Any) -> _CD:
    if isinstance(q, _Shallow) and len(q.d) < 4:
        return _Shallow(q.d + (x,))
    if isinstance(q, _Shallow):
        a, b, c, d = q.d
        return _Deep((a, b, c), _EMPTY, (d, x))
    f, m, r = q.f, q.m, q.r
    if len(r) < 4:
        return _Deep(f, m, r + (x,))
    a, b, c, d = r
    return _Deep(f, _snoc(m, (a, b, c)), (d, x))


def _head(q: _CD) -> Any:
    return q.d[0] if isinstance(q, _Shallow) else q.f[0]


def _last(q: _CD) -> Any:
    return q.d[-1] if isinstance(q, _Shallow) else q.r[-1]


def _tail(q: _CD) -> _CD:
    if isinstance(q, _Shallow):  # @line L7
        return _Shallow(q.d[1:])
    f, m, r = q.f, q.m, q.r  # @line L8
    if len(f) > 2:  # @line L9
        return _Deep(f[1:], m, r)
    b = f[1]  # @line L10
    if not _is_empty(m):  # @line L11
        return _Deep((b,) + _head(m), _tail(m), r)
    s = (b,) + r  # @line L12
    if len(s) <= 4:  # @line L13
        return _Shallow(s)
    return _Deep(s[:2], _EMPTY, s[2:])  # @line L14


def _init(q: _CD) -> _CD:
    if isinstance(q, _Shallow):
        return _Shallow(q.d[:-1])
    f, m, r = q.f, q.m, q.r
    if len(r) > 2:
        return _Deep(f, m, r[:-1])
    b = r[0]
    if not _is_empty(m):
        return _Deep(f, _init(m), _last(m) + (b,))
    s = f + (b,)
    if len(s) <= 4:
        return _Shallow(s)
    return _Deep(s[:3], _EMPTY, s[3:])


def _push_front(d: tuple[Any, ...], m: _CD) -> _CD:
    if len(d) <= 3:  # @line L15
        return _cons(d, m)
    return _cons(d[:2], _cons(d[2:], m))  # @line L16


def _push_back(m: _CD, d: tuple[Any, ...]) -> _CD:
    if len(d) <= 3:
        return _snoc(m, d)
    return _snoc(_snoc(m, d[:2]), d[2:])


def _cat(q1: _CD, q2: _CD) -> _CD:
    if isinstance(q1, _Shallow) and isinstance(q2, _Shallow):  # @line L17
        return _from_seq(q1.d + q2.d)
    if isinstance(q1, _Shallow):  # @line L18
        assert isinstance(q2, _Deep)
        if len(q1.d) <= 1:  # @line L19
            return _cons(q1.d[0], q2) if q1.d else q2
        return _Deep(q1.d, _push_front(q2.f, q2.m), q2.r)  # @line L20
    if isinstance(q2, _Shallow):  # @line L21
        if len(q2.d) <= 1:  # @line L22
            return _snoc(q1, q2.d[0]) if q2.d else q1
        return _Deep(q1.f, _push_back(q1.m, q1.r), q2.d)  # @line L23
    m = _cat(_push_back(q1.m, q1.r), _push_front(q2.f, q2.m))  # @line L24
    return _Deep(q1.f, m, q2.r)


def _items(x: Any, level: int) -> Iterator[Any]:
    if level == 0:
        yield x
    else:
        for y in x:
            yield from _items(y, level - 1)


def _walk(q: _CD, level: int) -> Iterator[Any]:
    if isinstance(q, _Shallow):
        for x in q.d:
            yield from _items(x, level)
        return
    for x in q.f:
        yield from _items(x, level)
    yield from _walk(q.m, level + 1)
    for x in q.r:
        yield from _items(x, level)


def _check(q: _CD, level: int) -> int:
    """Check the invariant and return the number of items below q."""

    def elem(x: Any, lv: int) -> int:
        if lv == 0:
            return 1
        assert isinstance(x, tuple) and 2 <= len(x) <= 3
        return sum(elem(y, lv - 1) for y in x)

    if isinstance(q, _Shallow):
        assert len(q.d) <= 4
        return sum(elem(x, level) for x in q.d)
    assert isinstance(q, _Deep)
    assert 2 <= len(q.f) <= 4 and 2 <= len(q.r) <= 4
    return (
        sum(elem(x, level) for x in q.f)
        + _check(q.m, level + 1)
        + sum(elem(x, level) for x in q.r)
    )


def _depth(q: _CD) -> int:
    k = 0
    while isinstance(q, _Deep):
        q = q.m
        k += 1
    return k


class CatenableDeque(Generic[T]):
    """An immutable catenable deque. Every method returns a new deque."""

    __slots__ = ("_q", "_n")

    def __init__(self, _q: _CD = _EMPTY, _n: int = 0) -> None:
        self._q = _q
        self._n = _n

    @classmethod
    def from_iterable(cls, xs: Iterable[T]) -> "CatenableDeque[T]":
        q: CatenableDeque[T] = cls()
        for x in xs:
            q = q.snoc(x)
        return q

    def __len__(self) -> int:
        return self._n

    def is_empty(self) -> bool:
        return self._n == 0

    def cons(self, x: T) -> "CatenableDeque[T]":
        return CatenableDeque(_cons(x, self._q), self._n + 1)

    def snoc(self, x: T) -> "CatenableDeque[T]":
        return CatenableDeque(_snoc(self._q, x), self._n + 1)

    def head(self) -> T:
        if self._n == 0:
            raise IndexError("head of an empty deque")
        return _head(self._q)

    def last(self) -> T:
        if self._n == 0:
            raise IndexError("last of an empty deque")
        return _last(self._q)

    def tail(self) -> "CatenableDeque[T]":
        if self._n == 0:
            raise IndexError("tail of an empty deque")
        return CatenableDeque(_tail(self._q), self._n - 1)

    def init(self) -> "CatenableDeque[T]":
        if self._n == 0:
            raise IndexError("init of an empty deque")
        return CatenableDeque(_init(self._q), self._n - 1)

    def concat(self, other: "CatenableDeque[T]") -> "CatenableDeque[T]":
        return CatenableDeque(_cat(self._q, other._q), self._n + other._n)

    def __add__(self, other: "CatenableDeque[T]") -> "CatenableDeque[T]":
        return self.concat(other)

    def __iter__(self) -> Iterator[T]:
        return _walk(self._q, 0)

    def to_list(self) -> list[T]:
        return list(self)

    def depth(self) -> int:
        """The number of Deep levels on the spine."""
        return _depth(self._q)

    def check_invariant(self) -> None:
        assert _check(self._q, 0) == self._n
test_catenable_deque.py 104 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/catenable-deque/. The authors are listed in prov/776W85.jsonl.

import math
import random

import pytest

from catenable_deque import CatenableDeque


def test_example() -> None:
    a = CatenableDeque.from_iterable([1, 2, 3, 4, 5])
    b = CatenableDeque.from_iterable([6, 7, 8, 9, 10])
    a.check_invariant()
    assert a.depth() == 1 and b.depth() == 1
    c = a + b
    c.check_invariant()
    assert c.to_list() == list(range(1, 11))
    assert c.depth() == 1
    assert c._q.f == (1, 2, 3) and c._q.r == (9, 10)
    assert c._q.m.d == ((4, 5), (6, 7, 8))
    assert c.head() == 1 and c.last() == 10
    t = c.tail().tail().tail()
    t.check_invariant()
    assert t.to_list() == list(range(4, 11))
    assert a.to_list() == [1, 2, 3, 4, 5]
    assert (c + c).to_list() == list(range(1, 11)) * 2


def test_empty() -> None:
    q: CatenableDeque[int] = CatenableDeque()
    q.check_invariant()
    assert len(q) == 0 and q.is_empty() and q.to_list() == []
    for op in (q.head, q.last, q.tail, q.init):
        with pytest.raises(IndexError):
            op()
    assert (q + q).to_list() == []
    one = q.cons(7)
    assert (one + q).to_list() == [7] and (q + one).to_list() == [7]


def test_ends() -> None:
    q = CatenableDeque.from_iterable(range(200))
    out = []
    while not q.is_empty():
        out.append(q.head())
        q = q.tail()
        q.check_invariant()
    assert out == list(range(200))
    q = CatenableDeque.from_iterable(range(200))
    out = []
    while not q.is_empty():
        out.append(q.last())
        q = q.init()
        q.check_invariant()
    assert out == list(range(199, -1, -1))


def test_depth_bound() -> None:
    q: CatenableDeque[int] = CatenableDeque()
    for k in range(1, 3000):
        q = q.cons(k) if k % 3 else q.snoc(k)
        if k % 97 == 0 and len(q) < 20000:
            q = q + q
        assert q.depth() <= max(0, math.log2(len(q)))
    q.check_invariant()


def test_random_persistent() -> None:
    rng = random.Random(776)
    versions: list[tuple[CatenableDeque[int], list[int]]] = [(CatenableDeque(), [])]
    checks = 0
    for step in range(3000):
        q, ref = rng.choice(versions)
        op = rng.randrange(7)
        if op == 0:
            x = rng.randrange(1000)
            q, ref = q.cons(x), [x] + ref
        elif op == 1:
            x = rng.randrange(1000)
            q, ref = q.snoc(x), ref + [x]
        elif op in (2, 3):
            q2, ref2 = rng.choice(versions)
            if len(ref) + len(ref2) > 400:
                continue
            q, ref = q + q2, ref + ref2
        elif op == 4 and ref:
            assert q.head() == ref[0]
            q, ref = q.tail(), ref[1:]
        elif op == 5 and ref:
            assert q.last() == ref[-1]
            q, ref = q.init(), ref[:-1]
        else:
            continue
        q.check_invariant()
        assert len(q) == len(ref)
        assert q.to_list() == ref
        checks += 1
        versions.append((q, ref))
    for q, ref in versions:
        assert q.to_list() == ref
        checks += 1
    assert checks >= 500

What the next level needs

  • the tests did not run in this build

Read the full page about Catenable deque