Add oldest and newest timestamps to moving window

Methods to retrieve the timestamps of the oldest and newest sample in
the moving window.

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

Author: cwasicki

RELEASE_NOTES.md

@@ -100,7 +100,9 @@ This release replaces the `@actor` decorator with a new `Actor` class.
 
 - `Actor`: This new class inherits from `BackgroundService` and it replaces the `@actor` decorator.
 
-- `MovingWindow`: Provide access to `capacity` (maximum number of elements).
+- `MovingWindow`: Provide access to `capacity` (maximum number of elements),
+  `oldest_timestamp` and `newest_timestamp` of the window. The names in the
+  `OrderedRingBuffer` have been changed to match these names.
 
 ## Bug Fixes
 

src/frequenz/sdk/timeseries/_moving_window.py

@@ -222,6 +222,32 @@ def capacity(self) -> int:
         """
         return self._buffer.maxlen
 
+    @property
+    def oldest_timestamp(self) -> datetime | None:
+        """
+        Return the oldest timestamp in the MovingWindow.
+
+        Returns:
+            The oldest timestamp in the MovingWindow.
+            If the MovingWindow is empty, None is returned.
+        """
+        if len(self._buffer) == 0:
+            return None
+        return self._buffer.oldest_timestamp
+
+    @property
+    def newest_timestamp(self) -> datetime | None:
+        """
+        Return the newest timestamp in the MovingWindow.
+
+        Returns:
+            The newest timestamp in the MovingWindow.
+            If the MovingWindow is empty, None is returned.
+        """
+        if len(self._buffer) == 0:
+            return None
+        return self._buffer.newest_timestamp
+
     async def _run_impl(self) -> None:
         """Awaits samples from the receiver and updates the underlying ring buffer.
 

tests/timeseries/test_moving_window.py

@@ -137,6 +137,25 @@ async def test_window_size() -> None:
         assert len(window) == 5, "Window should be full"
 
 
+async def test_timestamps() -> None:
+    """Test the timestamp methods of the window."""
+    window, sender = init_moving_window(timedelta(seconds=5))
+    async with window:
+        assert (
+            window.oldest_timestamp is None
+        ), "For an empty window, oldest timestamp should be None"
+        assert (
+            window.newest_timestamp is None
+        ), "For an empty window, newest timestamp should be None"
+        await push_logical_meter_data(sender, range(0, 20))
+        assert window.oldest_timestamp == UNIX_EPOCH + timedelta(
+            seconds=15
+        ), "Wrong oldest timestamp"
+        assert window.newest_timestamp == UNIX_EPOCH + timedelta(
+            seconds=19
+        ), "Wrong newest timestamp"
+
+
 # pylint: disable=redefined-outer-name
 async def test_resampling_window(fake_time: time_machine.Coordinates) -> None:
     """Test resampling in MovingWindow."""