Fix off-by-one in OrderedRingBuffer.count_covered and add consistency test (#1343)

### Summary

This PR fixes a subtle off-by-one issue in the
`OrderedRingBuffer.count_covered` method and adds a test to ensure
window count consistency in `MovingWindow`.

### Changes

1. **Fix `count_covered` calculation**
- Switched from floating-point division to integer `timedelta` division
(`timedelta // timedelta`).
- This eliminates rounding errors in sample count calculations and
ensures accurate coverage for all window sizes and sampling periods.

2. **Add `test_moving_window_length`**
- New test verifies that the moving window length is consistent without
resampling.
- Confirms that the count of covered samples matches the expected window
capacity.
- Helps prevent regressions related to off-by-one errors in the future.

### Motivation

Floating-point division could produce off-by-one errors for certain
window sizes and sampling periods, causing `count_covered` to be
inaccurate. Switching to integer-based `timedelta` division ensures
deterministic and exact counts. The new test provides coverage to catch
similar issues.

Author: cwasicki

RELEASE_NOTES.md

@@ -1,5 +1,17 @@
 # Frequenz Python SDK Release Notes
 
-## Bug fixes
+## Summary
 
-- `FormulaEngine` and `FormulaEngine3Phase` are now type aliases to `Formula` and `Formula3Phase`, fixing a typing issue introduced in `v1.0.0-rc2202`.
+<!-- Here goes a general summary of what this release is about -->
+
+## Upgrading
+
+<!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
+
+## New Features
+
+<!-- Here goes the main new features and examples or instructions on how to use them -->
+
+## Bug Fixes
+
+- Fixed an off-by-one calculation in `OrderedRingBuffer.count_covered` by switching to integer timedelta division, ensuring accurate sample counting for all window sizes and sampling periods.

benchmarks/timeseries/periodic_feature_extractor.py

@@ -71,8 +71,7 @@ def _calculate_avg_window(
     reshaped = feature_extractor._reshape_np_array(  # pylint: disable=protected-access
         window, window_size
     )
-    # ignoring the type because np.average returns Any
-    return np.average(reshaped[:, :window_size], axis=0)  # type: ignore[no-any-return]
+    return np.average(reshaped[:, :window_size], axis=0)
 
 
 def _calculate_avg_window_py(

pyproject.toml

@@ -30,7 +30,7 @@ dependencies = [
   # changing the version
   # (plugins.mkdocstrings.handlers.python.import)
   "frequenz-client-microgrid >= 0.18.1, < 0.19.0",
-  "frequenz-microgrid-component-graph >= 0.3.2, < 0.4",
+  "frequenz-microgrid-component-graph >= 0.3.2, < 0.3.4",
   "frequenz-client-common >= 0.3.6, < 0.4.0",
   "frequenz-channels >= 1.6.1, < 2.0.0",
   "frequenz-quantities[marshmallow] >= 1.0.0, < 2.0.0",

src/frequenz/sdk/timeseries/_periodic_feature_extractor.py

@@ -409,6 +409,4 @@ def avg(
             The averaged timeseries window.
         """
         (reshaped, window_size) = self._get_reshaped_np_array(start, end)
-        return np.average(  # type: ignore[no-any-return]
-            reshaped[:, :window_size], axis=0, weights=weights
-        )
+        return np.average(reshaped[:, :window_size], axis=0, weights=weights)

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

@@ -705,10 +705,7 @@ def count_covered(
             The count of samples between the oldest and newest (inclusive) valid samples
                 or 0 if there are is no time range covered.
         """
-        return int(
-            self._covered_time_range(since, until).total_seconds()
-            // self._sampling_period.total_seconds()
-        )
+        return self._covered_time_range(since, until) // self._sampling_period
 
     def count_valid(
         self, *, since: datetime | None = None, until: datetime | None = None

tests/timeseries/test_moving_window.py

@@ -560,6 +560,33 @@ async def test_resampling_window(fake_time: time_machine.Coordinates) -> None:
             assert 4.9 < value < 5.1
 
 
+async def test_moving_window_length(fake_time: time_machine.Coordinates) -> None:
+    """Test moving window length without resampling."""
+    channel = Broadcast[Sample[Quantity]](name="net_power")
+    sender = channel.new_sender()
+
+    window_size = timedelta(seconds=1)
+    input_sampling = timedelta(seconds=0.1)
+
+    async with MovingWindow(
+        size=window_size,
+        resampled_data_recv=channel.new_receiver(),
+        input_sampling_period=input_sampling,
+    ) as window:
+        assert window.capacity == window_size / input_sampling, "Wrong window capacity"
+        assert window.count_valid() == 0, "Window should be empty at the beginning"
+        stream_values = [4.0, 8.0, 2.0, 6.0, 5.0] * 100
+        for value in stream_values:
+            timestamp = datetime.now(tz=timezone.utc)
+            sample = Sample(timestamp, Quantity(float(value)))
+            await sender.send(sample)
+            await asyncio.sleep(0.1)
+            fake_time.shift(0.1)
+            assert window.count_valid() <= window.count_covered()
+
+        assert window.count_valid() == window_size / input_sampling
+
+
 async def test_timestamps() -> None:
     """Test indexing a window by timestamp."""
     window, sender = init_moving_window(timedelta(seconds=5))