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/scripts/example_uart.py b/scripts/example_uart.py new file mode 100644 index 0000000..77f5c82 --- /dev/null +++ b/scripts/example_uart.py @@ -0,0 +1,461 @@ +#!/usr/bin/env python3 +"""UART finger controller for Radxa / Dragon Q6A (sensor-side receiver). + +Combines ORY-FINGER processing (same path as ``examples/finger_pipeline.py``) +with a line-based UART command interface. A host (or ``uart_send.py``-style +transmitter) sends commands; this script replies and streams forces. + +Dependencies (beyond orisys): + pip install pyserial + +Default UART (Dragon Q6A): + port=/dev/ttyHS1 baud=115200 8N1 + +Commands (one line, case-insensitive): + search List cameras (/dev/video*) via orisys.capture.list_cameras + connect [source] Open FingerRuntime (camera index, /dev/videoN, or video file) + run Stream FNORMAL / FSHEARX / FSHEARY until stop (or EOF) + stop Stop force streaming; keep source connected + quit | exit Close runtime and serial, then exit + +Replies: + camera,,, + OK search + OK connect + OK run + force,,, + OK eof Video file ended (when --no-loop-video) + OK stop + OK quit + ERR + +Examples: + # Live camera over UART + python scripts/example_uart.py --port /dev/ttyHS1 --config finger + + # Offline debug from a recorded file (no camera needed) + python scripts/example_uart.py --port /dev/ttyHS1 --video /path/to/clip.mp4 + # then UART: connect (uses --video) or connect /path/to/clip.mp4 + # then UART: run +""" + +from __future__ import annotations + +import argparse +import os +import shlex +import sys +from pathlib import Path +from typing import Optional, Union + +os.environ.setdefault("OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS", "0") + +try: + import serial +except ImportError as exc: # pragma: no cover + raise SystemExit( + "pyserial is required for this script.\n" + " pip install pyserial\n" + f"Import error: {exc}" + ) from exc + +import cv2 +from orisys.capture import is_live_camera, list_cameras +from orisys.finger import FingerRuntime + +Source = Union[int, str] + + +def _parse_source(token: str) -> Optional[Source]: + """Accept camera index int, numeric string, V4L path, or filesystem path.""" + text = str(token).strip().strip('"').strip("'") + if not text: + return None + if text.isdigit() or (text.startswith("-") and text[1:].isdigit()): + return int(text) + return text + + +def _is_video_file(source: Source) -> bool: + """True when source is a filesystem video path (not a live camera).""" + if is_live_camera(source): + return False + if isinstance(source, int): + return False + path = Path(str(source)) + return path.is_file() or path.suffix.lower() in { + ".mp4", + ".avi", + ".mkv", + ".mov", + ".m4v", + ".wmv", + ".webm", + } + + +class UartFingerController: + """Serial command dispatcher around a single FingerRuntime.""" + + def __init__( + self, + ser: serial.Serial, + *, + config: str = "finger", + default_source: Source = 0, + max_camera_index: int = 10, + loop_video: bool = True, + show: bool = False, + verbose: bool = False, + ): + self.ser = ser + self.config = config + self.default_source = default_source + self.max_camera_index = max_camera_index + self.loop_video = loop_video + self.show = show + self.verbose = verbose + + self.runtime: Optional[FingerRuntime] = None + self._connected_source: Optional[Source] = None + self._last_search_source: Optional[Source] = None + self._streaming = False + self._should_quit = False + + def reply(self, line: str) -> None: + text = line.rstrip("\r\n") + "\n" + self.ser.write(text.encode("utf-8")) + self.ser.flush() + if self.verbose: + print(f"<< {text.rstrip()}", flush=True) + + def close_runtime(self) -> None: + self._streaming = False + if self.runtime is not None: + try: + self.runtime.close() + except Exception: # noqa: BLE001 + pass + self.runtime = None + self._connected_source = None + if self.show: + try: + cv2.destroyAllWindows() + except Exception: # noqa: BLE001 + pass + + def _resolve_connect_source(self, args: list[str]) -> Source: + if args: + # Join remaining tokens so unquoted paths with spaces still work. + joined = " ".join(args).strip() + parsed = _parse_source(joined) + if parsed is None: + raise ValueError("empty connect source") + return parsed + # Prefer --video default over last search so offline debug stays on file. + if isinstance(self.default_source, str) and not is_live_camera(self.default_source): + return self.default_source + if self._last_search_source is not None: + return self._last_search_source + return self.default_source + + def cmd_search(self, _args: list[str]) -> None: + try: + cameras = list_cameras(max_index=self.max_camera_index) + except Exception as exc: # noqa: BLE001 + self.reply(f"ERR search {exc}") + return + + self._last_search_source = None + for cam in cameras: + path = (cam.path or f"/dev/video{cam.index}").strip() or f"/dev/video{cam.index}" + size = f"{cam.width}x{cam.height}" + self.reply(f"camera,{cam.index},{path},{size}") + if self._last_search_source is None: + self._last_search_source = cam.index + self.reply(f"OK search {len(cameras)}") + + def cmd_connect(self, args: list[str]) -> None: + try: + source = self._resolve_connect_source(args) + except ValueError as exc: + self.reply(f"ERR connect {exc}") + return + + if isinstance(source, str) and not is_live_camera(source): + path = Path(source) + if not path.is_file(): + self.reply(f"ERR connect file not found: {source}") + return + + self._streaming = False + self.close_runtime() + try: + runtime = FingerRuntime.open( + config=self.config, + video=source, + verbose=self.verbose, + ) + runtime.warmup(color=False) + except Exception as exc: # noqa: BLE001 + self.reply(f"ERR connect {exc}") + return + + self.runtime = runtime + self._connected_source = source + kind = "file" if _is_video_file(source) else "camera" + self.reply(f"OK connect {source} ({kind})") + + def _rewind_video(self) -> bool: + """Seek connected file source back to frame 0. Return True on success.""" + if self.runtime is None or self._connected_source is None: + return False + if not _is_video_file(self._connected_source): + return False + cap = self.runtime.capture + if cap is None: + return False + try: + return bool(cap.set(cv2.CAP_PROP_POS_FRAMES, 0)) + except Exception: # noqa: BLE001 + return False + + def _poll_command_line(self, *, blocking: bool = True) -> Optional[str]: + """Read one command line if available; return None on timeout/empty.""" + if not blocking and self.ser.in_waiting <= 0: + return None + raw = self.ser.readline() + if not raw: + return None + try: + line = raw.decode("utf-8", errors="replace").strip() + except Exception: # noqa: BLE001 + return None + return line or None + + def cmd_run(self, _args: list[str]) -> None: + if self.runtime is None: + self.reply("ERR run not connected") + return + + self._streaming = True + self.reply("OK run") + tactile = self.runtime.tactile + is_file = self._connected_source is not None and _is_video_file(self._connected_source) + + while self._streaming and not self._should_quit: + ok, unrolled = self.runtime.grab_and_process(color=False) + if not ok: + if is_file and self.loop_video and self._rewind_video(): + if self.verbose: + print("video EOF: looping", flush=True) + continue + if is_file: + self.reply("OK eof") + else: + self.reply("ERR run grab failed") + self._streaming = False + break + + tactile.compute_deformation() + tactile.compute_contact() + fnormal, fshearx, fsheary = tactile.read_info( + tactile.info.FNORMAL, + tactile.info.FSHEARX, + tactile.info.FSHEARY, + ) + self.reply(f"force,{float(fnormal):.6f},{float(fshearx):.6f},{float(fsheary):.6f}") + + if self.show: + try: + import orisys + + arrows = orisys.util.draw_arrows( + unrolled, + tactile.flow, + threshold=0.4, + grid_spacing=10, + arrow_scale=1.0, + valid_mask=tactile.valid_mask, + ) + cv2.imshow("UV Unrolled", arrows) + key = cv2.waitKey(1) & 0xFF + if key == ord("q"): + self._should_quit = True + break + if key == ord("r"): + tactile.reset() + except Exception as exc: # noqa: BLE001 + if self.verbose: + print(f"show error: {exc}", file=sys.stderr) + + # Non-blocking command poll so force streaming is not delayed by UART timeout. + pending = self._poll_command_line(blocking=False) + if pending is not None: + if self.verbose: + print(f">> {pending}", flush=True) + if not self._dispatch(pending, allow_run=False): + break + + def cmd_stop(self, _args: list[str]) -> None: + self._streaming = False + self.reply("OK stop") + + def cmd_quit(self, _args: list[str]) -> None: + self._streaming = False + self._should_quit = True + self.reply("OK quit") + + def _dispatch(self, line: str, *, allow_run: bool = True) -> bool: + """Dispatch one command line. Return False if the main loop should exit.""" + try: + parts = shlex.split(line) + except ValueError: + parts = line.split() + if not parts: + return True + cmd = parts[0].lower() + args = parts[1:] + + if cmd == "search": + self.cmd_search(args) + elif cmd == "connect": + self.cmd_connect(args) + elif cmd == "run": + if not allow_run: + self.reply("ERR run already running") + else: + self.cmd_run(args) + elif cmd == "stop": + self.cmd_stop(args) + elif cmd in ("quit", "exit"): + self.cmd_quit(args) + return False + else: + self.reply(f"ERR unknown {cmd}") + return not self._should_quit + + def serve(self) -> None: + self.reply("OK ready") + while not self._should_quit: + line = self._poll_command_line() + if line is None: + continue + if self.verbose: + print(f">> {line}", flush=True) + if not self._dispatch(line): + break + + +def open_serial(port: str, baud: int, timeout: float) -> serial.Serial: + try: + return serial.Serial( + port=port, + baudrate=baud, + bytesize=serial.EIGHTBITS, + parity=serial.PARITY_NONE, + stopbits=serial.STOPBITS_ONE, + timeout=timeout, + ) + except serial.SerialException as exc: + print(f"Serial error: {exc}", file=sys.stderr) + print("Check:", file=sys.stderr) + print(f" 1. Device node {port} exists", file=sys.stderr) + print(" 2. User has access (sudo or dialout group)", file=sys.stderr) + print(" 3. Wiring / USB-TTL connection", file=sys.stderr) + raise SystemExit(1) from exc + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="UART finger controller (search/connect/run/stop force stream)" + ) + parser.add_argument("--port", default="/dev/ttyHS1", help="Serial device (default: /dev/ttyHS1)") + parser.add_argument("--baud", type=int, default=115200, help="Baud rate (default: 115200)") + parser.add_argument( + "--timeout", + type=float, + default=0.1, + help="Serial readline timeout seconds (default: 0.1)", + ) + parser.add_argument("--config", "-c", default="finger", help="Orisys config name or path") + parser.add_argument( + "--camera-id", + type=int, + default=0, + help="Default camera index when connect has no argument (ignored if --video is set)", + ) + parser.add_argument( + "--video", + default=None, + help="Default video file for connect (offline debug without a camera)", + ) + parser.add_argument( + "--max-camera-index", + type=int, + default=10, + help="Highest index probed by search", + ) + parser.add_argument( + "--loop-video", + dest="loop_video", + action="store_true", + default=True, + help="Loop video file on EOF during run (default: on)", + ) + parser.add_argument( + "--no-loop-video", + dest="loop_video", + action="store_false", + help="Stop run with OK eof when video file ends", + ) + parser.add_argument( + "--show", + action="store_true", + help="Show UV unrolled OpenCV window (local debug only)", + ) + parser.add_argument("--verbose", action="store_true", help="Echo UART traffic and SDK logs") + args = parser.parse_args(argv) + + if args.video: + video_path = Path(args.video) + if not video_path.is_file(): + print(f"Video file not found: {args.video}", file=sys.stderr) + return 1 + default_source: Source = str(video_path.resolve()) + else: + default_source = args.camera_id + + ser = open_serial(args.port, args.baud, args.timeout) + controller = UartFingerController( + ser, + config=args.config, + default_source=default_source, + max_camera_index=args.max_camera_index, + loop_video=args.loop_video, + show=args.show, + verbose=args.verbose, + ) + + print(f"UART finger controller on {args.port} @ {args.baud} bps", flush=True) + print(f"Default source: {default_source}", flush=True) + print("Commands: search | connect [source|file] | run | stop | quit", flush=True) + + try: + controller.serve() + except KeyboardInterrupt: + print("\nInterrupted.", flush=True) + try: + controller.reply("OK quit") + except Exception: # noqa: BLE001 + pass + finally: + controller.close_runtime() + if ser.is_open: + ser.close() + print("Serial closed.", flush=True) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) 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/DIS_no_integrate.py b/tests/DIS_no_integrate.py new file mode 100644 index 0000000..5e21dcf --- /dev/null +++ b/tests/DIS_no_integrate.py @@ -0,0 +1,420 @@ +#!/usr/bin/env python3 +"""Compare cumulative adaptive DIS vs fixed initial-reference DIS (no integration). + +Reproduces drift on static scenes where adaptive flow integrates incremental +displacement: final_flow = queue[-1] + prev_fine_flow (see orisys.tracker). + +Usage: + python tests/DIS_no_integrate.py --source /path/to/static_clip.mp4 --frames 500 + python tests/DIS_no_integrate.py --source 0 --frames 300 --plot drift.png + python tests/DIS_no_integrate.py --config box --source 0 --no-show + +Modes compared on identical grayscale frames: + cumulative Tracker(adaptive=True) — production default + non_adaptive Tracker(adaptive=False) — single ref, warm-started DIS + fixed_ref DIS calc(ref0, frame, None) — initial frame only, no queue +""" + +from __future__ import annotations + +import argparse +import os +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional + +import cv2 +import numpy as np + +REPO_ROOT = Path(__file__).resolve().parents[1] +SRC_ROOT = REPO_ROOT / "src" +for path in (REPO_ROOT, SRC_ROOT): + if str(path) not in sys.path: + sys.path.insert(0, str(path)) + +from orisys.tracker import Tracker # noqa: E402 + + +DEFAULT_CONFIG = "box" + + +def parse_source(value: str): + text = str(value).strip() + for prefix in ("camera:", "cam:", "webcam:"): + if text.lower().startswith(prefix): + text = text[len(prefix) :].strip() + break + try: + return int(text) + except ValueError: + return text + + +def load_tracker_config(config_ref: str) -> dict: + """Load tracker section from an Orisys config stem or JSON path.""" + path = Path(config_ref) + if path.suffix.lower() == ".json" and path.is_file(): + import json + + data = json.loads(path.read_text(encoding="utf-8")) + else: + import orisys + + data = orisys.configs.load_config(config_ref) + tracker = dict(data.get("tracker", {})) + tracker.setdefault("algorithm", "dis") + tracker.setdefault("dis_preset", "medium") + tracker.setdefault("dis_mode", 3) + tracker.setdefault("lower_threshold", 12.0) + tracker.setdefault("upper_threshold", 18.0) + tracker.setdefault("loop_closure_interval", 1) + return tracker + + +def dis_preset_from_name(name: str) -> int: + preset = (name or "medium").lower() + if preset == "ultrafast": + return cv2.DISOPTICAL_FLOW_PRESET_ULTRAFAST + if preset == "fast": + return cv2.DISOPTICAL_FLOW_PRESET_FAST + return cv2.DISOPTICAL_FLOW_PRESET_MEDIUM + + +def flow_magnitude_stats(flow: np.ndarray) -> tuple[float, float]: + """Return (mean magnitude, max magnitude) for HxWx2 flow.""" + if flow is None: + return float("nan"), float("nan") + mag = np.linalg.norm(flow.astype(np.float32), axis=-1) + return float(mag.mean()), float(mag.max()) + + +@dataclass +class FrameMetrics: + mean_mag: float + max_mag: float + queue_len: int = 0 + + +@dataclass +class ModeSeries: + name: str + mean_mags: list[float] = field(default_factory=list) + max_mags: list[float] = field(default_factory=list) + queue_lens: list[int] = field(default_factory=list) + + def append(self, metrics: FrameMetrics) -> None: + self.mean_mags.append(metrics.mean_mag) + self.max_mags.append(metrics.max_mag) + self.queue_lens.append(metrics.queue_len) + + def slope(self) -> float: + if len(self.mean_mags) < 2: + return float("nan") + x = np.arange(len(self.mean_mags), dtype=np.float64) + y = np.asarray(self.mean_mags, dtype=np.float64) + if not np.all(np.isfinite(y)): + y = np.nan_to_num(y, nan=0.0) + return float(np.polyfit(x, y, 1)[0]) + + +class FixedRefDIS: + """DIS from a fixed initial reference frame; no integration or queue.""" + + def __init__(self, config_tracker: dict): + preset = dis_preset_from_name(config_tracker.get("dis_preset", "medium")) + self._dis = cv2.DISOpticalFlow_create(preset) + self._dis.setFinestScale(int(config_tracker.get("dis_mode", 3))) + self._ref: Optional[np.ndarray] = None + + def reset(self) -> None: + self._ref = None + + def track(self, gray: np.ndarray) -> np.ndarray: + if self._ref is None: + self._ref = gray.copy() + h, w = gray.shape[:2] + return np.zeros((h, w, 2), dtype=np.float32) + return self._dis.calc(self._ref, gray, None) + + +def make_tracker(config_tracker: dict, *, adaptive: bool) -> Tracker: + cfg = dict(config_tracker) + return Tracker( + adaptive=adaptive, + cuda=False, + verbose=False, + config_tracker=cfg, + ) + + +def read_gray_frame(cap: cv2.VideoCapture, static_image: Optional[np.ndarray]) -> tuple[bool, Optional[np.ndarray]]: + if static_image is not None: + return True, static_image.copy() + ok, frame = cap.read() + if not ok or frame is None: + return False, None + if frame.ndim == 3: + frame = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) + return True, frame + + +def run_comparison( + source, + *, + config_tracker: dict, + max_frames: int, + reset_every: Optional[int], + show: bool, +) -> dict[str, ModeSeries]: + path = Path(str(source)) if isinstance(source, str) else None + static_image: Optional[np.ndarray] = None + if isinstance(source, str) and path is not None and path.is_file(): + img = cv2.imread(str(path), cv2.IMREAD_GRAYSCALE) + if img is not None: + static_image = img + + cap = None if static_image is not None else cv2.VideoCapture(source) + if cap is not None and not cap.isOpened(): + raise RuntimeError(f"Cannot open source: {source}") + + cumulative = make_tracker(config_tracker, adaptive=True) + non_adaptive = make_tracker(config_tracker, adaptive=False) + fixed_ref = FixedRefDIS(config_tracker) + + series = { + "cumulative": ModeSeries("cumulative"), + "non_adaptive": ModeSeries("non_adaptive"), + "fixed_ref": ModeSeries("fixed_ref"), + } + + window = "DIS compare (cumulative | non_adaptive | fixed_ref)" + if show: + cv2.namedWindow(window, cv2.WINDOW_NORMAL) + + try: + for frame_idx in range(max_frames): + ok, gray = read_gray_frame(cap, static_image) + if not ok or gray is None: + break + + if reset_every and reset_every > 0 and frame_idx > 0 and frame_idx % reset_every == 0: + cumulative.reset() + + flow_c = cumulative.track(gray.copy()) + flow_n = non_adaptive.track(gray.copy()) + flow_f = fixed_ref.track(gray) + + mean_c, max_c = flow_magnitude_stats(flow_c) + mean_n, max_n = flow_magnitude_stats(flow_n) + mean_f, max_f = flow_magnitude_stats(flow_f) + + series["cumulative"].append( + FrameMetrics(mean_c, max_c, queue_len=cumulative.getFrameQueueLength()) + ) + series["non_adaptive"].append(FrameMetrics(mean_n, max_n, queue_len=1)) + series["fixed_ref"].append(FrameMetrics(mean_f, max_f, queue_len=1)) + + if show and frame_idx % 2 == 0: + h, w = gray.shape[:2] + panels = [] + for flow, label, mean_val in ( + (flow_c, "cumulative", mean_c), + (flow_n, "non_adaptive", mean_n), + (flow_f, "fixed_ref", mean_f), + ): + vis = cv2.cvtColor(gray, cv2.COLOR_GRAY2BGR) + step = max(8, min(h, w) // 40) + ys, xs = np.mgrid[step // 2 : h : step, step // 2 : w : step] + for y, x in zip(ys.ravel(), xs.ravel()): + dx, dy = flow[y, x] + x2 = int(round(x + dx * 2)) + y2 = int(round(y + dy * 2)) + cv2.arrowedLine(vis, (x, y), (x2, y2), (0, 255, 0), 1, tipLength=0.3) + cv2.putText( + vis, + f"{label} m={mean_val:.3f}", + (8, 20), + cv2.FONT_HERSHEY_SIMPLEX, + 0.5, + (0, 255, 255), + 1, + cv2.LINE_AA, + ) + panels.append(vis) + combined = np.hstack(panels) + cv2.imshow(window, combined) + key = cv2.waitKey(1) & 0xFF + if key == ord("q"): + break + finally: + if cap is not None: + cap.release() + if show: + cv2.destroyAllWindows() + + return series + + +def print_summary(series: dict[str, ModeSeries], frames_processed: int) -> None: + print(f"\nProcessed {frames_processed} frame(s)\n") + header = f"{'mode':<14} {'initial':>10} {'final':>10} {'mean':>10} {'slope':>12} {'max_queue':>10}" + print(header) + print("-" * len(header)) + + for key in ("cumulative", "non_adaptive", "fixed_ref"): + s = series[key] + if not s.mean_mags: + print(f"{key:<14} {'(no data)':>10}") + continue + initial = s.mean_mags[0] + final = s.mean_mags[-1] + mean_all = float(np.mean(s.mean_mags)) + slope = s.slope() + max_q = max(s.queue_lens) if s.queue_lens else 0 + print( + f"{key:<14} {initial:10.4f} {final:10.4f} {mean_all:10.4f} {slope:12.6f} {max_q:10d}" + ) + + print("\nInterpretation on a static scene:") + print(" cumulative — rising mean/slope indicates integration drift") + print(" fixed_ref — should stay near a small bias if drift is removed") + + +def save_plot(series: dict[str, ModeSeries], plot_path: str) -> None: + import matplotlib.pyplot as plt + + fig, ax = plt.subplots(figsize=(10, 5)) + for key, color in ( + ("cumulative", "C0"), + ("non_adaptive", "C1"), + ("fixed_ref", "C2"), + ): + s = series[key] + if not s.mean_mags: + continue + ax.plot(s.mean_mags, label=key, color=color) + ax.set_xlabel("frame") + ax.set_ylabel("mean flow magnitude") + ax.set_title("DIS drift comparison (static scene expected: flat fixed_ref)") + ax.legend() + ax.grid(True, alpha=0.3) + fig.tight_layout() + path = Path(plot_path) + path.parent.mkdir(parents=True, exist_ok=True) + fig.savefig(path, dpi=120) + plt.close(fig) + print(f"Saved plot: {path.resolve()}") + + +def parse_args(): + parser = argparse.ArgumentParser( + description="Compare cumulative vs fixed-reference DIS optical flow drift" + ) + parser.add_argument( + "--source", + default="0", + help="Camera index, video path, or image path (default: 0)", + ) + parser.add_argument( + "--config", + "-c", + default=DEFAULT_CONFIG, + help=f"Orisys config stem or JSON path for DIS settings (default: {DEFAULT_CONFIG})", + ) + parser.add_argument( + "--frames", + type=int, + default=300, + help="Maximum frames to process (default: 300)", + ) + parser.add_argument( + "--reset-every", + type=int, + default=0, + help="Reset cumulative tracker every N frames (0 = disabled)", + ) + parser.add_argument( + "--plot", + nargs="?", + const="dis_drift.png", + default=None, + help="Save matplotlib plot (optional path, default: dis_drift.png)", + ) + parser.add_argument( + "--no-show", + action="store_true", + help="Skip OpenCV preview window", + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + source = parse_source(args.source) + config_tracker = load_tracker_config(args.config) + + if config_tracker.get("algorithm", "dis") != "dis": + print(f"Warning: config algorithm is {config_tracker.get('algorithm')}; forcing dis for test.") + + print("DIS no-integrate comparison") + print(f" source: {source}") + print(f" config: {args.config}") + print(f" dis_preset: {config_tracker.get('dis_preset')} dis_mode: {config_tracker.get('dis_mode')}") + print(f" frames: {args.frames} reset_every: {args.reset_every or 'off'}") + + series = run_comparison( + source, + config_tracker=config_tracker, + max_frames=max(args.frames, 1), + reset_every=args.reset_every or None, + show=not args.no_show, + ) + + n = len(series["cumulative"].mean_mags) + print_summary(series, n) + + if args.plot: + save_plot(series, args.plot) + + return 0 + + +def test_dis_no_integrate_smoke() -> None: + """Optional pytest entry: synthetic static frames, no camera required.""" + if os.getenv("ORISYS_RUN_DIS_DRIFT") != "1": + import pytest + + pytest.skip("Set ORISYS_RUN_DIS_DRIFT=1 to run DIS drift smoke test") + + gray = np.full((64, 64), 128, dtype=np.uint8) + # Slight per-frame noise to mimic sensor variation without true motion. + cap_frames = [] + rng = np.random.default_rng(0) + for _ in range(60): + noisy = gray.astype(np.int16) + rng.integers(-2, 3, gray.shape, dtype=np.int16) + cap_frames.append(np.clip(noisy, 0, 255).astype(np.uint8)) + + config_tracker = load_tracker_config(DEFAULT_CONFIG) + cumulative = make_tracker(config_tracker, adaptive=True) + fixed_ref = FixedRefDIS(config_tracker) + + cum_mags = [] + fix_mags = [] + for frame in cap_frames: + fc = cumulative.track(frame.copy()) + ff = fixed_ref.track(frame) + cum_mags.append(flow_magnitude_stats(fc)[0]) + fix_mags.append(flow_magnitude_stats(ff)[0]) + + assert len(cum_mags) >= 10 + # Fixed ref should not drift more than cumulative on identical static+noise input. + cum_slope = float(np.polyfit(np.arange(len(cum_mags)), cum_mags, 1)[0]) + fix_slope = float(np.polyfit(np.arange(len(fix_mags)), fix_mags, 1)[0]) + assert abs(fix_slope) <= abs(cum_slope) + 0.05, ( + f"fixed_ref slope {fix_slope:.4f} worse than cumulative {cum_slope:.4f}" + ) + + +if __name__ == "__main__": + raise SystemExit(main()) 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