OliviaCandClaude Opus 5.5
cc28127c9c
FrameAcquisition: a single thread calls wait_for_frames + align (the old
capture() body, now _grab_once) and keeps the latest frame, numbered.
capture() no longer touches the pipeline: it waits for the next frame of
that thread; requests waiting at the same moment get the same frame, so N
callers cost one grab. get_rgb()/get_depth() keep their 100 ms cache,
now on the thread's latest frame (_last_frame is gone).
The thread grabs on demand, only while someone waits for a newer frame:
grabbing continuously at 30 fps would align frames nobody reads, and align
on the Jetson CPU is the main suspect of the capture cost. A grab error is
logged and retried, the thread never dies. close() stops it before the
pipeline and joins it after (pipeline.stop() ends a wait in progress).
open/close and camera_open are unchanged, so the healthcheck + autoheal
recovery of ede70a2 sees the same state as before.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
103 lines
2.6 KiB
Python
103 lines
2.6 KiB
Python
"""FrameAcquisition: one thread grabs on demand; waiters share frames; errors never kill it."""
|
|
|
|
import threading
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from src.backend.services.frame_acquisition import FrameAcquisition
|
|
|
|
LATENCY_S = 0.1
|
|
|
|
|
|
class SlowGrab:
|
|
"""Returns 1, 2, 3, ... after LATENCY_S each; optionally raises on chosen calls."""
|
|
|
|
def __init__(self, fail_on=()):
|
|
self.calls = 0
|
|
self.fail_on = set(fail_on)
|
|
|
|
def __call__(self):
|
|
self.calls += 1
|
|
time.sleep(LATENCY_S)
|
|
if self.calls in self.fail_on:
|
|
raise RuntimeError("SDK error")
|
|
return self.calls
|
|
|
|
|
|
@pytest.fixture
|
|
def acquisition():
|
|
made = []
|
|
|
|
def make(grab):
|
|
acq = FrameAcquisition(grab, name="test-acquisition")
|
|
acq.start()
|
|
made.append(acq)
|
|
return acq
|
|
|
|
yield make
|
|
for acq in made:
|
|
acq.request_stop()
|
|
assert acq.join(2.0)
|
|
|
|
|
|
def test_three_waiters_share_one_grab(acquisition):
|
|
grab = SlowGrab()
|
|
acq = acquisition(grab)
|
|
results = []
|
|
|
|
def wait():
|
|
results.append(acq.wait_newer(0, timeout_s=2.0))
|
|
|
|
threads = [threading.Thread(target=wait) for _ in range(3)]
|
|
t0 = time.perf_counter()
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
elapsed = time.perf_counter() - t0
|
|
assert results == [(1, 1)] * 3
|
|
assert grab.calls == 1
|
|
assert elapsed < 2 * LATENCY_S
|
|
|
|
|
|
def test_no_grab_without_demand(acquisition):
|
|
grab = SlowGrab()
|
|
acquisition(grab)
|
|
time.sleep(3 * LATENCY_S)
|
|
assert grab.calls == 0
|
|
|
|
|
|
def test_an_exception_in_grab_does_not_stop_the_thread(acquisition):
|
|
grab = SlowGrab(fail_on={1})
|
|
acq = acquisition(grab)
|
|
seq, frame = acq.wait_newer(0, timeout_s=3.0)
|
|
assert (seq, frame) == (1, 2) # call 1 raised, call 2 delivered frame number 1
|
|
|
|
|
|
def test_timeout_returns_no_frame():
|
|
grab = SlowGrab()
|
|
acq = FrameAcquisition(grab) # never started: nothing will arrive
|
|
assert acq.wait_newer(0, timeout_s=0.05) == (0, None)
|
|
|
|
|
|
def test_stop_wakes_waiters_and_ends_the_thread():
|
|
acq = FrameAcquisition(lambda: None) # a grab that never delivers
|
|
acq.start()
|
|
got = []
|
|
waiter = threading.Thread(target=lambda: got.append(acq.wait_newer(0, timeout_s=5.0)))
|
|
waiter.start()
|
|
time.sleep(0.1)
|
|
acq.request_stop()
|
|
waiter.join(1.0)
|
|
assert got == [(0, None)]
|
|
assert acq.join(1.0)
|
|
|
|
|
|
def test_latest_is_the_last_delivered_frame(acquisition):
|
|
acq = acquisition(SlowGrab())
|
|
assert acq.latest() == (0, None)
|
|
acq.wait_newer(0, timeout_s=2.0)
|
|
acq.wait_newer(1, timeout_s=2.0)
|
|
assert acq.latest() == (2, 2)
|