OliviaCandClaude Opus 5.5
a8e380ed69
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>
98 lines
2.9 KiB
Python
98 lines
2.9 KiB
Python
"""A fake pyrealsense2 pipeline for tests: renders one 640x480 frameset per call after a fixed latency.
|
|
|
|
There is no RealSense on the development PC. The driver only uses a few
|
|
methods of the SDK objects, and these fakes provide exactly those.
|
|
"""
|
|
|
|
import threading
|
|
import time
|
|
|
|
import numpy as np
|
|
|
|
FAKE_LATENCY_S = 0.1 # one frame every 100 ms, as the request's test asks
|
|
WIDTH, HEIGHT = 640, 480
|
|
|
|
|
|
class FakeVideoFrame:
|
|
def __init__(self, data: np.ndarray):
|
|
self._data = data
|
|
|
|
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
|
|
|
|
def get_height(self) -> int:
|
|
return self._data.shape[0]
|
|
|
|
def get_width(self) -> int:
|
|
return self._data.shape[1]
|
|
|
|
def __bool__(self) -> bool:
|
|
return True
|
|
|
|
|
|
class FakeFrameset:
|
|
def __init__(self, index: int):
|
|
rng = np.random.default_rng(index)
|
|
# Textured images, so the JPEG/PNG encoders do realistic work.
|
|
self._color = FakeVideoFrame(rng.integers(0, 255, (HEIGHT, WIDTH, 3), dtype=np.uint8))
|
|
self._depth = FakeVideoFrame(rng.integers(300, 3000, (HEIGHT, WIDTH), dtype=np.uint16))
|
|
|
|
def get_color_frame(self):
|
|
return self._color
|
|
|
|
def get_depth_frame(self):
|
|
return self._depth
|
|
|
|
|
|
class FakePipeline:
|
|
"""wait_for_frames blocks FAKE_LATENCY_S and returns a new frameset; counts the calls."""
|
|
|
|
def __init__(self, latency_s: float = FAKE_LATENCY_S):
|
|
self.latency_s = latency_s
|
|
self.calls = 0
|
|
self.max_concurrent = 0
|
|
self._active = 0
|
|
self._lock = threading.Lock()
|
|
self.stopped = False
|
|
self.last_frameset = None
|
|
|
|
def wait_for_frames(self, timeout_ms: int = 5000):
|
|
with self._lock:
|
|
self._active += 1
|
|
self.max_concurrent = max(self.max_concurrent, self._active)
|
|
self.calls += 1
|
|
index = self.calls
|
|
try:
|
|
time.sleep(self.latency_s)
|
|
self.last_frameset = FakeFrameset(index)
|
|
return self.last_frameset
|
|
finally:
|
|
with self._lock:
|
|
self._active -= 1
|
|
|
|
def stop(self) -> None:
|
|
self.stopped = True
|
|
|
|
|
|
def open_fake_camera(camera, pipeline: FakePipeline) -> None:
|
|
"""Put a RealSenseD435 in the 'open' state on a fake pipeline (USB 3 path, no align).
|
|
|
|
Uses the driver's own start hook when it has one, so the same helper serves
|
|
the driver before and after the acquisition thread was introduced.
|
|
"""
|
|
import pyrealsense2 as rs
|
|
camera._pipeline = pipeline
|
|
camera._align = None
|
|
camera._color_format = rs.format.bgr8
|
|
camera._depth_scale = 0.001
|
|
camera._initialized = True
|
|
camera._open = True
|
|
start = getattr(camera, "_start_acquisition", None)
|
|
if start is not None:
|
|
start()
|