Add serializing ringbuffer

Signed-off-by: Mathias L. Baumann <[email protected]>

Author: Marenz

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/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/_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

tests/timeseries/test_serializable_ringbuffer.py

@@ -0,0 +1,108 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for the `SerializableRingBuffer` class."""
+
+from __future__ import annotations
+
+import random
+from datetime import datetime, timedelta
+from typing import Any
+
+import numpy as np
+import pytest
+
+from frequenz.sdk.timeseries import Sample
+from frequenz.sdk.timeseries._serializable_ringbuffer import SerializableRingBuffer
+
+FIVE_MINUTES = timedelta(minutes=5)
+_29_DAYS = 60 * 24 * 29
+ONE_MINUTE = timedelta(minutes=1)
+
+
+def load_dump_test(dumped: SerializableRingBuffer[Any]) -> None:
+    """Test ordered ring buffer."""
+    size = dumped.maxlen
+
+    random.seed(0)
+
+    # Fill with data so we have something to compare
+    # Avoiding .update() because it takes very long for 40k entries
+    for i in range(size):
+        dumped[i] = i
+
+    # But use update a bit so the timestamp and gaps are initialized
+    for i in range(0, size, 100):
+        dumped.update(
+            Sample(datetime.fromtimestamp(200 + i * FIVE_MINUTES.total_seconds()), i)
+        )
+
+    dumped.dump()
+
+    # Load old data
+    # pylint: disable=protected-access
+    loaded = SerializableRingBuffer.load(dumped._path)
+    assert loaded is not None
+
+    np.testing.assert_equal(dumped[:], loaded[:])
+
+    # pylint: disable=protected-access
+    assert dumped._datetime_oldest == loaded._datetime_oldest
+    # pylint: disable=protected-access
+    assert dumped._datetime_newest == loaded._datetime_newest
+    # pylint: disable=protected-access
+    assert len(dumped._gaps) == len(loaded._gaps)
+    # pylint: disable=protected-access
+    assert dumped._gaps == loaded._gaps
+    # pylint: disable=protected-access
+    assert dumped._time_range == loaded._time_range
+    # pylint: disable=protected-access
+    assert dumped._sampling_period == loaded._sampling_period
+    # pylint: disable=protected-access
+    assert dumped._time_index_alignment == loaded._time_index_alignment
+
+
+def test_load_dump_short(tmp_path_factory: pytest.TempPathFactory) -> None:
+    """Short test to perform loading & dumping."""
+    tmpdir = tmp_path_factory.mktemp("load_dump")
+
+    load_dump_test(
+        SerializableRingBuffer(
+            [0.0] * int(24 * FIVE_MINUTES.total_seconds()),
+            FIVE_MINUTES,
+            f"{tmpdir}/test_list.bin",
+            datetime(2, 2, 2),
+        )
+    )
+
+    load_dump_test(
+        SerializableRingBuffer(
+            np.empty(shape=(24 * int(FIVE_MINUTES.total_seconds()),), dtype=np.float64),
+            FIVE_MINUTES,
+            f"{tmpdir}/test_array.bin",
+            datetime(2, 2, 2),
+        )
+    )
+
+
+def test_load_dump(tmp_path_factory: pytest.TempPathFactory) -> None:
+    """Test to load/dump 29 days of 1-minute samples."""
+    tmpdir = tmp_path_factory.mktemp("load_dump")
+
+    load_dump_test(
+        SerializableRingBuffer(
+            [0.0] * _29_DAYS,
+            ONE_MINUTE,
+            f"{tmpdir}/test_list_29.bin",
+            datetime(2, 2, 2),
+        )
+    )
+
+    load_dump_test(
+        SerializableRingBuffer(
+            np.empty(shape=(_29_DAYS,), dtype=np.float64),
+            ONE_MINUTE,
+            f"{tmpdir}/test_array_29.bin",
+            datetime(2, 2, 2),
+        )
+    )