Serializing Ringbuffer (#167)

based on #139

Author: mathias-baumann-frequenz

RELEASE_NOTES.md

@@ -11,7 +11,7 @@
 * A `MovingWindow` class has been added that consumes a data stream from a logical meter and updates an `OrderedRingBuffer`.
 * Add EVChargerPool implementation. It has only streaming state changes for ev chargers, now.
 * Add 3-phase current formulas: `3-phase grid_current` and `3-phase ev_charger_current` to the LogicalMeter.
-
+* A new class `SerializableRingbuffer` is now available, extending the `OrderedRingBuffer` class with the ability to load & dump the data to disk.
 
 ## Bug Fixes
 

benchmarks/timeseries/benchmark_ringbuffer.py

@@ -70,7 +70,7 @@ def test_slices(days: int, buffer: OrderedRingBuffer[Any], median: bool) -> None
 def test_29_days_list(num_runs: int) -> Dict[str, float]:
     """Run the 29 day test on the list backend."""
     days = 29
-    buffer = OrderedRingBuffer([0] * MINUTES_IN_29_DAYS, timedelta(minutes=1))
+    buffer = OrderedRingBuffer([0.0] * MINUTES_IN_29_DAYS, timedelta(minutes=1))
 
     fill_time = timeit.Timer(lambda: fill_buffer(days, buffer)).timeit(number=1)
     test_time = timeit.Timer(lambda: test_days(days, buffer)).timeit(number=num_runs)
@@ -95,7 +95,7 @@ def test_29_days_array(num_runs: int) -> Dict[str, float]:
 def test_29_days_slicing_list(num_runs: int) -> Dict[str, float]:
     """Run slicing tests on list backend."""
     days = 29
-    buffer = OrderedRingBuffer([0] * MINUTES_IN_29_DAYS, timedelta(minutes=1))
+    buffer = OrderedRingBuffer([0.0] * MINUTES_IN_29_DAYS, timedelta(minutes=1))
 
     fill_time = timeit.Timer(lambda: fill_buffer(days, buffer)).timeit(number=1)
     median_test_time = timeit.Timer(

benchmarks/timeseries/serializable_ringbuffer.py

@@ -0,0 +1,81 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""Benchmarks the `SerializableRingbuffer` class."""
+
+from __future__ import annotations
+
+import fnmatch
+import os
+import time
+from datetime import datetime, timedelta
+from typing import Any
+
+import numpy as np
+
+from frequenz.sdk.timeseries import Sample
+from frequenz.sdk.timeseries._serializable_ringbuffer import SerializableRingBuffer
+
+FILE_NAME = "ringbuffer.pkl"
+FIVE_MINUTES = timedelta(minutes=5)
+
+# Size of the ringbuffer to dump/load
+SIZE = 4000_000
+# Number of iterations to run the benchmark
+ITERATIONS = 100
+
+
+def delete_files_with_prefix(prefix: str) -> None:
+    """Delete all files starting with the given prefix.
+
+    Args:
+        prefix: Prefix of the files to delete
+    """
+    for file in os.listdir():
+        if fnmatch.fnmatch(file, prefix + "*"):
+            os.remove(file)
+
+
+def benchmark_serialization(
+    ringbuffer: SerializableRingBuffer[Any], iterations: int
+) -> float:
+    """Benchmark the given buffer `iteration` times.
+
+    Args:
+        ringbuffer: Ringbuffer to benchmark to serialize.
+        iterations: amount of iterations to run.
+    """
+    total = 0.0
+    for _ in range(iterations):
+        start = time.time()
+        ringbuffer.dump()
+        SerializableRingBuffer.load(FILE_NAME)
+        end = time.time()
+        total += end - start
+        delete_files_with_prefix(FILE_NAME)
+
+    return total / iterations
+
+
+def main() -> None:
+    """Run Benchmark."""
+    ringbuffer = SerializableRingBuffer(
+        np.arange(0, SIZE, dtype=np.float64), timedelta(minutes=5), FILE_NAME
+    )
+
+    print("size:", SIZE)
+    print("iterations:", ITERATIONS)
+
+    for i in range(0, SIZE, 10000):
+        ringbuffer.update(
+            Sample(datetime.fromtimestamp(200 + i * FIVE_MINUTES.total_seconds()), i)
+        )
+
+    print(
+        "Avg time for Pickle dump/load:  "
+        f"{benchmark_serialization(ringbuffer, ITERATIONS)}s"
+    )
+
+
+if __name__ == "__main__":
+    main()

src/frequenz/sdk/timeseries/_ringbuffer.py

@@ -5,17 +5,18 @@
 
 from __future__ import annotations
 
+from collections.abc import Iterable
 from copy import deepcopy
 from dataclasses import dataclass
 from datetime import datetime, timedelta
-from typing import Any, Generic, List, Sequence, TypeVar
+from typing import Any, Generic, List, TypeVar, overload
 
 import numpy as np
-from numpy.lib import math
+import numpy.typing as npt
 
-from frequenz.sdk.timeseries import Sample
+from . import Sample
 
-Container = TypeVar("Container", list, np.ndarray)
+FloatArray = TypeVar("FloatArray", List[float], npt.NDArray[np.float64])
 
 
 @dataclass
@@ -42,12 +43,12 @@ def contains(self, timestamp: datetime):
         return False
 
 
-class OrderedRingBuffer(Generic[Container]):
+class OrderedRingBuffer(Generic[FloatArray]):
     """Time aware ringbuffer that keeps its entries sorted by time."""
 
     def __init__(
         self,
-        buffer: Container,
+        buffer: FloatArray,
         sampling_period: timedelta,
         time_index_alignment: datetime = datetime(1, 1, 1),
     ) -> None:
@@ -129,7 +130,7 @@ def update(self, sample: Sample) -> None:
         # Don't add outdated entries
         if timestamp < self._datetime_oldest and self._datetime_oldest != datetime.max:
             raise IndexError(
-                f"Timestamp too old (cut-off is at {self._datetime_oldest})."
+                f"Timestamp {timestamp} too old (cut-off is at {self._datetime_oldest})."
             )
 
         # Update timestamps
@@ -138,7 +139,7 @@ def update(self, sample: Sample) -> None:
         self._datetime_oldest = self._datetime_newest - self._time_range
 
         # Update data
-        value: float = math.nan if sample.value is None else sample.value
+        value: float = np.nan if sample.value is None else sample.value
         self._buffer[self.datetime_to_index(timestamp)] = value
 
         self._update_gaps(timestamp, prev_newest, sample.value is None)
@@ -179,7 +180,7 @@ def datetime_to_index(
 
     def window(
         self, start: datetime, end: datetime, force_copy: bool = False
-    ) -> Container:
+    ) -> FloatArray:
         """Request a view on the data between start timestamp and end timestamp.
 
         If the data is not used immediately it could be overwritten.
@@ -204,7 +205,7 @@ def window(
                 copy of the data.
 
         Raises:
-            IndexError: when requesting a window with invalid timestamps.
+            IndexError: When requesting a window with invalid timestamps.
 
         Returns:
             The requested window
@@ -228,14 +229,13 @@ def window(
                     window = np.concatenate((window, self._buffer[0:end_index]))
             return window
 
-        def in_gap(gap) -> bool:
-            return gap.contains(start) or gap.contains(end)
-
         # Return a copy if there are none-values in the data
-        if force_copy or any(map(in_gap, self._gaps)):
-            return deepcopy(self._buffer[start_index:end_index])
+        if force_copy or any(
+            map(lambda gap: gap.contains(start) or gap.contains(end), self._gaps)
+        ):
+            return deepcopy(self[start_index:end_index])
 
-        return self._buffer[start_index:end_index]
+        return self[start_index:end_index]
 
     def is_missing(self, timestamp: datetime) -> bool:
         """Check if the given timestamp falls within a gap.
@@ -386,7 +386,33 @@ def _wrap(self, index: int) -> int:
         """
         return index % self.maxlen
 
-    def __setitem__(self, index_or_slice: int | slice, value: float) -> None:
+    @overload
+    def __setitem__(self, index_or_slice: slice, value: Iterable[float]) -> None:
+        """Set values at the request slice positions.
+
+        No wrapping of the index will be done.
+        Create a feature request if you require this function.
+
+        Args:
+            index_or_slice: Slice specification of the requested data.
+            value: Sequence of value to set at the given range.
+        """
+
+    @overload
+    def __setitem__(self, index_or_slice: int, value: float) -> None:
+        """Set value at requested index.
+
+        No wrapping of the index will be done.
+        Create a feature request if you require this function.
+
+        Args:
+            index_or_slice: Index of the data.
+            value: Value to set at the given position.
+        """
+
+    def __setitem__(
+        self, index_or_slice: int | slice, value: float | Iterable[float]
+    ) -> None:
         """Set item or slice at requested position.
 
         No wrapping of the index will be done.
@@ -398,7 +424,29 @@ def __setitem__(self, index_or_slice: int | slice, value: float) -> None:
         """
         self._buffer.__setitem__(index_or_slice, value)
 
-    def __getitem__(self, index_or_slice: int | slice) -> float | Sequence[float]:
+    @overload
+    def __getitem__(self, index_or_slice: int) -> float:
+        """Get item at requested position.
+
+        No wrapping of the index will be done.
+        Create a feature request if you require this function.
+
+        Args:
+            index_or_slice: Index of the requested data.
+        """
+
+    @overload
+    def __getitem__(self, index_or_slice: slice) -> FloatArray:
+        """Get the data described by the given slice.
+
+        No wrapping of the index will be done.
+        Create a feature request if you require this function.
+
+        Args:
+            index_or_slice: Slice specification of where the requested data is.
+        """
+
+    def __getitem__(self, index_or_slice: int | slice) -> float | FloatArray:
         """Get item or slice at requested position.
 
         No wrapping of the index will be done.

src/frequenz/sdk/timeseries/_serializable_ringbuffer.py

@@ -0,0 +1,85 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""Ringbuffer implementation with serialization support."""
+
+# For use of the class type hint inside the class itself.
+from __future__ import annotations
+
+import pickle
+from datetime import datetime, timedelta
+from os.path import exists
+
+from ._ringbuffer import FloatArray, OrderedRingBuffer
+
+# Version of the file dumping/loading format
+FILE_FORMAT_VERSION: int = 1
+
+
+class SerializableRingBuffer(OrderedRingBuffer[FloatArray]):
+    """Sorted ringbuffer with serialization support."""
+
+    # pylint: disable=too-many-arguments
+    def __init__(
+        self,
+        buffer: FloatArray,
+        sampling_period: timedelta,
+        path: str,
+        time_index_alignment: datetime = datetime(1, 1, 1),
+    ) -> None:
+        """Initialize the time aware ringbuffer.
+
+        Args:
+            buffer: Instance of a buffer container to use internally.
+            sampling_period: Timedelta of the desired resampling period.
+            path: Path to where the data should be saved to and loaded from.
+            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.
+                For example, if the `time_index_alignment` is set to
+                "0001-01-01 12:00:00", and the `sampling_period` is set to
+                1 hour and the length of the buffer is 24, then the data
+                stored in the buffer could correspond to the time range from
+                "2022-01-01 12:00:00" to "2022-01-02 12:00:00" (date chosen
+                arbitrarily here).
+        """
+        super().__init__(buffer, sampling_period, time_index_alignment)
+        self._path = path
+        self._file_format_version = FILE_FORMAT_VERSION
+
+    def dump(self) -> None:
+        """Dump data to disk.
+
+        Raises:
+            I/O related exceptions when the file cannot be written.
+        """
+        with open(self._path, mode="wb+") as fileobj:
+            pickle.dump(self, fileobj)
+
+    @classmethod
+    def load(cls, path: str) -> SerializableRingBuffer[FloatArray] | None:
+        """Load data from disk.
+
+        Args:
+            path: Path to the file where the data is stored.
+
+        Raises:
+            I/O related exceptions when file exists but isn't accessable for any
+            reason.
+
+        Returns:
+            `None` when the file doesn't exist, otherwise an instance of the
+            `SerializableRingBuffer` class, loaded from disk.
+        """
+        if not exists(path):
+            return None
+
+        with open(path, mode="rb") as fileobj:
+            instance: SerializableRingBuffer = pickle.load(fileobj)
+            instance._path = path  # pylint: disable=protected-access
+            # Set latest file format version for next time it dumps.
+            # pylint: disable=protected-access
+            instance._file_format_version = FILE_FORMAT_VERSION
+
+            return instance