Move resampler configuration to its own class

This makes user code a little bit more verbose but makes the code much
more maintainable, as we avoid having to copy and passing around a lot
of configuration variables that are only (or mostly) only relevant to
the internal resampling class.

It also removes a lot of documentation duplication that can get easily
out of sync and cause confusion.

And we'll add quite a few more configuration variables in subsequent
commits, which will just exacerbate the above mentioned issues without
a config class.

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

Author: llucax

examples/resampling.py

@@ -18,7 +18,7 @@
 )
 from frequenz.sdk.microgrid.component import ComponentCategory, ComponentMetricId
 from frequenz.sdk.timeseries import Sample
-from frequenz.sdk.timeseries._resampling import Resampler, Sink, Source
+from frequenz.sdk.timeseries._resampling import Resampler, ResamplerConfig, Sink, Source
 
 HOST = "microgrid.sandbox.api.frequenz.io"
 PORT = 61060
@@ -65,7 +65,7 @@ async def run() -> None:  # pylint: disable=too-many-locals
         channel_registry=channel_registry,
         data_sourcing_request_sender=data_source_request_sender,
         resampling_request_receiver=resampling_request_receiver,
-        resampling_period_s=1,
+        config=ResamplerConfig(resampling_period_s=1),
     )
 
     components = await microgrid.get().api_client.components()
@@ -104,7 +104,7 @@ async def run() -> None:  # pylint: disable=too-many-locals
     # Create a channel to calculate an average for all the data
     average_chan = Broadcast[Sample]("average")
 
-    second_stage_resampler = Resampler(resampling_period_s=3.0)
+    second_stage_resampler = Resampler(ResamplerConfig(resampling_period_s=3.0))
     second_stage_resampler.add_timeseries(average_chan.new_receiver(), _print_sample)
 
     average_sender = average_chan.new_sender()

src/frequenz/sdk/actor/__init__.py

@@ -7,13 +7,14 @@
 from ._config_managing import ConfigManagingActor
 from ._data_sourcing import ComponentMetricRequest, DataSourcingActor
 from ._decorator import actor
-from ._resampling import ComponentMetricsResamplingActor
+from ._resampling import ComponentMetricsResamplingActor, ResamplerConfig
 
 __all__ = [
     "ChannelRegistry",
     "ComponentMetricRequest",
     "ComponentMetricsResamplingActor",
     "ConfigManagingActor",
     "DataSourcingActor",
+    "ResamplerConfig",
     "actor",
 ]

src/frequenz/sdk/actor/_resampling.py

@@ -14,12 +14,7 @@
 from frequenz.sdk.util.asyncio import cancel_and_await
 
 from ..timeseries import Sample
-from ..timeseries._resampling import (
-    Resampler,
-    ResamplingError,
-    ResamplingFunction,
-    average,
-)
+from ..timeseries._resampling import Resampler, ResamplerConfig, ResamplingError
 from ._channel_registry import ChannelRegistry
 from ._data_sourcing import ComponentMetricRequest
 from ._decorator import actor
@@ -37,9 +32,7 @@ def __init__(  # pylint: disable=too-many-arguments
         channel_registry: ChannelRegistry,
         data_sourcing_request_sender: Sender[ComponentMetricRequest],
         resampling_request_receiver: Receiver[ComponentMetricRequest],
-        resampling_period_s: float = 0.2,
-        max_data_age_in_periods: float = 3.0,
-        resampling_function: ResamplingFunction = average,
+        config: ResamplerConfig,
     ) -> None:
         """Initialize an instance.
 
@@ -51,34 +44,16 @@ def __init__(  # pylint: disable=too-many-arguments
                 to subscribe to component metrics.
             resampling_request_receiver: The receiver to use to receive new
                 resampmling subscription requests.
-            resampling_period_s: The time it passes between resampled data
-                should be calculated (in seconds).
-            max_data_age_in_periods: The maximum age a sample can have to be
-                considered *relevant* for resampling purposes, expressed in the
-                number of resampling periods. For exapmle is
-                `resampling_period_s` is 3 and `max_data_age_in_periods` is 2,
-                then data older than `3*2 = 6` secods will be discarded when
-                creating a new sample and never passed to the resampling
-                function.
-            resampling_function: The function to be applied to the sequence of
-                *relevant* samples at a given time. The result of the function
-                is what is sent as the resampled data.
+            config: The configuration for the resampler.
         """
         self._channel_registry: ChannelRegistry = channel_registry
