diff --git a/examples/basic_pipeline.py b/examples/basic_pipeline.py index 71c5624..03ea21c 100644 --- a/examples/basic_pipeline.py +++ b/examples/basic_pipeline.py @@ -28,12 +28,9 @@ def main(): # cal_path 为标定文件路径;每台设备建议使用独立的标定文件 # 输入既可以是摄像头编号,也可以是视频文件路径 - try: - input = int(args.video) - is_camera = True - sensor = orisys.Sensor(input, config_name=args.config, verbose=args.verbose) - except: - is_camera = False + if args.video.isdigit(): + sensor = orisys.Sensor(int(args.video), config_name=args.config, verbose=args.verbose) + else: sensor = orisys.Sensor(args.video, config_name=args.config, verbose=args.verbose) print("\n按键说明:'q' 退出程序,'r' 重置追踪器。") diff --git a/sensors/mini/force_calibration.json b/sensors/mini/force_calibration.json deleted file mode 100644 index 015fc07..0000000 --- a/sensors/mini/force_calibration.json +++ /dev/null @@ -1,30 +0,0 @@ -{ - "schema_version": 1, - "unit": "N", - "channels": { - "normal": { - "method": "piecewise_linear", - "raw_key": "fnormal", - "extrapolate": true, - "clamp_output_min": 0.0, - "points": [ - [1000.0, 0.0], - [10000.0, 0.377], - [30000.0, 1.377], - [62000.0, 2.377] - ] - }, - "shear": { - "method": "piecewise_linear", - "input": "components", - "signed": true, - "extrapolate": true, - "clamp_output_min": 0.0, - "points": [ - [1000.0, 0.0], - [36000.0, 1.19], - [68000.0, 2.06] - ] - } - } -} diff --git a/src/orisys/capture/__init__.py b/src/orisys/capture/__init__.py index c1f1015..a2f8148 100644 --- a/src/orisys/capture/__init__.py +++ b/src/orisys/capture/__init__.py @@ -1,14 +1,10 @@ -"""Capture subsystem: sources, buffering, reconnect, discovery, multi-camera.""" +"""Capture subsystem: OpenCV source, optional background thread, discovery.""" -from .buffer import FrameBuffer, FramePacket from .config import ( - MAX_QUEUE_SIZE, MAX_SIMULTANEOUS_CAPTURES, CaptureConfig, CaptureHealth, - CaptureMode, CaptureStatus, - OverflowPolicy, ReconnectPolicy, ) from .discovery import ( @@ -19,31 +15,25 @@ from .discovery import ( list_cameras, pick_default_camera, ) -from .manager import CaptureManager from .source import ( CaptureSource, + CaptureWorker, is_live_camera, normalize_source, + open_capture, open_opencv_capture, ) -from .worker import CaptureWorker, open_capture __all__ = [ - "MAX_QUEUE_SIZE", "MAX_SIMULTANEOUS_CAPTURES", "DEFAULT_MAX_CAMERA_INDEX", "DEFAULT_LAPTOP_WEBCAM_PATTERNS", "CaptureConfig", "CaptureHealth", - "CaptureMode", "CaptureStatus", - "OverflowPolicy", "ReconnectPolicy", - "FrameBuffer", - "FramePacket", "CaptureSource", "CaptureWorker", - "CaptureManager", "CameraInfo", "CameraFilter", "list_cameras", diff --git a/src/orisys/capture/buffer.py b/src/orisys/capture/buffer.py deleted file mode 100644 index c0b2895..0000000 --- a/src/orisys/capture/buffer.py +++ /dev/null @@ -1,127 +0,0 @@ -"""Bounded frame transport for capture workers.""" - -from __future__ import annotations - -import threading -import time -from collections import deque -from dataclasses import dataclass -from typing import Optional - -import numpy as np - -from .config import CaptureMode, OverflowPolicy - - -@dataclass -class FramePacket: - """One captured frame with sequencing metadata.""" - - frame: np.ndarray - seq: int - timestamp: float - - -class FrameBuffer: - """Thread-safe frame buffer supporting latest-only and bounded queue modes. - - * ``latest``: always overwrite the single published frame (low latency). - * ``queue``: FIFO up to ``maxsize``; overflow controlled by ``OverflowPolicy``. - """ - - def __init__( - self, - *, - mode: CaptureMode = CaptureMode.LATEST, - maxsize: int = 1, - overflow: OverflowPolicy = OverflowPolicy.DROP_OLDEST, - ): - self.mode = CaptureMode(mode) if isinstance(mode, str) else mode - self.maxsize = max(1, int(maxsize)) - self.overflow = OverflowPolicy(overflow) if isinstance(overflow, str) else overflow - self._lock = threading.Lock() - self._cond = threading.Condition(self._lock) - self._latest: Optional[FramePacket] = None - self._queue: deque[FramePacket] = deque() - self.dropped_frames = 0 - self._seq = 0 - - def clear(self) -> None: - with self._cond: - self._latest = None - self._queue.clear() - - def publish(self, frame: np.ndarray, *, timestamp: Optional[float] = None) -> int: - """Publish a frame. Returns assigned sequence number.""" - ts = time.perf_counter() if timestamp is None else float(timestamp) - with self._cond: - self._seq += 1 - packet = FramePacket(frame=frame, seq=self._seq, timestamp=ts) - if self.mode == CaptureMode.LATEST: - self._latest = packet - self._cond.notify_all() - return packet.seq - - if len(self._queue) >= self.maxsize: - if self.overflow == OverflowPolicy.DROP_NEWEST: - self.dropped_frames += 1 - return packet.seq - if self.overflow == OverflowPolicy.BLOCK: - while len(self._queue) >= self.maxsize: - self._cond.wait(timeout=0.05) - # If still full after wait, drop oldest to avoid deadlock if - # consumer stopped; count as dropped. - if len(self._queue) >= self.maxsize: - self._queue.popleft() - self.dropped_frames += 1 - break - else: - # drop_oldest - self._queue.popleft() - self.dropped_frames += 1 - self._queue.append(packet) - self._latest = packet - self._cond.notify_all() - return packet.seq - - def read_latest(self, *, copy: bool = True, wait_s: float = 0.0) -> tuple[bool, Optional[np.ndarray], int]: - """Return the newest frame without consuming a queue slot.""" - deadline = time.perf_counter() + max(0.0, wait_s) - with self._cond: - while self._latest is None: - remaining = deadline - time.perf_counter() - if remaining <= 0: - return False, None, 0 - self._cond.wait(timeout=remaining) - packet = self._latest - frame = packet.frame.copy() if copy else packet.frame - return True, frame, packet.seq - - def read_next(self, *, copy: bool = True, wait_s: float = 0.0) -> tuple[bool, Optional[np.ndarray], int]: - """Pop the next queued frame (queue mode) or return latest (latest mode).""" - if self.mode == CaptureMode.LATEST: - return self.read_latest(copy=copy, wait_s=wait_s) - - deadline = time.perf_counter() + max(0.0, wait_s) - with self._cond: - while not self._queue: - remaining = deadline - time.perf_counter() - if remaining <= 0: - return False, None, 0 - self._cond.wait(timeout=remaining) - packet = self._queue.popleft() - self._cond.notify_all() - frame = packet.frame.copy() if copy else packet.frame - return True, frame, packet.seq - - @property - def last_seq(self) -> int: - with self._lock: - return self._seq - - @property - def pending(self) -> int: - with self._lock: - if self.mode == CaptureMode.LATEST: - return 0 if self._latest is None else 1 - return len(self._queue) diff --git a/src/orisys/capture/config.py b/src/orisys/capture/config.py index 489d33b..536c00d 100644 --- a/src/orisys/capture/config.py +++ b/src/orisys/capture/config.py @@ -7,23 +7,6 @@ from enum import Enum from typing import Any, Optional, Sequence, Tuple, Union MAX_SIMULTANEOUS_CAPTURES = 10 -MAX_QUEUE_SIZE = 10 -DEFAULT_QUEUE_SIZE = 1 - - -class CaptureMode(str, Enum): - """Frame delivery mode for a live capture worker.""" - - LATEST = "latest" - QUEUE = "queue" - - -class OverflowPolicy(str, Enum): - """Behavior when a bounded queue is full.""" - - DROP_OLDEST = "drop_oldest" - DROP_NEWEST = "drop_newest" - BLOCK = "block" class CaptureHealth(str, Enum): @@ -83,11 +66,8 @@ class CaptureConfig: height: Optional[int] = None fps: Optional[float] = None fourcc: Optional[str] = "MJPG" - mode: CaptureMode = CaptureMode.LATEST - queue_size: int = DEFAULT_QUEUE_SIZE - overflow: OverflowPolicy = OverflowPolicy.DROP_OLDEST + background: bool = True reconnect: ReconnectPolicy = field(default_factory=ReconnectPolicy) - start_worker: bool = True verbose: bool = False # Optional discovery filters (used by list/open helpers, not open itself). name_include: Optional[Sequence[str]] = None @@ -97,19 +77,8 @@ class CaptureConfig: min_height: Optional[int] = None def __post_init__(self) -> None: - if isinstance(self.mode, str): - self.mode = CaptureMode(self.mode) - if isinstance(self.overflow, str): - self.overflow = OverflowPolicy(self.overflow) if not isinstance(self.reconnect, ReconnectPolicy): raise TypeError("reconnect must be a ReconnectPolicy") - size = int(self.queue_size) - if size < 1 or size > MAX_QUEUE_SIZE: - raise ValueError(f"queue_size must be in 1..{MAX_QUEUE_SIZE}, got {size}") - self.queue_size = size - if self.mode == CaptureMode.LATEST: - # Latest mode keeps only one published frame regardless of queue_size. - self.queue_size = 1 @classmethod def from_camera_dict( @@ -117,12 +86,17 @@ class CaptureConfig: source: SourceType, camera: dict[str, Any], *, - mode: CaptureMode | str = CaptureMode.LATEST, reconnect: Optional[ReconnectPolicy] = None, verbose: bool = False, **kwargs: Any, ) -> "CaptureConfig": """Build config from a sensor JSON ``camera`` section.""" + # Accept deprecated queue/worker kwargs from older callers without error. + kwargs.pop("mode", None) + kwargs.pop("queue_size", None) + kwargs.pop("overflow", None) + if "start_worker" in kwargs: + kwargs.setdefault("background", bool(kwargs.pop("start_worker"))) res = camera.get("resolution_raw") or [None, None] width = int(res[0]) if res and res[0] else None height = int(res[1]) if res and res[1] else None @@ -132,7 +106,6 @@ class CaptureConfig: width=width, height=height, fps=float(fps) if fps is not None else None, - mode=mode, reconnect=reconnect or ReconnectPolicy(), verbose=verbose, **kwargs, diff --git a/src/orisys/capture/manager.py b/src/orisys/capture/manager.py deleted file mode 100644 index 3c2b09d..0000000 --- a/src/orisys/capture/manager.py +++ /dev/null @@ -1,122 +0,0 @@ -"""Manage up to N simultaneous capture sources.""" - -from __future__ import annotations - -from typing import Dict, List, Optional, Sequence - -from .config import ( - MAX_SIMULTANEOUS_CAPTURES, - CaptureConfig, - CaptureStatus, - SourceType, -) -from .source import normalize_source -from .worker import CaptureWorker, open_capture - - -class CaptureManager: - """Own multiple ``CaptureWorker`` instances (default max 10). - - Each camera is isolated: one failed USB device does not stop the others. - Hardware USB bandwidth may still limit how many high-res streams work in - practice even when this manager allows up to ``max_sources`` software slots. - """ - - def __init__(self, *, max_sources: int = MAX_SIMULTANEOUS_CAPTURES): - if max_sources < 1 or max_sources > MAX_SIMULTANEOUS_CAPTURES: - raise ValueError( - f"max_sources must be in 1..{MAX_SIMULTANEOUS_CAPTURES}, got {max_sources}" - ) - self.max_sources = int(max_sources) - self._workers: Dict[str, CaptureWorker] = {} - - def __len__(self) -> int: - return len(self._workers) - - def __contains__(self, key: SourceType) -> bool: - return self._key(key) in self._workers - - @staticmethod - def _key(source: SourceType) -> str: - return str(normalize_source(source)) - - @property - def sources(self) -> List[str]: - return list(self._workers.keys()) - - def open(self, config: CaptureConfig, *, factory=None) -> CaptureWorker: - """Open one capture; raises if capacity exceeded or source already open.""" - key = self._key(config.source) - if key in self._workers: - raise RuntimeError(f"Capture already open for source {key!r}") - if len(self._workers) >= self.max_sources: - raise RuntimeError( - f"CaptureManager at capacity ({self.max_sources}). " - "Close a source before opening another." - ) - worker = open_capture(config, factory=factory) - self._workers[key] = worker - return worker - - def open_many( - self, - configs: Sequence[CaptureConfig], - *, - factory=None, - ) -> List[CaptureWorker]: - """Open several captures atomically (rolls back on failure).""" - if len(configs) + len(self._workers) > self.max_sources: - raise RuntimeError( - f"Opening {len(configs)} sources would exceed capacity " - f"{self.max_sources} (currently {len(self._workers)} open)." - ) - opened: List[CaptureWorker] = [] - try: - for cfg in configs: - opened.append(self.open(cfg, factory=factory)) - except Exception: - for worker in opened: - key = self._key(worker.source.config.source) - worker.close() - self._workers.pop(key, None) - raise - return opened - - def get(self, source: SourceType) -> Optional[CaptureWorker]: - return self._workers.get(self._key(source)) - - def read_latest(self, source: SourceType, *, wait_s: float = 0.0): - worker = self.get(source) - if worker is None: - return False, None - return worker.buffer.read_latest(wait_s=wait_s)[:2] - - def read_latest_all(self, *, wait_s: float = 0.0) -> Dict[str, tuple]: - """Return ``{source_key: (ok, frame)}`` for every open capture.""" - out: Dict[str, tuple] = {} - for key, worker in self._workers.items(): - ok, frame, _seq = worker.buffer.read_latest(wait_s=wait_s) - out[key] = (ok, frame) - return out - - def status_all(self) -> Dict[str, CaptureStatus]: - return {key: worker.source.status for key, worker in self._workers.items()} - - def close(self, source: SourceType) -> None: - key = self._key(source) - worker = self._workers.pop(key, None) - if worker is not None: - worker.close() - - def close_all(self) -> None: - keys = list(self._workers.keys()) - for key in keys: - worker = self._workers.pop(key, None) - if worker is not None: - worker.close() - - def __enter__(self) -> "CaptureManager": - return self - - def __exit__(self, exc_type, exc, tb) -> None: - self.close_all() diff --git a/src/orisys/capture/source.py b/src/orisys/capture/source.py index e38feea..6e16983 100644 --- a/src/orisys/capture/source.py +++ b/src/orisys/capture/source.py @@ -1,9 +1,10 @@ -"""OpenCV-backed capture source with optional reconnect.""" +"""OpenCV-backed capture source with optional background thread and reconnect.""" from __future__ import annotations import os import sys +import threading import time from typing import Any, Optional, Protocol, runtime_checkable @@ -28,6 +29,7 @@ class VideoCaptureLike(Protocol): def release(self) -> None: ... def get(self, propId: int) -> float: ... def set(self, propId: int, value: float) -> bool: ... + def grab(self) -> bool: ... def normalize_source(source: SourceType | None) -> SourceType: @@ -40,7 +42,6 @@ def normalize_source(source: SourceType | None) -> SourceType: return 0 if stripped.isdigit(): return int(stripped) - # Only strip trailing separators (Windows path quirk); keep leading '/'. return stripped.rstrip("/\\") or stripped return source @@ -69,12 +70,10 @@ def _backend_candidates(preferred: str = "auto") -> list[tuple[str, int]]: elif preferred in ("v4l2", "v4l") and sys.platform.startswith("linux"): add("V4L2", "CAP_V4L2") elif preferred not in ("auto", "any", ""): - # Unknown preference: fall through to platform defaults. pass if not candidates: if sys.platform == "win32": - # Prefer DSHOW to avoid MSMF HW-transform stalls on many USB cameras. add("DSHOW", "CAP_DSHOW") add("MSMF", "CAP_MSMF") elif sys.platform.startswith("linux"): @@ -101,6 +100,45 @@ def _decode_fourcc(value: float) -> str: return "".join(chr(b) if 32 <= b <= 126 else "." for b in bytes_) +def _read_stream_meta(cap: VideoCaptureLike) -> tuple[int, int, float, str]: + width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH) or 0) + height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0) + fps = float(cap.get(cv2.CAP_PROP_FPS) or 0.0) + fourcc = _decode_fourcc(cap.get(cv2.CAP_PROP_FOURCC) or 0) + return width, height, fps, fourcc + + +def _apply_stream_props( + cap: VideoCaptureLike, + *, + width: Optional[int], + height: Optional[int], + fps: Optional[float], + fourcc: Optional[str], +) -> None: + """Apply live-camera stream properties (width/height before fourcc on DSHOW).""" + code = _fourcc_code(fourcc) + if width is not None: + cap.set(cv2.CAP_PROP_FRAME_WIDTH, float(width)) + if height is not None: + cap.set(cv2.CAP_PROP_FRAME_HEIGHT, float(height)) + if code is not None: + cap.set(cv2.CAP_PROP_FOURCC, code) + if fps is not None: + cap.set(cv2.CAP_PROP_FPS, float(fps)) + # DSHOW often resets fourcc after size negotiation. + if code is not None: + cap.set(cv2.CAP_PROP_FOURCC, code) + + +def _warmup_capture(cap: VideoCaptureLike, *, grabs: int = 3) -> None: + for _ in range(max(0, grabs)): + try: + cap.grab() + except Exception: + break + + def open_opencv_capture( source: SourceType, *, @@ -115,13 +153,13 @@ def open_opencv_capture( Returns ``(capture, backend_name)``. """ - # Avoid MSMF HW transform stalls on Windows camera capture. if sys.platform == "win32": os.environ.setdefault("OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS", "0") source = normalize_source(source) live = is_live_camera(source) last_error = "Failed to open video source." + requested_fourcc = (fourcc or "").strip().upper() backends = _backend_candidates(backend) if live else [("ANY", cv2.CAP_ANY)] for backend_name, backend_api in backends: @@ -138,22 +176,28 @@ def open_opencv_capture( last_error = f"Backend {backend_name} failed to open {source!r}" continue if live and apply_stream_props: - code = _fourcc_code(fourcc) - if code is not None: - cap.set(cv2.CAP_PROP_FOURCC, code) - if fps is not None: - cap.set(cv2.CAP_PROP_FPS, float(fps)) - if width is not None: - cap.set(cv2.CAP_PROP_FRAME_WIDTH, float(width)) - if height is not None: - cap.set(cv2.CAP_PROP_FRAME_HEIGHT, float(height)) + _apply_stream_props( + cap, + width=width, + height=height, + fps=fps, + fourcc=fourcc, + ) + _warmup_capture(cap) + _, _, _, negotiated = _read_stream_meta(cap) + if requested_fourcc and negotiated.upper() != requested_fourcc: + # Retry once: some drivers need a second fourcc set after warmup. + code = _fourcc_code(fourcc) + if code is not None: + cap.set(cv2.CAP_PROP_FOURCC, code) + _warmup_capture(cap, grabs=2) return cap, backend_name raise RuntimeError(last_error) class CaptureSource: - """Owns a video capture handle, health state, and reconnect logic. + """Owns a video capture handle, optional background reader, and reconnect logic. Optionally accepts an injected ``factory`` for tests:: @@ -172,7 +216,13 @@ class CaptureSource: self._backend_name = "" self._status = CaptureStatus(source=normalize_source(config.source)) self._reconnect_attempt = 0 - self._lock = __import__("threading").Lock() + self._lock = threading.Lock() + self._frame_lock = threading.Lock() + self._frame_cond = threading.Condition(self._frame_lock) + self._latest_frame: Optional[np.ndarray] = None + self._latest_seq = 0 + self._thread: Optional[threading.Thread] = None + self._background_running = False @staticmethod def _default_factory(config: CaptureConfig) -> tuple[VideoCaptureLike, str]: @@ -186,6 +236,15 @@ class CaptureSource: apply_stream_props=is_live_camera(normalize_source(config.source)), ) + # Compatibility with older CaptureWorker API. + @property + def source(self) -> "CaptureSource": + return self + + @property + def running(self) -> bool: + return self._background_running + @property def status(self) -> CaptureStatus: with self._lock: @@ -240,12 +299,27 @@ class CaptureSource: f"OpenCV negotiated stream: FOURCC='{st.fourcc}', " f"size={st.width}x{st.height}, fps={st.fps:.2f}, backend={st.backend_name}" ) + requested = (self.config.fourcc or "").strip().upper() + if ( + requested + and is_live_camera(normalize_source(self.config.source)) + and st.fourcc.upper() != requested + ): + print( + f"Warning: requested FOURCC '{self.config.fourcc}' but camera " + f"negotiated '{st.fourcc}'." + ) def release(self) -> None: + self.stop_background() with self._lock: self._release_unlocked() self._status.health = CaptureHealth.CLOSED + def close(self) -> None: + """Stop background thread and release the capture handle.""" + self.release() + def _release_unlocked(self) -> None: if self._cap is not None: try: @@ -253,21 +327,32 @@ class CaptureSource: except Exception: pass self._cap = None + with self._frame_cond: + self._latest_frame = None + self._latest_seq = 0 def _refresh_stream_meta_unlocked(self) -> None: cap = self._cap if cap is None: return try: - self._status.width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH) or 0) - self._status.height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0) - self._status.fps = float(cap.get(cv2.CAP_PROP_FPS) or 0.0) - self._status.fourcc = _decode_fourcc(cap.get(cv2.CAP_PROP_FOURCC) or 0) + width, height, fps, fourcc = _read_stream_meta(cap) + self._status.width = width + self._status.height = height + self._status.fps = fps + self._status.fourcc = fourcc except Exception: pass + def warmup_frames(self, count: int = 3) -> None: + """Discard frames so the driver/format stabilizes (sync reads only).""" + if self._background_running: + return + for _ in range(max(0, int(count))): + self.read_raw() + def read_raw(self) -> tuple[bool, Optional[np.ndarray]]: - """Read one frame from the underlying capture (no reconnect).""" + """Read one frame directly from the underlying capture (no reconnect).""" cap = self._cap if cap is None: return False, None @@ -289,6 +374,92 @@ class CaptureSource: self._status.health = CaptureHealth.CONNECTED return True, frame + def read_latest( + self, + *, + copy: bool = True, + wait_s: float = 0.0, + ) -> tuple[bool, Optional[np.ndarray]]: + """Return the newest frame from the background thread, or read sync.""" + if not self._background_running: + ok, frame = self.read_raw() + if ok and copy and frame is not None: + return ok, frame.copy() + return ok, frame + + deadline = time.perf_counter() + max(0.0, wait_s) + with self._frame_cond: + while self._latest_frame is None: + remaining = deadline - time.perf_counter() + if remaining <= 0: + return False, None + self._frame_cond.wait(timeout=remaining) + frame = self._latest_frame.copy() if copy else self._latest_frame + return True, frame + + def read(self, *, copy: bool = True, wait_s: float = 1.0) -> tuple[bool, Optional[np.ndarray]]: + """Compat alias used by finger runtime and older worker callers.""" + wait = wait_s if wait_s > 0 else 1.0 + return self.read_latest(copy=copy, wait_s=wait) + + def start_background(self) -> None: + """Start a daemon thread that keeps only the latest frame.""" + if self._thread is not None: + return + if not self.is_open(): + self.open() + self._background_running = True + self._thread = threading.Thread( + target=self._background_loop, + daemon=True, + name="orisys-capture", + ) + self._thread.start() + if self.config.verbose: + print("Capture background thread started.") + + def start(self) -> None: + """Compat alias for ``start_background``.""" + self.start_background() + + def stop_background(self, timeout: float = 1.0) -> None: + self._background_running = False + if self._thread is not None: + self._thread.join(timeout=timeout) + self._thread = None + with self._frame_cond: + self._latest_frame = None + + def stop(self, timeout: float = 1.0) -> None: + """Compat alias for ``stop_background``.""" + self.stop_background(timeout=timeout) + + def _background_loop(self) -> None: + while self._background_running: + if self.should_reconnect(): + if not self.reconnect(): + time.sleep(0.05) + if self.status.health == CaptureHealth.FAILED: + time.sleep(0.2) + continue + + ok, frame = self.read_raw() + if not ok or frame is None: + if self.should_reconnect(): + continue + time.sleep(0.001) + continue + + with self._frame_cond: + if self._latest_frame is not None: + with self._lock: + self._status.dropped_frames += 1 + self._latest_seq += 1 + self._latest_frame = frame + with self._lock: + self._status.last_seq = self._latest_seq + self._frame_cond.notify_all() + def note_published(self, seq: int) -> None: with self._lock: self._status.last_seq = int(seq) @@ -334,7 +505,10 @@ class CaptureSource: delay = policy.next_backoff(attempt) if self.config.verbose: - print(f"Reconnecting to {self.config.source!r} (attempt {attempt + 1}, wait {delay:.2f}s)...") + print( + f"Reconnecting to {self.config.source!r} " + f"(attempt {attempt + 1}, wait {delay:.2f}s)..." + ) time.sleep(delay) try: self.open() @@ -347,3 +521,32 @@ class CaptureSource: else: self._status.health = CaptureHealth.RECONNECTING return False + + +# Backward-compatible alias for one release cycle. +CaptureWorker = CaptureSource + + +def open_capture( + config: CaptureConfig, + *, + factory=None, + start_background: Optional[bool] = None, + start_worker: Optional[bool] = None, +) -> CaptureSource: + """Open a capture source and optionally start its background thread. + + For live cameras, the background thread starts by default (``config.background``). + File/video sources stay synchronous unless ``start_background=True``. + """ + source = CaptureSource(config, factory=factory) + source.open() + should_start = config.background + if start_background is not None: + should_start = start_background + elif start_worker is not None: + should_start = start_worker + live = is_live_camera(normalize_source(config.source)) + if should_start and live: + source.start_background() + return source diff --git a/src/orisys/capture/worker.py b/src/orisys/capture/worker.py deleted file mode 100644 index 01d2a5e..0000000 --- a/src/orisys/capture/worker.py +++ /dev/null @@ -1,124 +0,0 @@ -"""Background capture worker that feeds a FrameBuffer.""" - -from __future__ import annotations - -import threading -import time -from typing import Optional - -from .buffer import FrameBuffer -from .config import CaptureConfig, CaptureHealth, CaptureMode -from .source import CaptureSource, is_live_camera, normalize_source - - -class CaptureWorker: - """One thread per live camera: read → publish → optional reconnect.""" - - def __init__( - self, - source: CaptureSource, - buffer: Optional[FrameBuffer] = None, - ): - self.source = source - self.config: CaptureConfig = source.config - self.buffer = buffer or FrameBuffer( - mode=self.config.mode, - maxsize=self.config.queue_size, - overflow=self.config.overflow, - ) - self._thread: Optional[threading.Thread] = None - self._running = False - self._last_consumed_seq = -1 - - @property - def running(self) -> bool: - return self._running - - def start(self) -> None: - if self._thread is not None: - return - if not self.source.is_open(): - self.source.open() - self._running = True - self._thread = threading.Thread(target=self._loop, daemon=True, name="orisys-capture") - self._thread.start() - if self.config.verbose: - mode = self.config.mode.value if isinstance(self.config.mode, CaptureMode) else self.config.mode - print(f"Capture worker started (mode={mode}).") - - def stop(self, timeout: float = 1.0) -> None: - self._running = False - if self._thread is not None: - self._thread.join(timeout=timeout) - self._thread = None - self.buffer.clear() - - def close(self) -> None: - self.stop() - self.source.release() - - def read(self, *, copy: bool = True, wait_s: float = 1.0) -> tuple[bool, object]: - """Read according to configured mode; mirrors Sensor._read_frame semantics.""" - if not self._running: - ok, frame = self.source.read_raw() - return ok, frame - - if self.config.mode == CaptureMode.QUEUE: - ok, frame, seq = self.buffer.read_next(copy=copy, wait_s=wait_s) - else: - # Startup race: wait briefly for first frame. - wait = wait_s if wait_s > 0 else 1.0 - ok, frame, seq = self.buffer.read_latest(copy=copy, wait_s=wait) - if ok: - self._last_consumed_seq = seq - return ok, frame - - def _loop(self) -> None: - while self._running: - if self.source.should_reconnect(): - if not self.source.reconnect(): - # Back off a bit before retrying the should_reconnect check. - time.sleep(0.05) - if self.source.status.health == CaptureHealth.FAILED: - time.sleep(0.2) - continue - - ok, frame = self.source.read_raw() - if not ok or frame is None: - if self.source.should_reconnect(): - continue - time.sleep(0.001) - continue - - before_dropped = self.buffer.dropped_frames - seq = self.buffer.publish(frame) - after_dropped = self.buffer.dropped_frames - if after_dropped > before_dropped: - self.source.note_dropped(after_dropped - before_dropped) - self.source.note_published(seq) - - -def open_capture( - config: CaptureConfig, - *, - factory=None, - start_worker: Optional[bool] = None, -) -> CaptureWorker: - """Open a capture source and optionally start its worker thread. - - For live cameras, the worker starts by default (``config.start_worker``). - File/video sources stay synchronous unless ``start_worker=True``. - """ - source = CaptureSource(config, factory=factory) - source.open() - buffer = FrameBuffer( - mode=config.mode, - maxsize=config.queue_size, - overflow=config.overflow, - ) - worker = CaptureWorker(source, buffer) - should_start = config.start_worker if start_worker is None else start_worker - live = is_live_camera(normalize_source(config.source)) - if should_start and live: - worker.start() - return worker diff --git a/src/orisys/finger/runtime.py b/src/orisys/finger/runtime.py index be92605..80be9f6 100644 --- a/src/orisys/finger/runtime.py +++ b/src/orisys/finger/runtime.py @@ -6,7 +6,7 @@ import numpy as np from ..capture import ( CaptureConfig, - CaptureWorker, + CaptureSource, ReconnectPolicy, is_live_camera, normalize_source, @@ -35,23 +35,21 @@ def normalize_video_source(video): def open_source(video, config, *, auto_reconnect: bool = False, verbose: bool = False): - """Open a capture worker for finger pipelines (compat: returns worker.cap-like via ``.cap``). + """Open a capture source for finger pipelines (compat: returns ``.cap``-like handle). Historically returned a bare ``cv2.VideoCapture``. Callers that need the raw - OpenCV handle can use ``worker.source.cap``; prefer ``worker.read()``. + OpenCV handle can use ``source.cap``; prefer ``source.read()``. """ capture_config = CaptureConfig.from_camera_dict( normalize_video_source(video), config.get("camera", {}), reconnect=ReconnectPolicy(enabled=auto_reconnect), verbose=verbose, - start_worker=False, + background=False, ) - worker = open_capture(capture_config, start_worker=False) - # Preserve prior attribute access: treat returned object like VideoCapture for - # simple .read()/.release()/.isOpened() callers, while exposing capture worker. - cap = worker.source.cap - cap._orisys_worker = worker # type: ignore[attr-defined] + source = open_capture(capture_config, start_background=False) + cap = source.cap + cap._orisys_worker = source # type: ignore[attr-defined] return cap @@ -220,7 +218,7 @@ class FingerRuntime: tactile: TactileProcessor, frame_processor: FrameUnwarper, capture, - capture_worker: CaptureWorker | None = None, + capture_source: CaptureSource | None = None, ): self.geometry = geometry self.domain = domain @@ -228,7 +226,9 @@ class FingerRuntime: self.sensor = tactile self.frame_processor = frame_processor self.capture = capture - self.capture_worker = capture_worker + self.capture_source = capture_source + # Deprecated alias. + self.capture_worker = capture_source @classmethod def open( @@ -248,6 +248,7 @@ class FingerRuntime: capture_mode: str = "latest", ) -> "FingerRuntime": """Open a live finger runtime from export bundle, config, and video source.""" + del capture_mode # deprecated; latest-frame capture only if camera_id is not None: video = camera_id video = normalize_video_source(video) @@ -280,19 +281,16 @@ class FingerRuntime: capture_config = CaptureConfig.from_camera_dict( video, geometry.config.get("camera", {}), - mode=capture_mode, reconnect=ReconnectPolicy(enabled=auto_reconnect), verbose=verbose, - start_worker=False, + background=False, ) - capture_worker = open_capture(capture_config, start_worker=False) - if not capture_worker.source.is_open(): + capture_source = open_capture(capture_config, start_background=False) + if not capture_source.is_open(): raise RuntimeError("Cannot open camera/video source") - # Start latest-frame worker for live cameras so processing speed does not - # stall the device buffer. if is_live_camera(video): - capture_worker.start() - capture = capture_worker.source.cap + capture_source.start_background() + capture = capture_source.cap sensor = Sensor("external", config_name=config, verbose=verbose, mirror=mirror) frame_processor = FrameUnwarper( @@ -310,7 +308,7 @@ class FingerRuntime: tactile=tactile, frame_processor=frame_processor, capture=capture, - capture_worker=capture_worker, + capture_source=capture_source, ) @property @@ -318,10 +316,10 @@ class FingerRuntime: return self.frame_processor.valid_mask def read_source(self): - if self.capture_worker is not None and self.capture_worker.running: - return self.capture_worker.read(copy=True, wait_s=1.0) - if self.capture_worker is not None: - return self.capture_worker.source.read_raw() + if self.capture_source is not None and self.capture_source.running: + return self.capture_source.read(copy=True, wait_s=1.0) + if self.capture_source is not None: + return self.capture_source.read_raw() return self.capture.read() def process_frame(self, frame, *, color=False): @@ -355,8 +353,9 @@ class FingerRuntime: return ok, unrolled def close(self): - if self.capture_worker is not None: - self.capture_worker.close() + if self.capture_source is not None: + self.capture_source.close() + self.capture_source = None self.capture_worker = None self.capture = None elif self.capture is not None: diff --git a/src/orisys/sensor.py b/src/orisys/sensor.py index 55d46bf..5632c0a 100644 --- a/src/orisys/sensor.py +++ b/src/orisys/sensor.py @@ -31,7 +31,7 @@ from .signal_processing import ( ) from .capture import ( CaptureConfig, - CaptureWorker, + CaptureSource, ReconnectPolicy, is_live_camera, normalize_source, @@ -76,7 +76,8 @@ class Sensor: config_name="prototype", cal_path=None, override=False, box=None, mirror=True, backend='auto', capture_config: Optional[CaptureConfig] = None, auto_reconnect: bool = False, capture_mode: str = "latest", - capture_queue_size: int = 1, capture_overflow: str = "drop_oldest"): + capture_queue_size: int = 1, capture_overflow: str = "drop_oldest", + **_deprecated_capture_kwargs): """ Initialize Sensor with video source and configuration. @@ -96,9 +97,9 @@ class Sensor: "cpu": Use CPU for computation capture_config: Optional ``CaptureConfig`` overriding capture kwargs below. auto_reconnect: Enable USB disconnect auto-reconnect for live cameras. - capture_mode: ``"latest"`` (low latency, default) or ``"queue"``. - capture_queue_size: Bounded queue size for queue mode (1..10). - capture_overflow: ``drop_oldest`` | ``drop_newest`` | ``block``. + capture_mode: Deprecated; ignored (latest-frame capture only). + capture_queue_size: Deprecated; ignored. + capture_overflow: Deprecated; ignored. """ self.CUPY_AVAILABLE = check_cupy_available() @@ -120,9 +121,6 @@ class Sensor: self.mirror = mirror self._capture_config_override = capture_config self._auto_reconnect = bool(auto_reconnect) - self._capture_mode = capture_mode - self._capture_queue_size = int(capture_queue_size) - self._capture_overflow = capture_overflow # Load configuration self.__config_cam = None @@ -231,7 +229,7 @@ class Sensor: # self._av_stream = None # self._av_decoder = None self.cap = None # compat alias for _capture_worker.source.cap - self._capture_worker: Optional[CaptureWorker] = None + self._capture_source: Optional[CaptureSource] = None self._is_live_camera = ( is_live_camera(self.vid_src) and not self._external_only ) @@ -705,9 +703,9 @@ class Sensor: def disconnect(self): """Release resources and stop async threads.""" - if self._capture_worker is not None: - self._capture_worker.close() - self._capture_worker = None + if self._capture_source is not None: + self._capture_source.close() + self._capture_source = None self.cap = None if self.verbose: print("Sensor disconnected.") @@ -934,25 +932,23 @@ class Sensor: capture_config = CaptureConfig.from_camera_dict( self.vid_src, self.__config_cam, - mode=self._capture_mode, - queue_size=self._capture_queue_size, - overflow=self._capture_overflow, reconnect=ReconnectPolicy(enabled=self._auto_reconnect), verbose=self.verbose, - start_worker=False, + background=False, ) try: - self._capture_worker = open_capture(capture_config, start_worker=False) + self._capture_source = open_capture(capture_config, start_background=False) except Exception as exc: raise Exception("Failed to open video source.") from exc - self.cap = self._capture_worker.source.cap - # self._use_pyav = False + self.cap = self._capture_source.cap if not self._is_source_open(): raise Exception("Failed to open video source.") + self._capture_source.warmup_frames(3) + # Wait for first valid frame frame = None while frame is None: @@ -967,16 +963,24 @@ class Sensor: if box is not None: self.stitcher.isstitch = self.isstitch = False self.stitcher.boxes = box + self.stitch_valid = bool(self.isstitch) + img = self.frame_processor.process(frame) + self.initial_img = self._make_initial_gray(img) else: print("Override=True but no box provided.") else: self._find_stitch_boxes(frame) + if not self.stitch_valid: + raise RuntimeError( + f"Failed to detect {self._config_units} vision tiles for stitching. " + "Check camera index, lighting, and that MJPG 1080p is negotiated." + ) img = self.frame_processor.process(frame) self.initial_img = self._make_initial_gray(img) - # For live cameras, start a capture worker to avoid latency buildup. - if self._is_live_camera and self._capture_worker is not None: - self._capture_worker.start() + # For live cameras, start background capture to avoid latency buildup. + if self._is_live_camera and self._capture_source is not None: + self._capture_source.start_background() def _make_initial_gray(self, img_bgr): gray = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2GRAY) @@ -990,6 +994,9 @@ class Sensor: def _find_stitch_boxes(self, img_src, show=False): """Detect vision tile boxes with camera stabilization.""" n_boxes, v_mean0, timeout = 0, 0, 5 + boxes = [] + best_n_boxes = 0 + best_boxes = [] start_time = cv2.getTickCount() while n_boxes < self._config_units: @@ -1000,12 +1007,18 @@ class Sensor: if abs(v_mean - v_mean0) < 5: n_boxes, boxes = self.stitcher.find_stitch_boxes(frame, show=show) + if n_boxes > best_n_boxes: + best_n_boxes = n_boxes + best_boxes = boxes if (cv2.getTickCount() - start_time) / cv2.getTickFrequency() > timeout: print("Timeout finding vision units.") - self.stitcher.isstitch = self.isstitch = False break v_mean0 = v_mean + + if n_boxes < self._config_units and best_n_boxes >= self._config_units: + n_boxes = best_n_boxes + boxes = best_boxes self.n_boxes = n_boxes self.boxes = boxes @@ -1023,8 +1036,8 @@ class Sensor: # return True def _is_source_open(self): - if self._capture_worker is not None: - return self._capture_worker.source.is_open() + if self._capture_source is not None: + return self._capture_source.is_open() return False # Legacy PyAV / direct-cap fallbacks (unused after orisys.capture refactor). # if self._use_pyav: @@ -1032,13 +1045,13 @@ class Sensor: # return self.cap is not None and self.cap.isOpened() def _read_frame(self): - if self._capture_worker is not None and self._capture_worker.running: - return self._capture_worker.read(copy=True, wait_s=1.0) + if self._capture_source is not None and self._capture_source.running: + return self._capture_source.read_latest(copy=True, wait_s=1.0) return self._read_frame_raw() def _read_frame_raw(self): - if self._capture_worker is not None: - return self._capture_worker.source.read_raw() + if self._capture_source is not None: + return self._capture_source.read_raw() return False, None # Legacy PyAV / direct-cap fallbacks (unused after orisys.capture refactor). # if self._use_pyav: @@ -1058,9 +1071,14 @@ class Sensor: @property def capture_status(self): """Return capture health snapshot, or None in external-frame mode.""" - if self._capture_worker is None: + if self._capture_source is None: return None - return self._capture_worker.source.status + return self._capture_source.status + + @property + def _capture_worker(self): + """Deprecated alias for ``_capture_source``.""" + return self._capture_source # ========================================================================= diff --git a/tests/test_capture.py b/tests/test_capture.py index 9bd730c..179c8ec 100644 --- a/tests/test_capture.py +++ b/tests/test_capture.py @@ -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