Make interface for serializing a ringbuffer more flexible. (#326)

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

Author: mathias-baumann-frequenz

RELEASE_NOTES.md

@@ -17,6 +17,7 @@
 * 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.
 
 ## New Features
 

…ks/timeseries/serializable_ringbuffer.py …s/timeseries/ringbuffer_serialization.pybenchmarks/timeseries/serializable_ringbuffer.py renamed to benchmarks/timeseries/ringbuffer_serialization.py benchmarks/timeseries/serializable_ringbuffer.py renamed to benchmarks/timeseries/ringbuffer_serialization.py

@@ -1,7 +1,7 @@
 # License: MIT
 # Copyright © 2023 Frequenz Energy-as-a-Service GmbH
 
-"""Benchmarks the `SerializableRingbuffer` class."""
+"""Benchmarks the serialization of the `OrderedRingBuffer` class."""
 
 from __future__ import annotations
 
@@ -13,8 +13,9 @@
 
 import numpy as np
 
+import frequenz.sdk.timeseries._ringbuffer_serialization as io
 from frequenz.sdk.timeseries import Sample
-from frequenz.sdk.timeseries._serializable_ringbuffer import SerializableRingBuffer
+from frequenz.sdk.timeseries._ringbuffer import OrderedRingBuffer
 
 FILE_NAME = "ringbuffer.pkl"
 FIVE_MINUTES = timedelta(minutes=5)
@@ -37,7 +38,7 @@ def delete_files_with_prefix(prefix: str) -> None:
 
 
 def benchmark_serialization(
-    ringbuffer: SerializableRingBuffer[Any], iterations: int
+    ringbuffer: OrderedRingBuffer[Any], iterations: int
 ) -> float:
     """Benchmark the given buffer `iteration` times.
 
@@ -48,8 +49,8 @@ def benchmark_serialization(
     total = 0.0
     for _ in range(iterations):
         start = time.time()
-        ringbuffer.dump()
-        SerializableRingBuffer.load(FILE_NAME)
+        io.dump(ringbuffer, FILE_NAME)
+        io.load(FILE_NAME)
         end = time.time()
         total += end - start
         delete_files_with_prefix(FILE_NAME)
@@ -59,8 +60,8 @@ def benchmark_serialization(
 
 def main() -> None:
     """Run Benchmark."""
-    ringbuffer = SerializableRingBuffer(
-        np.arange(0, SIZE, dtype=np.float64), timedelta(minutes=5), FILE_NAME
+    ringbuffer = OrderedRingBuffer(
+        np.arange(0, SIZE, dtype=np.float64), timedelta(minutes=5)
     )
 
     print("size:", SIZE)

src/frequenz/sdk/timeseries/_ringbuffer_serialization.py

@@ -0,0 +1,64 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""Ringbuffer dumping & loading functions."""
+
+# For use of the class type hint inside the class itself.
+from __future__ import annotations
+
+import pickle
+from os.path import exists
+
+from ._ringbuffer import FloatArray, OrderedRingBuffer
+
+# Version of the latest file dumping/loading format
+FILE_FORMAT_VERSION: int = 1
+
+
+def load(path: str) -> OrderedRingBuffer[FloatArray] | None:
+    """Load a ringbuffer from disk.
+
+    Args:
+        path: Path to the file where the data is stored.
+
+    Raises:
+        RuntimeError: when the file format version is unknown.
+
+    Returns:
+        `None` when the file doesn't exist, otherwise an instance of the
+        `OrderedRingBuffer` class, loaded from disk.
+    """
+    if not exists(path):
+        return None
+
+    with open(path, mode="rb") as fileobj:
+        instance: OrderedRingBuffer[FloatArray]
+        file_format_version: int
+
+        file_format_version, instance = pickle.load(fileobj)
+
+    if file_format_version != FILE_FORMAT_VERSION:
+        raise RuntimeError(
+            f"Unknown file format version: {file_format_version}. Can load: {FILE_FORMAT_VERSION}"
+        )
+
+    return instance
+
+
+def dump(
+    ringbuffer: OrderedRingBuffer[FloatArray],
+    path: str,
+    file_format_version: int = FILE_FORMAT_VERSION,
+) -> None:
+    """Dump a ringbuffer to disk.
+
+    Args:
+        ringbuffer: Instance of the ringbuffer to dump.
+        path: Path to where the data should be saved to.
+        file_format_version: Version of the file format, optional.
+
+    Raises:
+        I/O related exceptions when the file cannot be written.
+    """
+    with open(path, mode="wb+") as fileobj:
+        pickle.dump((file_format_version, ringbuffer), fileobj)

src/frequenz/sdk/timeseries/_serializable_ringbuffer.py

@@ -12,15 +12,16 @@
 import numpy as np
 import pytest
 
+import frequenz.sdk.timeseries._ringbuffer_serialization as io
 from frequenz.sdk.timeseries import Sample
-from frequenz.sdk.timeseries._serializable_ringbuffer import SerializableRingBuffer
+from frequenz.sdk.timeseries._ringbuffer import OrderedRingBuffer
 
 FIVE_MINUTES = timedelta(minutes=5)
 _29_DAYS = 60 * 24 * 29
 ONE_MINUTE = timedelta(minutes=1)
 
 
-def load_dump_test(dumped: SerializableRingBuffer[Any]) -> None:
+def load_dump_test(dumped: OrderedRingBuffer[Any], path: str) -> None:
     """Test ordered ring buffer."""
     size = dumped.maxlen
 
@@ -42,11 +43,11 @@ def load_dump_test(dumped: SerializableRingBuffer[Any]) -> None:
             )
         )
 
