ykcheungab
464 lines
14 KiB
Python
464 lines
14 KiB
Python
"""Unit tests for orisys.capture with fake VideoCapture objects (no hardware)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from orisys.capture import (
|
|
MAX_SIMULTANEOUS_CAPTURES,
|
|
CameraFilter,
|
|
CameraInfo,
|
|
CaptureConfig,
|
|
CaptureHealth,
|
|
CaptureManager,
|
|
CaptureMode,
|
|
CaptureSource,
|
|
FrameBuffer,
|
|
OverflowPolicy,
|
|
ReconnectPolicy,
|
|
list_cameras,
|
|
normalize_source,
|
|
open_capture,
|
|
pick_default_camera,
|
|
)
|
|
from orisys.capture.discovery import DEFAULT_LAPTOP_WEBCAM_PATTERNS
|
|
|
|
|
|
class FakeCap:
|
|
"""Deterministic stand-in for cv2.VideoCapture."""
|
|
|
|
def __init__(self, frames=None, *, fail_after=None, fail_reads=0, open_ok=True):
|
|
self._frames = list(frames or [])
|
|
self._idx = 0
|
|
self._open = bool(open_ok)
|
|
self._fail_after = fail_after
|
|
self._fail_reads_remaining = int(fail_reads)
|
|
self.read_count = 0
|
|
self.release_count = 0
|
|
self.width = 4
|
|
self.height = 3
|
|
self.fps = 30.0
|
|
self.fourcc = 0
|
|
|
|
def isOpened(self):
|
|
return self._open
|
|
|
|
def read(self):
|
|
self.read_count += 1
|
|
if not self._open:
|
|
return False, None
|
|
if self._fail_reads_remaining > 0:
|
|
self._fail_reads_remaining -= 1
|
|
return False, None
|
|
if self._fail_after is not None and self.read_count > self._fail_after:
|
|
self._open = False
|
|
return False, None
|
|
if self._idx >= len(self._frames):
|
|
# Repeat last frame like a live camera holding the stream.
|
|
if not self._frames:
|
|
return False, None
|
|
frame = self._frames[-1]
|
|
else:
|
|
frame = self._frames[self._idx]
|
|
self._idx += 1
|
|
return True, frame.copy()
|
|
|
|
def release(self):
|
|
self.release_count += 1
|
|
self._open = False
|
|
|
|
def get(self, prop_id):
|
|
# Values mirror common OpenCV property IDs without importing cv2 constants.
|
|
if prop_id in (3,): # CAP_PROP_FRAME_WIDTH
|
|
return float(self.width)
|
|
if prop_id in (4,): # CAP_PROP_FRAME_HEIGHT
|
|
return float(self.height)
|
|
if prop_id in (5,): # CAP_PROP_FPS
|
|
return float(self.fps)
|
|
if prop_id in (6,): # CAP_PROP_FOURCC
|
|
return float(self.fourcc)
|
|
return 0.0
|
|
|
|
def set(self, prop_id, value):
|
|
return True
|
|
|
|
|
|
def _frame(val: int) -> np.ndarray:
|
|
return np.full((3, 4, 3), val, dtype=np.uint8)
|
|
|
|
|
|
def _factory_from_caps(caps):
|
|
"""Return a factory that hands out FakeCap instances in order."""
|
|
lock = threading.Lock()
|
|
state = {"i": 0}
|
|
|
|
def factory(_config):
|
|
with lock:
|
|
if state["i"] >= len(caps):
|
|
raise RuntimeError("No more fake captures")
|
|
cap = caps[state["i"]]
|
|
state["i"] += 1
|
|
return cap, "FAKE"
|
|
|
|
return factory
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Source normalization / config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"video,expected",
|
|
[
|
|
(None, 0),
|
|
("0", 0),
|
|
(" 1 ", 1),
|
|
(2, 2),
|
|
("/dev/video0", "/dev/video0"),
|
|
(r".\clip.mp4" + "\\", r".\clip.mp4"),
|
|
],
|
|
)
|
|
def test_normalize_source(video, expected):
|
|
assert normalize_source(video) == expected
|
|
|
|
|
|
def test_capture_config_queue_bounds():
|
|
with pytest.raises(ValueError):
|
|
CaptureConfig(source=0, queue_size=0)
|
|
with pytest.raises(ValueError):
|
|
CaptureConfig(source=0, queue_size=11)
|
|
cfg = CaptureConfig(source=0, mode="queue", queue_size=10)
|
|
assert cfg.queue_size == 10
|
|
latest = CaptureConfig(source=0, mode="latest", queue_size=5)
|
|
assert latest.queue_size == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# FrameBuffer
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_frame_buffer_latest_overwrites():
|
|
buf = FrameBuffer(mode=CaptureMode.LATEST)
|
|
buf.publish(_frame(1))
|
|
buf.publish(_frame(2))
|
|
ok, frame, seq = buf.read_latest(copy=True)
|
|
assert ok and seq == 2
|
|
assert int(frame[0, 0, 0]) == 2
|
|
|
|
|
|
def test_frame_buffer_queue_drop_oldest():
|
|
buf = FrameBuffer(mode=CaptureMode.QUEUE, maxsize=2, overflow=OverflowPolicy.DROP_OLDEST)
|
|
buf.publish(_frame(1))
|
|
buf.publish(_frame(2))
|
|
buf.publish(_frame(3))
|
|
assert buf.dropped_frames == 1
|
|
ok, f1, _ = buf.read_next(copy=True)
|
|
ok2, f2, _ = buf.read_next(copy=True)
|
|
assert ok and ok2
|
|
assert int(f1[0, 0, 0]) == 2
|
|
assert int(f2[0, 0, 0]) == 3
|
|
|
|
|
|
def test_frame_buffer_queue_drop_newest():
|
|
buf = FrameBuffer(mode=CaptureMode.QUEUE, maxsize=1, overflow=OverflowPolicy.DROP_NEWEST)
|
|
buf.publish(_frame(1))
|
|
buf.publish(_frame(2))
|
|
assert buf.dropped_frames == 1
|
|
ok, frame, _ = buf.read_next(copy=True)
|
|
assert ok and int(frame[0, 0, 0]) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CaptureSource reconnect
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_capture_source_read_and_status():
|
|
frames = [_frame(7)]
|
|
cap = FakeCap(frames)
|
|
src = CaptureSource(
|
|
CaptureConfig(source=0, reconnect=ReconnectPolicy(enabled=False), start_worker=False),
|
|
factory=lambda cfg: (cap, "FAKE"),
|
|
)
|
|
src.open()
|
|
assert src.is_open()
|
|
assert src.status.health == CaptureHealth.CONNECTED
|
|
ok, frame = src.read_raw()
|
|
assert ok and int(frame[0, 0, 0]) == 7
|
|
src.release()
|
|
assert src.status.health == CaptureHealth.CLOSED
|
|
assert cap.release_count == 1
|
|
|
|
|
|
def test_auto_reconnect_after_failed_reads():
|
|
first = FakeCap([_frame(1)], fail_after=1)
|
|
second = FakeCap([_frame(9)])
|
|
factory = _factory_from_caps([first, second])
|
|
policy = ReconnectPolicy(
|
|
enabled=True,
|
|
fail_threshold=1,
|
|
initial_backoff_s=0.01,
|
|
max_backoff_s=0.01,
|
|
stale_timeout_s=None,
|
|
max_attempts=3,
|
|
)
|
|
src = CaptureSource(
|
|
CaptureConfig(source=0, reconnect=policy, start_worker=False, verbose=False),
|
|
factory=factory,
|
|
)
|
|
src.open()
|
|
ok, _ = src.read_raw()
|
|
assert ok
|
|
# Next read fails and closes first cap.
|
|
ok, _ = src.read_raw()
|
|
assert not ok
|
|
assert src.should_reconnect()
|
|
assert src.reconnect()
|
|
ok, frame = src.read_raw()
|
|
assert ok and int(frame[0, 0, 0]) == 9
|
|
assert src.status.health == CaptureHealth.CONNECTED
|
|
assert src.status.reconnect_attempts >= 1
|
|
|
|
|
|
def test_reconnect_disabled_does_not_reopen():
|
|
cap = FakeCap([_frame(1)], fail_after=1)
|
|
src = CaptureSource(
|
|
CaptureConfig(
|
|
source=0,
|
|
reconnect=ReconnectPolicy(enabled=False, fail_threshold=1),
|
|
start_worker=False,
|
|
),
|
|
factory=lambda cfg: (cap, "FAKE"),
|
|
)
|
|
src.open()
|
|
src.read_raw()
|
|
src.read_raw()
|
|
assert not src.should_reconnect()
|
|
|
|
|
|
def test_reconnect_attempts_exhausted():
|
|
good = FakeCap([_frame(1)], fail_after=1)
|
|
|
|
state = {"n": 0}
|
|
|
|
def factory(_cfg):
|
|
state["n"] += 1
|
|
if state["n"] == 1:
|
|
return good, "FAKE"
|
|
raise RuntimeError("camera missing")
|
|
|
|
policy = ReconnectPolicy(
|
|
enabled=True,
|
|
fail_threshold=1,
|
|
initial_backoff_s=0.01,
|
|
max_backoff_s=0.01,
|
|
stale_timeout_s=None,
|
|
max_attempts=2,
|
|
)
|
|
src = CaptureSource(
|
|
CaptureConfig(source=0, reconnect=policy, start_worker=False),
|
|
factory=factory,
|
|
)
|
|
src.open()
|
|
src.read_raw()
|
|
src.read_raw()
|
|
assert src.should_reconnect()
|
|
assert not src.reconnect()
|
|
assert not src.reconnect()
|
|
assert src.status.health == CaptureHealth.FAILED
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CaptureWorker
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_worker_latest_mode_publishes_frames():
|
|
caps = [FakeCap([_frame(i) for i in range(5)])]
|
|
worker = open_capture(
|
|
CaptureConfig(
|
|
source=0,
|
|
mode=CaptureMode.LATEST,
|
|
start_worker=True,
|
|
reconnect=ReconnectPolicy(enabled=False),
|
|
),
|
|
factory=_factory_from_caps(caps),
|
|
start_worker=True,
|
|
)
|
|
try:
|
|
deadline = time.perf_counter() + 2.0
|
|
ok = False
|
|
frame = None
|
|
while time.perf_counter() < deadline:
|
|
ok, frame = worker.read(wait_s=0.05)
|
|
if ok:
|
|
break
|
|
assert ok and frame is not None
|
|
finally:
|
|
worker.close()
|
|
|
|
|
|
def test_worker_reconnect_in_background():
|
|
first = FakeCap([_frame(1)], fail_after=2)
|
|
second = FakeCap([_frame(42) for _ in range(20)])
|
|
policy = ReconnectPolicy(
|
|
enabled=True,
|
|
fail_threshold=1,
|
|
initial_backoff_s=0.01,
|
|
max_backoff_s=0.02,
|
|
stale_timeout_s=None,
|
|
)
|
|
worker = open_capture(
|
|
CaptureConfig(source=0, mode="latest", reconnect=policy, start_worker=True),
|
|
factory=_factory_from_caps([first, second]),
|
|
start_worker=True,
|
|
)
|
|
try:
|
|
saw_reconnect = False
|
|
saw_new = False
|
|
deadline = time.perf_counter() + 3.0
|
|
while time.perf_counter() < deadline:
|
|
if worker.source.status.reconnect_attempts > 0:
|
|
saw_reconnect = True
|
|
ok, frame = worker.read(wait_s=0.05)
|
|
if ok and frame is not None and int(frame[0, 0, 0]) == 42:
|
|
saw_new = True
|
|
break
|
|
time.sleep(0.01)
|
|
assert saw_reconnect
|
|
assert saw_new
|
|
finally:
|
|
worker.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CaptureManager
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_capture_manager_capacity_and_status():
|
|
mgr = CaptureManager(max_sources=2)
|
|
caps = [FakeCap([_frame(1)]), FakeCap([_frame(2)]), FakeCap([_frame(3)])]
|
|
factory = _factory_from_caps(caps)
|
|
try:
|
|
mgr.open(CaptureConfig(source=0, start_worker=False), factory=factory)
|
|
mgr.open(CaptureConfig(source=1, start_worker=False), factory=factory)
|
|
assert len(mgr) == 2
|
|
with pytest.raises(RuntimeError):
|
|
mgr.open(CaptureConfig(source=2, start_worker=False), factory=factory)
|
|
statuses = mgr.status_all()
|
|
assert set(statuses) == {"0", "1"}
|
|
assert all(s.health == CaptureHealth.CONNECTED for s in statuses.values())
|
|
finally:
|
|
mgr.close_all()
|
|
assert len(mgr) == 0
|
|
|
|
|
|
def test_capture_manager_max_constant():
|
|
assert MAX_SIMULTANEOUS_CAPTURES == 10
|
|
with pytest.raises(ValueError):
|
|
CaptureManager(max_sources=11)
|
|
|
|
|
|
def test_capture_manager_read_latest_all():
|
|
mgr = CaptureManager(max_sources=3)
|
|
caps = [FakeCap([_frame(10)]), FakeCap([_frame(20)])]
|
|
factory = _factory_from_caps(caps)
|
|
try:
|
|
w0 = mgr.open(
|
|
CaptureConfig(source=0, mode="latest", start_worker=True),
|
|
factory=factory,
|
|
)
|
|
w1 = mgr.open(
|
|
CaptureConfig(source=1, mode="latest", start_worker=True),
|
|
factory=factory,
|
|
)
|
|
deadline = time.perf_counter() + 2.0
|
|
while time.perf_counter() < deadline:
|
|
if w0.buffer.last_seq > 0 and w1.buffer.last_seq > 0:
|
|
break
|
|
time.sleep(0.01)
|
|
frames = mgr.read_latest_all()
|
|
assert "0" in frames and "1" in frames
|
|
finally:
|
|
mgr.close_all()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Camera discovery filters (no hardware)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_camera_filter_excludes_laptop_webcams():
|
|
cams = [
|
|
CameraInfo(0, "DSHOW", 0, 640, 480, name="Integrated Webcam"),
|
|
CameraInfo(1, "DSHOW", 0, 1920, 1080, name="Orisys Tactile Cam"),
|
|
CameraInfo(2, "DSHOW", 0, 1280, 720, name="USB Camera"),
|
|
]
|
|
filt = CameraFilter(exclude_laptop_webcams=True)
|
|
kept = [c for c in cams if filt.matches(c)]
|
|
assert [c.index for c in kept] == [1, 2]
|
|
|
|
|
|
def test_camera_filter_name_include_and_vid_pid():
|
|
cams = [
|
|
CameraInfo(0, "V4L2", 0, 640, 480, name="foo", vendor_id=0x1234, product_id=0x5678),
|
|
CameraInfo(1, "V4L2", 0, 640, 480, name="orisys-finger", vendor_id=0xAAAA, product_id=0xBBBB),
|
|
]
|
|
filt = CameraFilter(name_include=[r"orisys"], vid_pid_allowlist=[(0xAAAA, 0xBBBB)])
|
|
kept = [c for c in cams if filt.matches(c)]
|
|
assert len(kept) == 1 and kept[0].index == 1
|
|
|
|
|
|
def test_camera_filter_min_resolution():
|
|
cams = [
|
|
CameraInfo(0, "DSHOW", 0, 320, 240, name="tiny"),
|
|
CameraInfo(1, "DSHOW", 0, 1280, 720, name="hd"),
|
|
]
|
|
filt = CameraFilter(min_width=640, min_height=480)
|
|
kept = [c for c in cams if filt.matches(c)]
|
|
assert [c.index for c in kept] == [1]
|
|
|
|
|
|
def test_pick_default_camera():
|
|
cams = [
|
|
CameraInfo(1, "DSHOW", 0, 640, 480),
|
|
CameraInfo(2, "DSHOW", 0, 640, 480),
|
|
]
|
|
assert pick_default_camera(cams, preferred="2") == "2"
|
|
assert pick_default_camera(cams, preferred=None) == "1"
|
|
assert pick_default_camera([], preferred="0") is None
|
|
|
|
|
|
def test_laptop_patterns_nonempty():
|
|
assert len(DEFAULT_LAPTOP_WEBCAM_PATTERNS) >= 5
|
|
|
|
|
|
def test_camera_info_label_includes_name():
|
|
cam = CameraInfo(3, "DSHOW", 700, 1280, 720, name="Orisys")
|
|
assert "Orisys" in cam.label()
|
|
assert cam.source == "3"
|
|
|
|
|
|
def test_list_cameras_is_callable():
|
|
# Smoke: function import works; may return [] without hardware.
|
|
assert callable(list_cameras)
|
|
|
|
|
|
def test_sensor_external_mode_still_works():
|
|
from orisys import Sensor
|
|
|
|
sensor = Sensor("external", config_name="box", verbose=False, backend="cpu")
|
|
try:
|
|
assert sensor._external_only
|
|
assert sensor.capture_status is None
|
|
assert sensor._capture_worker is None
|
|
finally:
|
|
sensor.disconnect()
|