Make resampling window constant

The resampling window depends on when the resampling timer was fired.
This is wrong, because it makes the resampling window not constant, as
in general the timer will fire a bit after the resampling period has
passed, so it tends to be a bit bigger than resampling period.

This commit fixes this and uses a constant resampling window, always
emitting new samples with an exact multiple of the resampling period.

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

Author: llucax

src/frequenz/sdk/timeseries/_resampling.py

@@ -12,7 +12,7 @@
 from bisect import bisect
 from collections import deque
 from dataclasses import dataclass
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import AsyncIterator, Callable, Coroutine, Optional, Sequence
 
 from frequenz.channels.util import Timer
@@ -354,6 +354,18 @@ def __init__(self, config: ResamplerConfig) -> None:
         self._timer: Timer = Timer(config.resampling_period_s)
         """The timer to trigger the next resampling."""
 
+        self._window_end: datetime = datetime.now(timezone.utc) + timedelta(
+            seconds=self._config.resampling_period_s
+        )
+        """The time in which the current window ends.
+
+        This is used to make sure every resampling window is generated at
+        precise times. We can't rely on the timer timestamp because timers will
+        never fire at the exact requested time, so if we don't use a precise
+        time for the end of the window, the resampling windows we produce will
+        have different sizes.
+        """
+
     @property
     def config(self) -> ResamplerConfig:
         """Get the resampler configuration.
@@ -435,11 +447,12 @@ async def resample(self, *, one_shot: bool = False) -> None:
                 timeseries from the resampler before calling this method
                 again).
         """
-        async for timer_timestamp in self._timer:
+        async for _ in self._timer:
             results = await asyncio.gather(
-                *[r.resample(timer_timestamp) for r in self._resamplers.values()],
+                *[r.resample(self._window_end) for r in self._resamplers.values()],
                 return_exceptions=True,
             )
+            self._update_window_end()
             exceptions = {
                 source: results[i]
                 for i, source in enumerate(self._resamplers)
@@ -452,6 +465,11 @@ async def resample(self, *, one_shot: bool = False) -> None:
             if one_shot:
                 break
 
+    def _update_window_end(self) -> None:
+        self._window_end = self._window_end + timedelta(
+            seconds=self._config.resampling_period_s
+        )
+
 
 class _ResamplingHelper:
     """Keeps track of *relevant* samples to pass them to the resampling function.

tests/timeseries/test_resampling.py

@@ -174,6 +174,88 @@ async def test_helper_buffer_too_big(
     assert helper._buffer.maxlen == DEFAULT_BUFFER_LEN_MAX
 
 
+async def test_resampling_window_size_is_constant(
+    fake_time: time_machine.Coordinates, source_chan: Broadcast[Sample]
+) -> None:
+    """Test resampling window size is consistent."""
+    timestamp = datetime.now(timezone.utc)
+
+    resampling_period_s = 2
+    expected_resampled_value = 42.0
+
+    resampling_fun_mock = MagicMock(
+        spec=ResamplingFunction, return_value=expected_resampled_value
+    )
+    config = ResamplerConfig(
+        resampling_period_s=resampling_period_s,
+        max_data_age_in_periods=1.0,
+        resampling_function=resampling_fun_mock,
+        initial_buffer_len=4,
+    )
+    resampler = Resampler(config)
+
+    source_receiver = source_chan.new_receiver()
+    source_sender = source_chan.new_sender()
+
+    sink_mock = AsyncMock(spec=Sink, return_value=True)
+
+    resampler.add_timeseries("test", source_receiver, sink_mock)
+    source_props = resampler.get_source_properties(source_receiver)
+
+    # Test timeline
+    #
+    # t(s)   0          1          2   2.5    3          4
+    #        |----------|----------R----|-----|----------R-----> (no more samples)
+    # value  5.0       12.0            2.0   4.0        5.0
+    #
+    # R = resampling is done
+
+    # Send a few samples and run a resample tick, advancing the fake time by one period
+    sample0s = Sample(timestamp, value=5.0)
+    sample1s = Sample(timestamp + timedelta(seconds=1), value=12.0)
+    await source_sender.send(sample0s)
+    await source_sender.send(sample1s)
+    fake_time.shift(resampling_period_s)  # timer matches resampling period
+    await resampler.resample(one_shot=True)
+
+    assert datetime.now(timezone.utc).timestamp() == 2
+    sink_mock.assert_called_once_with(
+        Sample(
+            timestamp + timedelta(seconds=resampling_period_s), expected_resampled_value
+        )
+    )
+    resampling_fun_mock.assert_called_once_with(
+        a_sequence(sample1s), config, source_props
+    )
+    sink_mock.reset_mock()
+    resampling_fun_mock.reset_mock()
+
+    # Second resampling run
+    sample2_5s = Sample(timestamp + timedelta(seconds=2.5), value=2.0)
+    sample3s = Sample(timestamp + timedelta(seconds=3), value=4.0)
+    sample4s = Sample(timestamp + timedelta(seconds=4), value=5.0)
+    await source_sender.send(sample2_5s)
+    await source_sender.send(sample3s)
+    await source_sender.send(sample4s)
+    fake_time.shift(resampling_period_s + 0.5)  # Timer fired with some delay
+    await resampler.resample(one_shot=True)
+
+    assert datetime.now(timezone.utc).timestamp() == 4.5
+    sink_mock.assert_called_once_with(
+        Sample(
+            # But the sample still gets 4s as timestamp, because we are keeping
+            # the window size constant, not dependent on when the timer fired
+            timestamp + timedelta(seconds=resampling_period_s * 2),
+            expected_resampled_value,
+        )
+    )
+    resampling_fun_mock.assert_called_once_with(
+        a_sequence(sample2_5s, sample3s, sample4s), config, source_props
+    )
+    sink_mock.reset_mock()
+    resampling_fun_mock.reset_mock()
+
+
 async def test_resampling_with_one_window(
     fake_time: time_machine.Coordinates, source_chan: Broadcast[Sample]
 ) -> None: