Fix single element access for moving window

RELEASE_NOTES.md

@@ -28,11 +28,12 @@
 - In `OrderedRingBuffer` and `MovingWindow`:
   - Support for integer indices is added.
   - Add `count_covered` method to count the number of elements covered by the used time range.
-
+- Add `at` method to `MovingWindow` to access a single element and use it in `__getitem__` magic to fully support single element access.
 
 
 
 ## Bug Fixes
 
 - Fix rendering of diagrams in the documentation.
 - The `__getitem__` magic of the `MovingWindow` is fixed to support the same functionality that the `window` method provides.
+- Fixes incorrect implementation of single element access in `__getitem__` magic of `MovingWindow`.

src/frequenz/sdk/timeseries/_moving_window.py

@@ -241,6 +241,48 @@ def capacity(self) -> int:
         """
         return self._buffer.maxlen
 
+    # pylint before 3.0 only accepts names with 3 or more chars
+    def at(self, key: int | datetime) -> float:  # pylint: disable=invalid-name
+        """
+        Return the sample at the given index or timestamp.
+
+        In contrast to the [`window`][frequenz.sdk.timeseries.MovingWindow.window] method,
+        which expects a slice as argument, this method expects a single index as argument
+        and returns a single value.
+
+        Args:
+            key: The index or timestamp of the sample to return.
+
+        Returns:
+            The sample at the given index or timestamp.
+
+        Raises:
+            IndexError: If the buffer is empty or the index is out of bounds.
+        """
+        if self._buffer.count_valid() == 0:
+            raise IndexError("The buffer is empty.")
+
+        if isinstance(key, datetime):
+            assert self._buffer.oldest_timestamp is not None
+            assert self._buffer.newest_timestamp is not None
+            if (
+                key < self._buffer.oldest_timestamp
+                or key > self._buffer.newest_timestamp
+            ):
+                raise IndexError(
+                    f"Timestamp {key} is out of range [{self._buffer.oldest_timestamp}, "
+                    f"{self._buffer.newest_timestamp}]"
+                )
+            return self._buffer[self._buffer.to_internal_index(key)]
+
+        if isinstance(key, int):
+            _logger.debug("Returning value at index %s ", key)
+            timestamp = self._buffer.get_timestamp(key)
+            assert timestamp is not None
+            return self._buffer[self._buffer.to_internal_index(timestamp)]
+
+        raise TypeError("Key has to be either a timestamp or an integer.")
+
     def window(
         self,
         start: datetime | int | None,
@@ -251,6 +293,10 @@ def window(
         """
         Return an array containing the samples in the given time interval.
 
+        In contrast to the [`at`][frequenz.sdk.timeseries.MovingWindow.at] method,
+        which expects a single index as argument, this method expects a slice as argument
+        and returns an array.
+
         Args:
             start: The start of the time interval. If `None`, the start of the
                 window is used.
@@ -372,15 +418,11 @@ def __getitem__(self, key: SupportsIndex | datetime | slice) -> float | ArrayLik
                 raise ValueError("Slicing with a step other than 1 is not supported.")
             return self.window(key.start, key.stop)
 
-        if self._buffer.count_valid() == 0:
-            raise IndexError("The buffer is empty.")
-
         if isinstance(key, datetime):
-            _logger.debug("Returning value at time %s ", key)
-            return self._buffer[self._buffer.to_internal_index(key)]
+            return self.at(key)
+
         if isinstance(key, SupportsIndex):
-            _logger.debug("Returning value at index %s ", key)
-            return self._buffer[key]
+            return self.at(key.__index__())
 
         raise TypeError(
             "Key has to be either a timestamp or an integer "

tests/timeseries/test_moving_window.py

@@ -80,18 +80,36 @@ def dt(i: int) -> datetime:  # pylint: disable=invalid-name
 
 async def test_access_window_by_index() -> None:
     """Test indexing a window by integer index."""
-    window, sender = init_moving_window(timedelta(seconds=1))
+    window, sender = init_moving_window(timedelta(seconds=2))
     async with window:
-        await push_logical_meter_data(sender, [1])
-        assert np.array_equal(window[0], 1.0)
+        await push_logical_meter_data(sender, [1, 2, 3])
+        assert np.array_equal(window[0], 2.0)
+        assert np.array_equal(window[1], 3.0)
+        assert np.array_equal(window[-1], 3.0)
+        assert np.array_equal(window[-2], 2.0)
+        with pytest.raises(IndexError):
+            _ = window[3]
+        with pytest.raises(IndexError):
+            _ = window[-3]
 
 
 async def test_access_window_by_timestamp() -> None:
     """Test indexing a window by timestamp."""
-    window, sender = init_moving_window(timedelta(seconds=1))
+    window, sender = init_moving_window(timedelta(seconds=2))
     async with window:
-        await push_logical_meter_data(sender, [1])
-        assert np.array_equal(window[UNIX_EPOCH], 1.0)
+        await push_logical_meter_data(sender, [0, 1, 2])
+        assert np.array_equal(window[dt(1)], 1.0)
+        assert np.array_equal(window.at(dt(1)), 1.0)
+        assert np.array_equal(window[dt(2)], 2.0)
+        assert np.array_equal(window.at(dt(2)), 2.0)
+        with pytest.raises(IndexError):
+            _ = window[dt(0)]
+        with pytest.raises(IndexError):
+            _ = window.at(dt(0))
+        with pytest.raises(IndexError):
+            _ = window[dt(3)]
+        with pytest.raises(IndexError):
+            _ = window.at(dt(3))
 
 
 async def test_access_window_by_int_slice() -> None: