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

@@ -10,7 +10,7 @@
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+- `MovingWindow`: Provide access to `capacity` (maximum number of elements).
 
 ## Bug Fixes
 

src/frequenz/sdk/timeseries/_moving_window.py

@@ -208,6 +208,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

@@ -121,8 +121,14 @@ async def test_access_empty_window() -> None:
 async def test_window_size() -> None:
     """Test the size of the window."""
     window, sender = init_moving_window(timedelta(seconds=5))
-    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
@@ -143,6 +149,9 @@ async def test_resampling_window(fake_time: time_machine.Coordinates) -> None:
         resampler_config=resampler_config,
     )
 
+    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)