examples/basic_pipeline.py

@@ -28,12 +28,9 @@ def main():
     # cal_path 为标定文件路径；每台设备建议使用独立的标定文件
 
     # 输入既可以是摄像头编号，也可以是视频文件路径
-    try:
+    if args.video.isdigit():
-        input = int(args.video)
+        sensor = orisys.Sensor(int(args.video), config_name=args.config, verbose=args.verbose)
-        is_camera = True
+    else:
-        sensor = orisys.Sensor(input, config_name=args.config, verbose=args.verbose)
-    except:
-        is_camera = False
         sensor = orisys.Sensor(args.video, config_name=args.config, verbose=args.verbose)
 
     print("\n按键说明：'q' 退出程序，'r' 重置追踪器。")

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,<index>,<path>,<WxH>
+  OK search <N>
+  OK connect <source>
+  OK run
+  force,<fnormal>,<fshearx>,<fsheary>
+  OK eof              Video file ended (when --no-loop-video)
+  OK stop
+  OK quit
+  ERR <context> <message>
+
+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())

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",

src/orisys/capture/buffer.py

@@ -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)

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
+    background: bool = True
-    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
@@ -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,

src/orisys/capture/manager.py

@@ -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()

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)
+            _apply_stream_props(
-            if code is not None:
+                cap,
-                cap.set(cv2.CAP_PROP_FOURCC, code)
+                width=width,
-            if fps is not None:
+                height=height,
-                cap.set(cv2.CAP_PROP_FPS, float(fps))
+                fps=fps,
-            if width is not None:
+                fourcc=fourcc,
-                cap.set(cv2.CAP_PROP_FRAME_WIDTH, float(width))
+            )
-            if height is not None:
+            _warmup_capture(cap)
-                cap.set(cv2.CAP_PROP_FRAME_HEIGHT, float(height))
+            _, _, _, 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)
+            width, height, fps, fourcc = _read_stream_meta(cap)
-            self._status.height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0)
+            self._status.width = width
-            self._status.fps = float(cap.get(cv2.CAP_PROP_FPS) or 0.0)
+            self._status.height = height
-            self._status.fourcc = _decode_fourcc(cap.get(cv2.CAP_PROP_FOURCC) or 0)
+            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

src/orisys/capture/worker.py

@@ -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

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)
+    source = open_capture(capture_config, start_background=False)
-    # Preserve prior attribute access: treat returned object like VideoCapture for
+    cap = source.cap
-    # simple .read()/.release()/.isOpened() callers, while exposing capture worker.
+    cap._orisys_worker = source  # type: ignore[attr-defined]
-    cap = worker.source.cap
-    cap._orisys_worker = worker  # 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)
+        capture_source = open_capture(capture_config, start_background=False)
-        if not capture_worker.source.is_open():
+        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_source.start_background()
-        capture = capture_worker.source.cap
+        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:
+        if self.capture_source is not None and self.capture_source.running:
-            return self.capture_worker.read(copy=True, wait_s=1.0)
+            return self.capture_source.read(copy=True, wait_s=1.0)
-        if self.capture_worker is not None:
+        if self.capture_source is not None:
-            return self.capture_worker.source.read_raw()
+            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:
+        if self.capture_source is not None:
-            self.capture_worker.close()
+            self.capture_source.close()
+            self.capture_source = None
             self.capture_worker = None
             self.capture = None
         elif self.capture is not None:

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_mode: Deprecated; ignored (latest-frame capture only).
-            capture_queue_size: Bounded queue size for queue mode (1..10).
+            capture_queue_size: Deprecated; ignored.
-            capture_overflow: ``drop_oldest`` | ``drop_newest`` | ``block``.
+            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:
+        if self._capture_source is not None:
-            self._capture_worker.close()
+            self._capture_source.close()
-            self._capture_worker = None
+            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.cap = self._capture_source.cap
-        # self._use_pyav = False
 
         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.
+        # For live cameras, start background capture to avoid latency buildup.
-        if self._is_live_camera and self._capture_worker is not None:
+        if self._is_live_camera and self._capture_source is not None:
-            self._capture_worker.start()
+            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,13 +1007,19 @@ 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
         self.stitch_valid = self.isstitch and n_boxes == self._config_units
@@ -1023,8 +1036,8 @@ class Sensor:
     #     return True
 
     def _is_source_open(self):
-        if self._capture_worker is not None:
+        if self._capture_source is not None:
-            return self._capture_worker.source.is_open()
+            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:
+        if self._capture_source is not None and self._capture_source.running:
-            return self._capture_worker.read(copy=True, wait_s=1.0)
+            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:
+        if self._capture_source is not None:
-            return self._capture_worker.source.read_raw()
+            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
 
 
     # =========================================================================

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())

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():
+def test_capture_config_from_camera_dict_ignores_deprecated_kwargs():
-    with pytest.raises(ValueError):
+    cfg = CaptureConfig.from_camera_dict(
-        CaptureConfig(source=0, queue_size=0)
+        0,
-    with pytest.raises(ValueError):
+        {"resolution_raw": [1920, 1080], "fps": 30},
-        CaptureConfig(source=0, queue_size=11)
+        mode="queue",
-    cfg = CaptureConfig(source=0, mode="queue", queue_size=10)
+        queue_size=5,
-    assert cfg.queue_size == 10
+        overflow="drop_newest",
-    latest = CaptureConfig(source=0, mode="latest", queue_size=5)
+        start_worker=False,
-    assert latest.queue_size == 1
+    )
+    assert cfg.width == 1920
+    assert cfg.height == 1080
+    assert cfg.background is False
 
 
-# ---------------------------------------------------------------------------
+def test_apply_stream_props_sets_size_before_fourcc():
-# FrameBuffer
+    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():
+def test_warmup_capture_grabs():
-    buf = FrameBuffer(mode=CaptureMode.LATEST)
+    cap = FakeCap([_frame(1)])
-    buf.publish(_frame(1))
+    _warmup_capture(cap, grabs=3)
-    buf.publish(_frame(2))
+    assert cap.grab_count == 3
-    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),
+        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
 
 
-# ---------------------------------------------------------------------------
+def test_background_thread_publishes_latest_frame():
-# CaptureWorker
-# ---------------------------------------------------------------------------
-
-
-def test_worker_latest_mode_publishes_frames():
     caps = [FakeCap([_frame(i) for i in range(5)])]
-    worker = open_capture(
+    src = open_capture(
         CaptureConfig(
             source=0,
-            mode=CaptureMode.LATEST,
+            background=True,
-            start_worker=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(
+    src = open_capture(
-        CaptureConfig(source=0, mode="latest", reconnect=policy, start_worker=True),
+        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()
 
 
-# ---------------------------------------------------------------------------
+def test_warmup_frames_discards_reads():
-# CaptureManager
+    cap = FakeCap([_frame(i) for i in range(10)])
-# ---------------------------------------------------------------------------
+    src = CaptureSource(
-
+        CaptureConfig(source=0),
-
+        factory=lambda cfg: (cap, "FAKE"),
-def test_capture_manager_capacity_and_status():
+    )
-    mgr = CaptureManager(max_sources=2)
+    src.open()
-    caps = [FakeCap([_frame(1)]), FakeCap([_frame(2)]), FakeCap([_frame(3)])]
+    src.warmup_frames(3)
-    factory = _factory_from_caps(caps)
+    ok, frame = src.read_raw()
-    try:
+    assert ok and int(frame[0, 0, 0]) == 3
-        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():
@@ -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