test: captured frames do not share memory with the SDK frame
The fake frames now hand out a buffer (memoryview) like pyrealsense2's BufData, so np.asanyarray() of it is a view, as on the robot; a control test proves it. Then: the captured rgb/depth share no memory with the fake SDK frame and own their data (without the fix both shared it), and _owned() copies views only. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
55e0a991b4
commit
a8e380ed69
2 files changed
+47
-1
No files matched your search
@@ -18,6 +18,11 @@ class FakeVideoFrame:
|
|||||||
self._data = data
|
self._data = data
|
||||||
|
|
||||||
def get_data(self):
|
def get_data(self):
|
||||||
|
# A buffer over the frame's memory, as pyrealsense2's BufData: np.asanyarray() of it is a view.
|
||||||
|
return memoryview(self._data)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def array(self) -> np.ndarray:
|
||||||
return self._data
|
return self._data
|
||||||
|
|
||||||
def get_height(self) -> int:
|
def get_height(self) -> int:
|
||||||
@@ -54,6 +59,7 @@ class FakePipeline:
|
|||||||
self._active = 0
|
self._active = 0
|
||||||
self._lock = threading.Lock()
|
self._lock = threading.Lock()
|
||||||
self.stopped = False
|
self.stopped = False
|
||||||
|
self.last_frameset = None
|
||||||
|
|
||||||
def wait_for_frames(self, timeout_ms: int = 5000):
|
def wait_for_frames(self, timeout_ms: int = 5000):
|
||||||
with self._lock:
|
with self._lock:
|
||||||
@@ -63,7 +69,8 @@ class FakePipeline:
|
|||||||
index = self.calls
|
index = self.calls
|
||||||
try:
|
try:
|
||||||
time.sleep(self.latency_s)
|
time.sleep(self.latency_s)
|
||||||
return FakeFrameset(index)
|
self.last_frameset = FakeFrameset(index)
|
||||||
|
return self.last_frameset
|
||||||
finally:
|
finally:
|
||||||
with self._lock:
|
with self._lock:
|
||||||
self._active -= 1
|
self._active -= 1
|
||||||
|
|||||||
@@ -0,0 +1,39 @@
|
|||||||
|
"""Frames acquired from the pipeline own their memory: nothing points into the SDK's buffers."""
|
||||||
|
|
||||||
|
import numpy as np
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
pytest.importorskip("pyrealsense2")
|
||||||
|
|
||||||
|
from src.backend.services.camera import RealSenseD435, _owned # noqa: E402
|
||||||
|
from tests.fake_realsense import FakeFrameset, FakePipeline, open_fake_camera # noqa: E402
|
||||||
|
|
||||||
|
|
||||||
|
def test_the_fake_hands_out_views_like_the_sdk():
|
||||||
|
"""Control: without the copy, the driver's arrays WOULD share the SDK frame's memory."""
|
||||||
|
color = FakeFrameset(1).get_color_frame()
|
||||||
|
assert np.shares_memory(np.asanyarray(color.get_data()), color.array)
|
||||||
|
|
||||||
|
|
||||||
|
def test_captured_frame_does_not_share_memory_with_the_sdk_frame():
|
||||||
|
cam = RealSenseD435()
|
||||||
|
pipe = FakePipeline()
|
||||||
|
open_fake_camera(cam, pipe)
|
||||||
|
try:
|
||||||
|
frame = cam.capture()
|
||||||
|
sdk = pipe.last_frameset
|
||||||
|
assert frame is not None
|
||||||
|
assert not np.shares_memory(frame.rgb, sdk.get_color_frame().array)
|
||||||
|
assert not np.shares_memory(frame.depth, sdk.get_depth_frame().array)
|
||||||
|
assert frame.rgb.base is None and frame.depth.base is None
|
||||||
|
np.testing.assert_array_equal(frame.rgb, sdk.get_color_frame().array) # same pixels
|
||||||
|
finally:
|
||||||
|
cam.close()
|
||||||
|
|
||||||
|
|
||||||
|
def test_owned_copies_views_only():
|
||||||
|
owned = np.zeros((4, 4), np.uint16)
|
||||||
|
view = np.asanyarray(memoryview(owned))
|
||||||
|
assert _owned(owned) is owned
|
||||||
|
copy = _owned(view)
|
||||||
|
assert copy is not view and not np.shares_memory(copy, owned)
|
||||||
Reference in new issue
Block a user