-    dumped.dump()
+    io.dump(dumped, path)
 
     # Load old data
     # pylint: disable=protected-access
-    loaded = SerializableRingBuffer.load(dumped._path)
+    loaded = io.load(path)
     assert loaded is not None
 
     np.testing.assert_equal(dumped[:], loaded[:])
@@ -72,21 +73,21 @@ def test_load_dump_short(tmp_path_factory: pytest.TempPathFactory) -> None:
     tmpdir = tmp_path_factory.mktemp("load_dump")
 
     load_dump_test(
-        SerializableRingBuffer(
+        OrderedRingBuffer(
             [0.0] * int(24 * FIVE_MINUTES.total_seconds()),
             FIVE_MINUTES,
-            f"{tmpdir}/test_list.bin",
             datetime(2, 2, 2, tzinfo=timezone.utc),
-        )
+        ),
+        f"{tmpdir}/test_list.bin",
     )
 
     load_dump_test(
-        SerializableRingBuffer(
+        OrderedRingBuffer(
             np.empty(shape=(24 * int(FIVE_MINUTES.total_seconds()),), dtype=np.float64),
             FIVE_MINUTES,
-            f"{tmpdir}/test_array.bin",
             datetime(2, 2, 2, tzinfo=timezone.utc),
-        )
+        ),
+        f"{tmpdir}/test_array.bin",
     )
 
 
@@ -95,19 +96,19 @@ def test_load_dump(tmp_path_factory: pytest.TempPathFactory) -> None:
     tmpdir = tmp_path_factory.mktemp("load_dump")
 
     load_dump_test(
-        SerializableRingBuffer(
+        OrderedRingBuffer(
             [0.0] * _29_DAYS,
             ONE_MINUTE,
-            f"{tmpdir}/test_list_29.bin",
             datetime(2, 2, 2, tzinfo=timezone.utc),
-        )
+        ),
+        f"{tmpdir}/test_list_29.bin",
     )
 
     load_dump_test(
-        SerializableRingBuffer(
+        OrderedRingBuffer(
             np.empty(shape=(_29_DAYS,), dtype=np.float64),
             ONE_MINUTE,
-            f"{tmpdir}/test_array_29.bin",
             datetime(2, 2, 2, tzinfo=timezone.utc),
-        )
+        ),
+        f"{tmpdir}/test_array_29.bin",
     )