Use Sample type for ringbuffer

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

Author: Marenz

benchmarks/timeseries/benchmark_ringbuffer.py

@@ -6,10 +6,11 @@
 import random
 import timeit
 from datetime import datetime, timedelta
-from typing import Any, TypeVar
+from typing import Any, Dict, TypeVar
 
 import numpy as np
 
+from frequenz.sdk.timeseries import Sample
 from frequenz.sdk.timeseries._ringbuffer import OrderedRingBuffer
 
 MINUTES_IN_A_DAY = 24 * 60
@@ -19,9 +20,7 @@
 T = TypeVar("T")
 
 
-def fill_buffer(
-    days: int, buffer: OrderedRingBuffer[T, Any], element_type: type
-) -> None:
+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)
@@ -31,14 +30,12 @@ def fill_buffer(
         # Push in random order
         for i in random.sample(range(MINUTES_IN_A_DAY), MINUTES_IN_A_DAY):
             buffer.update(
-                basetime + timedelta(days=day, minutes=i, seconds=i % 3),
-                element_type(i),
+                Sample(basetime + timedelta(days=day, minutes=i, seconds=i % 3))
             )
 
 
-def test_days(days: int, buffer: OrderedRingBuffer[T, 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)
 
     for day in range(days):
@@ -48,13 +45,12 @@ def test_days(days: int, buffer: OrderedRingBuffer[T, Any]) -> None:
         )
 
 
-def test_slices(days: int, buffer: OrderedRingBuffer[T, Any], median: bool) -> None:
+def test_slices(days: int, buffer: OrderedRingBuffer[Any], median: bool) -> None:
     """Benchmark slicing.
 
     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)
 
     total = 0.0
@@ -71,18 +67,17 @@ def test_slices(days: int, buffer: OrderedRingBuffer[T, Any], median: bool) -> N
                 total += float(np.average(minutes))
 
 
-def test_29_days_list(num_runs: int) -> dict:
+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))
 
-    fill_time = timeit.Timer(lambda: fill_buffer(days, buffer, int)).timeit(number=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)
     return {"fill": fill_time, "test": test_time}
 
 
-def test_29_days_array(num_runs: int) -> dict:
+def test_29_days_array(num_runs: int) -> Dict[str, float]:
     """Run the 29 day test on the array backend."""
     days = 29
     buffer = OrderedRingBuffer(
@@ -92,17 +87,17 @@ def test_29_days_array(num_runs: int) -> dict:
         timedelta(minutes=1),
     )
 
-    fill_time = timeit.Timer(lambda: fill_buffer(days, buffer, int)).timeit(number=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)
     return {"fill": fill_time, "test": test_time}
 
 
-def test_29_days_slicing_list(num_runs: int) -> dict:
+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))
 
-    fill_time = timeit.Timer(lambda: fill_buffer(days, buffer, int)).timeit(number=1)
+    fill_time = timeit.Timer(lambda: fill_buffer(days, buffer)).timeit(number=1)
     median_test_time = timeit.Timer(
         lambda: test_slices(days, buffer, median=True)
     ).timeit(number=num_runs)
@@ -113,7 +108,7 @@ def test_29_days_slicing_list(num_runs: int) -> dict:
     return {"fill": fill_time, "median": median_test_time, "avg": avg_test_time}
 
 
-def test_29_days_slicing_array(num_runs: int) -> dict:
+def test_29_days_slicing_array(num_runs: int) -> Dict[str, float]:
     """Run slicing tests on array backend."""
     days = 29
     buffer = OrderedRingBuffer(
@@ -123,7 +118,7 @@ def test_29_days_slicing_array(num_runs: int) -> dict:
         timedelta(minutes=1),
     )
 
-    fill_time = timeit.Timer(lambda: fill_buffer(days, buffer, int)).timeit(number=1)
+    fill_time = timeit.Timer(lambda: fill_buffer(days, buffer)).timeit(number=1)
     median_test_time = timeit.Timer(
         lambda: test_slices(days, buffer, median=True)
     ).timeit(number=num_runs)
@@ -139,31 +134,31 @@ def main() -> None:
 
     Result of previous run:
 
-    Date: Mi 1. Feb 17:15:02 CET 2023
+    Date: Mi 1. Feb 17:19:51 CET 2023
     Result:
 
            =====================
     Array: ..filling
     List:  ..filling
     Time to fill 29 days with data:
-            Array: 7.190492740017362 seconds
-            List:  7.209744154009968 seconds
-            Diff:  -0.019251413992606103
+            Array: 7.214875044999644 seconds
+            List:  7.174421657982748 seconds
+            Diff:  0.04045338701689616
     Day-Slices into 29 days with data:
-            Array: 0.0001254317001439631 seconds
-            List:  0.00017958255193661898 seconds
-            Diff:  -5.4150851792655874e-05
+            Array: 0.0001304591482039541 seconds
+            List:  0.00019963659869972616 seconds
+            Diff:  -6.917745049577205e-05
            =====================
     Array: ..filling
     List:  ..filling
     Avg of windows of 29 days and running average & mean on every day:
-            Array: 0.0007975498505402356 seconds
-            List:  0.0042349924508016555 seconds
-            Diff:  -0.0034374426002614198
+            Array: 0.0007829780981410295 seconds
+            List:  0.0042931242496706545 seconds
+            Diff:  -0.0035101461515296252
     Median of windows of 29 days and running average & mean on every day:
-            Array: 0.0021774103515781462 seconds
-            List:  0.004992740901070647 seconds
-            Diff:  -0.0028153305494925005
+            Array: 0.0021195551002165304 seconds
+            List:  0.00501448459981475 seconds
+            Diff:  -0.00289492949959822
     """
     num_runs = 20
 
@@ -208,7 +203,8 @@ def main() -> None:
         "Median of windows of 29 days and running average & mean on every day:\n\t"
         + f"Array: {slicing_array_times['median']/num_runs} seconds\n\t"
         + f"List:  {slicing_list_times['median']/num_runs} seconds\n\t"
-        + f"Diff:  {slicing_array_times['median']/num_runs -  slicing_list_times['median']/num_runs}"
+        + "Diff:  "
+        + f"{slicing_array_times['median']/num_runs -  slicing_list_times['median']/num_runs}"
     )
 
 

