Add a Moving Window Class (#190)

Author: matthias-wende-frequenz

RELEASE_NOTES.md

@@ -14,6 +14,7 @@
 
 * A new class `OrderedRingBuffer` is now available, providing a sorted ring buffer of datetime-value pairs with tracking of any values that have not yet been written.
 * Add logical meter formula for EV power.
+* A `MovingWindow` class has been added that consumes a data stream from a logical meter and updates an `OrderedRingBuffer`.
 
 ## Bug Fixes
 

src/frequenz/sdk/timeseries/_moving_window.py

@@ -0,0 +1,189 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""A data window that moves with the latest datapoints of a data stream."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from collections.abc import Sequence
+from datetime import datetime, timedelta
+
+import numpy as np
+from frequenz.channels import Receiver
+from numpy.typing import ArrayLike
+
+from .._internal.asyncio import cancel_and_await
+from . import Sample
+from ._ringbuffer import OrderedRingBuffer
+
+log = logging.getLogger(__name__)
+
+
+class MovingWindow(Sequence):
+    """
+    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
+    integers or timestamps.
+
+    Note that a numpy ndarray is returned and thus users can use
+    numpys operations directly on a window.
+
+    The window uses an ringbuffer for storage and the first element is aligned to
+    a fixed defined point in time. Since the moving nature of the window, the
+    date of the first and the last element are constantly changing and therefore
+    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
+    [`OrderedRingBuffer`][frequenz.sdk.timeseries._ringbuffer.OrderedRingBuffer]
+    documentation.
+
+
+    **Example1** (calculating the mean of a time interval):
+
+    ```
+    window = MovingWindow(size=100, resampled_data_recv=resampled_data_recv)
+
+    time_start = datetime.now()
+    time_end = time_start + timedelta(minutes=5)
+
+    # ... wait for 5 minutes until the buffer is filled
+    await asyncio.sleep(5)
+
+    # return an numpy array from the window
+    a = window[time_start:time_end]
+    # and use it to for example calculate the mean
+    mean = a.mean()
+    '''
+
+    **Example2** (create a polars data frame from a `MovingWindow`):
+
+    ```
+    import polars as pl
+
+    # 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)
+
+    # 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)
+    s = pl.Series("Jan_1", mv[time_start:time_end])
+    ```
+    """
+
+    def __init__(
+        self,
+        size: int,
+        resampled_data_recv: Receiver[Sample],
+        sampling_period: timedelta,
+        window_alignment: datetime = datetime(1, 1, 1),
+    ) -> None:
+        """
+        Initialize the MovingWindow.
+
+        This method creates the underlying ringbuffer and starts a
+        new task that updates the ringbuffer with new incoming samples.
+        The task stops running only if the channel receiver is closed.
+
+        Args:
+            size: The number of elements that are stored.
+            resampled_data_recv: A receiver that delivers samples with a
+                given sampling period.
+            sampling_period: The sampling period.
+            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)
+                For further information, consult the class level documentation.
+
+        Raises:
+            asyncio.CancelledError: when the task gets cancelled.
+        """
+        self._resampled_data_recv = resampled_data_recv
+        self._buffer = OrderedRingBuffer(
+            np.empty(shape=size, dtype=float),
+            sampling_period=sampling_period,
+            time_index_alignment=window_alignment,
+        )
+        self._copy_buffer = False
+        self._update_window_task: asyncio.Task = 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."""
+        try:
+            async for sample in self._resampled_data_recv:
+                log.debug("Received new sample: %s", sample)
+                self._buffer.update(sample)
+        except asyncio.CancelledError:
+            log.info("MovingWindow task has been cancelled.")
+            return
+
+        log.error("Channel has been closed")
+
+    async def stop(self) -> None:
+        """Cancel the running task and stop the MovingWindow."""
+        await cancel_and_await(self._update_window_task)
+
+    def __len__(self) -> int:
+        """
+        Return the size of the `MovingWindow`s underlying buffer.
+
+        Returns:
+            The size of the `MovingWindow`.
+        """
+        return len(self._buffer)
+
+    def __getitem__(self, key: int | datetime | slice) -> float | ArrayLike:
+        """
+        Return a sub window of the `MovingWindow`.
+
+        The `MovingWindow` is accessed either by timestamp or by index
+        or by a slice of timestamps or integers.
+
+        * If the key is an integer, the float value of that key
+          at the given position is returned.
+        * If the key is a datetime object, the float value of that key
+          that corresponds to the timestamp is returned.
+        * If the key is a slice of timestamps or integers, an ndarray is returned,
+          where the bounds correspond to the slice bounds.
+          Note that a half open interval, which is open at the end, is returned.
+
+        Args:
+            key: Either an integer or a timestamp or a slice of timestamps or integers.
+
+        Raises:
+            IndexError: when requesting an out of range timestamp or index
+            TypeError: when the key is not a datetime or slice object.
+
+        Returns:
+            A float if the key is a number or a timestamp.
+            an numpy array if the key is a slice.
+        """
+        if isinstance(key, slice):
+            log.debug("Returning slice for [%s:%s].", key.start, key.stop)
+            # 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)
+            if isinstance(key.start, int) and isinstance(key.stop, int):
+                return self._buffer[key]
+        elif isinstance(key, datetime):
+            log.debug("Returning value at time %s ", key)
+            return self._buffer[self._buffer.datetime_to_index(key)]
+        elif isinstance(key, int):
+            return self._buffer[key]
+
+        raise TypeError(
+            "Key has to be either a timestamp or an integer "
+            "or a slice of timestamps or integers"
+        )

src/frequenz/sdk/timeseries/_ringbuffer.py

@@ -28,7 +28,7 @@ class Gap:
     """End of the range, exclusive."""
 
     def contains(self, timestamp: datetime):
-        """Check if a given timestamp is inside this  gap.
+        """Check if a given timestamp is inside this gap.
 
         Args:
             timestamp: Timestamp to check.
@@ -55,8 +55,8 @@ def __init__(
 
         Args:
             buffer: Instance of a buffer container to use internally.
-            sampling_period: Timedelta of the desired resampling period.
-            time_index_alignment: Arbitary point in time used to align
+            sampling_period: Timedelta of the desired sampling period.
+            time_index_alignment: Arbitrary point in time used to align
                 timestamped data with the index position in the buffer.
                 Used to make the data stored in the buffer align with the
                 beginning and end of the buffer borders.

tests/timeseries/test_moving_window.py

@@ -0,0 +1,77 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for the moving window."""
+
+import asyncio
+from datetime import datetime, timedelta
+from typing import Sequence, Tuple
+
+import numpy as np
+from frequenz.channels import Broadcast, Sender
+
+from frequenz.sdk.timeseries import Sample
+from frequenz.sdk.timeseries._moving_window import MovingWindow
+
+
+async def push_lm_data(sender: Sender[Sample], test_seq: Sequence[float]) -> None:
+    """
+    Push data in the passed sender to mock `LogicalMeter` behaviour.
+    Starting with the First of January 2023.
+
+    Args:
+        sender: Sender for pushing resampled samples to the `MovingWindow`.
+        test_seq: The Sequence that is pushed into the `MovingWindow`.
+    """
+    start_ts: datetime = datetime(2023, 1, 1)
+    for i, j in zip(test_seq, range(0, len(test_seq))):
+        timestamp = start_ts + timedelta(seconds=j)
+        await sender.send(Sample(timestamp, float(i)))
+
+    await asyncio.sleep(0.0)
+
+
+def init_moving_window(shape: int) -> Tuple[MovingWindow, Sender[Sample]]:
+    """
+    Initialize the moving window with given shape
+
+    Args:
+        shape: The size of the `MovingWindow`
+
+    Returns:
+        tuple[MovingWindow, Sender[Sample]]: A pair of sender and `MovingWindow`.
+    """
+    lm_chan = Broadcast[Sample]("lm_net_power")
+    lm_tx = lm_chan.new_sender()
+    window = MovingWindow(shape, lm_chan.new_receiver(), timedelta(seconds=1))
+    return window, lm_tx
+
+
+async def test_access_window_by_index() -> None:
+    """Test indexing a window by integer index"""
+    window, sender = init_moving_window(1)
+    await push_lm_data(sender, [1])
+    assert np.array_equal(window[0], 1.0)
+
+
+async def test_access_window_by_timestamp() -> None:
+    """Test indexing a window by timestamp"""
+    window, sender = init_moving_window(1)
+    await push_lm_data(sender, [1])
+    assert np.array_equal(window[datetime(2023, 1, 1)], 1.0)
+
+
+async def test_access_window_by_int_slice() -> None:
+    """Test accessing a subwindow with an integer slice"""
+    window, sender = init_moving_window(5)
+    await push_lm_data(sender, range(0, 5))
+    assert np.array_equal(window[3:5], np.array([3.0, 4.0]))
+
+
+async def test_access_window_by_ts_slice() -> None:
+    """Test accessing a subwindow with a timestamp slice"""
+    window, sender = init_moving_window(5)
+    await push_lm_data(sender, range(0, 5))
+    time_start = datetime(2023, 1, 1) + timedelta(seconds=3)
+    time_end = time_start + timedelta(seconds=2)
+    assert np.array_equal(window[time_start:time_end], np.array([3.0, 4.0]))  # type: ignore