Update MovingWindow to add resampler (#272)

Author: daniel-zullo-frequenz

RELEASE_NOTES.md

@@ -12,6 +12,11 @@
   * 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.
 
 ## New Features
 
@@ -26,6 +31,7 @@
       )
       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.
 
 ## Bug Fixes

benchmarks/timeseries/benchmark_ringbuffer.py

@@ -5,7 +5,7 @@
 
 import random
 import timeit
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Any, Dict, TypeVar
 
 import numpy as np
@@ -23,7 +23,7 @@
 def fill_buffer(days: int, buffer: OrderedRingBuffer[Any]) -> None:
     """Fill the given buffer up to the given amount of days, one sample per minute."""
     random.seed(0)
-    basetime = datetime(2022, 1, 1)
+    basetime = datetime(2022, 1, 1, tzinfo=timezone.utc)
     print("..filling", end="", flush=True)
 
     for day in range(days):
@@ -36,7 +36,7 @@ def fill_buffer(days: int, buffer: OrderedRingBuffer[Any]) -> None:
 
 def test_days(days: int, buffer: OrderedRingBuffer[Any]) -> None:
     """Gets the data for each of the 29 days."""
-    basetime = datetime(2022, 1, 1)
+    basetime = datetime(2022, 1, 1, tzinfo=timezone.utc)
 
     for day in range(days):
         # pylint: disable=unused-variable
@@ -51,7 +51,7 @@ def test_slices(days: int, buffer: OrderedRingBuffer[Any], median: bool) -> None
     Takes a buffer, fills it up and then excessively gets
     the data for each day to calculate the average/median.
     """
-    basetime = datetime(2022, 1, 1)
+    basetime = datetime(2022, 1, 1, tzinfo=timezone.utc)
 
     total = 0.0
 

src/frequenz/sdk/timeseries/_moving_window.py

@@ -7,16 +7,18 @@
 
 import asyncio
 import logging
+import math
 from collections.abc import Sequence
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import SupportsIndex, overload
 
 import numpy as np
-from frequenz.channels import Receiver
+from frequenz.channels import Broadcast, Receiver, Sender
 from numpy.typing import ArrayLike
 
 from .._internal.asyncio import cancel_and_await
 from . import Sample
+from ._resampling import Resampler, ResamplerConfig
 from ._ringbuffer import OrderedRingBuffer
 
 log = logging.getLogger(__name__)
@@ -27,7 +29,7 @@ class MovingWindow:
     A data window that moves with the latest datapoints of a data stream.
 
     After initialization the `MovingWindow` can be accessed by an integer
-    index or a timestamp. A sub window can be accessed by using a slice of integers
+    index or a timestamp. A sub window can be accessed by using a slice of
     integers or timestamps.
 
     Note that a numpy ndarray is returned and thus users can use
@@ -39,19 +41,31 @@ class MovingWindow:
     the point in time that defines the alignment can be outside of the time window.
     Modulo arithmetic is used to move the `window_alignment` timestamp into the
     latest window.
-    If for example the `window_alignment` parameter is set to `datetime(1, 1, 1)`
-    and the window size is bigger than one day then the first element will always
-    be aligned to the midnight. For further information see also the
+    If for example the `window_alignment` parameter is set to
+    `datetime(1, 1, 1, tzinfo=timezone.utc)` and the window size is bigger than
+    one day then the first element will always be aligned to the midnight.
+    For further information see also the
     [`OrderedRingBuffer`][frequenz.sdk.timeseries._ringbuffer.OrderedRingBuffer]
     documentation.
 
+    Resampling might be required to reduce the number of samples to store, and
+    it can be set by specifying the resampler config parameter so that the user
+    can control the granularity of the samples to be stored in the underlying
+    buffer.
+
+    If resampling is not required, the resampler config parameter can be
+    set to None in which case the MovingWindow will not perform any resampling.
 
     **Example1** (calculating the mean of a time interval):
 
     ```
-    window = MovingWindow(size=100, resampled_data_recv=resampled_data_recv)
+    window = MovingWindow(
+        size=timedelta(minutes=5),
+        resampled_data_recv=resampled_data_recv,
+        input_sampling_period=timedelta(seconds=1),
+    )
 
-    time_start = datetime.now()
+    time_start = datetime.now(tz=timezone.utc)
     time_end = time_start + timedelta(minutes=5)
 
     # ... wait for 5 minutes until the buffer is filled
@@ -70,24 +84,29 @@ class MovingWindow:
 
     # create a window that stores two days of data
     # starting at 1.1.23 with samplerate=1
-    window = MovingWindow(size = (60 * 60 * 24 * 2), sample_receiver)
+    window = MovingWindow(
+        size=timedelta(days=2),
+        resampled_data_recv=sample_receiver,
+        input_sampling_period=timedelta(seconds=1),
+    )
 
     # wait for one full day until the buffer is filled
     asyncio.sleep(60*60*24)
 
     # create a polars series with one full day of data
-    time_start = datetime(2023, 1, 1)
-    time_end = datetime(2023, 1, 2)
+    time_start = datetime(2023, 1, 1, tzinfo=timezone.utc)
+    time_end = datetime(2023, 1, 2, tzinfo=timezone.utc)
     s = pl.Series("Jan_1", mv[time_start:time_end])
     ```
     """
 
