Fix copy behavior when extracting wrapped buffers

This reworks the method for extracting wrapped buffer to fix a bug
in the existing copy behavior for a specific case.

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

Author: cwasicki

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

@@ -254,7 +254,10 @@ def window(
 
     @staticmethod
     def _wrapped_buffer_window(
-        buffer: FloatArray, start_pos: int, end_pos: int, force_copy: bool = True
+        buffer: FloatArray,
+        start_pos: int,
+        end_pos: int,
+        force_copy: bool = True,
     ) -> FloatArray:
         """Get a wrapped window from the given buffer.
 
@@ -274,17 +277,20 @@ def _wrapped_buffer_window(
         """
         # Requested window wraps around the ends
         if start_pos >= end_pos:
+            if isinstance(buffer, list):
+                return buffer[start_pos:] + buffer[0:end_pos]
+            assert isinstance(
+                buffer, np.ndarray
+            ), f"Unsupported buffer type: {type(buffer)}"
             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:]
+                return np.concatenate((buffer[start_pos:], buffer[0:end_pos]))
+            arr = buffer[start_pos:]
+        else:
+            arr = buffer[start_pos:end_pos]
 
         if force_copy:
-            return deepcopy(buffer[start_pos:end_pos])
-        return buffer[start_pos:end_pos]
+            return deepcopy(arr)
+        return arr
 
     def is_missing(self, timestamp: datetime) -> bool:
         """Check if the given timestamp falls within a gap.

tests/timeseries/test_ringbuffer.py

@@ -511,5 +511,5 @@ def test_wrapped_buffer_window() -> None:
     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 [3, 4] == list(res2_copy)
     assert [4, 0] == list(res3_copy)