src/frequenz/sdk/timeseries/_ringbuffer.py

@@ -11,8 +11,9 @@
 from typing import Any, Generic, List, Sequence, TypeVar
 
 import numpy as np
+from numpy.lib import math
 
-T = TypeVar("T")
+from frequenz.sdk.timeseries import Sample
 
 Container = TypeVar("Container", list, np.ndarray)
 
@@ -41,7 +42,7 @@ def contains(self, timestamp: datetime):
         return False
 
 
-class OrderedRingBuffer(Generic[T, Container]):
+class OrderedRingBuffer(Generic[Container]):
     """Time aware ringbuffer that keeps its entries sorted by time."""
 
     def __init__(
@@ -105,28 +106,25 @@ def maxlen(self) -> int:
         """
         return len(self._buffer)
 
-    def update(self, timestamp: datetime, value: T, missing: bool = False) -> None:
+    def update(self, sample: Sample) -> None:
         """Update the buffer with a new value for the given timestamp.
 
+        Missing values are written as NaN. Be advised that when
+        `update()` is called with samples newer than the current time
+        + `sampling_period` (as is the case for loading historical data
+        after an app restart), a gap of missing data exists. This gap
+        does not contain NaN values but simply the old invalid values.
+        The list of gaps returned by `gaps()` will reflect this and
+        should be used as the only source of truth for unwritten data.
+
         Args:
-            timestamp: Timestamp of the new value.
-            value: value to add.
-            missing: if true, the given timestamp will be recorded as missing.
-                The value will still be written.
-
-                Note: When relying on your own values to mark missing data (e.g.
-                `math.nan`), be advised that when `update()` is called with
-                timestamps newer than the current time + `sampling_period`
-                (as is the case for loading historical data after an app
-                restart), a gap of missing data exists.
-                This gap does not contain NaN values but simply the old invalid values.
-                The list of gaps returned by `gaps()` will reflect this.
+            sample: Sample to add to the ringbuffer
 
         Raises:
             IndexError: When the timestamp to be added is too old.
         """
         # adjust timestamp to be exactly on the sample period time point
-        timestamp = self._normalize_timestamp(timestamp)
+        timestamp = self._normalize_timestamp(sample.timestamp)
 
         # Don't add outdated entries
         if timestamp < self._datetime_oldest and self._datetime_oldest != datetime.max:
@@ -140,9 +138,10 @@ def update(self, timestamp: datetime, value: T, missing: bool = False) -> None:
         self._datetime_oldest = self._datetime_newest - self._time_range
 
         # Update data
+        value: float = math.nan if sample.value is None else sample.value
         self._buffer[self.datetime_to_index(timestamp)] = value
 
-        self._update_gaps(timestamp, prev_newest, missing)
+        self._update_gaps(timestamp, prev_newest, sample.value is None)
 
     def datetime_to_index(
         self, timestamp: datetime, allow_outside_range: bool = False
@@ -387,7 +386,7 @@ def _wrap(self, index: int) -> int:
         """
         return index % self.maxlen
 
-    def __setitem__(self, index_or_slice: int | slice, value: T) -> None:
+    def __setitem__(self, index_or_slice: int | slice, value: float) -> None:
         """Set item or slice at requested position.
 
         No wrapping of the index will be done.
@@ -399,7 +398,7 @@ def __setitem__(self, index_or_slice: int | slice, value: T) -> None:
         """
         self._buffer.__setitem__(index_or_slice, value)
 
-    def __getitem__(self, index_or_slice: int | slice) -> T | Sequence[T]:
+    def __getitem__(self, index_or_slice: int | slice) -> float | Sequence[float]:
         """Get item or slice at requested position.
 
         No wrapping of the index will be done.