diff --git a/TODO b/TODO index 5ea88bb..dd4cd66 100644 --- a/TODO +++ b/TODO @@ -62,7 +62,7 @@ Description: Implement the Knuth-Morris-Pratt algorithm for string searching. ID: 2 Type: feature Title: implement circular buffer -Status: in-progress +Status: done Priority: high Created: 2025-05-04 Description: implement a simple circular buffer diff --git a/src/byteb4rb1e_utils/collections.py b/src/byteb4rb1e_utils/collections.py new file mode 100644 index 0000000..84cc82c --- /dev/null +++ b/src/byteb4rb1e_utils/collections.py @@ -0,0 +1,37 @@ +class CircularBuffer: + """circular buffer implementation for managing streamed data + """ + #: internal buffer storage maintaining a fixed size + buf: bytearray + #: maximum capacity of the buffer + size: int + #: index of the oldest element in the buffer + start: int + #: index where the next element will be inserted + end: int + #: indicates whether the buffer has overwritten older data + filled: bool + + def __init__(self, size: int): + """initializes the circular buffer with a fixed capacity + + :param size: maximum number of bytes the buffer can hold + """ + self.buf = bytearray(size) + self.size = size + self.start = 0 + self.end = 0 + self.filled = False + + def append(self, data: bytes): + """adds data to the circular buffer, overwriting old data if necessary + + :param data: byte sequence to append to the buffer + """ + for byte in data: + self.buf[self.end] = byte + self.end = (self.end + 1) % self.size + if self.end == self.start: # Overwriting case + self.start = (self.start + 1) % self.size + self.filled = True + diff --git a/tests/unit/byteb4rb1e_utils/collections/__init__.py b/tests/unit/byteb4rb1e_utils/collections/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/byteb4rb1e_utils/collections/test_circular_buffer.py b/tests/unit/byteb4rb1e_utils/collections/test_circular_buffer.py new file mode 100644 index 0000000..aeb0934 --- /dev/null +++ b/tests/unit/byteb4rb1e_utils/collections/test_circular_buffer.py @@ -0,0 +1,82 @@ +import unittest + +from byteb4rb1e_utils.collections import CircularBuffer + +class test_init(unittest.TestCase): + """CircularBuffer.__init__()""" + + def test_default(self): + """Test buffer initializes correctly.""" + buffer = CircularBuffer(10) + self.assertEqual(len(buffer.buf), 10) + self.assertEqual(buffer.size, 10) + self.assertEqual(buffer.start, 0) + self.assertEqual(buffer.end, 0) + self.assertFalse(buffer.filled) + + +class test_append(unittest.TestCase): + """CircularBuffer.append()""" + + def test_without_overflow(self): + """Test appending bytes without overwriting.""" + buffer = CircularBuffer(5) + buffer.append(b"abc") + self.assertEqual(buffer.buf[:3], bytearray(b"abc")) + self.assertEqual(buffer.start, 0) + self.assertEqual(buffer.end, 3) + self.assertFalse(buffer.filled) + + def test_with_overflow(self): + """appending bytes with overwriting""" + buffer = CircularBuffer(5) + buffer.append(b"abcde") # Fill buffer completely + buffer.append(b"fg") # Overwrite some elements + + self.assertEqual(buffer.buf, bytearray(b"fgcde")) # Last five elements should remain + self.assertEqual(buffer.start, 3) # Should have moved forward due to overwrite + self.assertEqual(buffer.end, 2) # Wrapped around + self.assertTrue(buffer.filled) # Buffer should indicate overwrite occurred + + def test_wraparound_behavior(self): + """correct handling of buffer wraparound""" + buffer = CircularBuffer(4) + buffer.append(b"abcd") # Fill buffer completely + buffer.append(b"e") # Overwrite first element + + self.assertEqual(buffer.buf, bytearray(b"ebcd")) # Should have wrapped around + self.assertEqual(buffer.start, 2) + self.assertEqual(buffer.end, 1) + self.assertTrue(buffer.filled) + + def test_consecutive_overwrites(self): + """repeated buffer overwrites across multiple cycles + + verifies that the circular buffer correctly handles multiple overwrite cycles. + + - Initially, the buffer is filled completely (`abcde`). + - New data (`fghij`) is appended, causing the oldest elements to be replaced. + - The buffer does not physically shift; only the `start` index moves forward as data is overwritten. + - When wrapping around, the `start` advances while `end` cycles back to zero. + + Expected: + - Buffer contents should be `bytearray(b"fghij")`, containing only + the most recent data + - `start` moves forward due to full overwriting. + - `end` correctly wraps around. + - `filled` flag indicates that overwrites have occurred. + + **Why `start = 1` instead of `0`?** + Well it took some brain juice for me to fully comprehend it, but in a + circular buffer, `start` always marks the oldest remaining element. + When a full overwrite occurs, `start` moves forward to indicate the next + oldest position, ensuring sequential ordering is preserved. + """ + buffer = CircularBuffer(5) + buffer.append(b"abcde") # Fill buffer + buffer.append(b"fghij") # Overwrite completely + + self.assertEqual(buffer.buf, bytearray(b"fghij")) # Only the latest data remains + self.assertEqual(buffer.start, 1) + self.assertEqual(buffer.end, 0) + self.assertTrue(buffer.filled)