Add an option to align resampling windows

The `ResamplerConfig` now has a new option `align_to` to align the
resampling windows to some arbitrary time. This means that the
timestamps of the produced samples will be a multiple of the resampling
period beginning at the `align_to` time.

By default `align_to` is the Unix epoch, and if it is `None` it will
have the current behaviour of aligning timestamps to the time when the
resampler is started.

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

Author: llucax

RELEASE_NOTES.md

@@ -58,6 +58,10 @@
 
   * 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.
 
+  * The periods are now aligned to the `UNIX_EPOCH` by default.
+
+    To use the old behaviour (aligning to the time the resampler was created), pass `align_to=None` to the `ResamplerConfig`.
+
 ## New Features
 
 * The core data-pipeline actors are now created automatically (#270).
@@ -76,6 +80,10 @@
 
 * The `Result` class (and subclasses) for the `PowerDistributingActor` are now `dataclass`es, so logging them will produce a more detailed output.
 
+* The `Resampler` can now can align the resampling period to an arbitrary `datetime`.
+
+  This can be configured via the new `align_to` option in the `ResamplerConfig`. By default the resampling period is aligned to the `UNIX_EPOCH`.
+
 ## Bug Fixes
 
 * Change `PowerDistributor` to use all batteries when none are working (#258)

src/frequenz/sdk/timeseries/__init__.py

@@ -6,8 +6,35 @@
 
 A timeseries is a stream (normally an async iterator) of
 [`Sample`][frequenz.sdk.timeseries.Sample]s.
+
+# Periodicity and alignment
+
+All the data produced by this package is always periodic and aligned to the
+`UNIX_EPOCH` (by default).
+
+Classes normally take a (re)sampling period as and argument and, optionally, an
+`align_to` argument.
+
+This means timestamps are always separated exaclty by a period, and that this
+timestamp falls always at multiples of the period, starting at the `align_to`.
+
+This ensures that the data is predictable and consistent among restarts.
+
+Example:
+    If we have a period of 10 seconds, and are aligning to the UNIX
+    epoch. Assuming the following timeline starts in 1970-01-01 00:00:00
+    UTC and our current `now` is 1970-01-01 00:00:32 UTC, then the next
+    timestamp will be at 1970-01-01 00:00:40 UTC:
+
+    ```
+    align_to = 1970-01-01 00:00:00         next event = 1970-01-01 00:00:40
+    |                                       |
+    |---------|---------|---------|-|-------|---------|---------|---------|
+    0        10        20        30 |      40        50        60        70
+                                   now = 1970-01-01 00:00:32
+    ```
 """
 
-from ._base_types import Sample, Sample3Phase
+from ._base_types import UNIX_EPOCH, Sample, Sample3Phase
 
-__all__ = ["Sample", "Sample3Phase"]
+__all__ = ["Sample", "Sample3Phase", "UNIX_EPOCH"]

src/frequenz/sdk/timeseries/_base_types.py

@@ -7,9 +7,12 @@
 
 import functools
 from dataclasses import dataclass, field
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Callable, Iterator, Optional, overload
 
+UNIX_EPOCH = datetime.fromtimestamp(0.0, tz=timezone.utc)
+"""The UNIX epoch (in UTC)."""
+
 
 # Ordering by timestamp is a bit arbitrary, and it is not always what might be
 # wanted. We are using this order now because usually we need to do binary

src/frequenz/sdk/timeseries/_resampling.py

@@ -17,6 +17,7 @@
 
 from .._internal.asyncio import cancel_and_await
 from . import Sample
+from ._base_types import UNIX_EPOCH
 
 _logger = logging.getLogger(__name__)
 
@@ -191,6 +192,17 @@ class ResamplerConfig:
     It must be at bigger than `warn_buffer_len`.
     """
 
+    align_to: datetime | None = UNIX_EPOCH
+    """The time to align the resampling period to.
+
+    The resampling period will be aligned to this time, so the first resampled
+    sample will be at the first multiple of `resampling_period` starting from
+    `align_to`. It must be an aware datetime and can be in the future too.
+
+    If `align_to` is `None`, the resampling period will be aligned to the
+    time the resampler is created.
+    """
+
     def __post_init__(self) -> None:
         """Check that config values are valid.
 
@@ -231,6 +243,10 @@ def __post_init__(self) -> None:
                 self.initial_buffer_len,
                 self.warn_buffer_len,
             )
+        if self.align_to is not None and self.align_to.tzinfo is None:
+            raise ValueError(
+                f"align_to ({self.align_to}) should be a timezone aware datetime"
+            )
 
 
 class SourceStoppedError(RuntimeError):
@@ -348,16 +364,17 @@ 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) + self._config.resampling_period
-        )
+        self._window_end: datetime = self._calculate_window_end()
         """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.
+
+        The window end will also be aligned to the `config.align_to` time, so
+        the window end is deterministic.
         """
 
     @property
@@ -491,6 +508,29 @@ async def _wait_for_next_resampling_period(self) -> None:
                 self._config.resampling_period,
             )
 
+    def _calculate_window_end(self) -> datetime:
+        """Calculate the end of the current resampling window.
+
+        The calculated resampling window end is a multiple of
+        `self._config.resampling_period` starting at `self._config.align_to`.
+
+        if `self._config.align_to` is `None`, the current time is used.
+
+        Returns:
+            The end of the current resampling window aligned to
+                `self._config.align_to`.
+        """
+        now = datetime.now(timezone.utc)
+        period = self._config.resampling_period
+        align_to = self._config.align_to
+
+        if align_to is None:
+            return now + period
+
+        elapsed = (now - align_to) % period
+
+        return now + period - elapsed
+
 
 class _ResamplingHelper:
     """Keeps track of *relevant* samples to pass them to the resampling function.

tests/timeseries/mock_microgrid.py

@@ -119,7 +119,14 @@ async def start(self, mocker: MockerFixture) -> None:
 
         # pylint: disable=protected-access
         _data_pipeline._DATA_PIPELINE = _data_pipeline._DataPipeline(
-            ResamplerConfig(resampling_period=timedelta(seconds=self._sample_rate_s))
+            ResamplerConfig(
+                resampling_period=timedelta(seconds=self._sample_rate_s),
+                # Align to the time the resampler is created to avoid flakiness
+                # in the tests, it seems test using the mock microgrid assume
+                # that the resampling window is aligned to the start of the
+                # test.
+                align_to=None,
+            )
         )
         # pylint: enable=protected-access
 

tests/timeseries/test_resampling.py

@@ -34,7 +34,8 @@
 
 from ..utils import a_sequence
 
-# pylint: disable=too-many-locals,redefined-outer-name
+# We relax some pylint checks as for tests they don't make a lot of sense.
+# pylint: disable=too-many-lines,disable=too-many-locals,redefined-outer-name
 
 
 # Setting 'autouse' has no effect as this method replaces the event loop for all tests in the file.
@@ -168,6 +169,78 @@ async def test_helper_buffer_too_big(
     assert helper._buffer.maxlen == DEFAULT_BUFFER_LEN_MAX
 
 
+@pytest.mark.parametrize(
+    "resampling_period_s,now,align_to,result",
+    (
+        (
+            1.0,
+            datetime(2020, 1, 1, 2, 3, 5, 300000, tzinfo=timezone.utc),
+            datetime(2020, 1, 1, tzinfo=timezone.utc),
+            datetime(2020, 1, 1, 2, 3, 6, tzinfo=timezone.utc),
+        ),
+        (
+            3.0,
+            datetime(2020, 1, 1, 2, 3, 5, 300000, tzinfo=timezone.utc),
+            datetime(2020, 1, 1, 0, 0, 5, tzinfo=timezone.utc),
+            datetime(2020, 1, 1, 2, 3, 8, tzinfo=timezone.utc),
+        ),
+        (
+            10.0,
+            datetime(2020, 1, 1, 2, 3, 5, 300000, tzinfo=timezone.utc),
+            datetime(2020, 1, 1, 0, 0, 5, tzinfo=timezone.utc),
+            datetime(2020, 1, 1, 2, 3, 15, tzinfo=timezone.utc),
+        ),
+        # Future align_to
+        (
+            10.0,
+            datetime(2020, 1, 1, 2, 3, 5, 300000, tzinfo=timezone.utc),
+            datetime(2020, 1, 1, 2, 3, 18, tzinfo=timezone.utc),
+            datetime(2020, 1, 1, 2, 3, 8, tzinfo=timezone.utc),
+        ),
+    ),
+)
+def test_calculate_window_end_trivial_cases(
+    fake_time: time_machine.Coordinates,
+    resampling_period_s: float,
+    now: datetime,
+    align_to: datetime,
+    result: datetime,
+) -> None:
+    """Test the calculation of the resampling window end for simple cases."""
+    resampling_period = timedelta(seconds=resampling_period_s)
+    resampler = Resampler(
+        ResamplerConfig(
+            resampling_period=resampling_period,
+            align_to=align_to,
+        )
+    )
+    fake_time.move_to(now)
+    # pylint: disable=protected-access
+    assert resampler._calculate_window_end() == result
+
+    # Repeat the test with align_to=None, so the result should be align to now
+    # instead
+    resampler_now = Resampler(
+        ResamplerConfig(
+            resampling_period=resampling_period,
+            align_to=now,
+        )
+    )
+    resampler_none = Resampler(
+        ResamplerConfig(
+            resampling_period=resampling_period,
+            align_to=None,
+        )
+    )
+    fake_time.move_to(now)
+    # pylint: disable=protected-access
+    assert (
+        resampler_now._calculate_window_end() == resampler_none._calculate_window_end()
+    )
+    # pylint: disable=protected-access
+    assert resampler_none._calculate_window_end() == now + resampling_period
+
+
 async def test_resampling_window_size_is_constant(
     fake_time: time_machine.Coordinates, source_chan: Broadcast[Sample]
 ) -> None: