Make the resampling period a timedelta

The `ResamplerConfig` now takes the resampling period as a `timedelta`.

The configuration was renamed from `resampling_period_s` to
`resampling_period` accordingly.

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

Author: llucax

RELEASE_NOTES.md

@@ -18,6 +18,7 @@
   Notice that the parameter `sampling_period` has been renamed to `input_sampling_period`
   to better distinguish it from the sampling period parameter in the resampler.
 * The serialization feature for the ringbuffer was made more flexible. The `dump` and `load` methods can now work directly with a ringbuffer instance.
+* The `ResamplerConfig` now takes the resampling period as a `timedelta`. The configuration was renamed from `resampling_period_s` to `resampling_period` accordingly.
 
 ## New Features
 

benchmarks/power_distribution/power_distributor.py

@@ -8,6 +8,7 @@
 import random
 import timeit
 from dataclasses import dataclass
+from datetime import timedelta
 from typing import Any, Coroutine, Dict, List, Set  # pylint: disable=unused-import
 
 from frequenz.channels import Bidirectional, Broadcast
@@ -154,7 +155,9 @@ async def run() -> None:
     """Create microgrid api and run tests."""
     # pylint: disable=protected-access
 
-    await microgrid.initialize(HOST, PORT, ResamplerConfig(resampling_period_s=1.0))
+    await microgrid.initialize(
+        HOST, PORT, ResamplerConfig(resampling_period=timedelta(seconds=1.0))
+    )
 
     all_batteries: Set[Component] = connection_manager.get().component_graph.components(
         component_category={ComponentCategory.BATTERY}

benchmarks/timeseries/resampling.py

@@ -29,7 +29,7 @@ def _benchmark_resampling_helper(resamples: int, samples: int) -> None:
     helper = _ResamplingHelper(
         "benchmark",
         ResamplerConfig(
-            resampling_period_s=1.0,
+            resampling_period=timedelta(seconds=1.0),
             max_data_age_in_periods=3.0,
             resampling_function=nop,
             initial_buffer_len=samples * 3,

examples/battery_pool.py

@@ -7,6 +7,7 @@
 
 import asyncio
 import logging
+from datetime import timedelta
 from typing import Any, Dict
 
 from frequenz.channels import Receiver
@@ -25,7 +26,9 @@ async def main() -> None:
         level=logging.DEBUG, format="%(asctime)s %(name)s %(levelname)s:%(message)s"
     )
     await microgrid.initialize(
-        host=HOST, port=PORT, resampler_config=ResamplerConfig(resampling_period_s=1.0)
+        host=HOST,
+        port=PORT,
+        resampler_config=ResamplerConfig(resampling_period=timedelta(seconds=1.0)),
     )
 
     battery_pool = microgrid.battery_pool()

examples/battery_status.py

@@ -9,6 +9,7 @@
 
 import asyncio
 import logging
+from datetime import timedelta
 
 from frequenz.channels import Broadcast
 
@@ -31,7 +32,9 @@ async def main() -> None:
     logging.basicConfig(
         level=logging.DEBUG, format="%(asctime)s %(name)s %(levelname)s:%(message)s"
     )
-    await microgrid.initialize(HOST, PORT, ResamplerConfig(resampling_period_s=1.0))
+    await microgrid.initialize(
+        HOST, PORT, ResamplerConfig(resampling_period=timedelta(seconds=1.0))
+    )
     batteries = {
         bat.component_id
         for bat in connection_manager.get().component_graph.components(

examples/power_distribution.py

@@ -12,7 +12,7 @@
 
 import asyncio
 import logging
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from queue import Queue
 from typing import List, Optional, Set
 
@@ -159,7 +159,9 @@ async def run() -> None:
     logging.basicConfig(
         level=logging.DEBUG, format="%(asctime)s %(name)s %(levelname)s:%(message)s"
     )
-    await microgrid.initialize(HOST, PORT, ResamplerConfig(resampling_period_s=1.0))
+    await microgrid.initialize(
+        HOST, PORT, ResamplerConfig(resampling_period=timedelta(seconds=1.0))
+    )
 
     channel_registry = ChannelRegistry(name="Microgrid Channel Registry")
 
@@ -180,7 +182,7 @@ async def run() -> None:
         channel_registry=channel_registry,
         data_sourcing_request_sender=data_source_request_channel.new_sender(),
         resampling_request_receiver=resampling_actor_request_channel.new_receiver(),
-        config=ResamplerConfig(resampling_period_s=1.0),
+        config=ResamplerConfig(resampling_period=timedelta(seconds=1.0)),
     )
 
     logical_meter = LogicalMeter(

examples/resampling.py

@@ -4,7 +4,7 @@
 """Frequenz Python SDK resampling example."""
 
 import asyncio
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 
 from frequenz.channels import Broadcast
 from frequenz.channels.util import Merge
@@ -43,7 +43,9 @@ async def _print_sample(sample: Sample) -> None:
 
 async def run() -> None:  # pylint: disable=too-many-locals
     """Run main functions that initializes and creates everything."""
-    await microgrid.initialize(HOST, PORT, ResamplerConfig(resampling_period_s=0.2))
+    await microgrid.initialize(
+        HOST, PORT, ResamplerConfig(resampling_period=timedelta(seconds=0.2))
+    )
 
     channel_registry = ChannelRegistry(name="data-registry")
 
@@ -66,7 +68,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,
-        config=ResamplerConfig(resampling_period_s=1),
+        config=ResamplerConfig(resampling_period=timedelta(seconds=1)),
     )
 
     components = await connection_manager.get().api_client.components()
@@ -105,7 +107,9 @@ 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(ResamplerConfig(resampling_period_s=3.0))
+    second_stage_resampler = Resampler(
+        ResamplerConfig(resampling_period=timedelta(seconds=3.0))
+    )
     second_stage_resampler.add_timeseries(
         "avg", average_chan.new_receiver(), _print_sample
     )

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -13,7 +13,6 @@
 import typing
 from collections import abc
 from dataclasses import dataclass
-from datetime import timedelta
 
 from frequenz.channels import Bidirectional, Broadcast, Sender
 
@@ -175,9 +174,7 @@ def battery_pool(
                 batteries_status_receiver=self._battery_status_channel.new_receiver(
                     maxsize=1
                 ),
-                min_update_interval=timedelta(
-                    seconds=self._resampler_config.resampling_period_s
-                ),
+                min_update_interval=self._resampler_config.resampling_period,
                 batteries_id=battery_ids,
             )
 

src/frequenz/sdk/timeseries/_moving_window.py

@@ -142,13 +142,12 @@ def __init__(  # pylint: disable=too-many-arguments
         self._resampler_task: asyncio.Task[None] | None = None
 
         if resampler_config:
-            resampling_period = timedelta(seconds=resampler_config.resampling_period_s)
             assert (
-                resampling_period <= size
+                resampler_config.resampling_period <= size
             ), "The resampling period should be equal to or lower than the window size."
 
             self._resampler = Resampler(resampler_config)
-            sampling = resampling_period
+            sampling = resampler_config.resampling_period
 
         # Sampling period might not fit perfectly into the window size.
         num_samples = math.ceil(size.total_seconds() / sampling.total_seconds())

src/frequenz/sdk/timeseries/_resampling.py

@@ -76,10 +76,10 @@
 
 A resampling function produces a new sample based on a list of pre-existing
 samples. It can do "upsampling" when there data rate of the `input_samples`
-period is smaller than the `resampling_period_s`, or "downsampling" if it is
+period is smaller than the `resampling_period`, or "downsampling" if it is
 bigger.
 
-In general a resampling window is the same as the `resampling_period_s`, and
+In general a resampling window is the same as the `resampling_period`, and
 this function might receive input samples from multiple windows in the past to
 enable extrapolation, but no samples from the future (so the timestamp of the
 new sample that is going to be produced will always be bigger than the biggest
@@ -123,31 +123,31 @@ def average(
 class ResamplerConfig:
     """Resampler configuration."""
 
-    resampling_period_s: float
-    """The resampling period in seconds.
+    resampling_period: timedelta
+    """The resampling period.
 
     This is the time it passes between resampled data should be calculated.
 
-    It must be a positive number.
+    It must be a positive time span.
     """
 
     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 periods, where period is the `resampling_period_s`
+    Expressed in number of periods, where period is the `resampling_period`
     if we are downsampling (resampling period bigger than the input period) or
     the input sampling period if we are upsampling (input period bigger than
     the resampling period).
 
     It must be bigger than 1.0.
 
     Example:
-        If `resampling_period_s` is 3, the input sampling period is
+        If `resampling_period` is 3 seconds, the input sampling period is
         1 and `max_data_age_in_periods` is 2, then data older than 3*2
         = 6 seconds will be discarded when creating a new sample and never
         passed to the resampling function.
 
-        If `resampling_period_s` is 3, the input sampling period is
+        If `resampling_period` is 3 seconds, the input sampling period is
         5 and `max_data_age_in_periods` is 2, then data older than 5*2
         = 10 seconds will be discarded when creating a new sample and never
         passed to the resampling function.
@@ -197,9 +197,9 @@ def __post_init__(self) -> None:
         Raises:
             ValueError: If any value is out of range.
         """
-        if self.resampling_period_s < 0.0:
+        if self.resampling_period.total_seconds() < 0.0:
             raise ValueError(
-                f"resampling_period_s ({self.resampling_period_s}) must be positive"
+                f"resampling_period ({self.resampling_period}) must be positive"
             )
         if self.max_data_age_in_periods < 1.0:
             raise ValueError(
@@ -348,8 +348,8 @@ def __init__(self, config: ResamplerConfig) -> None:
         self._resamplers: dict[Source, _StreamingHelper] = {}
         """A mapping between sources and the streaming helper handling that source."""
 
-        self._window_end: datetime = datetime.now(timezone.utc) + timedelta(
-            seconds=self._config.resampling_period_s
+        self._window_end: datetime = (
+            datetime.now(timezone.utc) + self._config.resampling_period
         )
         """The time in which the current window ends.
 
@@ -449,9 +449,7 @@ async def resample(self, *, one_shot: bool = False) -> None:
                 return_exceptions=True,
             )
 
-            self._window_end = self._window_end + timedelta(
-                seconds=self._config.resampling_period_s
-            )
+            self._window_end = self._window_end + self._config.resampling_period
             exceptions = {
                 source: results[i]
                 for i, source in enumerate(self._resamplers)
@@ -476,25 +474,30 @@ async def _wait_for_next_resampling_period(self) -> None:
             sleep_for = self._window_end - now
             await asyncio.sleep(sleep_for.total_seconds())
 
-        timer_error_s = (now - self._window_end).total_seconds()
-        if timer_error_s > (self._config.resampling_period_s / 10.0):
+        timer_error = now - self._window_end
+        # We use a tolerance of 10% of the resampling period
+        tolerance = timedelta(
+            seconds=self._config.resampling_period.total_seconds() / 10.0
+        )
+        if timer_error > tolerance:
             _logger.warning(
-                "The resampling task woke up too late. Resampling should have started "
-                "at %s, but it started at %s (%s seconds difference; resampling "
-                "period is %s seconds)",
+                "The resampling task woke up too late. Resampling should have "
+                "started at %s, but it started at %s (tolerance: %s, "
+                "difference: %s; resampling period: %s)",
                 self._window_end,
                 now,
-                timer_error_s,
-                self._config.resampling_period_s,
+                tolerance,
+                timer_error,
+                self._config.resampling_period,
             )
 
 
 class _ResamplingHelper:
     """Keeps track of *relevant* samples to pass them to the resampling function.
 
     Samples are stored in an internal ring buffer. All collected samples that
-    are newer than `max(resampling_period_s, input_period_s)
-    * max_data_age_in_periods` seconds are considered *relevant* and are passed
+    are newer than `max(resampling_period, input_period_s)
+    * max_data_age_in_periods` are considered *relevant* and are passed
     to the provided `resampling_function` when calling the `resample()` method.
     All older samples are discarded.
     """
@@ -552,7 +555,7 @@ def _update_source_sample_period(self, now: datetime) -> bool:
             props.sampling_period_s is not None
             or props.sampling_start is None
             or props.received_samples
-            < config.resampling_period_s * config.max_data_age_in_periods
+            < config.resampling_period.total_seconds() * config.max_data_age_in_periods
             or len(self._buffer) < self._buffer.maxlen
             # There might be a race between the first sample being received and
             # this function being called
@@ -589,14 +592,14 @@ def _update_buffer_len(self) -> bool:
         # If we are upsampling, one sample could be enough for back-filling, but
         # we store max_data_age_in_periods for input periods, so resampling
         # functions can do more complex inter/extrapolation if they need to.
-        if input_sampling_period_s > config.resampling_period_s:
+        if input_sampling_period_s > config.resampling_period.total_seconds():
             new_buffer_len = input_sampling_period_s * config.max_data_age_in_periods
         # If we are upsampling, we want a buffer that can hold
-        # max_data_age_in_periods * resampling_period_s seconds of data, and we
+        # max_data_age_in_periods * resampling_period of data, and we
         # one sample every input_sampling_period_s.
         else:
             new_buffer_len = (
-                config.resampling_period_s
+                config.resampling_period.total_seconds()
                 / input_sampling_period_s
                 * config.max_data_age_in_periods
             )
@@ -652,9 +655,9 @@ def resample(self, timestamp: datetime) -> Sample:
         # To see which samples are relevant we need to consider if we are down
         # or upsampling.
         period = (
-            max(conf.resampling_period_s, props.sampling_period_s)
+            max(conf.resampling_period.total_seconds(), props.sampling_period_s)
             if props.sampling_period_s is not None
-            else conf.resampling_period_s
+            else conf.resampling_period.total_seconds()
         )
         minimum_relevant_timestamp = timestamp - timedelta(
             seconds=period * conf.max_data_age_in_periods