Replace len magic with count_valid for moving window

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

Author: cwasicki

benchmarks/timeseries/periodic_feature_extractor.py

@@ -115,8 +115,16 @@ def _num_windows(
         Returns:
             The number of windows that are fully contained in the MovingWindow.
         """
-        num_windows = len(window) // period
-        if len(window) - num_windows * period >= window_size:
+
+        def length(window: NDArray[np.float_] | MovingWindow) -> int:
+            return (
+                window.count_valid()
+                if isinstance(window, MovingWindow)
+                else len(window)
+            )
+
+        num_windows = length(window) // period
+        if length(window) - num_windows * period >= window_size:
             num_windows += 1
 
         return num_windows

src/frequenz/sdk/timeseries/_moving_window.py

@@ -376,7 +376,7 @@ def __getitem__(self, key: SupportsIndex | datetime | slice) -> float | ArrayLik
         if isinstance(key, slice):
             if isinstance(key.start, int) or isinstance(key.stop, int):
                 if key.start is None or key.stop is None:
-                    key = slice(slice(key.start, key.stop).indices(self.__len__()))
+                    key = slice(slice(key.start, key.stop).indices(self.count_valid()))
             elif isinstance(key.start, datetime) or isinstance(key.stop, datetime):
                 if key.start is None:
                     key = slice(self._buffer.time_bound_oldest, key.stop)

src/frequenz/sdk/timeseries/_periodic_feature_extractor.py

@@ -128,13 +128,13 @@ def __init__(
         """Distance between two succeeding intervals in samples."""
 
         _logger.debug("Initializing PeriodicFeatureExtractor!")
-        _logger.debug("MovingWindow size: %i", len(self._moving_window))
+        _logger.debug("MovingWindow size: %i", self._moving_window.count_valid())
         _logger.debug(
             "Period between two succeeding intervals (in samples): %i",
             self._period,
         )
 
-        if not len(self._moving_window) % self._period == 0:
+        if not self._moving_window.count_valid() % self._period == 0:
             raise ValueError(
                 "The MovingWindow size is not a integer multiple of the period."
             )
@@ -323,7 +323,7 @@ def _get_buffer_bounds(
 
         rel_pos = self._get_relative_positions(start, window_size)
 
-        if window_size > len(self._moving_window):
+        if window_size > self._moving_window.count_valid():
             raise ValueError(
                 "The window size must be smaller than the size of the `MovingWindow`"
             )
@@ -379,7 +379,7 @@ def _get_reshaped_np_array(
         (start_pos, end_pos, window_size) = self._get_buffer_bounds(start, end)
 
         if start_pos >= end_pos:
-            window_start = self._buffer[start_pos : len(self._moving_window)]
+            window_start = self._buffer[start_pos : self._moving_window.count_valid()]
             window_end = self._buffer[0:end_pos]
             # make the linter happy
             assert isinstance(window_start, np.ndarray)

tests/timeseries/test_moving_window.py

@@ -143,13 +143,13 @@ async def test_window_size() -> None:
     window, sender = init_moving_window(timedelta(seconds=5))
     async with window:
         assert window.capacity == 5, "Wrong window capacity"
-        assert len(window) == 0, "Window should be empty"
+        assert window.count_valid() == 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"
+        assert window.count_valid() == 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"
+        assert window.count_valid() == 5, "Window should be full"
 
 
 # pylint: disable=redefined-outer-name
@@ -170,7 +170,7 @@ async def test_resampling_window(fake_time: time_machine.Coordinates) -> None:
         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"
+        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)
@@ -179,7 +179,7 @@ async def test_resampling_window(fake_time: time_machine.Coordinates) -> None:
             await asyncio.sleep(0.1)
             fake_time.shift(0.1)
 
-        assert len(window) == window_size / output_sampling
+        assert window.count_valid() == window_size / output_sampling
         for value in window:  # type: ignore
             assert 4.9 < value < 5.1