Add get_timestamp 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,33 @@ def to_internal_index(
             )
         )
 
+    def get_timestamp(self, index: int) -> datetime | None:
+        """Convert the given index to the underlying timestamp.
+
+        Index 0 corresponds to the oldest timestamp in the buffer.
+        If negative indices are used, the newest timestamp is used as reference.
+
+        !!!warning
+
+            The resulting timestamp 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_get_timestamp() -> None:
+    """Test the get_timestamp 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.get_timestamp(-1)
+    assert dt(0) == buffer.get_timestamp(-5)
+    assert dt(-1) == buffer.get_timestamp(-6)
+    assert dt(0) == buffer.get_timestamp(0)
+    assert dt(5) == buffer.get_timestamp(5)
+    assert dt(6) == buffer.get_timestamp(6)
+
+    for i in range(10, 15):
+        buffer.update(Sample(dt(i), Quantity(i)))
+    assert dt(14) == buffer.get_timestamp(-1)
+    assert dt(10) == buffer.get_timestamp(-5)
+    assert dt(9) == buffer.get_timestamp(-6)
+    assert dt(10) == buffer.get_timestamp(0)
+    assert dt(15) == buffer.get_timestamp(5)
+    assert dt(16) == buffer.get_timestamp(6)