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

@@ -11,7 +11,8 @@
 ## New Features
 
 - DFS for compentent graph
-- `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.
 
 ## Bug Fixes
 

src/frequenz/sdk/timeseries/_moving_window.py

@@ -221,6 +221,32 @@ def capacity(self) -> int:
         """
         return self._buffer.maxlen
 
+    @property
+    def oldest_timestamp(self) -> datetime:
+        """
+        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.time_bound_oldest
+
+    @property
+    def newest_timestamp(self) -> datetime:
+        """
+        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.time_bound_newest
+
     async def _run_impl(self) -> None:
         """Awaits samples from the receiver and updates the underlying ring buffer.
 

tests/timeseries/test_moving_window.py

@@ -131,6 +131,24 @@ 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))
+    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."""