Make source properties' sampling period a timedelta

The `SourceProperties` of the resampler now uses a `timedelta`
for the input sampling period. The attribute was renamed from
`sampling_period_s` to `sampling_period` accordingly.

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

Author: llucax

RELEASE_NOTES.md

@@ -19,6 +19,7 @@
   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.
+* The `SourceProperties` of the resampler now uses a `timedelta` for the input sampling period. The attribute was renamed from `sampling_period_s` to `sampling_period` accordingly.
 
 ## New Features
 

src/frequenz/sdk/timeseries/_resampling.py

@@ -311,11 +311,11 @@ class SourceProperties:
     received_samples: int = 0
     """Total samples received by this source so far."""
 
-    sampling_period_s: Optional[float] = None
-    """The sampling period of this source (in seconds).
+    sampling_period: Optional[timedelta] = None
+    """The sampling period of this source.
 
-    This is we receive (on average) one sample for this source every
-    `sampling_period_s` seconds.
+    This means we receive (on average) one sample for this source every
+    `sampling_period` time.
 
     `None` means it is unknown.
     """
@@ -496,7 +496,7 @@ 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, input_period_s)
+    are newer than `max(resampling_period, input_period)
     * 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 +552,7 @@ def _update_source_sample_period(self, now: datetime) -> bool:
 
         # We only update it if we didn't before and we have enough data
         if (
-            props.sampling_period_s is not None
+            props.sampling_period is not None
             or props.sampling_start is None
             or props.received_samples
             < config.resampling_period.total_seconds() * config.max_data_age_in_periods
@@ -564,14 +564,14 @@ def _update_source_sample_period(self, now: datetime) -> bool:
             return False
 
         samples_time_delta = now - props.sampling_start
-        props.sampling_period_s = (
-            samples_time_delta.total_seconds()
-        ) / props.received_samples
+        props.sampling_period = timedelta(
+            seconds=samples_time_delta.total_seconds() / props.received_samples
+        )
 
         _logger.debug(
             "New input sampling period calculated for %r: %ss",
             self._name,
-            props.sampling_period_s,
+            props.sampling_period,
         )
         return True
 
@@ -581,26 +581,28 @@ def _update_buffer_len(self) -> bool:
         Returns:
             Whether the buffer length was changed (was really updated).
         """
-        input_sampling_period_s = self._source_properties.sampling_period_s
-
         # To make type checking happy
-        assert input_sampling_period_s is not None
         assert self._buffer.maxlen is not None
+        assert self._source_properties.sampling_period is not None
+
+        input_sampling_period = self._source_properties.sampling_period
 
         config = self._config
 
         # 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.total_seconds():
-            new_buffer_len = input_sampling_period_s * config.max_data_age_in_periods
+        if input_sampling_period > config.resampling_period:
+            new_buffer_len = (
+                input_sampling_period.total_seconds() * config.max_data_age_in_periods
+            )
         # If we are upsampling, we want a buffer that can hold
         # max_data_age_in_periods * resampling_period of data, and we
-        # one sample every input_sampling_period_s.
+        # one sample every input_sampling_period.
         else:
             new_buffer_len = (
                 config.resampling_period.total_seconds()
-                / input_sampling_period_s
+                / input_sampling_period.total_seconds()
                 * config.max_data_age_in_periods
             )
 
@@ -655,13 +657,14 @@ 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.total_seconds(), props.sampling_period_s)
-            if props.sampling_period_s is not None
-            else conf.resampling_period.total_seconds()
-        )
-        minimum_relevant_timestamp = timestamp - timedelta(
-            seconds=period * conf.max_data_age_in_periods
+            max(
+                conf.resampling_period,
+                props.sampling_period,
+            )
+            if props.sampling_period is not None
+            else conf.resampling_period
         )
+        minimum_relevant_timestamp = timestamp - period * conf.max_data_age_in_periods
 
         # We need to pass a dummy Sample to bisect because it only support
         # specifying a key extraction function in Python 3.10, so we need to

tests/timeseries/test_resampling.py

@@ -407,7 +407,7 @@ async def test_future_samples_not_included(
         a_sequence(sample0s, sample1s), config, source_props  # sample2_1s is not here
     )
     assert source_props == SourceProperties(
-        sampling_start=timestamp, received_samples=3, sampling_period_s=None
+        sampling_start=timestamp, received_samples=3, sampling_period=None
     )
     assert _get_buffer_len(resampler, source_receiver) == config.initial_buffer_len
     sink_mock.reset_mock()
@@ -489,7 +489,7 @@ async def test_resampling_with_one_window(
         a_sequence(sample1s), config, source_props
     )
     assert source_props == SourceProperties(
-        sampling_start=timestamp, received_samples=2, sampling_period_s=None
+        sampling_start=timestamp, received_samples=2, sampling_period=None
     )
     assert _get_buffer_len(resampler, source_recvr) == config.initial_buffer_len
     sink_mock.reset_mock()
