Add separate method for extracting wrapped buffers

This just moves the logic into a separate function to allow better testing
of the logic. Bug found in the logic will be fixed in later commit.

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

Author: cwasicki

src/frequenz/sdk/timeseries/_ringbuffer/buffer.py

@@ -248,22 +248,43 @@ def window(
         start_index = self.datetime_to_index(start)
         end_index = self.datetime_to_index(end)
 
+        return self._wrapped_buffer_window(
+            self._buffer, start_index, end_index, force_copy
+        )
+
+    @staticmethod
+    def _wrapped_buffer_window(
+        buffer: FloatArray, start_pos: int, end_pos: int, force_copy: bool = True
+    ) -> FloatArray:
+        """Get a wrapped window from the given buffer.
+
+        If start_pos == end_pos, the full wrapped buffer is returned starting at start_pos.
+
+        Copies can only be avoided for numpy arrays and when the window is not wrapped.
+        Lists of floats are always copies.
+
+        Args:
+            buffer: The buffer to get the window from.
+            start_pos: The start position of the window in the buffer.
+            end_pos: The end position of the window in the buffer (exclusive).
+            force_copy: If True, will always create a copy of the data.
+
+        Returns:
+            The requested window.
+        """
         # Requested window wraps around the ends
-        if start_index >= end_index:
-            if end_index > 0:
-                if isinstance(self._buffer, list):
-                    return self._buffer[start_index:] + self._buffer[0:end_index]
-                if isinstance(self._buffer, np.ndarray):
-                    return np.concatenate(
-                        (self._buffer[start_index:], self._buffer[0:end_index])
-                    )
-                assert False, f"Unknown _buffer type: {type(self._buffer)}"
-            return self._buffer[start_index:]
+        if start_pos >= end_pos:
+            if end_pos > 0:
+                if isinstance(buffer, list):
+                    return buffer[start_pos:] + buffer[0:end_pos]
+                if isinstance(buffer, np.ndarray):
+                    return np.concatenate((buffer[start_pos:], buffer[0:end_pos]))
+                assert False, f"Unknown buffer type: {type(buffer)}"
+            return buffer[start_pos:]
 
         if force_copy:
-            return deepcopy(self[start_index:end_index])
-
-        return self[start_index:end_index]
+            return deepcopy(buffer[start_pos:end_pos])
+        return buffer[start_pos:end_pos]
 
     def is_missing(self, timestamp: datetime) -> bool:
         """Check if the given timestamp falls within a gap.

tests/timeseries/test_ringbuffer.py

@@ -452,3 +452,64 @@ def test_window() -> None:
     assert [0, 1, 2] == list(win)
     win = buffer.window(dt(0), dt(3), force_copy=False)
     assert [0, 1, 2] == list(win)
+
+
+def test_wrapped_buffer_window() -> None:
+    """Test the wrapped buffer window function."""
+    wbw = OrderedRingBuffer._wrapped_buffer_window  # pylint: disable=protected-access
+
+    #
+    # Tests for list buffer
+    #
+    buffer = [0.0, 1.0, 2.0, 3.0, 4.0]
+    # start = end
+    assert [0, 1, 2, 3, 4] == wbw(buffer, 0, 0, force_copy=False)
+    assert [4, 0, 1, 2, 3] == wbw(buffer, 4, 4, force_copy=False)
+    # start < end
+    assert [0] == wbw(buffer, 0, 1, force_copy=False)
+    assert [0, 1, 2, 3, 4] == wbw(buffer, 0, 5, force_copy=False)
+    # start > end, end = 0
+    assert [4] == wbw(buffer, 4, 0, force_copy=False)
+    # start > end, end > 0
+    assert [4, 0, 1] == wbw(buffer, 4, 2, force_copy=False)
+
+    # Lists are always shallow copies
+    res_copy = wbw(buffer, 0, 5, force_copy=False)
+    assert [0, 1, 2, 3, 4] == res_copy
+    buffer[0] = 9
+    assert [0, 1, 2, 3, 4] == res_copy
+
+    #
+    # Tests for array buffer
+    #
+    buffer = np.array([0, 1, 2, 3, 4])  # type: ignore
+    # start = end
+    assert [0, 1, 2, 3, 4] == list(wbw(buffer, 0, 0, force_copy=False))
+    assert [4, 0, 1, 2, 3] == list(wbw(buffer, 4, 4, force_copy=False))
+    # start < end
+    assert [0] == list(wbw(buffer, 0, 1, force_copy=False))
+    assert [0, 1, 2, 3, 4] == list(wbw(buffer, 0, 5, force_copy=False))
+    # start > end, end = 0
+    assert [4] == list(wbw(buffer, 4, 0, force_copy=False))
+    # start > end, end > 0
+    assert [4, 0, 1] == list(wbw(buffer, 4, 2, force_copy=False))
+
+    # Get a view and a copy before modifying the buffer
+    res1_view = wbw(buffer, 3, 5, force_copy=False)
+    res1_copy = wbw(buffer, 3, 5, force_copy=True)
+    res2_view = wbw(buffer, 3, 0, force_copy=False)
+    res2_copy = wbw(buffer, 3, 0, force_copy=True)
+    res3_copy = wbw(buffer, 4, 1, force_copy=False)
+    assert [3, 4] == list(res1_view)
+    assert [3, 4] == list(res1_copy)
+    assert [3, 4] == list(res2_view)
+    assert [3, 4] == list(res2_copy)
+    assert [4, 0] == list(res3_copy)
+
+    # Modify the buffer and check that only the view is updated
+    buffer[4] = 9
+    assert [3, 9] == list(res1_view)
+    assert [3, 4] == list(res1_copy)
+    assert [3, 9] == list(res2_view)
+    # assert [3, 4] == list(res2_copy) #Fails because of a bug
+    assert [4, 0] == list(res3_copy)