diff --git a/assets/finger/30fps-0703/camera_intrinsics.npz b/assets/finger/30fps-0703/camera_intrinsics.npz deleted file mode 100644 index 888b1eb..0000000 Binary files a/assets/finger/30fps-0703/camera_intrinsics.npz and /dev/null differ diff --git a/assets/finger/30fps-0703/mesh.npz b/assets/finger/30fps-0703/mesh.npz deleted file mode 100644 index 4da6bb1..0000000 Binary files a/assets/finger/30fps-0703/mesh.npz and /dev/null differ diff --git a/assets/finger/30fps-0703/pixel_to_xyz.npz b/assets/finger/30fps-0703/pixel_to_xyz.npz deleted file mode 100644 index 15eee73..0000000 Binary files a/assets/finger/30fps-0703/pixel_to_xyz.npz and /dev/null differ diff --git a/assets/finger/30fps-0703/unroll_map_uv_s0.4.npz b/assets/finger/30fps-0703/unroll_map_uv_s0.4.npz deleted file mode 100644 index 7cbb8cb..0000000 Binary files a/assets/finger/30fps-0703/unroll_map_uv_s0.4.npz and /dev/null differ diff --git a/assets/finger/30fps-0703/world_to_cam.npy b/assets/finger/30fps-0703/world_to_cam.npy deleted file mode 100644 index 3d6f897..0000000 Binary files a/assets/finger/30fps-0703/world_to_cam.npy and /dev/null differ diff --git a/assets/finger/finger/camera_intrinsics.npz b/assets/finger/finger/camera_intrinsics.npz deleted file mode 100644 index 888b1eb..0000000 Binary files a/assets/finger/finger/camera_intrinsics.npz and /dev/null differ diff --git a/assets/finger/finger/mesh.npz b/assets/finger/finger/mesh.npz deleted file mode 100644 index 4da6bb1..0000000 Binary files a/assets/finger/finger/mesh.npz and /dev/null differ diff --git a/assets/finger/finger/pixel_to_xyz.npz b/assets/finger/finger/pixel_to_xyz.npz deleted file mode 100644 index 15eee73..0000000 Binary files a/assets/finger/finger/pixel_to_xyz.npz and /dev/null differ diff --git a/assets/finger/finger/unroll_map_uv_s0.4.npz b/assets/finger/finger/unroll_map_uv_s0.4.npz deleted file mode 100644 index 7cbb8cb..0000000 Binary files a/assets/finger/finger/unroll_map_uv_s0.4.npz and /dev/null differ diff --git a/assets/finger/finger/world_to_cam.npy b/assets/finger/finger/world_to_cam.npy deleted file mode 100644 index 3d6f897..0000000 Binary files a/assets/finger/finger/world_to_cam.npy and /dev/null differ diff --git a/assets/finger/visualization/camera_intrinsics.npz b/assets/finger/visualization/camera_intrinsics.npz deleted file mode 100644 index 1d211d5..0000000 Binary files a/assets/finger/visualization/camera_intrinsics.npz and /dev/null differ diff --git a/assets/finger/visualization/flatten_remap.npz b/assets/finger/visualization/flatten_remap.npz deleted file mode 100644 index 02ca287..0000000 Binary files a/assets/finger/visualization/flatten_remap.npz and /dev/null differ diff --git a/assets/finger/visualization/mesh.npz b/assets/finger/visualization/mesh.npz deleted file mode 100644 index ad81542..0000000 Binary files a/assets/finger/visualization/mesh.npz and /dev/null differ diff --git a/assets/finger/visualization/pixel_to_xyz.npz b/assets/finger/visualization/pixel_to_xyz.npz deleted file mode 100644 index 4096205..0000000 Binary files a/assets/finger/visualization/pixel_to_xyz.npz and /dev/null differ diff --git a/assets/finger/visualization/world_to_cam.npy b/assets/finger/visualization/world_to_cam.npy deleted file mode 100644 index 2d5e411..0000000 Binary files a/assets/finger/visualization/world_to_cam.npy and /dev/null differ diff --git a/build/common/verify/gui.py b/build/common/verify/gui.py index 491a0a9..5306b6b 100644 --- a/build/common/verify/gui.py +++ b/build/common/verify/gui.py @@ -19,8 +19,9 @@ GUI_PRODUCTS: dict[str, dict] = { "exe_name": "OrisysDashboard.exe", "required_relpaths": [ "config/dd02-ov.json", - "assets/finger/30fps-0703/mesh.npz", - "assets/finger/30fps-0703/world_to_cam.npy", + "sensors/finger/assets/30fps-0703/mesh.npz", + "sensors/finger/assets/30fps-0703/world_to_cam.npy", + "sensors/finger/assets/visualization/world_to_cam.npy", ], }, "ory_mini": { diff --git a/examples/gui/preference.json b/examples/gui/preference.json index c81d1e7..9612335 100644 --- a/examples/gui/preference.json +++ b/examples/gui/preference.json @@ -6,7 +6,7 @@ "csv_save_dir": "./outputs/csv", "show_log": true, "show_flow_arrows": false, - "arrow_background": "unrolled", + "arrow_background": "pure_color", "free_rotate": true, "log_display_mode": "hidden" } diff --git a/examples/gui/viewer3d/camera_discovery.py b/examples/gui/viewer3d/camera_discovery.py index b161abe..f43ef13 100644 --- a/examples/gui/viewer3d/camera_discovery.py +++ b/examples/gui/viewer3d/camera_discovery.py @@ -1,139 +1,22 @@ -"""Discover USB cameras for the dashboard (Windows DSHOW/MSMF, Linux V4L2).""" +"""Discover USB cameras for the dashboard. + +Thin compatibility shim over ``orisys.capture`` discovery helpers. +""" from __future__ import annotations -import contextlib -import os -import sys -from dataclasses import dataclass +from orisys.capture import ( + DEFAULT_MAX_CAMERA_INDEX, + CameraFilter, + CameraInfo, + list_cameras, + pick_default_camera, +) -import cv2 - -DEFAULT_MAX_CAMERA_INDEX = 10 - - -@dataclass(frozen=True) -class CameraInfo: - index: int - backend_name: str - backend_api: int - width: int - height: int - - @property - def source(self) -> str: - return str(self.index) - - def label(self) -> str: - return f"{self.index} ({self.width}x{self.height}) [{self.backend_name}]" - - -@contextlib.contextmanager -def _suppress_opencv_noise(): - """Hide OpenCV VideoCapture probe warnings (stderr + OpenCV log).""" - prev_level = None - set_level = None - try: - logging = getattr(getattr(cv2, "utils", None), "logging", None) - if logging is not None: - set_level = getattr(logging, "setLogLevel", None) - get_level = getattr(logging, "getLogLevel", None) - silent = getattr(logging, "LOG_LEVEL_SILENT", 0) - if set_level is not None: - if get_level is not None: - prev_level = get_level() - set_level(silent) - except Exception: - prev_level = None - - devnull = open(os.devnull, "w") - try: - with contextlib.redirect_stderr(devnull): - stderr_fd = sys.stderr.fileno() - saved_fd = os.dup(stderr_fd) - try: - os.dup2(devnull.fileno(), stderr_fd) - yield - finally: - os.dup2(saved_fd, stderr_fd) - os.close(saved_fd) - finally: - devnull.close() - if set_level is not None and prev_level is not None: - try: - set_level(prev_level) - except Exception: - pass - - -def _backend_candidates() -> list[tuple[str, int]]: - candidates: list[tuple[str, int]] = [] - if sys.platform == "win32": - if hasattr(cv2, "CAP_DSHOW"): - candidates.append(("DSHOW", cv2.CAP_DSHOW)) - if hasattr(cv2, "CAP_MSMF"): - candidates.append(("MSMF", cv2.CAP_MSMF)) - elif sys.platform.startswith("linux"): - # Prefer v4l2 for USB camera indices on Linux. - if hasattr(cv2, "CAP_V4L2"): - candidates.append(("V4L2", cv2.CAP_V4L2)) - if not candidates: - candidates.append(("ANY", cv2.CAP_ANY)) - return candidates - - -def _try_open(index: int, backend_name: str, backend_api: int) -> CameraInfo | None: - cap = cv2.VideoCapture(index, backend_api) - if not cap.isOpened(): - cap.release() - return None - ok, frame = cap.read() - width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH) or 0) - height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0) - if not ok or frame is None: - if width <= 0 or height <= 0: - cap.release() - return None - else: - h, w = frame.shape[:2] - width = width or int(w) - height = height or int(h) - cap.release() - return CameraInfo( - index=index, - backend_name=backend_name, - backend_api=backend_api, - width=width, - height=height, - ) - - -def list_cameras(max_index: int = DEFAULT_MAX_CAMERA_INDEX) -> list[CameraInfo]: - """Probe indices ``0 .. max_index`` and return working cameras.""" - found: list[CameraInfo] = [] - seen_indices: set[int] = set() - backends = _backend_candidates() - - with _suppress_opencv_noise(): - for index in range(0, max_index + 1): - for backend_name, backend_api in backends: - info = _try_open(index, backend_name, backend_api) - if info is None: - continue - if index in seen_indices: - break - found.append(info) - seen_indices.add(index) - break - return found - - -def pick_default_camera(cameras: list[CameraInfo], preferred: str | None = None) -> str | None: - if not cameras: - return None - preferred = (preferred or "").strip() - if preferred: - for cam in cameras: - if cam.source == preferred: - return cam.source - return cameras[0].source +__all__ = [ + "DEFAULT_MAX_CAMERA_INDEX", + "CameraFilter", + "CameraInfo", + "list_cameras", + "pick_default_camera", +] diff --git a/examples/viz/mesh_camera.py b/examples/viz/mesh_camera.py index 8d05524..d1a9856 100644 --- a/examples/viz/mesh_camera.py +++ b/examples/viz/mesh_camera.py @@ -6,7 +6,34 @@ from pathlib import Path import numpy as np -DEFAULT_FINGER_VIZ_POSE = "assets/finger/visualization/world_to_cam.npy" +DEFAULT_FINGER_VIZ_POSE = "sensors/finger/assets/visualization/world_to_cam.npy" + + +def _finger_bundle_viz_pose_candidates() -> list[Path]: + """Resolve viz pose paths from the packaged sensors/finger bundle when present.""" + try: + from orisys.configs import resolve_sensor_bundle, resolve_sensor_resource + except ImportError: + return [] + + found: list[Path] = [] + bundle = resolve_sensor_bundle("finger") + if bundle is not None and bundle.manifest_json is not None: + try: + import json + + manifest = json.loads(bundle.manifest_json.read_text(encoding="utf-8")) + except Exception: + manifest = {} + rel = manifest.get("viz_pose") + if isinstance(rel, str) and rel.strip(): + found.append((bundle.root / rel).resolve()) + + packaged = resolve_sensor_resource("finger", "assets", "visualization", "world_to_cam.npy") + if packaged is not None: + found.append(packaged.resolve()) + return found + _CAMERA_X_ROT_90 = np.array( [ @@ -28,27 +55,42 @@ _CAMERA_Z_ROT_180 = np.array( def resolve_viz_pose_path(path: str | Path | None = None) -> Path: - """Resolve world_to_cam.npy from cwd or packaged asset layout.""" + """Resolve world_to_cam.npy from sensor bundle, cwd, or packaged layout.""" raw = Path(path or DEFAULT_FINGER_VIZ_POSE).expanduser() if raw.is_file(): return raw.resolve() here = Path(__file__).resolve() - candidates = [ - Path.cwd() / raw, - here.parents[2] / raw, - ] + candidates: list[Path] = [] + + # Prefer the sensors/finger bundle for the default (and matching relative) path. + if path is None or raw.as_posix() == DEFAULT_FINGER_VIZ_POSE or not raw.is_absolute(): + candidates.extend(_finger_bundle_viz_pose_candidates()) + try: from gui.app_paths import app_root - candidates.insert(0, app_root() / raw) + candidates.append(app_root() / raw) except ImportError: pass - for candidate in candidates: - if candidate.is_file(): - return candidate.resolve() + candidates.extend( + [ + Path.cwd() / raw, + here.parents[2] / raw, + ] + ) - checked = ", ".join(str(p) for p in candidates) + unique: list[Path] = [] + for candidate in candidates: + resolved = candidate.resolve() + if resolved not in unique: + unique.append(resolved) + + for candidate in unique: + if candidate.is_file(): + return candidate + + checked = ", ".join(str(p) for p in unique) raise FileNotFoundError(f"Visualization pose file not found: {raw}. Checked: {checked}") diff --git a/src/orisys/__init__.py b/src/orisys/__init__.py index d464157..310285c 100644 --- a/src/orisys/__init__.py +++ b/src/orisys/__init__.py @@ -13,6 +13,7 @@ from . import util from . import configs from . import finger from . import signal_processing +from . import capture from .frame_processors import FFCFrameProcessor CompoundEyeSensor = Sensor @@ -37,6 +38,9 @@ __all__ = [ # signal helpers 'signal_processing', + # capture helpers + 'capture', + # configs utilities functions 'list_config', 'export_config', diff --git a/src/orisys/capture/__init__.py b/src/orisys/capture/__init__.py new file mode 100644 index 0000000..c1f1015 --- /dev/null +++ b/src/orisys/capture/__init__.py @@ -0,0 +1,55 @@ +"""Capture subsystem: sources, buffering, reconnect, discovery, multi-camera.""" + +from .buffer import FrameBuffer, FramePacket +from .config import ( + MAX_QUEUE_SIZE, + MAX_SIMULTANEOUS_CAPTURES, + CaptureConfig, + CaptureHealth, + CaptureMode, + CaptureStatus, + OverflowPolicy, + ReconnectPolicy, +) +from .discovery import ( + DEFAULT_LAPTOP_WEBCAM_PATTERNS, + DEFAULT_MAX_CAMERA_INDEX, + CameraFilter, + CameraInfo, + list_cameras, + pick_default_camera, +) +from .manager import CaptureManager +from .source import ( + CaptureSource, + is_live_camera, + normalize_source, + 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", + "pick_default_camera", + "normalize_source", + "is_live_camera", + "open_opencv_capture", + "open_capture", +] diff --git a/src/orisys/capture/buffer.py b/src/orisys/capture/buffer.py new file mode 100644 index 0000000..c0b2895 --- /dev/null +++ b/src/orisys/capture/buffer.py @@ -0,0 +1,127 @@ +"""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 new file mode 100644 index 0000000..489d33b --- /dev/null +++ b/src/orisys/capture/config.py @@ -0,0 +1,158 @@ +"""Capture configuration, health state, and reconnect policy.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +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): + """Lifecycle / health state for a capture source.""" + + CLOSED = "closed" + CONNECTING = "connecting" + CONNECTED = "connected" + RECONNECTING = "reconnecting" + FAILED = "failed" + + +SourceType = Union[int, str] + + +@dataclass(frozen=True) +class ReconnectPolicy: + """Auto-reconnect settings for live cameras. + + Parameters + ---------- + enabled: + When False, failed reads do not attempt reopen. + max_attempts: + Maximum reopen attempts per disconnect episode. ``None`` means unlimited. + initial_backoff_s: + First wait before a reconnect attempt. + max_backoff_s: + Upper bound for exponential backoff. + fail_threshold: + Consecutive failed reads that trigger reconnect. + stale_timeout_s: + If no frame arrives for this many seconds while connected, treat as stale + and reconnect. ``None`` disables stale detection. + """ + + enabled: bool = False + max_attempts: Optional[int] = None + initial_backoff_s: float = 0.25 + max_backoff_s: float = 5.0 + fail_threshold: int = 5 + stale_timeout_s: Optional[float] = 2.0 + + def next_backoff(self, attempt: int) -> float: + """Exponential backoff capped at ``max_backoff_s`` (attempt is 0-based).""" + delay = self.initial_backoff_s * (2 ** max(0, attempt)) + return min(delay, self.max_backoff_s) + + +@dataclass +class CaptureConfig: + """Configuration for opening and running a capture source.""" + + source: SourceType + backend: str = "auto" + width: Optional[int] = None + 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 + 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 + name_exclude: Optional[Sequence[str]] = None + vid_pid_allowlist: Optional[Sequence[Tuple[int, int]]] = None + min_width: Optional[int] = None + 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( + cls, + 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.""" + 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 + fps = camera.get("fps") + return cls( + source=source, + width=width, + height=height, + fps=float(fps) if fps is not None else None, + mode=mode, + reconnect=reconnect or ReconnectPolicy(), + verbose=verbose, + **kwargs, + ) + + +@dataclass +class CaptureStatus: + """Snapshot of capture health for diagnostics / multi-camera UIs.""" + + health: CaptureHealth = CaptureHealth.CLOSED + source: Optional[SourceType] = None + last_error: Optional[str] = None + last_frame_time: Optional[float] = None + last_seq: int = 0 + dropped_frames: int = 0 + reconnect_attempts: int = 0 + consecutive_failures: int = 0 + width: int = 0 + height: int = 0 + fps: float = 0.0 + backend_name: str = "" + fourcc: str = "" diff --git a/src/orisys/capture/discovery.py b/src/orisys/capture/discovery.py new file mode 100644 index 0000000..0ae7285 --- /dev/null +++ b/src/orisys/capture/discovery.py @@ -0,0 +1,334 @@ +"""Discover cameras with optional platform metadata and filters.""" + +from __future__ import annotations + +import contextlib +import os +import re +import sys +from dataclasses import dataclass, field +from typing import Iterable, Optional, Sequence, Tuple + +import cv2 + +from .config import MAX_SIMULTANEOUS_CAPTURES +from .source import _backend_candidates + +DEFAULT_MAX_CAMERA_INDEX = MAX_SIMULTANEOUS_CAPTURES + +# Common built-in laptop webcam name fragments (case-insensitive). +DEFAULT_LAPTOP_WEBCAM_PATTERNS: Tuple[str, ...] = ( + r"integrated", + r"built[- ]?in", + r"laptop", + r"facetime", + r"ir\s*camera", + r"rgb\s*camera", + r"hd\s*webcam", + r"hd\s*camera", + r"user[- ]facing", + r"front\s*camera", + r"dell\s*webcam", + r"lenovo\s*easycamera", + r"hp\s*(truevision|webcam|hd)", + r"microsoft\s*life(cam|studio)", + r"realtek", + r"chicony", + r"sonix", +) + + +@dataclass(frozen=True) +class CameraInfo: + """Description of a probed camera device.""" + + index: int + backend_name: str + backend_api: int + width: int + height: int + name: str = "" + path: str = "" + vendor_id: Optional[int] = None + product_id: Optional[int] = None + bus: str = "" + + @property + def source(self) -> str: + return str(self.index) + + def label(self) -> str: + name = self.name.strip() + if name: + return f"{self.index}: {name} ({self.width}x{self.height}) [{self.backend_name}]" + return f"{self.index} ({self.width}x{self.height}) [{self.backend_name}]" + + @property + def vid_pid(self) -> Optional[Tuple[int, int]]: + if self.vendor_id is None or self.product_id is None: + return None + return (int(self.vendor_id), int(self.product_id)) + + +@dataclass(frozen=True) +class CameraFilter: + """Optional filters for ``list_cameras`` / ``pick_default_camera``.""" + + name_include: Optional[Sequence[str]] = None + name_exclude: Optional[Sequence[str]] = None + exclude_laptop_webcams: bool = False + laptop_patterns: Sequence[str] = field(default_factory=lambda: DEFAULT_LAPTOP_WEBCAM_PATTERNS) + vid_pid_allowlist: Optional[Sequence[Tuple[int, int]]] = None + min_width: Optional[int] = None + min_height: Optional[int] = None + prefer_named: bool = True + + def matches(self, cam: CameraInfo) -> bool: + name = (cam.name or "").lower() + if self.min_width is not None and cam.width < int(self.min_width): + return False + if self.min_height is not None and cam.height < int(self.min_height): + return False + if self.vid_pid_allowlist: + allowed = {(int(v), int(p)) for v, p in self.vid_pid_allowlist} + if cam.vid_pid is None or cam.vid_pid not in allowed: + return False + if self.name_include: + if not any(re.search(pat, name, re.I) for pat in self.name_include): + # If name unknown, keep the device unless prefer_named forces drop. + if name: + return False + if self.prefer_named: + return False + if self.name_exclude: + if name and any(re.search(pat, name, re.I) for pat in self.name_exclude): + return False + if self.exclude_laptop_webcams and name: + if any(re.search(pat, name, re.I) for pat in self.laptop_patterns): + return False + return True + + +@contextlib.contextmanager +def _suppress_opencv_noise(): + """Hide OpenCV VideoCapture probe warnings (stderr + OpenCV log).""" + prev_level = None + set_level = None + try: + logging = getattr(getattr(cv2, "utils", None), "logging", None) + if logging is not None: + set_level = getattr(logging, "setLogLevel", None) + get_level = getattr(logging, "getLogLevel", None) + silent = getattr(logging, "LOG_LEVEL_SILENT", 0) + if set_level is not None: + if get_level is not None: + prev_level = get_level() + set_level(silent) + except Exception: + prev_level = None + + devnull = open(os.devnull, "w") + try: + with contextlib.redirect_stderr(devnull): + stderr_fd = sys.stderr.fileno() + saved_fd = os.dup(stderr_fd) + try: + os.dup2(devnull.fileno(), stderr_fd) + yield + finally: + os.dup2(saved_fd, stderr_fd) + os.close(saved_fd) + finally: + devnull.close() + if set_level is not None and prev_level is not None: + try: + set_level(prev_level) + except Exception: + pass + + +def _linux_device_metadata(index: int) -> dict: + """Best-effort V4L2 name / USB VID:PID from sysfs.""" + meta = {"name": "", "path": f"/dev/video{index}", "vendor_id": None, "product_id": None, "bus": ""} + name_path = f"/sys/class/video4linux/video{index}/name" + try: + with open(name_path, "r", encoding="utf-8", errors="ignore") as fh: + meta["name"] = fh.read().strip() + except OSError: + pass + + # Walk up to a USB device directory containing idVendor/idProduct. + device_link = f"/sys/class/video4linux/video{index}/device" + try: + cur = os.path.realpath(device_link) + for _ in range(8): + vend = os.path.join(cur, "idVendor") + prod = os.path.join(cur, "idProduct") + if os.path.isfile(vend) and os.path.isfile(prod): + with open(vend, "r", encoding="utf-8") as fh: + meta["vendor_id"] = int(fh.read().strip(), 16) + with open(prod, "r", encoding="utf-8") as fh: + meta["product_id"] = int(fh.read().strip(), 16) + meta["bus"] = "usb" + break + parent = os.path.dirname(cur) + if parent == cur: + break + cur = parent + except (OSError, ValueError): + pass + return meta + + +def _windows_device_names() -> dict[int, dict]: + """Best-effort mapping of camera index → name / hardware id via PowerShell. + + Index order is not guaranteed to match OpenCV; names are matched later by + probing order best-effort. Returns ``{ordinal: meta}`` in enumeration order. + """ + if sys.platform != "win32": + return {} + try: + import json + import subprocess + + # Camera devices from PnP; keep this soft-fail for machines without cameras. + cmd = [ + "powershell", + "-NoProfile", + "-Command", + ( + "Get-PnpDevice -Class Camera,Image -Status OK -ErrorAction SilentlyContinue | " + "Select-Object FriendlyName, InstanceId | ConvertTo-Json -Compress" + ), + ] + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=5, check=False) + if proc.returncode != 0 or not proc.stdout.strip(): + return {} + data = json.loads(proc.stdout) + if isinstance(data, dict): + data = [data] + out: dict[int, dict] = {} + for i, item in enumerate(data): + name = str(item.get("FriendlyName") or "").strip() + instance = str(item.get("InstanceId") or "") + vid = pid = None + m = re.search(r"VID_([0-9A-Fa-f]{4}).*PID_([0-9A-Fa-f]{4})", instance) + if m: + vid = int(m.group(1), 16) + pid = int(m.group(2), 16) + out[i] = { + "name": name, + "path": "", + "vendor_id": vid, + "product_id": pid, + "bus": "usb" if vid is not None else "", + } + return out + except Exception: + return {} + + +def _device_metadata(index: int, windows_names: Optional[dict[int, dict]] = None) -> dict: + if sys.platform.startswith("linux"): + return _linux_device_metadata(index) + if sys.platform == "win32" and windows_names is not None: + # Best-effort: OpenCV index often matches enumeration order, but not always. + return windows_names.get(index, {"name": "", "path": "", "vendor_id": None, "product_id": None, "bus": ""}) + return {"name": "", "path": "", "vendor_id": None, "product_id": None, "bus": ""} + + +def _try_open(index: int, backend_name: str, backend_api: int, meta: dict) -> Optional[CameraInfo]: + cap = cv2.VideoCapture(index, backend_api) + if not cap.isOpened(): + cap.release() + return None + ok, frame = cap.read() + width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH) or 0) + height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0) + if not ok or frame is None: + if width <= 0 or height <= 0: + cap.release() + return None + else: + h, w = frame.shape[:2] + width = width or int(w) + height = height or int(h) + cap.release() + return CameraInfo( + index=index, + backend_name=backend_name, + backend_api=backend_api, + width=width, + height=height, + name=str(meta.get("name") or ""), + path=str(meta.get("path") or ""), + vendor_id=meta.get("vendor_id"), + product_id=meta.get("product_id"), + bus=str(meta.get("bus") or ""), + ) + + +def list_cameras( + max_index: int = DEFAULT_MAX_CAMERA_INDEX, + *, + camera_filter: Optional[CameraFilter] = None, + exclude_laptop_webcams: bool = False, + name_include: Optional[Sequence[str]] = None, + name_exclude: Optional[Sequence[str]] = None, + vid_pid_allowlist: Optional[Sequence[Tuple[int, int]]] = None, + min_width: Optional[int] = None, + min_height: Optional[int] = None, + backend: str = "auto", +) -> list[CameraInfo]: + """Probe indices ``0 .. max_index`` and return working cameras. + + Laptop webcam filtering is best-effort and requires device names from the + OS. When names are unavailable, devices are kept unless other filters reject + them. Pass ``exclude_laptop_webcams=True`` or a ``CameraFilter`` to opt in. + """ + filt = camera_filter or CameraFilter( + name_include=name_include, + name_exclude=name_exclude, + exclude_laptop_webcams=exclude_laptop_webcams, + vid_pid_allowlist=vid_pid_allowlist, + min_width=min_width, + min_height=min_height, + ) + + found: list[CameraInfo] = [] + seen_indices: set[int] = set() + backends = _backend_candidates(backend) + windows_names = _windows_device_names() if sys.platform == "win32" else None + + with _suppress_opencv_noise(): + for index in range(0, max_index + 1): + meta = _device_metadata(index, windows_names) + for backend_name, backend_api in backends: + info = _try_open(index, backend_name, backend_api, meta) + if info is None: + continue + if index in seen_indices: + break + if not filt.matches(info): + seen_indices.add(index) + break + found.append(info) + seen_indices.add(index) + break + return found + + +def pick_default_camera( + cameras: Iterable[CameraInfo], + preferred: str | None = None, +) -> str | None: + cameras = list(cameras) + if not cameras: + return None + preferred = (preferred or "").strip() + if preferred: + for cam in cameras: + if cam.source == preferred: + return cam.source + return cameras[0].source diff --git a/src/orisys/capture/manager.py b/src/orisys/capture/manager.py new file mode 100644 index 0000000..3c2b09d --- /dev/null +++ b/src/orisys/capture/manager.py @@ -0,0 +1,122 @@ +"""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 new file mode 100644 index 0000000..e38feea --- /dev/null +++ b/src/orisys/capture/source.py @@ -0,0 +1,349 @@ +"""OpenCV-backed capture source with optional reconnect.""" + +from __future__ import annotations + +import os +import sys +import time +from typing import Any, Optional, Protocol, runtime_checkable + +import cv2 +import numpy as np + +from .config import ( + CaptureConfig, + CaptureHealth, + CaptureStatus, + ReconnectPolicy, + SourceType, +) + + +@runtime_checkable +class VideoCaptureLike(Protocol): + """Minimal protocol so tests can inject fake captures.""" + + def isOpened(self) -> bool: ... + def read(self) -> tuple[bool, Any]: ... + def release(self) -> None: ... + def get(self, propId: int) -> float: ... + def set(self, propId: int, value: float) -> bool: ... + + +def normalize_source(source: SourceType | None) -> SourceType: + """Accept camera index, numeric string, V4L path, or filesystem path.""" + if source is None: + return 0 + if isinstance(source, str): + stripped = source.strip() + if not stripped: + return 0 + if stripped.isdigit(): + return int(stripped) + # Only strip trailing separators (Windows path quirk); keep leading '/'. + return stripped.rstrip("/\\") or stripped + return source + + +def is_live_camera(source: SourceType) -> bool: + """True for camera indices and Linux V4L2 device paths.""" + if isinstance(source, int): + return True + if isinstance(source, str) and source.startswith("/dev/video"): + return True + return False + + +def _backend_candidates(preferred: str = "auto") -> list[tuple[str, int]]: + preferred = (preferred or "auto").lower() + candidates: list[tuple[str, int]] = [] + + def add(name: str, attr: str) -> None: + if hasattr(cv2, attr): + candidates.append((name, getattr(cv2, attr))) + + if preferred in ("dshow", "directshow") and sys.platform == "win32": + add("DSHOW", "CAP_DSHOW") + elif preferred in ("msmf",) and sys.platform == "win32": + add("MSMF", "CAP_MSMF") + 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"): + add("V4L2", "CAP_V4L2") + if not candidates: + candidates.append(("ANY", cv2.CAP_ANY)) + return candidates + + +def _fourcc_code(fourcc: Optional[str]) -> Optional[int]: + if not fourcc: + return None + text = str(fourcc).strip() + if len(text) != 4: + return None + return cv2.VideoWriter.fourcc(*text) + + +def _decode_fourcc(value: float) -> str: + fourcc_int = int(value) + if fourcc_int <= 0: + return "N/A" + bytes_ = [(fourcc_int >> (8 * i)) & 0xFF for i in range(4)] + return "".join(chr(b) if 32 <= b <= 126 else "." for b in bytes_) + + +def open_opencv_capture( + source: SourceType, + *, + backend: str = "auto", + width: Optional[int] = None, + height: Optional[int] = None, + fps: Optional[float] = None, + fourcc: Optional[str] = "MJPG", + apply_stream_props: bool = True, +) -> tuple[VideoCaptureLike, str]: + """Open a ``cv2.VideoCapture`` trying preferred backends in order. + + 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." + + backends = _backend_candidates(backend) if live else [("ANY", cv2.CAP_ANY)] + for backend_name, backend_api in backends: + try: + if live and backend_api != cv2.CAP_ANY: + cap = cv2.VideoCapture(source, backend_api) + else: + cap = cv2.VideoCapture(source) + except Exception as exc: # pragma: no cover - OpenCV quirks + last_error = str(exc) + continue + if not cap.isOpened(): + cap.release() + 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)) + return cap, backend_name + + raise RuntimeError(last_error) + + +class CaptureSource: + """Owns a video capture handle, health state, and reconnect logic. + + Optionally accepts an injected ``factory`` for tests:: + + CaptureSource(config, factory=lambda cfg: (FakeCap(), "FAKE")) + """ + + def __init__( + self, + config: CaptureConfig, + *, + factory=None, + ): + self.config = config + self._factory = factory or self._default_factory + self._cap: Optional[VideoCaptureLike] = None + self._backend_name = "" + self._status = CaptureStatus(source=normalize_source(config.source)) + self._reconnect_attempt = 0 + self._lock = __import__("threading").Lock() + + @staticmethod + def _default_factory(config: CaptureConfig) -> tuple[VideoCaptureLike, str]: + return open_opencv_capture( + config.source, + backend=config.backend, + width=config.width, + height=config.height, + fps=config.fps, + fourcc=config.fourcc, + apply_stream_props=is_live_camera(normalize_source(config.source)), + ) + + @property + def status(self) -> CaptureStatus: + with self._lock: + return CaptureStatus( + health=self._status.health, + source=self._status.source, + last_error=self._status.last_error, + last_frame_time=self._status.last_frame_time, + last_seq=self._status.last_seq, + dropped_frames=self._status.dropped_frames, + reconnect_attempts=self._status.reconnect_attempts, + consecutive_failures=self._status.consecutive_failures, + width=self._status.width, + height=self._status.height, + fps=self._status.fps, + backend_name=self._status.backend_name, + fourcc=self._status.fourcc, + ) + + @property + def cap(self) -> Optional[VideoCaptureLike]: + """Raw capture handle (compat with Sensor.cap).""" + return self._cap + + def is_open(self) -> bool: + cap = self._cap + return cap is not None and cap.isOpened() + + def open(self) -> None: + with self._lock: + self._status.health = CaptureHealth.CONNECTING + self._status.last_error = None + try: + cap, backend_name = self._factory(self.config) + except Exception as exc: + with self._lock: + self._status.health = CaptureHealth.FAILED + self._status.last_error = str(exc) + raise + with self._lock: + self._release_unlocked() + self._cap = cap + self._backend_name = backend_name + self._refresh_stream_meta_unlocked() + self._status.health = CaptureHealth.CONNECTED + self._status.backend_name = backend_name + self._status.consecutive_failures = 0 + self._reconnect_attempt = 0 + if self.config.verbose: + st = self.status + print( + f"OpenCV negotiated stream: FOURCC='{st.fourcc}', " + f"size={st.width}x{st.height}, fps={st.fps:.2f}, backend={st.backend_name}" + ) + + def release(self) -> None: + with self._lock: + self._release_unlocked() + self._status.health = CaptureHealth.CLOSED + + def _release_unlocked(self) -> None: + if self._cap is not None: + try: + self._cap.release() + except Exception: + pass + self._cap = None + + 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) + except Exception: + pass + + def read_raw(self) -> tuple[bool, Optional[np.ndarray]]: + """Read one frame from the underlying capture (no reconnect).""" + cap = self._cap + if cap is None: + return False, None + try: + ok, frame = cap.read() + except Exception as exc: + with self._lock: + self._status.consecutive_failures += 1 + self._status.last_error = str(exc) + return False, None + if not ok or frame is None: + with self._lock: + self._status.consecutive_failures += 1 + return False, None + with self._lock: + self._status.consecutive_failures = 0 + self._status.last_frame_time = time.perf_counter() + if self._status.health != CaptureHealth.CONNECTED: + self._status.health = CaptureHealth.CONNECTED + return True, frame + + def note_published(self, seq: int) -> None: + with self._lock: + self._status.last_seq = int(seq) + + def note_dropped(self, n: int = 1) -> None: + with self._lock: + self._status.dropped_frames += int(n) + + def should_reconnect(self) -> bool: + policy = self.config.reconnect + if not policy.enabled: + return False + with self._lock: + if self._status.health == CaptureHealth.CLOSED: + return False + if not self.is_open(): + return True + if self._status.consecutive_failures >= max(1, policy.fail_threshold): + return True + if policy.stale_timeout_s is not None and self._status.last_frame_time is not None: + age = time.perf_counter() - self._status.last_frame_time + if age >= policy.stale_timeout_s and self._status.health == CaptureHealth.CONNECTED: + return True + return False + + def reconnect(self) -> bool: + """Attempt to reopen the capture. Returns True on success.""" + policy: ReconnectPolicy = self.config.reconnect + if not policy.enabled: + return False + if policy.max_attempts is not None and self._reconnect_attempt >= policy.max_attempts: + with self._lock: + self._status.health = CaptureHealth.FAILED + self._status.last_error = "Reconnect attempts exhausted" + return False + + with self._lock: + self._status.health = CaptureHealth.RECONNECTING + self._status.reconnect_attempts += 1 + attempt = self._reconnect_attempt + self._reconnect_attempt += 1 + self._release_unlocked() + + delay = policy.next_backoff(attempt) + if self.config.verbose: + print(f"Reconnecting to {self.config.source!r} (attempt {attempt + 1}, wait {delay:.2f}s)...") + time.sleep(delay) + try: + self.open() + return True + except Exception as exc: + with self._lock: + self._status.last_error = str(exc) + if policy.max_attempts is not None and self._reconnect_attempt >= policy.max_attempts: + self._status.health = CaptureHealth.FAILED + else: + self._status.health = CaptureHealth.RECONNECTING + return False diff --git a/src/orisys/capture/worker.py b/src/orisys/capture/worker.py new file mode 100644 index 0000000..01d2a5e --- /dev/null +++ b/src/orisys/capture/worker.py @@ -0,0 +1,124 @@ +"""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 0393797..be92605 100644 --- a/src/orisys/finger/runtime.py +++ b/src/orisys/finger/runtime.py @@ -2,9 +2,16 @@ from __future__ import annotations from dataclasses import dataclass -import cv2 import numpy as np +from ..capture import ( + CaptureConfig, + CaptureWorker, + ReconnectPolicy, + is_live_camera, + normalize_source, + open_capture, +) from ..configs import load_config from ..frame_processors import FrameUnwarper from ..sensor import Sensor @@ -24,25 +31,28 @@ from .exports import ( def normalize_video_source(video): """Accept camera index, numeric string, or filesystem video path.""" - if video is None: - return 0 - if isinstance(video, str): - stripped = video.strip().strip("/\\") - if not stripped: - return 0 - if stripped.isdigit(): - return int(stripped) - return stripped - return video + return normalize_source(video) -def open_source(video, config): - capture = cv2.VideoCapture(normalize_video_source(video)) - capture.set(cv2.CAP_PROP_FRAME_WIDTH, config.get("camera", {}).get("resolution_raw")[0]) - capture.set(cv2.CAP_PROP_FRAME_HEIGHT, config.get("camera", {}).get("resolution_raw")[1]) - capture.set(cv2.CAP_PROP_FPS, config.get("camera", {}).get("fps", 30)) - capture.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter.fourcc('M', 'J', 'P', 'G')) - return capture +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``). + + Historically returned a bare ``cv2.VideoCapture``. Callers that need the raw + OpenCV handle can use ``worker.source.cap``; prefer ``worker.read()``. + """ + capture_config = CaptureConfig.from_camera_dict( + normalize_video_source(video), + config.get("camera", {}), + reconnect=ReconnectPolicy(enabled=auto_reconnect), + verbose=verbose, + start_worker=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] + return cap class TactileProcessor: @@ -210,6 +220,7 @@ class FingerRuntime: tactile: TactileProcessor, frame_processor: FrameUnwarper, capture, + capture_worker: CaptureWorker | None = None, ): self.geometry = geometry self.domain = domain @@ -217,6 +228,7 @@ class FingerRuntime: self.sensor = tactile self.frame_processor = frame_processor self.capture = capture + self.capture_worker = capture_worker @classmethod def open( @@ -232,6 +244,8 @@ class FingerRuntime: flip_source_vertical=True, verbose=False, mirror=False, + auto_reconnect: bool = False, + capture_mode: str = "latest", ) -> "FingerRuntime": """Open a live finger runtime from export bundle, config, and video source.""" if camera_id is not None: @@ -263,9 +277,22 @@ class FingerRuntime: ) domain.remap.ensure_fast_maps() - capture = open_source(video, geometry.config) - if not capture.isOpened(): + capture_config = CaptureConfig.from_camera_dict( + video, + geometry.config.get("camera", {}), + mode=capture_mode, + reconnect=ReconnectPolicy(enabled=auto_reconnect), + verbose=verbose, + start_worker=False, + ) + capture_worker = open_capture(capture_config, start_worker=False) + if not capture_worker.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 sensor = Sensor("external", config_name=config, verbose=verbose, mirror=mirror) frame_processor = FrameUnwarper( @@ -283,6 +310,7 @@ class FingerRuntime: tactile=tactile, frame_processor=frame_processor, capture=capture, + capture_worker=capture_worker, ) @property @@ -290,6 +318,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() return self.capture.read() def process_frame(self, frame, *, color=False): @@ -323,6 +355,11 @@ class FingerRuntime: return ok, unrolled def close(self): - if self.capture is not None: + if self.capture_worker is not None: + self.capture_worker.close() + self.capture_worker = None + self.capture = None + elif self.capture is not None: self.capture.release() + self.capture = None self.tactile.disconnect() diff --git a/src/orisys/sensor.py b/src/orisys/sensor.py index a8dc898..e6a2c93 100644 --- a/src/orisys/sensor.py +++ b/src/orisys/sensor.py @@ -6,13 +6,12 @@ optical tactile sensor data with support for stitching, optical flow tracking, and Helmholtz-Hodge decomposition. """ -import os +# import os # only used by commented-out stitch_cal import numpy as np import cv2 import time -import threading from enum import Enum, auto -from typing import Tuple +from typing import Optional, Tuple # from threading import Thread, Event # from queue import Queue, Empty @@ -24,6 +23,14 @@ from .get_depth import Contact3D from .backend_availability import (check_cupy_available, check_opencl_available) from .frame_processors import create_frame_processor from .signal_processing import DEFAULT_SAMPLE_RATE_HZ, LowpassFilter +from .capture import ( + CaptureConfig, + CaptureWorker, + ReconnectPolicy, + is_live_camera, + normalize_source, + open_capture, +) class Sensor: """ @@ -61,7 +68,9 @@ class Sensor: def __init__(self, vid_src, cuda=True, isstitch=True, verbose=True, debug=False, config_name="prototype", cal_path=None, override=False, box=None, mirror=True, - backend='auto'): + 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"): """ Initialize Sensor with video source and configuration. @@ -79,8 +88,11 @@ class Sensor: "cupy": Use CuPy for GPU acceleration "opencl": Use OpenCL for GPU acceleration "cpu": Use CPU for computation - hhd_downsample: Downsample factor for flow field before HHD (default: 1, no downsampling). - Factor of 2 halves each dimension, giving ~4x speedup in Poisson solve. + 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``. """ self.CUPY_AVAILABLE = check_cupy_available() @@ -92,7 +104,7 @@ class Sensor: self._external_only = isinstance(vid_src, str) and vid_src.lower() == "external" # Store settings - self.vid_src = vid_src + self.vid_src = normalize_source(vid_src) if not self._external_only else vid_src self.isstitch = isstitch self.cuda = cuda self.verbose = verbose @@ -100,6 +112,11 @@ class Sensor: self.config_name = config_name self.cal_path = cal_path 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 @@ -194,21 +211,16 @@ class Sensor: self.masked_poisson_max_iters = self.__config_tracker.get("masked_poisson_max_iters", 3) self.masked_poisson_tol = self.__config_tracker.get("masked_poisson_tol", 1e-4) self.masked_poisson_reg = self.__config_tracker.get("masked_poisson_reg", 1e-8) - self._use_pyav = False - self._av_container = None - self._av_stream = None - self._av_decoder = None - self.cap = None + # Legacy PyAV capture state (unused after orisys.capture refactor). + # self._use_pyav = False + # self._av_container = None + # 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._is_live_camera = ( - isinstance(self.vid_src, int) - or (isinstance(self.vid_src, str) and self.vid_src.startswith("/dev/video")) - ) and not self._external_only - self._capture_thread = None - self._capture_running = False - self._latest_frame = None - self._latest_frame_seq = 0 - self._last_consumed_seq = -1 - self._latest_frame_lock = threading.Lock() + is_live_camera(self.vid_src) and not self._external_only + ) # Setup video capture unless this sensor is fed externally. if self._external_only: @@ -679,23 +691,30 @@ class Sensor: def disconnect(self): """Release resources and stop async threads.""" - self._capture_running = False - if self._capture_thread is not None: - self._capture_thread.join(timeout=1.0) - self._capture_thread = None - - if self._use_pyav: - if self._av_container is not None: - self._av_container.close() - self._av_container = None - self._av_stream = None - self._av_decoder = None - if self.verbose: - print("Sensor disconnected.") - elif self.cap is not None and self.cap.isOpened(): - self.cap.release() + if self._capture_worker is not None: + self._capture_worker.close() + self._capture_worker = None + self.cap = None if self.verbose: print("Sensor disconnected.") + # Legacy PyAV / direct-cap fallbacks (unused after orisys.capture refactor). + # elif self._use_pyav: + # if self._av_container is not None: + # self._av_container.close() + # self._av_container = None + # self._av_stream = None + # self._av_decoder = None + # if self.verbose: + # print("Sensor disconnected.") + # elif self.cap is not None: + # try: + # if self.cap.isOpened(): + # self.cap.release() + # except Exception: + # pass + # self.cap = None + # if self.verbose: + # print("Sensor disconnected.") cv2.destroyAllWindows() # ========================================================================= @@ -830,12 +849,13 @@ class Sensor: "compound_eye" if "stitcher" in self._config else "single_camera", ) self.__config_cam = self._config["camera"] - self._config_resolution_raw = self.__config_cam["resolution_raw"] - self._config_fps = self.__config_cam["fps"] + # Capture resolution/fps now applied via CaptureConfig.from_camera_dict(__config_cam). + # self._config_resolution_raw = self.__config_cam["resolution_raw"] + # self._config_fps = self.__config_cam["fps"] self._config_units = self.__config_cam.get("units", 1) self._config_resolution_stitch = self.__config_cam["resolution_stitch"] - self._config_dis_preset = self._config["tracker"]["dis_preset"] + # self._config_dis_preset = self._config["tracker"]["dis_preset"] self.__config_tracker = self._config["tracker"] self.__config_stitcher = self._config.get("stitcher") @@ -850,71 +870,34 @@ class Sensor: def _open_video_source(self, override, box): """Initialize video capture and calibration.""" - # _is_camera_index = isinstance(self.vid_src, int) - # _is_v4l2_path = isinstance(self.vid_src, str) and self.vid_src.startswith("/dev/video") - # use_linux_camera_pyav = ( - # platform.system() == "Linux" - # and (_is_camera_index or _is_v4l2_path) - # and av is not None - # ) - - # if use_linux_camera_pyav: - # device = self.vid_src if _is_v4l2_path else f"/dev/video{self.vid_src}" - # if self.verbose: - # print(f"Using PyAV Linux camera backend: {device}") - # try: - # self._av_container = av.open( - # device, - # format="v4l2", - # options={ - # "input_format": "mjpeg", - # "video_size": f"{self._config_resolution_raw[0]}x{self._config_resolution_raw[1]}", - # "framerate": str(self._config_fps), - # }, - # ) - # self._av_stream = self._av_container.streams.video[0] - # self._av_decoder = self._av_container.decode(self._av_stream) - # self._use_pyav = True - # except Exception as exc: - # if self.verbose: - # print(f"PyAV camera open failed ({exc}), fallback to OpenCV backend.") - # self._use_pyav = False - - self.use_pyav = False - if not self._use_pyav: - self.cap = cv2.VideoCapture(self.vid_src) - if self._is_live_camera: - self.cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter.fourcc('M', 'J', 'P', 'G')) - self.cap.set(cv2.CAP_PROP_FPS, self._config_fps) - self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, self._config_resolution_raw[0]) - self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self._config_resolution_raw[1]) - - if self.verbose: print("Connecting to video source...") - - + + if self._capture_config_override is not None: + capture_config = self._capture_config_override + else: + 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, + ) + + try: + self._capture_worker = open_capture(capture_config, start_worker=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 + if not self._is_source_open(): raise Exception("Failed to open video source.") - if self.verbose and not self._use_pyav and self.cap is not None: - fourcc_int = int(self.cap.get(cv2.CAP_PROP_FOURCC)) - fourcc_bytes = [(fourcc_int >> (8 * i)) & 0xFF for i in range(4)] - # Show printable FOURCC only; some backends return 0 or non-printable bytes. - if fourcc_int <= 0: - fourcc = "N/A" - else: - fourcc = "".join(chr(b) if 32 <= b <= 126 else "." for b in fourcc_bytes) - width = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH)) - height = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) - fps = self.cap.get(cv2.CAP_PROP_FPS) - backend_id = int(self.cap.get(cv2.CAP_PROP_BACKEND)) - print( - f"OpenCV negotiated stream: FOURCC='{fourcc}' " - f"(int={fourcc_int}, hex=0x{fourcc_int:08X}), " - f"size={width}x{height}, fps={fps:.2f}, backend_id={backend_id}" - ) - # Wait for first valid frame frame = None while frame is None: @@ -936,10 +919,9 @@ class Sensor: img = self.frame_processor.process(frame) self.initial_img = self._make_initial_gray(img) - # For live cameras, start a latest-frame capture thread to avoid latency buildup. - if self._is_live_camera: - self._start_capture_thread() - + # 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() def _make_initial_gray(self, img_bgr): gray = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2GRAY) @@ -978,73 +960,52 @@ class Sensor: print(f"Sensor: {status}") return n_boxes - def _extract_tiles(self, img_src): - """Extract tiles from source image.""" - self.stitcher.get_stitch_unit(img_src) - self.stitcher.extract_tiles() - return True + # Unused after frame_processor owns stitch extract/process. + # def _extract_tiles(self, img_src): + # """Extract tiles from source image.""" + # self.stitcher.get_stitch_unit(img_src) + # self.stitcher.extract_tiles() + # return True def _is_source_open(self): - if self._use_pyav: - return self._av_container is not None - return self.cap is not None and self.cap.isOpened() + if self._capture_worker is not None: + return self._capture_worker.source.is_open() + return False + # Legacy PyAV / direct-cap fallbacks (unused after orisys.capture refactor). + # if self._use_pyav: + # return self._av_container is not None + # return self.cap is not None and self.cap.isOpened() def _read_frame(self): - if self._capture_running: - # Startup race guard: wait briefly until capture thread publishes first frame. - if self._latest_frame is None: - deadline = time.perf_counter() + 1.0 - while time.perf_counter() < deadline: - with self._latest_frame_lock: - if self._latest_frame is not None: - break - time.sleep(0.001) - - # Return latest frame immediately (do not wait for next camera tick). - # This keeps end-to-end latency minimal; caller may receive repeated - # frames if processing runs faster than camera FPS. - with self._latest_frame_lock: - frame = self._latest_frame - seq = self._latest_frame_seq - if frame is not None: - self._last_consumed_seq = seq - return True, frame.copy() - return False, None + if self._capture_worker is not None and self._capture_worker.running: + return self._capture_worker.read(copy=True, wait_s=1.0) return self._read_frame_raw() def _read_frame_raw(self): - if self._use_pyav: - if self._av_decoder is None: - return False, None - try: - frame = next(self._av_decoder) - except StopIteration: - return False, None - except Exception: - return False, None - return True, frame.to_ndarray(format="bgr24") - if self.cap is None: - return False, None - return self.cap.read() + if self._capture_worker is not None: + return self._capture_worker.source.read_raw() + return False, None + # Legacy PyAV / direct-cap fallbacks (unused after orisys.capture refactor). + # if self._use_pyav: + # if self._av_decoder is None: + # return False, None + # try: + # frame = next(self._av_decoder) + # except StopIteration: + # return False, None + # except Exception: + # return False, None + # return True, frame.to_ndarray(format="bgr24") + # if self.cap is None: + # return False, None + # return self.cap.read() - def _start_capture_thread(self): - if self._capture_thread is not None: - return - self._capture_running = True - self._capture_thread = threading.Thread(target=self._capture_loop, daemon=True) - self._capture_thread.start() - if self.verbose: - print("Latest-frame capture thread started.") - - def _capture_loop(self): - while self._capture_running: - ret, frame = self._read_frame_raw() - if not ret: - time.sleep(0.001) - continue - with self._latest_frame_lock: - self._latest_frame = frame - self._latest_frame_seq += 1 + @property + def capture_status(self): + """Return capture health snapshot, or None in external-frame mode.""" + if self._capture_worker is None: + return None + return self._capture_worker.source.status # ========================================================================= diff --git a/tests/test_capture.py b/tests/test_capture.py new file mode 100644 index 0000000..9bd730c --- /dev/null +++ b/tests/test_capture.py @@ -0,0 +1,463 @@ +"""Unit tests for orisys.capture with fake VideoCapture objects (no hardware).""" + +from __future__ import annotations + +import threading +import time + +import numpy as np +import pytest + +from orisys.capture import ( + MAX_SIMULTANEOUS_CAPTURES, + CameraFilter, + CameraInfo, + CaptureConfig, + CaptureHealth, + CaptureManager, + CaptureMode, + CaptureSource, + FrameBuffer, + OverflowPolicy, + ReconnectPolicy, + list_cameras, + normalize_source, + open_capture, + pick_default_camera, +) +from orisys.capture.discovery import DEFAULT_LAPTOP_WEBCAM_PATTERNS + + +class FakeCap: + """Deterministic stand-in for cv2.VideoCapture.""" + + def __init__(self, frames=None, *, fail_after=None, fail_reads=0, open_ok=True): + self._frames = list(frames or []) + self._idx = 0 + self._open = bool(open_ok) + self._fail_after = fail_after + self._fail_reads_remaining = int(fail_reads) + self.read_count = 0 + self.release_count = 0 + self.width = 4 + self.height = 3 + self.fps = 30.0 + self.fourcc = 0 + + def isOpened(self): + return self._open + + def read(self): + self.read_count += 1 + if not self._open: + return False, None + if self._fail_reads_remaining > 0: + self._fail_reads_remaining -= 1 + return False, None + if self._fail_after is not None and self.read_count > self._fail_after: + self._open = False + return False, None + if self._idx >= len(self._frames): + # Repeat last frame like a live camera holding the stream. + if not self._frames: + return False, None + frame = self._frames[-1] + else: + frame = self._frames[self._idx] + self._idx += 1 + return True, frame.copy() + + def release(self): + self.release_count += 1 + self._open = False + + def get(self, prop_id): + # Values mirror common OpenCV property IDs without importing cv2 constants. + if prop_id in (3,): # CAP_PROP_FRAME_WIDTH + return float(self.width) + if prop_id in (4,): # CAP_PROP_FRAME_HEIGHT + return float(self.height) + if prop_id in (5,): # CAP_PROP_FPS + return float(self.fps) + if prop_id in (6,): # CAP_PROP_FOURCC + return float(self.fourcc) + return 0.0 + + def set(self, prop_id, value): + return True + + +def _frame(val: int) -> np.ndarray: + return np.full((3, 4, 3), val, dtype=np.uint8) + + +def _factory_from_caps(caps): + """Return a factory that hands out FakeCap instances in order.""" + lock = threading.Lock() + state = {"i": 0} + + def factory(_config): + with lock: + if state["i"] >= len(caps): + raise RuntimeError("No more fake captures") + cap = caps[state["i"]] + state["i"] += 1 + return cap, "FAKE" + + return factory + + +# --------------------------------------------------------------------------- +# Source normalization / config +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "video,expected", + [ + (None, 0), + ("0", 0), + (" 1 ", 1), + (2, 2), + ("/dev/video0", "/dev/video0"), + (r".\clip.mp4" + "\\", r".\clip.mp4"), + ], +) +def test_normalize_source(video, expected): + assert normalize_source(video) == expected + + +def test_capture_config_queue_bounds(): + with pytest.raises(ValueError): + CaptureConfig(source=0, queue_size=0) + with pytest.raises(ValueError): + CaptureConfig(source=0, queue_size=11) + cfg = CaptureConfig(source=0, mode="queue", queue_size=10) + assert cfg.queue_size == 10 + latest = CaptureConfig(source=0, mode="latest", queue_size=5) + assert latest.queue_size == 1 + + +# --------------------------------------------------------------------------- +# FrameBuffer +# --------------------------------------------------------------------------- + + +def test_frame_buffer_latest_overwrites(): + buf = FrameBuffer(mode=CaptureMode.LATEST) + buf.publish(_frame(1)) + buf.publish(_frame(2)) + ok, frame, seq = buf.read_latest(copy=True) + assert ok and seq == 2 + assert int(frame[0, 0, 0]) == 2 + + +def test_frame_buffer_queue_drop_oldest(): + buf = FrameBuffer(mode=CaptureMode.QUEUE, maxsize=2, overflow=OverflowPolicy.DROP_OLDEST) + buf.publish(_frame(1)) + buf.publish(_frame(2)) + buf.publish(_frame(3)) + assert buf.dropped_frames == 1 + ok, f1, _ = buf.read_next(copy=True) + ok2, f2, _ = buf.read_next(copy=True) + assert ok and ok2 + assert int(f1[0, 0, 0]) == 2 + assert int(f2[0, 0, 0]) == 3 + + +def test_frame_buffer_queue_drop_newest(): + buf = FrameBuffer(mode=CaptureMode.QUEUE, maxsize=1, overflow=OverflowPolicy.DROP_NEWEST) + buf.publish(_frame(1)) + buf.publish(_frame(2)) + assert buf.dropped_frames == 1 + ok, frame, _ = buf.read_next(copy=True) + assert ok and int(frame[0, 0, 0]) == 1 + + +# --------------------------------------------------------------------------- +# CaptureSource reconnect +# --------------------------------------------------------------------------- + + +def test_capture_source_read_and_status(): + frames = [_frame(7)] + cap = FakeCap(frames) + src = CaptureSource( + CaptureConfig(source=0, reconnect=ReconnectPolicy(enabled=False), start_worker=False), + factory=lambda cfg: (cap, "FAKE"), + ) + src.open() + assert src.is_open() + assert src.status.health == CaptureHealth.CONNECTED + ok, frame = src.read_raw() + assert ok and int(frame[0, 0, 0]) == 7 + src.release() + assert src.status.health == CaptureHealth.CLOSED + assert cap.release_count == 1 + + +def test_auto_reconnect_after_failed_reads(): + first = FakeCap([_frame(1)], fail_after=1) + second = FakeCap([_frame(9)]) + factory = _factory_from_caps([first, second]) + policy = ReconnectPolicy( + enabled=True, + fail_threshold=1, + initial_backoff_s=0.01, + max_backoff_s=0.01, + stale_timeout_s=None, + max_attempts=3, + ) + src = CaptureSource( + CaptureConfig(source=0, reconnect=policy, start_worker=False, verbose=False), + factory=factory, + ) + src.open() + ok, _ = src.read_raw() + assert ok + # Next read fails and closes first cap. + ok, _ = src.read_raw() + assert not ok + assert src.should_reconnect() + assert src.reconnect() + ok, frame = src.read_raw() + assert ok and int(frame[0, 0, 0]) == 9 + assert src.status.health == CaptureHealth.CONNECTED + assert src.status.reconnect_attempts >= 1 + + +def test_reconnect_disabled_does_not_reopen(): + cap = FakeCap([_frame(1)], fail_after=1) + src = CaptureSource( + CaptureConfig( + source=0, + reconnect=ReconnectPolicy(enabled=False, fail_threshold=1), + start_worker=False, + ), + factory=lambda cfg: (cap, "FAKE"), + ) + src.open() + src.read_raw() + src.read_raw() + assert not src.should_reconnect() + + +def test_reconnect_attempts_exhausted(): + good = FakeCap([_frame(1)], fail_after=1) + + state = {"n": 0} + + def factory(_cfg): + state["n"] += 1 + if state["n"] == 1: + return good, "FAKE" + raise RuntimeError("camera missing") + + policy = ReconnectPolicy( + enabled=True, + fail_threshold=1, + initial_backoff_s=0.01, + max_backoff_s=0.01, + stale_timeout_s=None, + max_attempts=2, + ) + src = CaptureSource( + CaptureConfig(source=0, reconnect=policy, start_worker=False), + factory=factory, + ) + src.open() + src.read_raw() + src.read_raw() + assert src.should_reconnect() + assert not src.reconnect() + assert not src.reconnect() + assert src.status.health == CaptureHealth.FAILED + + +# --------------------------------------------------------------------------- +# CaptureWorker +# --------------------------------------------------------------------------- + + +def test_worker_latest_mode_publishes_frames(): + caps = [FakeCap([_frame(i) for i in range(5)])] + worker = open_capture( + CaptureConfig( + source=0, + mode=CaptureMode.LATEST, + start_worker=True, + reconnect=ReconnectPolicy(enabled=False), + ), + factory=_factory_from_caps(caps), + start_worker=True, + ) + try: + deadline = time.perf_counter() + 2.0 + ok = False + frame = None + while time.perf_counter() < deadline: + ok, frame = worker.read(wait_s=0.05) + if ok: + break + assert ok and frame is not None + finally: + worker.close() + + +def test_worker_reconnect_in_background(): + first = FakeCap([_frame(1)], fail_after=2) + second = FakeCap([_frame(42) for _ in range(20)]) + policy = ReconnectPolicy( + enabled=True, + fail_threshold=1, + initial_backoff_s=0.01, + max_backoff_s=0.02, + stale_timeout_s=None, + ) + worker = open_capture( + CaptureConfig(source=0, mode="latest", reconnect=policy, start_worker=True), + factory=_factory_from_caps([first, second]), + start_worker=True, + ) + try: + saw_reconnect = False + saw_new = False + deadline = time.perf_counter() + 3.0 + while time.perf_counter() < deadline: + if worker.source.status.reconnect_attempts > 0: + saw_reconnect = True + ok, frame = worker.read(wait_s=0.05) + if ok and frame is not None and int(frame[0, 0, 0]) == 42: + saw_new = True + break + time.sleep(0.01) + assert saw_reconnect + assert saw_new + finally: + worker.close() + + +# --------------------------------------------------------------------------- +# CaptureManager +# --------------------------------------------------------------------------- + + +def test_capture_manager_capacity_and_status(): + mgr = CaptureManager(max_sources=2) + caps = [FakeCap([_frame(1)]), FakeCap([_frame(2)]), FakeCap([_frame(3)])] + factory = _factory_from_caps(caps) + try: + mgr.open(CaptureConfig(source=0, start_worker=False), factory=factory) + mgr.open(CaptureConfig(source=1, start_worker=False), factory=factory) + assert len(mgr) == 2 + with pytest.raises(RuntimeError): + mgr.open(CaptureConfig(source=2, start_worker=False), factory=factory) + statuses = mgr.status_all() + assert set(statuses) == {"0", "1"} + assert all(s.health == CaptureHealth.CONNECTED for s in statuses.values()) + finally: + mgr.close_all() + assert len(mgr) == 0 + + +def test_capture_manager_max_constant(): + assert MAX_SIMULTANEOUS_CAPTURES == 10 + with pytest.raises(ValueError): + CaptureManager(max_sources=11) + + +def test_capture_manager_read_latest_all(): + mgr = CaptureManager(max_sources=3) + caps = [FakeCap([_frame(10)]), FakeCap([_frame(20)])] + factory = _factory_from_caps(caps) + try: + w0 = mgr.open( + CaptureConfig(source=0, mode="latest", start_worker=True), + factory=factory, + ) + w1 = mgr.open( + CaptureConfig(source=1, mode="latest", start_worker=True), + factory=factory, + ) + deadline = time.perf_counter() + 2.0 + while time.perf_counter() < deadline: + if w0.buffer.last_seq > 0 and w1.buffer.last_seq > 0: + break + time.sleep(0.01) + frames = mgr.read_latest_all() + assert "0" in frames and "1" in frames + finally: + mgr.close_all() + + +# --------------------------------------------------------------------------- +# Camera discovery filters (no hardware) +# --------------------------------------------------------------------------- + + +def test_camera_filter_excludes_laptop_webcams(): + cams = [ + CameraInfo(0, "DSHOW", 0, 640, 480, name="Integrated Webcam"), + CameraInfo(1, "DSHOW", 0, 1920, 1080, name="Orisys Tactile Cam"), + CameraInfo(2, "DSHOW", 0, 1280, 720, name="USB Camera"), + ] + filt = CameraFilter(exclude_laptop_webcams=True) + kept = [c for c in cams if filt.matches(c)] + assert [c.index for c in kept] == [1, 2] + + +def test_camera_filter_name_include_and_vid_pid(): + cams = [ + CameraInfo(0, "V4L2", 0, 640, 480, name="foo", vendor_id=0x1234, product_id=0x5678), + CameraInfo(1, "V4L2", 0, 640, 480, name="orisys-finger", vendor_id=0xAAAA, product_id=0xBBBB), + ] + filt = CameraFilter(name_include=[r"orisys"], vid_pid_allowlist=[(0xAAAA, 0xBBBB)]) + kept = [c for c in cams if filt.matches(c)] + assert len(kept) == 1 and kept[0].index == 1 + + +def test_camera_filter_min_resolution(): + cams = [ + CameraInfo(0, "DSHOW", 0, 320, 240, name="tiny"), + CameraInfo(1, "DSHOW", 0, 1280, 720, name="hd"), + ] + filt = CameraFilter(min_width=640, min_height=480) + kept = [c for c in cams if filt.matches(c)] + assert [c.index for c in kept] == [1] + + +def test_pick_default_camera(): + cams = [ + CameraInfo(1, "DSHOW", 0, 640, 480), + CameraInfo(2, "DSHOW", 0, 640, 480), + ] + assert pick_default_camera(cams, preferred="2") == "2" + assert pick_default_camera(cams, preferred=None) == "1" + assert pick_default_camera([], preferred="0") is None + + +def test_laptop_patterns_nonempty(): + assert len(DEFAULT_LAPTOP_WEBCAM_PATTERNS) >= 5 + + +def test_camera_info_label_includes_name(): + cam = CameraInfo(3, "DSHOW", 700, 1280, 720, name="Orisys") + assert "Orisys" in cam.label() + assert cam.source == "3" + + +def test_list_cameras_is_callable(): + # Smoke: function import works; may return [] without hardware. + assert callable(list_cameras) + + +def test_sensor_external_mode_still_works(): + from orisys import Sensor + + sensor = Sensor("external", config_name="box", verbose=False, backend="cpu") + try: + assert sensor._external_only + assert sensor.capture_status is None + assert sensor._capture_worker is None + finally: + sensor.disconnect()