ykcheungab
136 lines
4.4 KiB
Python
136 lines
4.4 KiB
Python
"""Live Finger consumes current frames without a FIFO of pending work."""
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from orisys_app_shared.acquisition import finger_worker as fw
|
|
|
|
|
|
@pytest.mark.parametrize("is_file, expected", [(False, "serial"), (True, "stream")])
|
|
def test_default_path_keeps_only_offline_processing_queued(
|
|
monkeypatch, is_file, expected
|
|
):
|
|
worker = fw.FingerAcquisitionWorker()
|
|
paths = []
|
|
monkeypatch.setattr(worker, "_open_runtime", lambda *_: (None, None, None, is_file))
|
|
monkeypatch.setattr(worker, "_run_serial_loop", lambda *_: paths.append("serial"))
|
|
monkeypatch.setattr(
|
|
worker, "_run_stream_pipeline", lambda *_: paths.append("stream")
|
|
)
|
|
worker._run_session(1, None)
|
|
assert paths == [expected]
|
|
|
|
|
|
def _live_loop(monkeypatch, *, sequences=None, compute_s=0.0, sink_s=0.0, pause=False):
|
|
clock = [10.0]
|
|
worker = fw.FingerAcquisitionWorker()
|
|
worker._session_id = 1
|
|
worker._mono_t0 = clock[0]
|
|
packets = []
|
|
resets = []
|
|
source = SimpleNamespace(
|
|
running=True, status=SimpleNamespace(last_seq=0, dropped_frames=7)
|
|
)
|
|
flow = np.zeros((8, 8, 2), np.float32)
|
|
depth = np.zeros((8, 8), np.float32)
|
|
tactile = SimpleNamespace(reset=lambda: resets.append(clock[0]))
|
|
runtime = SimpleNamespace(
|
|
tactile=tactile, capture_source=source, valid_mask=np.ones((8, 8), bool)
|
|
)
|
|
reads = [0]
|
|
paused_once = [False]
|
|
|
|
def read():
|
|
reads[0] += 1
|
|
if reads[0] > 40:
|
|
worker._stop.set()
|
|
return None
|
|
if sequences is not None:
|
|
seq = sequences[min(reads[0] - 1, len(sequences) - 1)]
|
|
else:
|
|
seq = 1 + int((clock[0] - 10.0) * 30 + 1e-6)
|
|
source.status.last_seq = seq
|
|
return SimpleNamespace(
|
|
image=np.full((8, 8, 3), seq, np.uint8),
|
|
sequence=seq,
|
|
host_mono_ns=int(clock[0] * 1e9),
|
|
host_utc_ns=int(clock[0] * 1e9),
|
|
media_time_ms=None,
|
|
)
|
|
|
|
def process(frame, **_):
|
|
flow.fill(frame[0, 0, 0])
|
|
depth.fill(frame[0, 0, 0])
|
|
return frame[:, :, 0]
|
|
|
|
def chain(_):
|
|
clock[0] += compute_s
|
|
if pause and not paused_once[0]:
|
|
paused_once[0] = True
|
|
worker.pause()
|
|
return depth, ""
|
|
|
|
def sleep(seconds):
|
|
clock[0] += max(seconds, 0.001)
|
|
if worker.paused:
|
|
worker.resume()
|
|
|
|
def sink(packet):
|
|
packets.append(packet)
|
|
clock[0] += sink_s
|
|
if len(packets) >= 3:
|
|
worker._stop.set()
|
|
|
|
runtime.read_source_stamped = read
|
|
runtime.process_frame = process
|
|
monkeypatch.setattr(
|
|
fw, "time", SimpleNamespace(perf_counter=lambda: clock[0], sleep=sleep)
|
|
)
|
|
monkeypatch.setattr(fw, "process_chain", chain)
|
|
monkeypatch.setattr(fw, "_contact_from_sensor", lambda *_, **__: None)
|
|
monkeypatch.setattr(
|
|
fw,
|
|
"read_forces_from_sensor",
|
|
lambda *_, **__: {
|
|
"normal_force": 1.0,
|
|
"shear_x": 2.0,
|
|
"shear_y": 3.0,
|
|
"normal_force_raw": 1.0,
|
|
"shear_x_raw": 2.0,
|
|
"shear_y_raw": 3.0,
|
|
"force_unit": "raw",
|
|
"force_calibrated": False,
|
|
"flow": flow,
|
|
"depth": depth,
|
|
},
|
|
)
|
|
worker.add_sink(sink)
|
|
worker._run_serial_loop(1, SimpleNamespace(config_ref=None), 0, runtime, False)
|
|
return packets, worker, resets
|
|
|
|
|
|
def test_repeated_camera_frames_are_not_processed_or_recorded(monkeypatch):
|
|
packets, _, _ = _live_loop(monkeypatch, sequences=[1, 1, 2, 2, 3])
|
|
assert [p.source_frame_id for p in packets] == [1, 2, 3]
|
|
assert [p.frame_id for p in packets] == [0, 1, 2]
|
|
|
|
|
|
def test_slow_compute_and_sink_take_current_input_without_a_backlog(monkeypatch):
|
|
packets, _, _ = _live_loop(monkeypatch, compute_s=0.07, sink_s=0.10)
|
|
assert [p.source_frame_id for p in packets] == [1, 6, 11]
|
|
for packet in packets:
|
|
assert packet.flow_frame_id == packet.depth_frame_id == packet.frame_id
|
|
assert np.all(packet.flow == packet.source_frame_id)
|
|
assert np.all(packet.depth == packet.source_frame_id)
|
|
assert packet.capture_dropped == 7
|
|
|
|
|
|
def test_pause_during_compute_does_not_publish_and_resume_rebases(monkeypatch):
|
|
packets, _, resets = _live_loop(monkeypatch, compute_s=0.07, pause=True)
|
|
assert len(packets) == 3
|
|
assert packets[0].source_frame_id > 1
|
|
assert packets[0].frame_id == 0
|
|
assert len(resets) == 1
|