-    def __init__(
+    def __init__(  # pylint: disable=too-many-arguments
         self,
-        size: int,
+        size: timedelta,
         resampled_data_recv: Receiver[Sample],
-        sampling_period: timedelta,
-        window_alignment: datetime = datetime(1, 1, 1),
+        input_sampling_period: timedelta,
+        resampler_config: ResamplerConfig | None = None,
+        window_alignment: datetime = datetime(1, 1, 1, tzinfo=timezone.utc),
     ) -> None:
         """
         Initialize the MovingWindow.
@@ -97,45 +116,97 @@ def __init__(
         The task stops running only if the channel receiver is closed.
 
         Args:
-            size: The number of elements that are stored.
+            size: The time span of the moving window over which samples will be stored.
             resampled_data_recv: A receiver that delivers samples with a
                 given sampling period.
-            sampling_period: The sampling period.
+            input_sampling_period: The time interval between consecutive input samples.
+            resampler_config: The resampler configuration in case resampling is required.
             window_alignment: A datetime object that defines a point in time to which
                 the window is aligned to modulo window size.
-                (default is midnight 01.01.01)
+                (default is 0001-01-01T00:00:00+00:00)
                 For further information, consult the class level documentation.
 
         Raises:
             asyncio.CancelledError: when the task gets cancelled.
         """
+        assert (
+            input_sampling_period.total_seconds() > 0
+        ), "The input sampling period should be greater than zero."
+        assert (
+            input_sampling_period <= size
+        ), "The input sampling period should be equal to or lower than the window size."
+
+        sampling = input_sampling_period
+        self._resampler: Resampler | None = None
+        self._resampler_sender: Sender[Sample] | None = None
+        self._resampler_task: asyncio.Task[None] | None = None
+
+        if resampler_config:
+            resampling_period = timedelta(seconds=resampler_config.resampling_period_s)
+            assert (
+                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 period might not fit perfectly into the window size.
+        num_samples = math.ceil(size.total_seconds() / sampling.total_seconds())
+
         self._resampled_data_recv = resampled_data_recv
         self._buffer = OrderedRingBuffer(
-            np.empty(shape=size, dtype=float),
-            sampling_period=sampling_period,
+            np.empty(shape=num_samples, dtype=float),
+            sampling_period=sampling,
             time_index_alignment=window_alignment,
         )
-        self._copy_buffer = False
+
+        if self._resampler:
+            self._configure_resampler()
+
         self._update_window_task: asyncio.Task[None] = asyncio.create_task(
             self._run_impl()
         )
-        log.debug("Cancelling MovingWindow task: %s", __name__)
 
     async def _run_impl(self) -> None:
-        """Awaits samples from the receiver and updates the underlying ringbuffer."""
+        """Awaits samples from the receiver and updates the underlying ringbuffer.
+
+        Raises:
+            asyncio.CancelledError: if the MovingWindow task is cancelled.
+        """
         try:
             async for sample in self._resampled_data_recv:
                 log.debug("Received new sample: %s", sample)
-                self._buffer.update(sample)
+                if self._resampler and self._resampler_sender:
+                    await self._resampler_sender.send(sample)
+                else:
+                    self._buffer.update(sample)
+
         except asyncio.CancelledError:
             log.info("MovingWindow task has been cancelled.")
-            return
+            raise
 
         log.error("Channel has been closed")
 
     async def stop(self) -> None:
-        """Cancel the running task and stop the MovingWindow."""
+        """Cancel the running tasks and stop the MovingWindow."""
         await cancel_and_await(self._update_window_task)
+        if self._resampler_task:
+            await cancel_and_await(self._resampler_task)
+
+    def _configure_resampler(self) -> None:
+        """Configure the components needed to run the resampler."""
+        assert self._resampler is not None
+
+        async def sink_buffer(sample: Sample) -> None:
+            if sample.value is not None:
+                self._buffer.update(sample)
+
+        resampler_channel = Broadcast[Sample]("average")
+        self._resampler_sender = resampler_channel.new_sender()
+        self._resampler.add_timeseries(
+            "avg", resampler_channel.new_receiver(), sink_buffer
+        )
+        self._resampler_task = asyncio.create_task(self._resampler.resample())
 
     def __len__(self) -> int:
         """
@@ -198,7 +269,7 @@ def __getitem__(self, key: SupportsIndex | datetime | slice) -> float | ArrayLik
             # we are doing runtime typechecks since there is no abstract slice type yet
             # see also (https://peps.python.org/pep-0696)
             if isinstance(key.start, datetime) and isinstance(key.stop, datetime):
-                return self._buffer.window(key.start, key.stop, self._copy_buffer)
+                return self._buffer.window(key.start, key.stop)
             if isinstance(key.start, int) and isinstance(key.stop, int):
                 return self._buffer[key]
         elif isinstance(key, datetime):

src/frequenz/sdk/timeseries/_resampling.py

@@ -146,12 +146,12 @@ class ResamplerConfig:
     Example:
         If `resampling_period_s` is 3, the input sampling period is
         1 and `max_data_age_in_periods` is 2, then data older than 3*2
-        = 6 secods will be discarded when creating a new sample and never
+        = 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
         5 and `max_data_age_in_periods` is 2, then data older than 5*2
-        = 10 secods will be discarded when creating a new sample and never
+        = 10 seconds will be discarded when creating a new sample and never
         passed to the resampling function.
     """
 
@@ -177,7 +177,7 @@ class ResamplerConfig:
     """The minimum length of the resampling buffer that will emit a warning.
 
     If a buffer grows bigger than this value, it will emit a warning in the
-    logs, so buffers don't grow too big inadvertly.
+    logs, so buffers don't grow too big inadvertently.
 
     It must be at least 1 and at most `max_buffer_len`.
     """
@@ -242,12 +242,11 @@ def __init__(self, source: Source) -> None:
         """Create an instance.
 
         Args:
-            source: The source of the timeseries that stopped producting
-                samples.
+            source: The source of the timeseries that stopped producing samples.
         """
         super().__init__(f"Timeseries stopped producing samples, source: {source}")
         self.source = source
-        """The source of the timeseries that stopped producting samples."""
+        """The source of the timeseries that stopped producing samples."""
 
     def __repr__(self) -> str:
         """Return the representation of the instance.
@@ -672,7 +671,7 @@ def resample(self, timestamp: datetime) -> Sample:
         min_index = bisect(self._buffer, Sample(minimum_relevant_timestamp, None))
         max_index = bisect(self._buffer, Sample(timestamp, None))
         # Using itertools for slicing doesn't look very efficient, but
-        # experiements with a custom (ring) buffer that can slice showed that
+        # experiments with a custom (ring) buffer that can slice showed that
         # it is not that bad. See:
         # https://github.com/frequenz-floss/frequenz-sdk-python/pull/130
         # So if we need more performance beyond this point, we probably need to

src/frequenz/sdk/timeseries/_ringbuffer.py

@@ -8,7 +8,7 @@
 from collections.abc import Iterable
 from copy import deepcopy
 from dataclasses import dataclass
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Generic, List, SupportsFloat, SupportsIndex, TypeVar, overload
 
 import numpy as np
@@ -46,11 +46,14 @@ def contains(self, timestamp: datetime) -> bool:
 class OrderedRingBuffer(Generic[FloatArray]):
     """Time aware ringbuffer that keeps its entries sorted by time."""
 
+    _DATETIME_MIN = datetime.min.replace(tzinfo=timezone.utc)
+    _DATETIME_MAX = datetime.max.replace(tzinfo=timezone.utc)
+
     def __init__(
         self,
         buffer: FloatArray,
         sampling_period: timedelta,
-        time_index_alignment: datetime = datetime(1, 1, 1),
+        time_index_alignment: datetime = datetime(1, 1, 1, tzinfo=timezone.utc),
     ) -> None:
         """Initialize the time aware ringbuffer.
 
@@ -76,8 +79,8 @@ def __init__(
         self._time_index_alignment: datetime = time_index_alignment
 
         self._gaps: list[Gap] = []
-        self._datetime_newest: datetime = datetime.min
-        self._datetime_oldest: datetime = datetime.max
+        self._datetime_newest: datetime = self._DATETIME_MIN
+        self._datetime_oldest: datetime = self._DATETIME_MAX
         self._time_range: timedelta = (len(self._buffer) - 1) * sampling_period
 
     @property
@@ -130,7 +133,10 @@ def update(self, sample: Sample) -> None:
         timestamp = self._normalize_timestamp(sample.timestamp)
 
         # Don't add outdated entries
-        if timestamp < self._datetime_oldest and self._datetime_oldest != datetime.max:
+        if (
+            timestamp < self._datetime_oldest
+            and self._datetime_oldest != self._DATETIME_MAX
+        ):
             raise IndexError(
                 f"Timestamp {timestamp} too old (cut-off is at {self._datetime_oldest})."
             )
@@ -485,7 +491,7 @@ def __len__(self) -> int:
         Returns:
             The length.
         """
-        if self._datetime_newest == datetime.min:
+        if self._datetime_newest == self._DATETIME_MIN:
             return 0
 
         start_index = self.datetime_to_index(self._datetime_oldest)