Calculate the input sampling period

The input sampling period is calculated by counting the total received
samples, and dividing the total resampling time until the internal
buffer is full by the total received samples.

To store all the source properties (sampling period, sampling start,
total received samples) a new class is used (SourceProperties) and this
information is passed to the resampling function, so it can have more
information about the source to perform a proper resampling.

This calculation is performed once and then remains constant, but at
some point we could add a timer to re-calculate it.

This also improves slightly the documentation and validation of the
ResamplingConfig class.

The add_timeseries() methods now also takes a name as a way to identify
different sources in log messages (the actor uses the channel name as
timeseries name).

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

Author: llucax

benchmarks/timeseries/resampling.py

@@ -8,11 +8,17 @@
 from typing import Sequence
 
 from frequenz.sdk.timeseries import Sample
-from frequenz.sdk.timeseries._resampling import ResamplerConfig, _ResamplingHelper
+from frequenz.sdk.timeseries._resampling import (
+    ResamplerConfig,
+    SourceProperties,
+    _ResamplingHelper,
+)
 
 
 def nop(  # pylint: disable=unused-argument
-    samples: Sequence[Sample], resampler_config: ResamplerConfig
+    samples: Sequence[Sample],
+    resampler_config: ResamplerConfig,
+    source_properties: SourceProperties,
 ) -> float:
     """Return 0.0."""
     return 0.0
@@ -21,14 +27,15 @@ def nop(  # pylint: disable=unused-argument
 def _benchmark_resampling_helper(resamples: int, samples: int) -> None:
     """Benchmark the resampling helper."""
     helper = _ResamplingHelper(
+        "benchmark",
         ResamplerConfig(
             resampling_period_s=1.0,
             max_data_age_in_periods=3.0,
             resampling_function=nop,
             initial_buffer_len=samples * 3,
-            max_buffer_len=samples * 3,
-            warn_buffer_len=samples * 3,
-        )
+            warn_buffer_len=samples * 3 + 2,
+            max_buffer_len=samples * 3 + 3,
+        ),
     )
     now = datetime.now(timezone.utc)
 

examples/resampling.py

@@ -105,7 +105,9 @@ async def run() -> None:  # pylint: disable=too-many-locals
     average_chan = Broadcast[Sample]("average")
 
     second_stage_resampler = Resampler(ResamplerConfig(resampling_period_s=3.0))
-    second_stage_resampler.add_timeseries(average_chan.new_receiver(), _print_sample)
+    second_stage_resampler.add_timeseries(
+        "avg", average_chan.new_receiver(), _print_sample
+    )
 
     average_sender = average_chan.new_sender()
     # Needed until channels Senders raises exceptions on errors

src/frequenz/sdk/actor/_resampling.py

@@ -85,7 +85,7 @@ async def sink_adapter(sample: Sample) -> None:
             if not await sender.send(sample):
                 raise Exception(f"Error while sending with sender {sender}", sender)
 
-        self._resampler.add_timeseries(receiver, sink_adapter)
+        self._resampler.add_timeseries(request_channel_name, receiver, sink_adapter)
 
     async def _process_resampling_requests(self) -> None:
         """Process resampling data requests."""