Add index_to_datetime method

This method maps an index passed by the user to its corresponding
timestamp of the ring buffer.

Signed-off-by: cwasicki <[email protected]>

Author: cwasicki

src/frequenz/sdk/timeseries/_ringbuffer/buffer.py

@@ -257,6 +257,29 @@ def to_internal_index(
             )
         )
 
+    def index_to_datetime(self, index: int) -> datetime | None:
+        """Convert the given index to the underlying datetime.
+
+        Index 0 corresponds to the oldest timestamp in the buffer. The
+        resulting datetime can be outside the range of the buffer.
+
+        Args:
+            index: Index to convert.
+
+        Returns:
+            Datetime index where the value for the given index can be found.
+                Or None if the buffer is empty.
+        """
+        if self.oldest_timestamp is None:
+            return None
+        assert self.newest_timestamp is not None
+        ref_ts = (
+            self.oldest_timestamp
+            if index >= 0
+            else self.newest_timestamp + self._sampling_period
+        )
+        return ref_ts + index * self._sampling_period
+
     def window(
         self, start: datetime, end: datetime, *, force_copy: bool = True
     ) -> FloatArray:

tests/timeseries/test_ringbuffer.py

@@ -590,3 +590,28 @@ def test_wrapped_buffer_window() -> None:
     assert [3, 9] == list(res2_view)
     assert [3, 4] == list(res2_copy)
     assert [4, 0] == list(res3_copy)
+
+
+def test_index_to_datetime() -> None:
+    """Test the index_to_datetime function."""
+    buffer = OrderedRingBuffer(
+        np.empty(shape=5, dtype=float),
+        sampling_period=timedelta(seconds=1),
+    )
+    for i in range(5):
+        buffer.update(Sample(dt(i), Quantity(i)))
+    assert dt(4) == buffer.index_to_datetime(-1)
+    assert dt(0) == buffer.index_to_datetime(-5)
+    assert dt(-1) == buffer.index_to_datetime(-6)
+    assert dt(0) == buffer.index_to_datetime(0)
+    assert dt(5) == buffer.index_to_datetime(5)
+    assert dt(6) == buffer.index_to_datetime(6)
+
+    for i in range(10, 15):
+        buffer.update(Sample(dt(i), Quantity(i)))
+    assert dt(14) == buffer.index_to_datetime(-1)
+    assert dt(10) == buffer.index_to_datetime(-5)
+    assert dt(9) == buffer.index_to_datetime(-6)
+    assert dt(10) == buffer.index_to_datetime(0)
+    assert dt(15) == buffer.index_to_datetime(5)
+    assert dt(16) == buffer.index_to_datetime(6)