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