Expose oldest and newest timestamp in moving window

Author: cwasicki

RELEASE_NOTES.md

@@ -20,10 +20,10 @@
 
 - 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` and `OrderedRingBuffer`:
   - NaN values are treated as missing when gaps are determined in the `OrderedRingBuffer`.
   - Provide access to `capacity` (maximum number of elements) in `MovingWindow`.
+  - Methods to retrieve oldest and newest timestamp of valid samples are added to both.
 
 
 ## Bug Fixes

src/frequenz/sdk/timeseries/_moving_window.py

@@ -209,6 +209,26 @@ def sampling_period(self) -> timedelta:
         """
         return self._sampling_period
 
+    @property
+    def oldest_timestamp(self) -> datetime | None:
+        """
+        Return the oldest timestamp of the MovingWindow.
+
+        Returns:
+            The oldest timestamp of the MovingWindow or None if the buffer is empty.
+        """
+        return self._buffer.oldest_timestamp
+
+    @property
+    def newest_timestamp(self) -> datetime | None:
+        """
+        Return the newest timestamp of the MovingWindow.
+
+        Returns:
+            The newest timestamp of the MovingWindow or None if the buffer is empty.
+        """
+        return self._buffer.newest_timestamp
+
     @property
     def capacity(self) -> int:
         """

tests/timeseries/test_moving_window.py

@@ -36,7 +36,9 @@ def fake_time() -> Iterator[time_machine.Coordinates]:
 
 
 async def push_logical_meter_data(
-    sender: Sender[Sample[Quantity]], test_seq: Sequence[float]
+    sender: Sender[Sample[Quantity]],
+    test_seq: Sequence[float],
+    start_ts: datetime = UNIX_EPOCH,
 ) -> None:
     """Push data in the passed sender to mock `LogicalMeter` behaviour.
 
@@ -45,8 +47,8 @@ async def push_logical_meter_data(
     Args:
         sender: Sender for pushing resampled samples to the `MovingWindow`.
         test_seq: The Sequence that is pushed into the `MovingWindow`.
+        start_ts: The start timestamp of the `MovingWindow`.
     """
-    start_ts: datetime = UNIX_EPOCH
     for i, j in zip(test_seq, range(0, len(test_seq))):
         timestamp = start_ts + timedelta(seconds=j)
         await sender.send(Sample(timestamp, Quantity(float(i))))
@@ -165,3 +167,14 @@ async def test_resampling_window(fake_time: time_machine.Coordinates) -> None:
         assert len(window) == window_size / output_sampling
         for value in window:  # type: ignore
             assert 4.9 < value < 5.1
+
+
+async def test_timestamps() -> None:
+    """Test indexing a window by timestamp."""
+    window, sender = init_moving_window(timedelta(seconds=5))
+    async with window:
+        await push_logical_meter_data(
+            sender, [1, 2], start_ts=UNIX_EPOCH + timedelta(seconds=1)
+        )
+        assert window.oldest_timestamp == UNIX_EPOCH + timedelta(seconds=1)
+        assert window.newest_timestamp == UNIX_EPOCH + timedelta(seconds=2)