Merge branch 'feature/2' into dev

ID: 2
Type: feature
Title: implement circular buffer
Status: done
Priority: high
Created: 2025-05-04
Description: implement a simple circular buffer

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

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
+

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)