@@ -518,7 +518,9 @@ async def test_resampling_with_one_window(
     # By now we have a full buffer (5 samples and a buffer of length 4), which
     # we received in 4 seconds, so we have an input period of 0.8s.
     assert source_props == SourceProperties(
-        sampling_start=timestamp, received_samples=5, sampling_period_s=0.8
+        sampling_start=timestamp,
+        received_samples=5,
+        sampling_period=timedelta(seconds=0.8),
     )
     # The buffer should be able to hold 2 seconds of data, and data is coming
     # every 0.8 seconds, so we should be able to store 3 samples.
@@ -536,7 +538,9 @@ async def test_resampling_with_one_window(
         current_iteration=3,
     )
     assert source_props == SourceProperties(
-        sampling_start=timestamp, received_samples=5, sampling_period_s=0.8
+        sampling_start=timestamp,
+        received_samples=5,
+        sampling_period=timedelta(seconds=0.8),
     )
     assert _get_buffer_len(resampler, source_recvr) == 3
 
@@ -598,7 +602,7 @@ async def test_resampling_with_one_and_a_half_windows(  # pylint: disable=too-ma
         a_sequence(sample0s, sample1s), config, source_props
     )
     assert source_props == SourceProperties(
-        sampling_start=timestamp, received_samples=2, sampling_period_s=None
+        sampling_start=timestamp, received_samples=2, sampling_period=None
     )
     assert _get_buffer_len(resampler, source_recvr) == config.initial_buffer_len
     sink_mock.reset_mock()
@@ -626,7 +630,7 @@ async def test_resampling_with_one_and_a_half_windows(  # pylint: disable=too-ma
         a_sequence(sample2_5s, sample3s, sample4s), config, source_props
     )
     assert source_props == SourceProperties(
-        sampling_start=timestamp, received_samples=5, sampling_period_s=None
+        sampling_start=timestamp, received_samples=5, sampling_period=None
     )
     assert _get_buffer_len(resampler, source_recvr) == config.initial_buffer_len
     sink_mock.reset_mock()
@@ -654,7 +658,9 @@ async def test_resampling_with_one_and_a_half_windows(  # pylint: disable=too-ma
     # By now we have a full buffer (7 samples and a buffer of length 6), which
     # we received in 4 seconds, so we have an input period of 6/7s.
     assert source_props == SourceProperties(
-        sampling_start=timestamp, received_samples=7, sampling_period_s=6 / 7
+        sampling_start=timestamp,
+        received_samples=7,
+        sampling_period=timedelta(seconds=6 / 7),
     )
     # The buffer should be able to hold 2 * 1.5 (3) seconds of data, and data
     # is coming every 6/7 seconds (~0.857s), so we should be able to store
@@ -693,7 +699,9 @@ async def test_resampling_with_one_and_a_half_windows(  # pylint: disable=too-ma
         current_iteration=5,
     )
     assert source_props == SourceProperties(
-        sampling_start=timestamp, received_samples=7, sampling_period_s=6 / 7
+        sampling_start=timestamp,
+        received_samples=7,
+        sampling_period=timedelta(seconds=6 / 7),
     )
     assert _get_buffer_len(resampler, source_recvr) == 4
 
@@ -755,7 +763,7 @@ async def test_resampling_with_two_windows(  # pylint: disable=too-many-statemen
         a_sequence(sample0s, sample1s), config, source_props
     )
     assert source_props == SourceProperties(
-        sampling_start=timestamp, received_samples=2, sampling_period_s=None
+        sampling_start=timestamp, received_samples=2, sampling_period=None
     )
     assert _get_buffer_len(resampler, source_recvr) == config.initial_buffer_len
     sink_mock.reset_mock()
@@ -783,7 +791,7 @@ async def test_resampling_with_two_windows(  # pylint: disable=too-many-statemen
         a_sequence(sample1s, sample2_5s, sample3s, sample4s), config, source_props
     )
     assert source_props == SourceProperties(
-        sampling_start=timestamp, received_samples=5, sampling_period_s=None
+        sampling_start=timestamp, received_samples=5, sampling_period=None
     )
     assert _get_buffer_len(resampler, source_recvr) == config.initial_buffer_len
     sink_mock.reset_mock()
@@ -811,7 +819,7 @@ async def test_resampling_with_two_windows(  # pylint: disable=too-many-statemen
         source_props,
     )
     assert source_props == SourceProperties(
-        sampling_start=timestamp, received_samples=7, sampling_period_s=None
+        sampling_start=timestamp, received_samples=7, sampling_period=None
     )
     assert _get_buffer_len(resampler, source_recvr) == config.initial_buffer_len
     sink_mock.reset_mock()
@@ -833,7 +841,7 @@ async def test_resampling_with_two_windows(  # pylint: disable=too-many-statemen
         a_sequence(sample5s, sample6s), config, source_props
     )
     assert source_props == SourceProperties(
-        sampling_start=timestamp, received_samples=7, sampling_period_s=None
+        sampling_start=timestamp, received_samples=7, sampling_period=None
     )
     assert _get_buffer_len(resampler, source_recvr) == config.initial_buffer_len
     sink_mock.reset_mock()
@@ -849,7 +857,7 @@ async def test_resampling_with_two_windows(  # pylint: disable=too-many-statemen
         current_iteration=5,
     )
     assert source_props == SourceProperties(
-        sampling_start=timestamp, received_samples=7, sampling_period_s=None
+        sampling_start=timestamp, received_samples=7, sampling_period=None
     )
     assert _get_buffer_len(resampler, source_recvr) == config.initial_buffer_len