ring_buffer.py

Go to the documentation of this file.

"""Implementation of a ring buffer using bytearray."""
 
 
class RingBuffer:
    """Basic ring buffer using a bytearray.
 
    Not threadsafe.
    """
 
    def __init__(self, maxlen: int) -> None:
        """Initialize empty buffer."""
        self._buffer_buffer = bytearray(maxlen)
        self._pos_pos = 0
        self._length_length = 0
        self._maxlen_maxlen = maxlen
 
    @property
    def maxlen(self) -> int:
        """Return the maximum size of the buffer."""
        return self._maxlen_maxlen
 
    @property
    def pos(self) -> int:
        """Return the current put position."""
        return self._pos_pos
 
    def __len__(self) -> int:
        """Return the length of data stored in the buffer."""
        return self._length_length
 
    def put(self, data: bytes) -> None:
        """Put a chunk of data into the buffer, possibly wrapping around."""
        data_len = len(data)
        new_pos = self._pos_pos + data_len
        if new_pos >= self._maxlen_maxlen:
            # Split into two chunks
            num_bytes_1 = self._maxlen_maxlen - self._pos_pos
            num_bytes_2 = new_pos - self._maxlen_maxlen
 
            self._buffer_buffer[self._pos_pos : self._maxlen_maxlen] = data[:num_bytes_1]
            self._buffer_buffer[:num_bytes_2] = data[num_bytes_1:]
            new_pos = new_pos - self._maxlen_maxlen
        else:
            # Entire chunk fits at current position
            self._buffer_buffer[self._pos_pos : self._pos_pos + data_len] = data
 
        self._pos_pos = new_pos
        self._length_length = min(self._maxlen_maxlen, self._length_length + data_len)
 
    def getvalue(self) -> bytes:
        """Get bytes written to the buffer."""
        if (self._pos_pos + self._length_length) <= self._maxlen_maxlen:
            # Single chunk
            return bytes(self._buffer_buffer[: self._length_length])
 
        # Two chunks
        return bytes(self._buffer_buffer[self._pos_pos :] + self._buffer_buffer[: self._pos_pos])