Dig Deeper
Built In Types
Built In Types
class CircularBuffer:
def __init__(self, capacity: int) -> None:
self.capacity = capacity
self.content = []
def read(self) -> str:
if not self.content:
raise BufferEmptyException("Circular buffer is empty")
return self.content.pop(0)
def write(self, data: str) -> None:
if len(self.content) == self.capacity:
raise BufferFullException("Circular buffer is full")
self.content.append(data)
def overwrite(self, data: str) -> None:
if len(self.content) == self.capacity:
self.content.pop(0)
self.write(data)
def clear(self) -> None:
self.content = []
In Python, the list type is ubiquitous and exceptionally versatile.
Code similar to that shown above is a very common way to implement this exercise.
Though lists can do much more, here we use append() to add an entry to the end of the list, and pop(0) to remove an entry from the beginning.
By design, lists have no built-in length limit and can grow arbitrarily, so the main task of the programmer here is to keep track of capacity, and limit it when needed.
A list is also designed to hold an arbitrary mix of Python objects, and this flexibility in content is emphasized over performance.
For more precise control, at the price of some increased programming complexity, it is possible to use a bytearray, or the array.array type from the [array][array-module module.
For details on using array.array, see the standard library approach.
In the case of a bytearray, entries are of fixed type: integers in the range 0 <= n < 256.
The tests are designed such that this is sufficient to solve the exercise, and byte handling may be quite a realistic view of how circular buffers are often used in practice.
The code below shows an implementation using this lower-level collection class:
class CircularBuffer:
def __init__(self, capacity):
self.capacity = bytearray(capacity)
self.read_start = 0
self.write_start = 0
def read(self):
if not any(self.capacity):
raise BufferEmptyException('Circular buffer is empty')
data = chr(self.capacity[self.read_start])
self.capacity[self.read_start] = 0
self.read_start = (self.read_start + 1) % len(self.capacity)
return data
def write(self, data):
if all(self.capacity):
raise BufferFullException('Circular buffer is full')
try:
self.capacity[self.write_start] = data
except TypeError:
self.capacity[self.write_start] = ord(data)
self.write_start = (self.write_start + 1) % len(self.capacity)
def overwrite(self, data):
try:
self.capacity[self.write_start] = data
except TypeError:
self.capacity[self.write_start] = ord(data)
if all(self.capacity) and self.write_start == self.read_start:
self.read_start = (self.read_start + 1) % len(self.capacity)
self.write_start = (self.write_start + 1) % len(self.capacity)
def clear(self):
self.capacity = bytearray(len(self.capacity))
Standard Library
Standard Library
from queue import Queue
class CircularBuffer:
def __init__(self, capacity):
self.buffer = Queue(capacity)
def read(self):
if self.buffer.empty():
raise BufferEmptyException("Circular buffer is empty")
return self.buffer.get()
def write(self, data):
if self.buffer.full():
raise BufferFullException("Circular buffer is full")
self.buffer.put(data)
def overwrite(self, data):
if self.buffer.full():
_ = self.buffer.get()
self.buffer.put(data)
def clear(self):
while not self.buffer.empty():
_ = self.buffer.get()
The above code uses a Queue object to “implement” the buffer, a collection class which assumes entries will be added at the end and removed at the beginning.
This is a “queue” in British English, though Americans call it a “line”.
Alternatively, the collections module provides a deque object, a double-ended queue class.
A deque allows adding and removing entries at both ends, which is not something we need for a circular buffer.
However, the syntax may be even more concise than for a queue:
from collections import deque
from typing import Any
class CircularBuffer:
def __init__(self, capacity: int):
self.buffer = deque(maxlen=capacity)
def read(self) -> Any:
if len(self.buffer) == 0:
raise BufferEmptyException("Circular buffer is empty")
return self.buffer.popleft()
def write(self, data: Any) -> None:
if len(self.buffer) == self.buffer.maxlen:
raise BufferFullException("Circular buffer is full")
self.buffer.append(data)
def overwrite(self, data: Any) -> None:
self.buffer.append(data)
def clear(self) -> None:
if len(self.buffer) > 0:
self.buffer.popleft()
Both Queue and deque have the ability to limit the queues length by declaring a ‘capacity’ or ‘maxlen’ attribute.
This simplifies empty/full and read/write tracking.
Finally, the array class from the array module can be used to initialize a ‘buffer’ that works similarly to a built-in list or bytearray, but with efficiencies in storage and access:
from array import array
class CircularBuffer:
def __init__(self, capacity):
self.buffer = array('u')
self.capacity = capacity
self.marker = 0
def read(self):
if not self.buffer:
raise BufferEmptyException('Circular buffer is empty')
else:
data = self.buffer.pop(self.marker)
if self.marker > len(self.buffer)-1: self.marker = 0
return data
def write(self, data):
if len(self.buffer) < self.capacity:
try:
self.buffer.append(data)
except TypeError:
self.buffer.append(data)
else: raise BufferFullException('Circular buffer is full')
def overwrite(self, data):
if len(self.buffer) < self.capacity: self.buffer.append(data)
else:
self.buffer[self.marker] = data
if self.marker < self.capacity - 1: self.marker += 1
else: self.marker = 0
def clear(self):
self.marker = 0
self.buffer = array('u')
Source: Exercism python/circular-buffer