orisys-tech/orisys-sdk
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

hotfix/camera not connected error

Browse Source
This commit is contained in:
ykcheungab
2026-09-02 21:37:40 +08:00
parent 32823a7666
commit 5be2d9759a
11 changed files with 427 additions and 677 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_capture.py
+115 -142
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import threading
import time
import cv2
import numpy as np
import pytest
@@ -14,35 +15,47 @@ from orisys.capture import (
CameraInfo,
CaptureConfig,
CaptureHealth,
CaptureManager,
CaptureMode,
CaptureSource,
FrameBuffer,
OverflowPolicy,
ReconnectPolicy,
list_cameras,
normalize_source,
open_capture,
pick_default_camera,
)
from orisys.capture.source import _apply_stream_props, _warmup_capture
from orisys.capture.discovery import DEFAULT_LAPTOP_WEBCAM_PATTERNS
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):
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
@@ -58,7 +71,6 @@ class FakeCap:
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]
@@ -67,12 +79,18 @@ class FakeCap:
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):
# 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
@@ -84,6 +102,19 @@ class FakeCap:
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
@@ -92,7 +123,6 @@ def _frame(val: int) -> np.ndarray:
def _factory_from_caps(caps):
"""Return a factory that hands out FakeCap instances in order."""
lock = threading.Lock()
state = {"i": 0}
@@ -107,11 +137,6 @@ def _factory_from_caps(caps):
return factory
# ---------------------------------------------------------------------------
# Source normalization / config
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"video,expected",
[
@@ -127,63 +152,42 @@ 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
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
# ---------------------------------------------------------------------------
# FrameBuffer
# ---------------------------------------------------------------------------
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_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_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), start_worker=False),
CaptureConfig(source=0, reconnect=ReconnectPolicy(enabled=False)),
factory=lambda cfg: (cap, "FAKE"),
)
src.open()
@@ -209,13 +213,12 @@ def test_auto_reconnect_after_failed_reads():
max_attempts=3,
)
src = CaptureSource(
CaptureConfig(source=0, reconnect=policy, start_worker=False, verbose=False),
CaptureConfig(source=0, reconnect=policy, 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()
@@ -232,7 +235,6 @@ def test_reconnect_disabled_does_not_reopen():
CaptureConfig(
source=0,
reconnect=ReconnectPolicy(enabled=False, fail_threshold=1),
start_worker=False,
),
factory=lambda cfg: (cap, "FAKE"),
)
@@ -244,7 +246,6 @@ def test_reconnect_disabled_does_not_reopen():
def test_reconnect_attempts_exhausted():
good = FakeCap([_frame(1)], fail_after=1)
state = {"n": 0}
def factory(_cfg):
@@ -262,7 +263,7 @@ def test_reconnect_attempts_exhausted():
max_attempts=2,
)
src = CaptureSource(
CaptureConfig(source=0, reconnect=policy, start_worker=False),
CaptureConfig(source=0, reconnect=policy),
factory=factory,
)
src.open()
@@ -274,37 +275,32 @@ def test_reconnect_attempts_exhausted():
assert src.status.health == CaptureHealth.FAILED
# ---------------------------------------------------------------------------
# CaptureWorker
# ---------------------------------------------------------------------------
def test_worker_latest_mode_publishes_frames():
def test_background_thread_publishes_latest_frame():
caps = [FakeCap([_frame(i) for i in range(5)])]
worker = open_capture(
src = open_capture(
CaptureConfig(
source=0,
mode=CaptureMode.LATEST,
start_worker=True,
background=True,
reconnect=ReconnectPolicy(enabled=False),
),
factory=_factory_from_caps(caps),
start_worker=True,
start_background=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)
ok, frame = src.read(wait_s=0.05)
if ok:
break
assert ok and frame is not None
assert src.running
finally:
worker.close()
src.close()
def test_worker_reconnect_in_background():
def test_background_reconnect():
first = FakeCap([_frame(1)], fail_after=2)
second = FakeCap([_frame(42) for _ in range(20)])
policy = ReconnectPolicy(
@@ -314,19 +310,19 @@ def test_worker_reconnect_in_background():
max_backoff_s=0.02,
stale_timeout_s=None,
)
worker = open_capture(
CaptureConfig(source=0, mode="latest", reconnect=policy, start_worker=True),
src = open_capture(
CaptureConfig(source=0, reconnect=policy, background=True),
factory=_factory_from_caps([first, second]),
start_worker=True,
start_background=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:
if src.status.reconnect_attempts > 0:
saw_reconnect = True
ok, frame = worker.read(wait_s=0.05)
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
@@ -334,65 +330,19 @@ def test_worker_reconnect_in_background():
assert saw_reconnect
assert saw_new
finally:
worker.close()
src.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_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():
@@ -447,10 +397,13 @@ def test_camera_info_label_includes_name():
def test_list_cameras_is_callable():
# Smoke: function import works; may return [] without hardware.
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
@@ -458,6 +411,26 @@ def test_sensor_external_mode_still_works():
try:
assert sensor._external_only
assert sensor.capture_status is None
assert sensor._capture_worker 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