-        self._resampling_period_s: float = resampling_period_s
-        self._max_data_age_in_periods: float = max_data_age_in_periods
-        self._resampling_function: ResamplingFunction = resampling_function
         self._data_sourcing_request_sender: Sender[
             ComponentMetricRequest
         ] = data_sourcing_request_sender
         self._resampling_request_receiver: Receiver[
             ComponentMetricRequest
         ] = resampling_request_receiver
-        self._resampler: Resampler = Resampler(
-            resampling_period_s=resampling_period_s,
-            max_data_age_in_periods=max_data_age_in_periods,
-            resampling_function=resampling_function,
-        )
+        self._resampler: Resampler = Resampler(config)
         self._active_req_channels: set[str] = set()
 
     async def _subscribe(self, request: ComponentMetricRequest) -> None:

src/frequenz/sdk/timeseries/_resampling.py

@@ -9,6 +9,7 @@
 import logging
 import math
 from collections import deque
+from dataclasses import dataclass
 from datetime import datetime, timedelta
 from typing import AsyncIterator, Callable, Coroutine, Sequence
 
@@ -85,6 +86,34 @@ def average(samples: Sequence[Sample], resampling_period_s: float) -> float:
     return sum(values) / len(values)
 
 
+@dataclass(frozen=True)
+class ResamplerConfig:
+    """Resampler configuration."""
+
+    resampling_period_s: float
+    """The resapmling period in seconds.
+
+    This is the time it passes between resampled data should be calculated.
+    """
+
+    max_data_age_in_periods: float = 3.0
+    """The maximum age a sample can have to be considered *relevant* for resampling.
+
+    Expressed in number of resampling periods. For example if
+    `resampling_period_s` is 3 and `max_data_age_in_periods` is 2, then data
+    older than 3*2 = 6 secods will be discarded when creating a new sample and
+    never passed to the resampling function.
+    """
+
+    resampling_function: ResamplingFunction = average
+    """The resampling function.
+
+    This function will be applied to the sequence of relevant samples at
+    a given time. The result of the function is what is sent as the resampled
+    value.
+    """
+
+
 class SourceStoppedError(RuntimeError):
     """A timeseries stopped producing samples."""
 
@@ -166,34 +195,24 @@ class Resampler:
     no way to produce meaningful samples with the available data.
     """
 
-    def __init__(
-        self,
-        *,
-        resampling_period_s: float,
-        resampling_function: ResamplingFunction = average,
-        max_data_age_in_periods: float = 3.0,
-    ) -> None:
+    def __init__(self, config: ResamplerConfig) -> None:
         """Initialize an instance.
 
         Args:
-            resampling_period_s: The time it passes between resampled data
-                should be calculated (in seconds).
-            max_data_age_in_periods: The maximum age a sample can have to be
-                considered *relevant* for resampling purposes, expressed in the
-                number of resampling periods. For exapmle is
-                `resampling_period_s` is 3 and `max_data_age_in_periods` is 2,
-                then data older than `3*2 = 6` secods will be discarded when
-                creating a new sample and never passed to the resampling
-                function.
-            resampling_function: The function to be applied to the sequence of
-                *relevant* samples at a given time. The result of the function
-                is what is sent as the resampled data.
+            config: The configuration for the resampler.
         """
-        self._resampling_period_s = resampling_period_s
-        self._max_data_age_in_periods: float = max_data_age_in_periods
-        self._resampling_function: ResamplingFunction = resampling_function
+        self._config = config
         self._resamplers: dict[Source, _StreamingHelper] = {}
-        self._timer: Timer = Timer(self._resampling_period_s)
+        self._timer: Timer = Timer(config.resampling_period_s)
+
+    @property
+    def config(self) -> ResamplerConfig:
+        """Get the resampler configuration.
+
+        Returns:
+            The resampler configuration.
+        """
+        return self._config
 
     async def stop(self) -> None:
         """Cancel all receiving tasks."""
@@ -214,15 +233,7 @@ def add_timeseries(self, source: Source, sink: Sink) -> bool:
         if source in self._resamplers:
             return False
 
-        resampler = _StreamingHelper(
-            _ResamplingHelper(
-                resampling_period_s=self._resampling_period_s,
-                max_data_age_in_periods=self._max_data_age_in_periods,
-                resampling_function=self._resampling_function,
-            ),
-            source,
-            sink,
-        )
+        resampler = _StreamingHelper(_ResamplingHelper(self._config), source, sink)
         self._resamplers[source] = resampler
         return True
 
@@ -288,33 +299,14 @@ class _ResamplingHelper:
     when calling the `resample()` method. All older samples are discarded.
     """
 
