diff --git a/tests/fake_realsense.py b/tests/fake_realsense.py index b3c4309..4a99c59 100644 --- a/tests/fake_realsense.py +++ b/tests/fake_realsense.py @@ -1,7 +1,12 @@ -"""A fake pyrealsense2 pipeline for tests: renders one 640x480 frameset per call after a fixed latency. +"""A fake pyrealsense2 pipeline for tests (there is no RealSense on the development PC). -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. +Two ways to deliver frames: +- per call (default): wait_for_frames blocks `latency_s` and returns a new frameset; +- clocked (`fps=...`): the camera produces a frameset every 1/fps s whether anyone + reads it or not, and wait_for_frames returns the next one, like the real device. +Frame N carries its number in the pixels (a 16x16 color block = N % 256, depth[0, 0] = N) +so a test can tell which frame an image came from. FakeAlign stands in for rs.align. +The driver only uses a few methods of the SDK objects, and these fakes provide exactly those. """ import threading @@ -9,13 +14,16 @@ import time import numpy as np -FAKE_LATENCY_S = 0.1 # one frame every 100 ms, as the request's test asks +FAKE_LATENCY_S = 0.1 # one frame every 100 ms, as the R01 test asks WIDTH, HEIGHT = 640, 480 +MARK = 16 # side of the color block that carries the frame number class FakeVideoFrame: - def __init__(self, data: np.ndarray): + def __init__(self, data: np.ndarray, index: int, fps: float): self._data = data + self._index = index + self._fps = fps def get_data(self): # A buffer over the frame's memory, as pyrealsense2's BufData: np.asanyarray() of it is a view. @@ -31,16 +39,28 @@ class FakeVideoFrame: def get_width(self) -> int: return self._data.shape[1] + def get_timestamp(self) -> float: + """Camera time of the frame, milliseconds.""" + return self._index * 1000.0 / self._fps + + def get_frame_timestamp_domain(self) -> str: + return "timestamp_domain.hardware_clock" + def __bool__(self) -> bool: return True class FakeFrameset: - def __init__(self, index: int): + def __init__(self, index: int, fps: float = 1.0 / FAKE_LATENCY_S): + self.index = index 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)) + color = rng.integers(0, 255, (HEIGHT, WIDTH, 3), dtype=np.uint8) + color[:MARK, :MARK] = index % 256 + depth = rng.integers(300, 3000, (HEIGHT, WIDTH), dtype=np.uint16) + depth[0, 0] = index + self._color = FakeVideoFrame(color, index, fps) + self._depth = FakeVideoFrame(depth, index, fps) def get_color_frame(self): return self._color @@ -49,27 +69,45 @@ class FakeFrameset: return self._depth -class FakePipeline: - """wait_for_frames blocks FAKE_LATENCY_S and returns a new frameset; counts the calls.""" +def frame_number_of_color(bgr: np.ndarray) -> int: + """The frame number a fake color image carries (JPEG-tolerant: median of the block).""" + return int(np.median(bgr[2:MARK - 2, 2:MARK - 2])) - def __init__(self, latency_s: float = FAKE_LATENCY_S): + +class FakePipeline: + """wait_for_frames returns framesets numbered 1, 2, 3, ...; counts the calls.""" + + def __init__(self, latency_s: float = FAKE_LATENCY_S, fps: float | None = None): self.latency_s = latency_s + self.fps = fps self.calls = 0 self.max_concurrent = 0 self._active = 0 self._lock = threading.Lock() + self._start = time.monotonic() self.stopped = False self.last_frameset = None + def _next_clocked_index(self) -> int: + """Sleep until the camera's next frame and return its number.""" + period = 1.0 / self.fps + elapsed = time.monotonic() - self._start + index = int(elapsed / period) + 1 + time.sleep(max(0.0, self._start + index * period - time.monotonic())) + return index + 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 + call = self.calls try: - time.sleep(self.latency_s) - self.last_frameset = FakeFrameset(index) + if self.fps: + self.last_frameset = FakeFrameset(self._next_clocked_index(), self.fps) + else: + time.sleep(self.latency_s) + self.last_frameset = FakeFrameset(call) return self.last_frameset finally: with self._lock: @@ -79,15 +117,28 @@ class FakePipeline: 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). +class FakeAlign: + """rs.align stand-in: takes `delay_s` and returns the frameset (the fake depth is already on the color grid).""" + + def __init__(self, delay_s: float): + self.delay_s = delay_s + self.calls = 0 + + def process(self, frameset): + self.calls += 1 + time.sleep(self.delay_s) + return frameset + + +def open_fake_camera(camera, pipeline: FakePipeline, align: FakeAlign | None = None) -> None: + """Put a RealSenseD435 in the 'open' state on a fake pipeline (USB 3 path; align only if given). 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. + every version of the driver. """ import pyrealsense2 as rs camera._pipeline = pipeline - camera._align = None + camera._align = align camera._color_format = rs.format.bgr8 camera._depth_scale = 0.001 camera._initialized = True diff --git a/tests/test_two_channels.py b/tests/test_two_channels.py new file mode 100644 index 0000000..f035d2f --- /dev/null +++ b/tests/test_two_channels.py @@ -0,0 +1,134 @@ +"""Two channels on a fake 30 fps camera: fast color (no align) and aligned color+depth on request.""" + +import asyncio +import base64 +import time + +import cv2 +import numpy as np +import pytest + +pytest.importorskip("pyrealsense2") + +from httpx import ASGITransport, AsyncClient # noqa: E402 +from starlette.testclient import TestClient # noqa: E402 + +from src.backend.api import routes, websocket # noqa: E402 +from src.backend.main import app # noqa: E402 +from src.backend.services.camera import RealSenseD435 # noqa: E402 +from tests.fake_realsense import FakeAlign, FakePipeline, frame_number_of_color, open_fake_camera # noqa: E402 + +FPS = 30 + + +def _open(cam, align_s): + pipe, align = FakePipeline(fps=FPS), FakeAlign(delay_s=align_s) + open_fake_camera(cam, pipe, align) + routes.set_camera(cam) + websocket.set_camera(cam) + return pipe, align + + +@pytest.fixture +def camera(): + cams = [] + + def make(align_s=0.0, frame_cache_ms=100.0): + cam = RealSenseD435(frame_cache_ms=frame_cache_ms) + cams.append(cam) + return (cam, *_open(cam, align_s)) + + yield make + for cam in cams: + cam.close() + routes.set_camera(None) + websocket.set_camera(None) + + +def _decode(content: bytes, flags=cv2.IMREAD_COLOR) -> np.ndarray: + return cv2.imdecode(np.frombuffer(content, np.uint8), flags) + + +def test_rgb_then_capture_then_depth_come_from_one_frame(camera): + """Criterion B: /api/capture in between must not move the frame /api/rgb and /api/depth share. + + frame_cache_ms is raised so the test does not depend on this PC's timing: + what it checks is WHICH frame the cache serves, not the 100 ms. + """ + camera(frame_cache_ms=1000.0) + client = TestClient(app) + rgb_number = frame_number_of_color(_decode(client.get("/api/rgb").content)) + capture = client.get("/api/capture").json()["data"] + depth = _decode(client.get("/api/depth").content, cv2.IMREAD_UNCHANGED) + capture_number = frame_number_of_color(_decode(base64.b64decode(capture["rgb_jpeg_b64"]))) + assert capture_number != rgb_number # /api/capture did get a newer frame ... + assert int(depth[0, 0]) % 256 == rgb_number # ... and /api/depth still pairs with /api/rgb + + +def test_color_channel_serves_consecutive_frames_with_both_clocks(camera): + camera() + client = TestClient(app) + first = client.get("/api/color/latest") + seq = int(first.headers["X-Frame-Seq"]) + seqs, marks = [seq], [] + for _ in range(10): + r = client.get("/api/color/latest", params={"after": seq, "timeout_ms": 1000}) + assert r.status_code == 200 + seq = int(r.headers["X-Frame-Seq"]) + seqs.append(seq) + camera_index = round(float(r.headers["X-Camera-Timestamp-Ms"]) * FPS / 1000.0) + marks.append((frame_number_of_color(_decode(r.content)), camera_index % 256)) + assert r.headers["X-Camera-Timestamp-Domain"] == "hardware_clock" + assert abs(float(r.headers["X-Host-Timestamp"]) - time.time()) < 5.0 + assert seqs == sorted(set(seqs)) # strictly newer each time + assert all(pixel == clock for pixel, clock in marks) # the image is the frame its header names + + +def test_color_channel_times_out_with_504(camera): + camera() + client = TestClient(app) + seq = int(client.get("/api/color/latest").headers["X-Frame-Seq"]) + t0 = time.perf_counter() + r = client.get("/api/color/latest", params={"after": seq + 10_000, "timeout_ms": 100}) + assert r.status_code == 504 + assert time.perf_counter() - t0 < 1.0 + + +def test_color_frame_is_encoded_once(camera): + cam, _, _ = camera() + seq, color = cam.color_frame() + first = routes.encoded_color(seq, color) + assert routes.encoded_color(seq, color) is first + assert first.jpeg() is first.jpeg() + + +def test_color_channel_keeps_camera_rate_while_capture_runs(camera): + """Criteria A1/A2 in short (3 s): color >= 25 new frames/s while /api/capture loops with a 100 ms align.""" + _, _, align = camera(align_s=0.1) + duration = 3.0 + + async def run(): + seqs, stamps = set(), set() + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://t", timeout=10) as c: + stop_at = time.monotonic() + duration + + async def color(): + after = int((await c.get("/api/color/latest")).headers["X-Frame-Seq"]) + while time.monotonic() < stop_at: + r = await c.get("/api/color/latest", params={"after": after, "timeout_ms": 1000}) + if r.status_code == 200: + after = int(r.headers["X-Frame-Seq"]) + seqs.add(after) + + async def capture(): + while time.monotonic() < stop_at: + body = (await c.get("/api/capture")).json() + if body["success"]: + stamps.add(body["data"]["timestamp"]) + + await asyncio.gather(color(), capture()) + return seqs, stamps + + seqs, stamps = asyncio.run(run()) + assert len(seqs) / duration >= 25 + assert len(stamps) >= 1 and align.calls >= len(stamps)