Add capacity property to moving window

The capacity is the maximum number of values that the moving window can
hold.

Signed-off-by: cwasicki <[email protected]>

Author: cwasicki

RELEASE_NOTES.md

@@ -19,6 +19,8 @@
 - A new class `Fuse` has been added to represent fuses. This class has a member variable `max_current` which represents the maximum current that can course through the fuse. If the current flowing through a fuse is greater than this limit, then the fuse will break the circuit.
 
 
+- `MovingWindow`: Provide access to `capacity` (maximum number of elements).
+
 ## Bug Fixes
 
 <!-- Here goes notable bug fixes that are worth a special mention or explanation -->

src/frequenz/sdk/timeseries/_moving_window.py

@@ -209,6 +209,19 @@ def sampling_period(self) -> timedelta:
         """
         return self._sampling_period
 
+    @property
+    def capacity(self) -> int:
+        """
+        Return the capacity of the MovingWindow.
+
+        Capacity is the maximum number of samples that can be stored in the
+        MovingWindow.
+
+        Returns:
+            The capacity of the MovingWindow.
+        """
+        return self._buffer.maxlen
+
     async def _run_impl(self) -> None:
         """Awaits samples from the receiver and updates the underlying ring buffer.
 

tests/timeseries/test_moving_window.py

@@ -125,8 +125,14 @@ async def test_window_size() -> None:
     """Test the size of the window."""
     window, sender = init_moving_window(timedelta(seconds=5))
     async with window:
-        await push_logical_meter_data(sender, range(0, 20))
-        assert len(window) == 5
+        assert window.capacity == 5, "Wrong window capacity"
+        assert len(window) == 0, "Window should be empty"
+        await push_logical_meter_data(sender, range(0, 2))
+        assert window.capacity == 5, "Wrong window capacity"
+        assert len(window) == 2, "Window should be partially full"
+        await push_logical_meter_data(sender, range(2, 20))
+        assert window.capacity == 5, "Wrong window capacity"
+        assert len(window) == 5, "Window should be full"
 
 
 # pylint: disable=redefined-outer-name
@@ -146,6 +152,8 @@ async def test_resampling_window(fake_time: time_machine.Coordinates) -> None:
         input_sampling_period=input_sampling,
         resampler_config=resampler_config,
     ) as window:
+        assert window.capacity == window_size / output_sampling, "Wrong window capacity"
+        assert len(window) == 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)