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