Fix ring buffer len (#445)

- RingBuffer: Fix off-by-one error in gap calculation
- RingBuffer: Rename variables and add comments for more clarity
- RingBuffer: Decrease test samples to speedup a long test
- RingBuffer: Fix gaps being ignored in __len__()

fixes #442

Author: Marenz

RELEASE_NOTES.md

@@ -6,7 +6,7 @@
 
 ## Upgrading
 
-<!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with --> 
+<!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
 
 ## New Features
 
@@ -16,4 +16,8 @@
 
 ## Bug Fixes
 
-<!-- Here goes notable bug fixes that are worth a special mention or explanation -->
+- Two bugs in the ring buffer which is used by the `MovingWindow` class were fixed:
+  - `len(buffer)` was not considering potentially existing gaps (areas without elements) in the buffer.
+  - A off-by-one error in the gap calculation logic was fixed that recorded a
+    gap when there was none if an element with a future timestamp was added that
+    would create a gap of exactly 1.

src/frequenz/sdk/timeseries/_ringbuffer/buffer.py

@@ -82,7 +82,7 @@ def __init__(
         self._gaps: list[Gap] = []
         self._datetime_newest: datetime = self._DATETIME_MIN
         self._datetime_oldest: datetime = self._DATETIME_MAX
-        self._time_range: timedelta = (len(self._buffer) - 1) * sampling_period
+        self._full_time_range: timedelta = len(self._buffer) * self._sampling_period
 
     @property
     def sampling_period(self) -> timedelta:
@@ -145,7 +145,9 @@ def update(self, sample: Sample[QuantityT]) -> None:
         # Update timestamps
         prev_newest = self._datetime_newest
         self._datetime_newest = max(self._datetime_newest, timestamp)
-        self._datetime_oldest = self._datetime_newest - self._time_range
+        self._datetime_oldest = self._datetime_newest - (
+            self._full_time_range - self._sampling_period
+        )
 
         # Update data
         value: float = np.nan if sample.value is None else sample.value.base_value
@@ -279,39 +281,39 @@ def is_missing(self, timestamp: datetime) -> bool:
         return any(map(lambda gap: gap.contains(timestamp), self._gaps))
 
     def _update_gaps(
-        self, timestamp: datetime, newest: datetime, new_missing: bool
+        self, timestamp: datetime, newest: datetime, record_as_missing: bool
     ) -> None:
         """Update gap list with new timestamp.
 
         Args:
             timestamp: Timestamp of the new value.
             newest: Timestamp of the newest value before the current update.
-            new_missing: if true, the given timestamp will be recorded as missing.
-
+            record_as_missing: if `True`, the given timestamp will be recorded as missing.
         """
-        currently_missing = self.is_missing(timestamp)
+        found_in_gaps = self.is_missing(timestamp)
 
-        if not new_missing:
+        if not record_as_missing:
             # Replace all gaps with one if we went far into then future
-            if self._datetime_newest - newest >= self._time_range:
+            if self._datetime_newest - newest >= self._full_time_range:
                 self._gaps = [
                     Gap(start=self._datetime_oldest, end=self._datetime_newest)
                 ]
                 return
 
-            if not currently_missing and timestamp > newest + self._sampling_period:
+            # Check if we created a gap with the addition of the new value
+            if not found_in_gaps and timestamp > newest + self._sampling_period:
                 self._gaps.append(
                     Gap(start=newest + self._sampling_period, end=timestamp)
                 )
 
         # New missing entry that is not already in a gap?
-        if new_missing:
-            if not currently_missing:
+        if record_as_missing:
+            if not found_in_gaps:
                 self._gaps.append(
                     Gap(start=timestamp, end=timestamp + self._sampling_period)
                 )
         elif len(self._gaps) > 0:
-            if currently_missing:
+            if found_in_gaps:
                 self._remove_gap(timestamp)
 
         self._cleanup_gaps()
@@ -515,10 +517,24 @@ def __len__(self) -> int:
         if self._datetime_newest == self._DATETIME_MIN:
             return 0
 
+        # Sum of all elements in the gap ranges
+        sum_missing_entries = max(
+            0,
+            sum(
+                (
+                    gap.end
+                    # Don't look further back than oldest timestamp
+                    - max(gap.start, self._datetime_oldest)
+                )
+                // self._sampling_period
+                for gap in self._gaps
+            ),
+        )
+
         start_index = self.datetime_to_index(self._datetime_oldest)
         end_index = self.datetime_to_index(self._datetime_newest)
 
         if end_index < start_index:
-            return len(self._buffer) - start_index + end_index + 1
+            return len(self._buffer) - start_index + end_index + 1 - sum_missing_entries
 
-        return end_index + 1 - start_index
+        return end_index + 1 - start_index - sum_missing_entries

tests/timeseries/test_ringbuffer.py

@@ -195,14 +195,14 @@ def test_timestamp_ringbuffer_missing_parameter(
 @pytest.mark.parametrize(
     "buffer",
     [
-        OrderedRingBuffer([0.0] * 24 * int(ONE_MINUTE.total_seconds()), ONE_MINUTE),
+        OrderedRingBuffer([0.0] * 10 * int(ONE_MINUTE.total_seconds()), ONE_MINUTE),
         OrderedRingBuffer(
-            np.empty(shape=(24 * int(ONE_MINUTE.total_seconds()),), dtype=np.float64),
+            np.empty(shape=(12 * int(ONE_MINUTE.total_seconds()),), dtype=np.float64),
             ONE_MINUTE,
         ),
-        OrderedRingBuffer([0.0] * 24 * int(FIVE_MINUTES.total_seconds()), FIVE_MINUTES),
+        OrderedRingBuffer([0.0] * 5 * int(FIVE_MINUTES.total_seconds()), FIVE_MINUTES),
         OrderedRingBuffer(
-            np.empty(shape=(24 * int(FIVE_MINUTES.total_seconds())), dtype=np.float64),
+            np.empty(shape=(5 * int(FIVE_MINUTES.total_seconds())), dtype=np.float64),
             FIVE_MINUTES,
         ),
     ],
@@ -289,6 +289,21 @@ def test_len_ringbuffer_samples_fit_buffer_size() -> None:
         assert len(buffer) == len(test_samples)
 
 
+def test_len_with_gaps() -> None:
+    """Test the length when there are gaps in the buffer."""
+    buffer = OrderedRingBuffer(
+        np.empty(shape=10, dtype=float),
+        sampling_period=timedelta(seconds=1),
+        align_to=datetime(1, 1, 1, tzinfo=timezone.utc),
+    )
+
+    for i in range(10):
+        buffer.update(
+            Sample(datetime(2, 2, 2, 0, 0, i, tzinfo=timezone.utc), Quantity(float(i)))
+        )
+        assert len(buffer) == i + 1
+
+
 def test_len_ringbuffer_samples_overwrite_buffer() -> None:
     """Test the length of ordered ring buffer.
 
@@ -335,3 +350,22 @@ def test_ringbuffer_empty_buffer() -> None:
             sampling_period=timedelta(seconds=1),
             align_to=datetime(1, 1, 1),
         )
+
+
+def test_off_by_one_gap_logic_bug() -> None:
+    """Test off by one bug in the gap calculation."""
+    buffer = OrderedRingBuffer(
+        np.empty(shape=2, dtype=float),
+        sampling_period=timedelta(seconds=1),
+        align_to=datetime(1, 1, 1, tzinfo=timezone.utc),
+    )
+
+    base_time = datetime(2023, 1, 1, tzinfo=timezone.utc)
+
+    times = [base_time, base_time + timedelta(seconds=1)]
+
+    buffer.update(Sample(times[0], Quantity(1.0)))
+    buffer.update(Sample(times[1], Quantity(2.0)))
+
+    assert buffer.is_missing(times[0]) is False
+    assert buffer.is_missing(times[1]) is False

tests/timeseries/test_ringbuffer_serialization.py

@@ -61,8 +61,6 @@ def load_dump_test(dumped: rb.OrderedRingBuffer[Any], path: str) -> None:
     # pylint: disable=protected-access
     assert dumped._gaps == loaded._gaps
     # pylint: disable=protected-access
-    assert dumped._time_range == loaded._time_range
-    # pylint: disable=protected-access
     assert dumped._sampling_period == loaded._sampling_period
     # pylint: disable=protected-access
     assert dumped._time_index_alignment == loaded._time_index_alignment