ykcheungab
436 lines
12 KiB
Python
436 lines
12 KiB
Python
"""Unit tests for orisys.capture with fake VideoCapture objects (no hardware)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
|
|
import cv2
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from orisys.capture import (
|
|
MAX_SIMULTANEOUS_CAPTURES,
|
|
CameraFilter,
|
|
CameraInfo,
|
|
CaptureConfig,
|
|
CaptureHealth,
|
|
CaptureSource,
|
|
ReconnectPolicy,
|
|
list_cameras,
|
|
normalize_source,
|
|
open_capture,
|
|
pick_default_camera,
|
|
)
|
|
from orisys.capture.discovery import DEFAULT_LAPTOP_WEBCAM_PATTERNS
|
|
from orisys.capture.source import _apply_stream_props, _warmup_capture
|
|
|
|
MJPG = cv2.VideoWriter_fourcc(*"MJPG")
|
|
YUY2 = cv2.VideoWriter_fourcc(*"YUY2")
|
|
|
|
|
|
class FakeCap:
|
|
"""Deterministic stand-in for cv2.VideoCapture."""
|
|
|
|
def __init__(
|
|
self,
|
|
frames=None,
|
|
*,
|
|
fail_after=None,
|
|
fail_reads=0,
|
|
open_ok=True,
|
|
reset_fourcc_on_resize=False,
|
|
):
|
|
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.grab_count = 0
|
|
self.release_count = 0
|
|
self.width = 4
|
|
self.height = 3
|
|
self.fps = 30.0
|
|
self.fourcc = 0
|
|
self.reset_fourcc_on_resize = reset_fourcc_on_resize
|
|
self.set_calls: list[tuple[int, float]] = []
|
|
|
|
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):
|
|
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 grab(self):
|
|
self.grab_count += 1
|
|
if not self._open:
|
|
return False
|
|
ok, _ = self.read()
|
|
return ok
|
|
|
|
def release(self):
|
|
self.release_count += 1
|
|
self._open = False
|
|
|
|
def get(self, prop_id):
|
|
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):
|
|
self.set_calls.append((prop_id, float(value)))
|
|
if prop_id in (3,): # CAP_PROP_FRAME_WIDTH
|
|
self.width = int(value)
|
|
if self.reset_fourcc_on_resize:
|
|
self.fourcc = YUY2
|
|
if prop_id in (4,): # CAP_PROP_FRAME_HEIGHT
|
|
self.height = int(value)
|
|
if self.reset_fourcc_on_resize:
|
|
self.fourcc = YUY2
|
|
if prop_id in (5,): # CAP_PROP_FPS
|
|
self.fps = float(value)
|
|
if prop_id in (6,): # CAP_PROP_FOURCC
|
|
self.fourcc = int(value)
|
|
return True
|
|
|
|
|
|
def _frame(val: int) -> np.ndarray:
|
|
return np.full((3, 4, 3), val, dtype=np.uint8)
|
|
|
|
|
|
def _factory_from_caps(caps):
|
|
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
|
|
|
|
|
|
@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_from_camera_dict_ignores_deprecated_kwargs():
|
|
cfg = CaptureConfig.from_camera_dict(
|
|
0,
|
|
{"resolution_raw": [1920, 1080], "fps": 30},
|
|
mode="queue",
|
|
queue_size=5,
|
|
overflow="drop_newest",
|
|
start_worker=False,
|
|
)
|
|
assert cfg.width == 1920
|
|
assert cfg.height == 1080
|
|
assert cfg.background is False
|
|
|
|
|
|
def test_apply_stream_props_sets_size_before_fourcc():
|
|
cap = FakeCap([_frame(1)], reset_fourcc_on_resize=True)
|
|
_apply_stream_props(cap, width=1920, height=1080, fps=30, fourcc="MJPG")
|
|
assert cap.fourcc == MJPG
|
|
width_idx = [p for p, _ in cap.set_calls if p == 3]
|
|
fourcc_calls = [v for p, v in cap.set_calls if p == 6]
|
|
assert width_idx
|
|
assert fourcc_calls
|
|
assert cap.set_calls.index((3, 1920.0)) < cap.set_calls.index((6, float(MJPG)))
|
|
|
|
|
|
def test_warmup_capture_grabs():
|
|
cap = FakeCap([_frame(1)])
|
|
_warmup_capture(cap, grabs=3)
|
|
assert cap.grab_count == 3
|
|
|
|
|
|
def test_capture_source_read_and_status():
|
|
frames = [_frame(7)]
|
|
cap = FakeCap(frames)
|
|
src = CaptureSource(
|
|
CaptureConfig(source=0, reconnect=ReconnectPolicy(enabled=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, verbose=False),
|
|
factory=factory,
|
|
)
|
|
src.open()
|
|
ok, _ = src.read_raw()
|
|
assert ok
|
|
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),
|
|
),
|
|
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),
|
|
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
|
|
|
|
|
|
def test_background_thread_publishes_latest_frame():
|
|
caps = [FakeCap([_frame(i) for i in range(5)])]
|
|
src = open_capture(
|
|
CaptureConfig(
|
|
source=0,
|
|
background=True,
|
|
reconnect=ReconnectPolicy(enabled=False),
|
|
),
|
|
factory=_factory_from_caps(caps),
|
|
start_background=True,
|
|
)
|
|
try:
|
|
deadline = time.perf_counter() + 2.0
|
|
ok = False
|
|
frame = None
|
|
while time.perf_counter() < deadline:
|
|
ok, frame = src.read(wait_s=0.05)
|
|
if ok:
|
|
break
|
|
assert ok and frame is not None
|
|
assert src.running
|
|
finally:
|
|
src.close()
|
|
|
|
|
|
def test_background_reconnect():
|
|
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,
|
|
)
|
|
src = open_capture(
|
|
CaptureConfig(source=0, reconnect=policy, background=True),
|
|
factory=_factory_from_caps([first, second]),
|
|
start_background=True,
|
|
)
|
|
try:
|
|
saw_reconnect = False
|
|
saw_new = False
|
|
deadline = time.perf_counter() + 3.0
|
|
while time.perf_counter() < deadline:
|
|
if src.status.reconnect_attempts > 0:
|
|
saw_reconnect = True
|
|
ok, frame = src.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:
|
|
src.close()
|
|
|
|
|
|
def test_warmup_frames_discards_reads():
|
|
cap = FakeCap([_frame(i) for i in range(10)])
|
|
src = CaptureSource(
|
|
CaptureConfig(source=0),
|
|
factory=lambda cfg: (cap, "FAKE"),
|
|
)
|
|
src.open()
|
|
src.warmup_frames(3)
|
|
ok, frame = src.read_raw()
|
|
assert ok and int(frame[0, 0, 0]) == 3
|
|
|
|
|
|
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():
|
|
assert callable(list_cameras)
|
|
|
|
|
|
def test_max_simultaneous_captures_constant():
|
|
assert MAX_SIMULTANEOUS_CAPTURES == 10
|
|
|
|
|
|
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_source is None
|
|
finally:
|
|
sensor.disconnect()
|
|
|
|
|
|
def test_capture_source_open_applies_stream_props_via_default_factory(monkeypatch):
|
|
cap = FakeCap([_frame(1)], reset_fourcc_on_resize=True)
|
|
|
|
def fake_open(*args, **kwargs):
|
|
_apply_stream_props(
|
|
cap,
|
|
width=kwargs.get("width"),
|
|
height=kwargs.get("height"),
|
|
fps=kwargs.get("fps"),
|
|
fourcc=kwargs.get("fourcc"),
|
|
)
|
|
_warmup_capture(cap)
|
|
return cap, "FAKE"
|
|
|
|
monkeypatch.setattr("orisys.capture.source.open_opencv_capture", fake_open)
|
|
src = CaptureSource(CaptureConfig(source=0, width=1920, height=1080, fourcc="MJPG"))
|
|
src.open()
|
|
assert cap.fourcc == MJPG
|