Add an option to align resampling windows (#335)

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.

Fixes #328.

Author: llucax

RELEASE_NOTES.md

@@ -12,12 +12,10 @@
   battery_power_receiver = microgrid.battery_pool().power.new_receiver()
   ```
 
-+ Formulas composition has changed (#327) -
-  - receivers from formulas are no longer composable.
-  - formula composition is now done by composing FormulaEngine instances.
-  - Automatic formulas from the logical meter and *pools, are now
-    properties, and return `FormulaEngine` instances, which can be
-    composed further, or can provide a receiver to fetch values.
+* Formulas composition has changed (#327)
+  * Receivers from formulas are no longer composable.
+  * Formula composition is now done by composing FormulaEngine instances.
+  * Automatic formulas from the logical meter and \*pools, are now properties, and return `FormulaEngine` instances, which can be composed further, or can provide a receiver to fetch values.
 
   ``` python
   grid_power_receiver = microgrid.logical_meter().grid_power.new_receiver()
@@ -30,26 +28,47 @@
   inverter_power_receiver = self._inverter_power.new_receiver()
   ```
 
-* Update BatteryStatus to mark battery with unknown capacity as not working (#263)
+* Update `BatteryStatus` to mark battery with unknown capacity as not working (#263)
+
 * The channels dependency was updated to v0.14.0 (#292)
+
 * Some properties for `PowerDistributingActor` results were renamed to be more consistent between `Success` and `PartialFailure`:
+
   * The `Success.used_batteries` property was renamed to `succeeded_batteries`.
   * The `PartialFailure.success_batteries` property was renamed to `succeeded_batteries`.
   * The `succeed_power` property was renamed to `succeeded_power` for both `Success` and `PartialFailure`.
-* Update MovingWindow to accept size parameter as timedelta instead of int (#269).
-  This change allows users to define the time span of the moving window more intuitively, representing the duration over which samples will be stored.
-* Add a resampler in the MovingWindow to control the granularity of the samples to be stored in the underlying buffer (#269).
-  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.
-* 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.
+
+* `MovingWindow`
+
+  * Accept the `size` parameter as `timedelta` instead of `int` (#269).
+
+    This change allows users to define the time span of the moving window more intuitively, representing the duration over which samples will be stored.
+
+  * The input data will be resampled if a `resampler_config` is passed (#269).
+
+    This allows controlling the granularity of the samples to be stored in the underlying buffer.
+
+    Note that the parameter `sampling_period` has been renamed to `input_sampling_period` to better distinguish it from the sampling period parameter in the `resampler_config`.
+
+* `Resampler`
+
+  * 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.
+
+  * 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
 
-* Automatic creation of core data-pipeline actors, to eliminate a lot
-  of boiler plate code.  This makes it much simpler to deploy apps
-  (#270).  For example:
+* The core data-pipeline actors are now created automatically (#270).
+
+  This eliminates a lot of boiler plate code and makes it much simpler to deploy apps.
+
+  For example:
 
   ``` python
   async def run():
@@ -59,9 +78,14 @@
       grid_power = microgrid.logical_meter().grid_power()
   ```
 
-* The `Result` class (and subclasses) for the `PowerDistributingActor` are now dataclasses, so logging them will produce a more detailed output.
+* 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 if none is working (#258)
-* Update the ordered ring buffer to fix the len() function so that it returns a value equal to or greater than zero, as expected (#274)
+* Change `PowerDistributor` to use all batteries when none are working (#258)
+
+* Update the ordered ring buffer to fix the `len()` function so that it returns a value equal to or greater than zero, as expected (#274)

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: