OliviaCandClaude Opus 5.5
e11565de04
The number of the camera frameset an aligned frame comes from, in the same numbering as /api/raw/latest and /api/color/latest: it lets a tool compare Intel's align on PC2 with an align on the host on the SAME frameset. Existing fields unchanged (frame_id still counts responses); the field-set tests now say "the R01 fields plus frameset_seq only". Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
110 lines
4.3 KiB
Python
110 lines
4.3 KiB
Python
"""The API on a fake 100 ms camera: parallel requests are not serialized, the contract is unchanged."""
|
|
|
|
import asyncio
|
|
import statistics
|
|
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 FakePipeline, open_fake_camera # noqa: E402
|
|
|
|
# Fields of /api/capture before optimize-streams R01 (the multi-camera service parses them),
|
|
# and the only one added since (R05): existing fields never change.
|
|
CAPTURE_FIELDS = {
|
|
"frame_id", "timestamp", "rgb_shape", "depth_shape", "depth_min_mm", "depth_max_mm",
|
|
"depth_valid_pixels", "rgb_jpeg_b64", "depth_png_b64", "depth_color_b64",
|
|
}
|
|
ADDED_FIELDS = {"frameset_seq"}
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_camera():
|
|
cam = RealSenseD435()
|
|
pipe = FakePipeline()
|
|
open_fake_camera(cam, pipe)
|
|
routes.set_camera(cam) # the lifespan (not run in tests) injects the camera into both
|
|
websocket.set_camera(cam)
|
|
yield cam, pipe
|
|
cam.close()
|
|
routes.set_camera(None)
|
|
websocket.set_camera(None)
|
|
|
|
|
|
async def _wall_time(client, n: int) -> tuple[float, list]:
|
|
t0 = time.perf_counter()
|
|
responses = await asyncio.gather(*(client.get("/api/capture") for _ in range(n)))
|
|
return time.perf_counter() - t0, [r.json() for r in responses]
|
|
|
|
|
|
def test_three_parallel_captures_take_less_than_twice_one(fake_camera):
|
|
"""The criterion declared for R01: t(3 parallel) < 2 x t(1 alone), medians of 5."""
|
|
async def run():
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://t") as client:
|
|
await _wall_time(client, 1)
|
|
one = [(await _wall_time(client, 1))[0] for _ in range(5)]
|
|
three = [(await _wall_time(client, 3))[0] for _ in range(5)]
|
|
return statistics.median(one), statistics.median(three)
|
|
|
|
one, three = asyncio.run(run())
|
|
assert three < 2 * one, f"1 alone {one*1000:.0f} ms, 3 parallel {three*1000:.0f} ms"
|
|
|
|
|
|
def test_capture_fields_unchanged_and_one_frame_shared(fake_camera):
|
|
_, pipe = fake_camera
|
|
|
|
async def run():
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://t") as client:
|
|
return (await _wall_time(client, 3))[1]
|
|
|
|
bodies = asyncio.run(run())
|
|
assert all(b["success"] for b in bodies)
|
|
data = [b["data"] for b in bodies]
|
|
assert all(set(d) == CAPTURE_FIELDS | ADDED_FIELDS for d in data)
|
|
assert len({d["frame_id"] for d in data}) == 3 # frame_id still counts responses
|
|
assert len({d["timestamp"] for d in data}) == 1 # ... all three served by one frame
|
|
assert len({d["rgb_jpeg_b64"] for d in data}) == 1 # ... with the same bytes
|
|
assert pipe.max_concurrent == 1 # only one thread calls the pipeline
|
|
|
|
|
|
def test_rgb_and_depth_endpoints_decode(fake_camera):
|
|
client = TestClient(app)
|
|
rgb = cv2.imdecode(np.frombuffer(client.get("/api/rgb").content, np.uint8), cv2.IMREAD_COLOR)
|
|
depth = cv2.imdecode(np.frombuffer(client.get("/api/depth").content, np.uint8), cv2.IMREAD_UNCHANGED)
|
|
color = cv2.imdecode(np.frombuffer(client.get("/api/depth/color").content, np.uint8), cv2.IMREAD_COLOR)
|
|
assert rgb.shape == (480, 640, 3)
|
|
assert depth.shape == (480, 640) and depth.dtype == np.uint16
|
|
assert color.shape == (480, 640, 3)
|
|
|
|
|
|
def test_websocket_streams_frames(fake_camera):
|
|
with TestClient(app).websocket_connect("/ws") as ws:
|
|
ws.send_text("start")
|
|
first, second = ws.receive_json(), ws.receive_json()
|
|
ws.send_text("close")
|
|
assert set(first) == CAPTURE_FIELDS | ADDED_FIELDS
|
|
assert second["timestamp"] > first["timestamp"]
|
|
|
|
|
|
def test_close_stops_the_acquisition_threads():
|
|
cam = RealSenseD435()
|
|
pipe = FakePipeline()
|
|
open_fake_camera(cam, pipe)
|
|
assert cam.capture() is not None
|
|
threads = [cam._raw, cam._aligned]
|
|
cam.close()
|
|
assert not cam.is_open()
|
|
assert pipe.stopped
|
|
assert not any(t._thread.is_alive() for t in threads)
|
|
assert cam.capture() is None
|
|
assert cam.color_frame() == (0, None)
|