Add newest timestamp to moving window

Method to retrieve the timestamp of the newest sample in
the moving window.

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

Author: cwasicki

RELEASE_NOTES.md

@@ -19,7 +19,9 @@
 - A new class `Fuse` has been added to represent fuses. This class has a member variable `max_current` which represents the maximum current that can course through the fuse. If the current flowing through a fuse is greater than this limit, then the fuse will break the circuit.
 
 
-- `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,19 @@ def capacity(self) -> int:
         """
         return self._buffer.maxlen
 
+    @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.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

@@ -135,6 +135,19 @@ async def test_window_size() -> None:
         assert len(window) == 5, "Window should be full"
 
 
+async def test_timestamps() -> None:
+    """Test the newest_timestamp method of the window."""
+    window, sender = init_moving_window(timedelta(seconds=5))
+    async with window:
+        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.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."""