-    def __init__(
-        self,
-        *,
-        resampling_period_s: float,
-        max_data_age_in_periods: float,
-        resampling_function: ResamplingFunction,
-    ) -> None:
+    def __init__(self, config: ResamplerConfig) -> None:
         """Initialize an instance.
 
         Args:
-            resampling_period_s: The time it passes between resampled data
-                should be calculated (in seconds).
-            max_data_age_in_periods: The maximum age a sample can have to be
-                considered *relevant* for resampling purposes, expressed in the
-                number of resampling periods. For exapmle is
-                `resampling_period_s` is 3 and `max_data_age_in_periods` is 2,
-                then data older than 3*2 = 6 secods will be discarded when
-                creating a new sample and never passed to the resampling
-                function.
-            resampling_function: The function to be applied to the sequence of
-                relevant samples at a given time. The result of the function is
-                what is sent as the resampled data.
+            config: The configuration for the resampler.
         """
-        self._resampling_period_s = resampling_period_s
-        self._max_data_age_in_periods: float = max_data_age_in_periods
+        self._config = config
         self._buffer: deque[Sample] = deque()
-        self._resampling_function: ResamplingFunction = resampling_function
 
     def add_sample(self, sample: Sample) -> None:
         """Add a new sample to the internal buffer.
@@ -361,14 +353,17 @@ def resample(self, timestamp: datetime) -> Sample:
                 have `None` as `value`.
         """
         threshold = timestamp - timedelta(
-            seconds=self._max_data_age_in_periods * self._resampling_period_s
+            seconds=self._config.max_data_age_in_periods
+            * self._config.resampling_period_s
         )
         self._remove_outdated_samples(threshold=threshold)
 
         value = (
             None
             if not self._buffer
-            else self._resampling_function(self._buffer, self._resampling_period_s)
+            else self._config.resampling_function(
+                self._buffer, self._config.resampling_period_s
+            )
         )
         return Sample(timestamp, value)
 

tests/actor/test_resampling.py

@@ -16,6 +16,7 @@
     ChannelRegistry,
     ComponentMetricRequest,
     ComponentMetricsResamplingActor,
+    ResamplerConfig,
 )
 from frequenz.sdk.microgrid.component import ComponentMetricId
 from frequenz.sdk.timeseries import Sample
@@ -123,8 +124,10 @@ async def test_single_request(
         channel_registry=channel_registry,
         data_sourcing_request_sender=data_source_req_chan.new_sender(),
         resampling_request_receiver=resampling_req_chan.new_receiver(),
-        resampling_period_s=0.2,
-        max_data_age_in_periods=2,
+        config=ResamplerConfig(
+            resampling_period_s=0.2,
+            max_data_age_in_periods=2,
+        ),
     )
 
     subs_req = ComponentMetricRequest(
@@ -167,8 +170,10 @@ async def test_duplicate_request(
         channel_registry=channel_registry,
         data_sourcing_request_sender=data_source_req_chan.new_sender(),
         resampling_request_receiver=resampling_req_chan.new_receiver(),
-        resampling_period_s=0.2,
-        max_data_age_in_periods=2,
+        config=ResamplerConfig(
+            resampling_period_s=0.2,
+            max_data_age_in_periods=2,
+        ),
     )
 
     subs_req = ComponentMetricRequest(

tests/timeseries/mock_microgrid.py

@@ -27,6 +27,7 @@
     ComponentMetricRequest,
     ComponentMetricsResamplingActor,
     DataSourcingActor,
+    ResamplerConfig,
 )
 from tests.microgrid import mock_api
 
@@ -236,7 +237,7 @@ async def _init_client_and_actors(
             channel_registry=channel_registry,
             data_sourcing_request_sender=data_source_request_sender,
             resampling_request_receiver=resampling_actor_request_receiver,
-            resampling_period_s=0.1,
+            config=ResamplerConfig(resampling_period_s=0.1),
         )
 
         return (resampling_actor_request_sender, channel_registry)