Algorithm Search⌘K

Circular buffer in Python

Fixed-size array used as a wrap-around FIFO buffer.

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

from typing import Any, Generic, Iterator, TypeVar

T = TypeVar("T")


class CircularBuffer(Generic[T]):
    """A bounded queue and deque in an array of fixed capacity.

    enqueue, dequeue, push_front, pop_back and get cost O(1) worst case.
    Space: capacity cells. Precondition: capacity >= 1.
    enqueue and push_front raise OverflowError on a full buffer.
    """

    def __init__(self, capacity: int) -> None:
        if capacity < 1:
            raise ValueError("capacity must be at least 1")
        self.cap = capacity
        self.a: list[Any] = [None] * capacity
        self.h = 0
        self.n = 0

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

    def __iter__(self) -> Iterator[T]:
        for i in range(self.n):
            yield self.get(i)

    def is_full(self) -> bool:
        return self.n == self.cap

    def enqueue(self, x: T) -> None:
        """Add x at the back. Cost O(1)."""
        if self.n == self.cap:  # @line L1
            raise OverflowError("buffer is full")
        self.a[(self.h + self.n) % self.cap] = x  # @line L2
        self.n += 1  # @line L3

    def dequeue(self) -> T:
        """Remove and return the front item. Cost O(1)."""
        if self.n == 0:  # @line L4
            raise IndexError("dequeue from empty buffer")
        x = self.a[self.h]  # @line L5
        self.h = (self.h + 1) % self.cap  # @line L6
        self.n -= 1
        return x  # @line L7

    def push_front(self, x: T) -> None:
        """Add x at the front. Cost O(1)."""
        if self.n == self.cap:
            raise OverflowError("buffer is full")
        self.h = (self.h - 1) % self.cap  # @line L8
        self.a[self.h] = x
        self.n += 1

    def pop_back(self) -> T:
        """Remove and return the back item. Cost O(1)."""
        if self.n == 0:
            raise IndexError("pop from empty buffer")
        self.n -= 1  # @line L9
        return self.a[(self.h + self.n) % self.cap]  # @line L10

    def get(self, i: int) -> T:
        """Return the item at position i from the front. Cost O(1)."""
        if not 0 <= i < self.n:
            raise IndexError("position out of range")
        return self.a[(self.h + i) % self.cap]  # @line L11

    def check_invariant(self) -> None:
        assert len(self.a) == self.cap
        assert 0 <= self.h < self.cap
        assert 0 <= self.n <= self.cap
test_circular_buffer.py 76 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/circular-buffer/. The authors are listed in prov/9NS9WP.jsonl.

import random

import pytest

from circular_buffer import CircularBuffer


def test_page_example() -> None:
    B: CircularBuffer[int] = CircularBuffer(4)
    for x in (1, 2, 3):
        B.enqueue(x)
    assert B.dequeue() == 1 and B.h == 1
    B.enqueue(4)
    B.enqueue(5)
    assert B.a == [5, 2, 3, 4] and list(B) == [2, 3, 4, 5]
    with pytest.raises(OverflowError):
        B.enqueue(6)
    assert B.dequeue() == 2
    B.push_front(9)
    assert B.h == 1 and B.a == [5, 9, 3, 4] and list(B) == [9, 3, 4, 5]
    assert B.pop_back() == 5
    assert B.get(2) == 4 and list(B) == [9, 3, 4]
    B.check_invariant()


def test_edges() -> None:
    with pytest.raises(ValueError):
        CircularBuffer(0)
    B: CircularBuffer[str] = CircularBuffer(1)
    with pytest.raises(IndexError):
        B.dequeue()
    with pytest.raises(IndexError):
        B.pop_back()
    with pytest.raises(IndexError):
        B.get(0)
    B.push_front("a")
    assert B.is_full()
    with pytest.raises(OverflowError):
        B.push_front("b")
    assert B.pop_back() == "a" and len(B) == 0
    B.check_invariant()


def test_random_against_list() -> None:
    rng = random.Random(99)
    for _ in range(600):
        cap = rng.randrange(1, 8)
        B: CircularBuffer[int] = CircularBuffer(cap)
        model: list[int] = []
        for _ in range(rng.randrange(1, 50)):
            op = rng.randrange(5)
            x = rng.randrange(100)
            if op == 0:
                if len(model) == cap:
                    with pytest.raises(OverflowError):
                        B.enqueue(x)
                else:
                    B.enqueue(x); model.append(x)
            elif op == 1:
                if len(model) == cap:
                    with pytest.raises(OverflowError):
                        B.push_front(x)
                else:
                    B.push_front(x); model.insert(0, x)
            elif op == 2 and model:
                assert B.dequeue() == model.pop(0)
            elif op == 3 and model:
                assert B.pop_back() == model.pop()
            elif op == 4 and model:
                i = rng.randrange(len(model))
                assert B.get(i) == model[i]
            B.check_invariant()
            assert list(B) == model

What the next level needs

  • the tests did not run in this build

Read the full page about Circular buffer