Make the MovingWindow and PeriodicFeatureExtractor generic

Because the resampler is now generic, we can also make the
`MovingWindow` and `PeriodicFeatureExtractor` generic so they can return
a specialized quantity instead of a unit-less quantity, again improving
performance and safety.

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

src/frequenz/sdk/timeseries/_moving_window.py

@@ -9,22 +9,21 @@
 import math
 from collections.abc import Sequence
 from datetime import datetime, timedelta
-from typing import SupportsIndex, overload
+from typing import Generic, SupportsIndex, overload
 
 import numpy as np
 from frequenz.channels import Broadcast, Receiver, Sender
 from numpy.typing import ArrayLike
 
 from ..actor._background_service import BackgroundService
-from ._base_types import UNIX_EPOCH, Sample
-from ._quantities import Quantity
+from ._base_types import UNIX_EPOCH, Sample, SupportsFloatT
 from ._resampling import Resampler, ResamplerConfig
 from ._ringbuffer import OrderedRingBuffer
 
 _logger = logging.getLogger(__name__)
 
 
-class MovingWindow(BackgroundService):
+class MovingWindow(BackgroundService, Generic[SupportsFloatT]):
     """
     A data window that moves with the latest datapoints of a data stream.
 
@@ -130,9 +129,9 @@ async def run() -> None:
     def __init__(  # pylint: disable=too-many-arguments
         self,
         size: timedelta,
-        resampled_data_recv: Receiver[Sample[Quantity]],
+        resampled_data_recv: Receiver[Sample[SupportsFloatT]],
         input_sampling_period: timedelta,
-        resampler_config: ResamplerConfig | None = None,
+        resampler_config: ResamplerConfig[SupportsFloatT] | None = None,
         align_to: datetime = UNIX_EPOCH,
         *,
         name: str | None = None,
@@ -166,8 +165,8 @@ def __init__(  # pylint: disable=too-many-arguments
 
         self._sampling_period = input_sampling_period
 
-        self._resampler: Resampler | None = None
-        self._resampler_sender: Sender[Sample[Quantity]] | None = None
+        self._resampler: Resampler[SupportsFloatT] | None = None
+        self._resampler_sender: Sender[Sample[SupportsFloatT]] | None = None
 
         if resampler_config:
             assert (
@@ -182,7 +181,9 @@ def __init__(  # pylint: disable=too-many-arguments
             size.total_seconds() / self._sampling_period.total_seconds()
         )
 
-        self._resampled_data_recv = resampled_data_recv
+        self._resampled_data_recv: Receiver[Sample[SupportsFloatT]] = (
+            resampled_data_recv
+        )
         self._buffer = OrderedRingBuffer(
             np.empty(shape=num_samples, dtype=float),
             sampling_period=self._sampling_period,
@@ -341,11 +342,11 @@ def _configure_resampler(self) -> None:
         """Configure the components needed to run the resampler."""
         assert self._resampler is not None
 
-        async def sink_buffer(sample: Sample[Quantity]) -> None:
+        async def sink_buffer(sample: Sample[SupportsFloatT]) -> None:
             if sample.value is not None:
                 self._buffer.update(sample)
 
-        resampler_channel = Broadcast[Sample[Quantity]]("average")
+        resampler_channel = Broadcast[Sample[SupportsFloatT]]("average")
         self._resampler_sender = resampler_channel.new_sender()
         self._resampler.add_timeseries(
             "avg", resampler_channel.new_receiver(), sink_buffer

src/frequenz/sdk/timeseries/_periodic_feature_extractor.py

@@ -16,12 +16,14 @@
 import logging
 from dataclasses import dataclass
 from datetime import datetime, timedelta
+from typing import Generic
 
 import numpy as np
 from numpy.typing import NDArray
 
 from .._internal._math import is_close_to_zero
 from ._moving_window import MovingWindow
+from ._quantities import SupportsFloatT
 from ._ringbuffer import OrderedRingBuffer
 
 _logger = logging.getLogger(__name__)
@@ -48,7 +50,7 @@ class RelativePositions:
     """The relative position of the next incoming sample."""
 
 
-class PeriodicFeatureExtractor:
+class PeriodicFeatureExtractor(Generic[SupportsFloatT]):
     """
     A feature extractor for historical timeseries data.
 
@@ -106,7 +108,7 @@ class PeriodicFeatureExtractor:
 
     def __init__(
         self,
-        moving_window: MovingWindow,
+        moving_window: MovingWindow[SupportsFloatT],
         period: timedelta,
     ) -> None:
         """
@@ -119,7 +121,7 @@ def __init__(
         Raises:
             ValueError: If the MovingWindow size is not a integer multiple of the period.
         """
-        self._moving_window = moving_window
+        self._moving_window: MovingWindow[SupportsFloatT] = moving_window
 
         self._sampling_period = self._moving_window.sampling_period
         """The sampling_period as float to use it for indexing of samples."""