From 07339c337425049036ee2c2cb1bc050554466e26 Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Wed, 14 Oct 2026 15:39:21 +0800 Subject: [PATCH] =?UTF-8?q?=E7=BB=9F=E4=B8=80=E5=A4=9A=E7=9B=B8=E6=9C=BA?= =?UTF-8?q?=E8=BF=9E=E6=8E=A5=E4=B8=8E=20UART=20=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=8D=8F=E8=AE=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 启动时完整连接所有配置相机，移除旧命令与单路数据格式。通过 22 项回归测试、真实 SDK 的 1–4 路录像验证、Ruff 和文档检查。 --- docs/uart.md | 200 +++++++++++---------- serial/example_uart.py | 245 +++++++------------------ serial/finger_api.py | 94 ++++++---- tests/check_backend.py | 78 ++++++++ tests/test_uart.py | 400 ++++++++++++++++++----------------------- 5 files changed, 476 insertions(+), 541 deletions(-) create mode 100644 tests/check_backend.py diff --git a/docs/uart.md b/docs/uart.md index 06d6f32..4dc5f32 100644 --- a/docs/uart.md +++ b/docs/uart.md @@ -1,141 +1,145 @@ # UART finger service -`serial/example_uart.py` processes 1–4 finger cameras or recorded videos and -publishes the SDK's normal and shear values over a separate data UART. Python -3.10+, the board's installed `orisys` SDK and `pyserial` are required. Run inside -the board environment prepared by `setup/sdk_setup.sh`; install `pyserial` there -with `python -m pip install pyserial` if needed. +Start the service with an explicit list of 1-4 camera sources. It opens and +warms up **every source first**, then opens the data UART and starts publishing. +If any source fails to prepare, all opened sources are closed and the process +exits with status 1. No partial camera set is published. -## Launch from the login console +## Connect the cameras and start once -The TTL login console and data UART must use **different device ports**. Start -the script in the login shell; connect the data receiver to the port selected -by `--port`. A shell/getty or another serial program must not own that data port. -The default is `/dev/ttyHS1`, 115200 baud, 8N1, with no flow control. This service -does not configure pin multiplexing or enable the board's UART hardware. +Plug the USB cameras into the Q6A. In its TTL login shell, identify their actual +capture device paths before launching; one USB camera can expose more than one +`/dev/video*` node, so do not assume every node represents a different camera. +For example, `v4l2-ctl --list-devices` shows device groups. Select one usable +capture node per camera. Use the board environment prepared by `setup/sdk_setup.sh`; +it needs Python 3.10+, the `orisys` SDK and `pyserial`. -From the board's `dragon-q6a` checkout: +Run from the board's `dragon-q6a` directory. These paths are examples, not detected +hardware identities: ```bash -# Host commands control connection and streaming (default: one camera). -python serial/example_uart.py --port /dev/ttyHS1 --config finger +# One camera. +python serial/example_uart.py --port /dev/ttyHS1 --sources /dev/video0 -# Four configured sources; use 1, 2, 3 or 4 entries as required. -python serial/example_uart.py --port /dev/ttyHS1 --sources 0 2 4 6 +# Two cameras. +python serial/example_uart.py --port /dev/ttyHS1 --sources /dev/video0 /dev/video2 -# A passive receiver needs only to read: connection and streaming start automatically. -python serial/example_uart.py --port /dev/ttyHS1 --sources 0 2 --read-only +# Three cameras. +python serial/example_uart.py --port /dev/ttyHS1 --sources /dev/video0 /dev/video2 /dev/video4 -# Start streaming immediately, while still accepting UART commands. -python serial/example_uart.py --port /dev/ttyHS1 --sources 0 2 --autostart +# Four cameras. +python serial/example_uart.py --port /dev/ttyHS1 --sources /dev/video0 /dev/video2 /dev/video4 /dev/video6 ``` -Use actual camera indices/device paths discovered on the board; `0 2 4 6` is -only an example. Slot IDs are fixed by `--sources` order, not camera index. -Repeat `--config` once per source for different sensor configurations, or supply -one shared configuration. Existing `--camera-id` and `--video` remain available -for a single source. `--sources` replaces those options. `--show` requires a local -GUI and is off by default. Logs go to the login console, replies to the data UART. +The first source is sensor 1, the second sensor 2, and so on. The number of listed +sources determines the sensor count. Camera indices such as `--sources 0 2` are +also accepted. Sources must be distinct. A shared `--config finger` is the +default; repeat `--config` once per source to use different configurations. +The login console prints each sensor's source/config mapping during preparation. +There is no UART connection command: physical connection and source selection +happen before startup, and software connection happens automatically. -In `--read-only` mode, all incoming UART commands are ignored. End the process -with Ctrl-C in the login shell. Startup still emits `OK`/`ERR` records; the receiver -should select `force,` lines. Without `--autostart` or `--read-only`, a passive -reader receives only `OK ready` until a host sends `connect` and `run`. +The TTL login console and data UART must use **different ports**. The data +UART defaults to `/dev/ttyHS1`, 115200 baud, 8N1 without flow control. It must +not be occupied by getty or another serial program. This script does not enable +UART pins or configure the board's hardware. Logs stay in the login console; +protocol records go to the data UART. `--show` is optional desktop debugging. -## Command protocol +For a passive receiver, append `--read-only` to the same startup command. Startup +and output are identical, but incoming UART commands are discarded. Exit using +Ctrl-C in the login shell. The receiver should filter `force,` records because +startup and failures also produce status records. -Commands are UTF-8, case-insensitive, terminated by LF, CR or CRLF. Incomplete -lines remain buffered across serial timeouts. Lines exceeding 1024 bytes are -discarded through their terminator and produce `ERR command ...` in control mode. -Quote file paths containing spaces. Linux device paths are recommended on Q6A. +## One protocol for every sensor count -| Command | Behavior | -|---|---| -| `search` | List cameras, then `OK search N`; use before connecting. Rejected while streaming. | -| `connect` | Connect slot 1 using its configured source; in single-source mode, prefer the latest search result for a camera. | -| `connect SOURCE` | Legacy form: connect slot 1 to this camera index, device path or file. | -| `connect ID SOURCE` | Connect/replace one slot; stop streaming that slot. Other slots keep their connections. | -| `connect all` | Connect all configured slots; report each success/failure independently. | -| `run [ID\|all]` | Stream the selected slot, or all connected slots when omitted. Requires a connection. | -| `read [ID\|all]` | Publish one newly processed measurement per selected slot; defaults to all configured slots. Does not start streaming. | -| `stop [ID\|all]` | Stop selected streaming, or all when omitted; keep cameras connected. | -| `quit` / `exit` | Exit and release all camera and UART resources. | - -`connect 2` means **camera index 2 on slot 1**, preserving the old protocol. -To connect slot 2, send `connect 2 /dev/video2` or use `connect all`. -There are no motor-control commands; these commands control acquisition only. - -Single-source measurements retain the original format: +After all sources are ready, the data UART reports `OK ready N`, then immediately +streams each complete processing round in source order. Every measurement has +five CSV fields, including a sensor ID even when only one camera is configured: ```text -force,1.250000,-2.500000,3.750000 -``` - -When configured with multiple sources, measurements include the slot ID even if -only one source connected successfully: - -```text -OK connect 2 /dev/video2 (camera) -OK run 1 2 +OK ready 2 force,1,1.250000,-2.500000,3.750000 force,2,0.500000,0.100000,-0.200000 -ERR run 2 processing failed -OK eof 1 ``` -Force fields are `FNORMAL`, `FSHEARX`, `FSHEARY` in SDK axes. They may be raw -or calibrated; **do not assume Newtons**. The Python API exposes SDK units per -reading, using `unknown` if an older SDK does not expose them. CSV preserves -the original fields and carries no units, timestamps or cross-sensor synchronization. +The fields are `force,sensor_id,FNORMAL,FSHEARX,FSHEARY`. Axes follow the SDK; +units depend on calibration and must not be assumed to be Newtons. CSV carries +no timestamp or units. Frames are processed sequentially, not synchronized +across cameras. All inputs for a round must succeed before that round is +published. A transport failure can still interrupt transmission between lines. -Each processing failure or EOF stops only the affected slot. Processing exceptions -produce `ERR`, never fabricated zero force. `--no-loop-video` stops files at EOF; -the default tries one rewind, resets the tracker and retries one frame. Failed -preparation releases the newly opened runtime; replacing a connection releases -the old one before opening the replacement. A failed replacement requires retrying -`connect`. Startup continues with any successfully connected slots and fails if -none connect. Reconnect and run explicitly after a camera failure. +Control mode accepts only four commands, each acting on the entire camera set: + +| Command | Result | +|---|---| +| `stop` | Pause continuous output and reply `OK stop`; keep cameras open. | +| `start` | Resume continuous output and reply `OK start`. | +| `read` | Process and publish one complete round; preserve streaming state on success. | +| `quit` | Reply `OK quit`, close all cameras and UART, then exit. | + +Commands take no arguments. They are case-insensitive UTF-8 lines terminated by +LF, CR or CRLF. Partial lines survive read timeouts; commands exceeding 1024 +bytes are discarded through their terminator. Unknown commands or arguments +produce `ERR`. The former protocol and source-selection options have been +removed; receivers must use this single CSV format and explicit `--sources`. + +A capture or processing failure produces `ERR read sensor ID: ...` and pauses +output for the whole set. Measurements are never replaced with zeros. Sources +remain open; `start` retries processing, or restart the service to reopen them. +For recorded files, `--no-loop-video` reports `OK eof sensor ID: video ended` +and pauses the whole set at EOF. Otherwise, one rewind and tracker reset is +attempted per read. Ctrl-C always ends the process from the login console. ## Callable acquisition API -`serial/finger_api.py` owns acquisition and runtime cleanup; importing it opens -no UART or camera. Place that directory on the caller's module search path: +`serial/finger_api.py` has no UART dependency and opens no resources on import. +Put its directory on the caller's module search path. Use the same connection-first +lifecycle as the service: ```python from finger_api import FingerSensorBank bank = FingerSensorBank([0, 2], configs=["finger"]) try: - bank.connect(1) - bank.connect(2) - reading = bank.read(1) - if reading is not None: - print(reading.normal, reading.shear_x, reading.shear_y, reading.units) + bank.connect_all() + for reading in bank.read_all(): + print(reading.sensor_id, reading.normal, reading.shear_x, reading.shear_y, reading.units) finally: bank.close() ``` -The caller owns one thread and must close the bank. `read()` returns `None` on -capture failure/EOF and raises on processing failure. It returns an image and -three scalar forces from the same processed frame. One file or camera is read -at a time; the API introduces no workers beyond those owned by the SDK. +One thread owns the bank. `connect_all()` rolls back every opened runtime on +failure. `read_all()` returns a tuple only when every source succeeds; it raises +`EOFError` for file EOF and `RuntimeError` for capture/processing failure. Each +reading contains the unrolled image, three scalar forces and SDK units from its +own runtime. Older SDKs without a units property report `unknown`. Reconnection +requires closing the bank before calling `connect_all()` again. -## Verification and limits +## Backend evidence and remaining limits -Hardware-free regression tests cover command fragmentation, CR/LF, legacy output, -1–4 slot identities, selective control, failure isolation, read-only startup and -cleanup. They use in-memory serial and runtime fakes, not physical devices: +The SDK's `FingerRuntime.open()` constructs a separate capture object, +`FrameUnwarper`, `TactileProcessor` and `Sensor` for each source. Live sources +have separate SDK capture workers. The service holds all runtimes open together +and reads them sequentially. SDK implementation is unchanged. + +Verification on 2026-10-14 used local SDK revision +`3953747f1726b197a3f327a38c65d9878e333e12`: real SDK instances processed 1, 2, 3 +and 4 distinct recorded inputs, three complete rounds for each count. Capture +handles and sensor objects were distinct; optical-flow arrays did not share +memory. All capture resources were released after each case. Run the same check +with a recording containing at least 12 frames; put generated clips and caches +in an isolated task scratch directory: ```bash +python tests/check_backend.py --recording /path/to/recording.avi --scratch .scratch/uart-connect-first/backend-run python -m unittest discover -s tests -v ``` -Q6A UART wiring, installed ARM SDK compatibility, camera bandwidth, sustained -frame rates and latency remain unverified. The shipped ARM wheel is compiled -`orisys 0.5.0`; local tests do not execute it. SDK capture and processing remain -synchronous from this service's perspective. Commands are checked before each -round; a slow grab, processing call, connection or camera scan can delay them. -Serial writes have a finite timeout (default 1 second), and transport failure -terminates the service with cleanup. At 115200 baud/8N1, the data budget is -11,520 bytes/s, shared by all measurements and replies. Measure actual frame -rates and stop latency on the board before choosing a higher baud rate. +Unit tests separately verify startup ordering, full rollback, a uniform protocol, +removed commands/options, command framing and failure behavior. These tests do +not prove Q6A USB bandwidth or the shipped compiled ARM `orisys 0.5.0` wheel. +Run the backend check in the board's installed environment to verify that SDK. +Four live cameras, serial wiring, sustained frame rates and command latency +still require hardware validation. A slow capture/processing call delays command +handling. Serial writes time out after a configurable interval (default 1 second). +At 115200 baud/8N1, all data and replies share 11,520 bytes/s. diff --git a/serial/example_uart.py b/serial/example_uart.py index 43480b2..67c5d35 100644 --- a/serial/example_uart.py +++ b/serial/example_uart.py @@ -1,8 +1,8 @@ #!/usr/bin/env python3 """Headless UART finger service for 1-4 sensors; see ../docs/uart.md. -The login console and data UART must be separate ports. Single-sensor force -records retain the original format; multi-sensor records add a 1-based ID. +The login console and data UART must be separate ports. Every configured camera +is prepared before UART output. All force records include a 1-based sensor ID. Importing this module opens no devices; acquisition is owned by finger_api. """ @@ -12,14 +12,12 @@ import argparse import logging import math import os -import shlex import sys from collections import deque -from pathlib import Path os.environ.setdefault("OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS", "0") -from finger_api import FingerSensorBank, is_video_file, parse_source +from finger_api import FingerSensorBank, parse_source logger = logging.getLogger(__name__) @@ -28,32 +26,21 @@ class UartFingerController: """Single-threaded command service; the caller owns serial and final cleanup.""" def __init__( - self, ser, *, config="finger", default_source=0, sources=None, - configs=None, max_camera_index=10, loop_video=True, show=False, - verbose=False, read_only=False, autostart=False, + self, ser, *, sources, configs=("finger",), loop_video=True, + show=False, verbose=False, read_only=False, ): self.ser = ser - self.bank = FingerSensorBank( - sources if sources is not None else (default_source,), - configs=configs if configs is not None else (config,), verbose=verbose, - ) - self.max_camera_index = max_camera_index + self.bank = FingerSensorBank(sources, configs=configs, verbose=verbose) self.loop_video = loop_video self.show = show self.verbose = verbose self.read_only = read_only - self.autostart = autostart or read_only - self._streaming = set() + self._streaming = False self._should_quit = False - self._last_search_source = None self._partial = bytearray() self._lines = deque() self._discard_line = False - @property - def multi_sensor(self): - return len(self.bank.sources) > 1 - def reply(self, line): # SDK exceptions can contain newlines; every reply must be one record. text = " ".join(str(line).splitlines()) + "\n" @@ -64,7 +51,7 @@ class UartFingerController: print(f"<< {text.rstrip()}", file=sys.stderr, flush=True) def close_runtime(self): - self._streaming.clear() + self._streaming = False try: self.bank.close() finally: @@ -99,112 +86,27 @@ class UartFingerController: self._discard_line = True return self._lines.popleft() if self._lines else None - def _targets(self, args, *, connected=False): - if len(args) > 1: - raise ValueError("expected [sensor-id|all]") - if args and args[0].lower() != "all": - sensor_id = int(args[0]) - self.bank.validate_id(sensor_id) - return [sensor_id] - return sorted(self.bank.runtimes if connected else self.bank.sources) - - def _sensor_suffix(self, sensor_id): - return f" {sensor_id}" if self.multi_sensor else "" - - def cmd_search(self, args): - if args: - raise ValueError("search takes no arguments") - if self._streaming: - raise ValueError("stop streaming before camera search") + def _read_round(self): try: - from orisys.capture import list_cameras - - cameras = list_cameras(max_index=self.max_camera_index) + readings = self.bank.read_all(loop_video=self.loop_video) + except EOFError as exc: + self._streaming = False + self.reply(f"OK eof {exc}") + return except Exception as exc: - logger.exception("Camera search failed") - self.reply(f"ERR search {exc}") + logger.exception("Acquisition round failed") + self._streaming = False + self.reply(f"ERR read {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}" - self.reply(f"camera,{cam.index},{path},{cam.width}x{cam.height}") - 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): - # Legacy `connect ` targets sensor 1. `connect ` - # selects a slot explicitly; quote source paths containing spaces. - if args and args[0].lower() == "all": - if len(args) != 1: - raise ValueError("expected connect all") - for sensor_id in self.bank.sources: - self._connect_one(sensor_id) - return - sensor_id = 1 - if len(args) >= 2 and args[0].isdigit(): - sensor_id = int(args[0]) - args = args[1:] - source = parse_source(" ".join(args)) if args else None - if (source is None and not self.multi_sensor and self._last_search_source is not None - and not is_video_file(self.bank.sources[1])): - source = self._last_search_source - self._connect_one(sensor_id, source) - - def _connect_one(self, sensor_id, source=None): - self.bank.validate_id(sensor_id) - self._streaming.discard(sensor_id) - try: - source = self.bank.connect(sensor_id, source) - except Exception as exc: - logger.exception("Sensor %s connection failed", sensor_id) - self.reply(f"ERR connect{self._sensor_suffix(sensor_id)} {exc}") - return - kind = "file" if is_video_file(source) else "camera" - self.reply(f"OK connect{self._sensor_suffix(sensor_id)} {source} ({kind})") - - def cmd_run(self, args): - targets = self._targets(args, connected=True) - if not targets or any(i not in self.bank.runtimes for i in targets): - raise ValueError("not connected") - if set(targets).issubset(self._streaming): - raise ValueError("already running") - self._streaming.update(targets) - self.reply("OK run" + (f" {' '.join(map(str, targets))}" if self.multi_sensor else "")) - - def cmd_stop(self, args): - targets = self._targets(args) - self._streaming.difference_update(targets) - self.reply("OK stop" + (f" {' '.join(map(str, targets))}" if self.multi_sensor else "")) - - def cmd_quit(self, args): - if args: - raise ValueError("quit takes no arguments") - self._streaming.clear() - self._should_quit = True - self.reply("OK quit") - - def _read_one(self, sensor_id, *, context): - try: - reading = self.bank.read(sensor_id, loop_video=self.loop_video) - except Exception as exc: - logger.exception("Sensor %s acquisition failed", sensor_id) - self._streaming.discard(sensor_id) - self.reply(f"ERR {context}{self._sensor_suffix(sensor_id)} {exc}") - return - if reading is None: - self._streaming.discard(sensor_id) - if is_video_file(self.bank.sources[sensor_id]): - self.reply(f"OK eof{self._sensor_suffix(sensor_id)}") - else: - self.reply(f"ERR {context}{self._sensor_suffix(sensor_id)} grab failed") - return - sensor_prefix = f"{sensor_id}," if self.multi_sensor else "" - self.reply( - f"force,{sensor_prefix}{reading.normal:.6f},{reading.shear_x:.6f},{reading.shear_y:.6f}" - ) - if self.show: - self._show(reading) + for reading in readings: + self.reply( + f"force,{reading.sensor_id},{reading.normal:.6f}," + f"{reading.shear_x:.6f},{reading.shear_y:.6f}" + ) + if self.show: + self._show(reading) + if self._should_quit: + return def _show(self, reading): import cv2 @@ -219,7 +121,9 @@ class UartFingerController: cv2.imshow(f"UV Unrolled {reading.sensor_id}", arrows) key = cv2.waitKey(1) & 0xFF if key == ord("q"): - self.cmd_quit([]) + self._streaming = False + self._should_quit = True + self.reply("OK quit") elif key == ord("r"): tactile.reset() except Exception: @@ -228,56 +132,42 @@ class UartFingerController: def _dispatch(self, line): if self.read_only: return True - try: - parts = shlex.split(line) - except ValueError as exc: - self.reply(f"ERR command {exc}") - return True + parts = line.split() if not parts: return True - cmd, args = parts[0].lower(), parts[1:] - try: - if cmd == "search": - self.cmd_search(args) - elif cmd == "connect": - self.cmd_connect(args) - elif cmd == "run": - self.cmd_run(args) - elif cmd == "read": - for sensor_id in self._targets(args): - self._read_one(sensor_id, context="read") - elif cmd == "stop": - self.cmd_stop(args) - elif cmd in ("quit", "exit"): - self.cmd_quit(args) - else: - self.reply(f"ERR unknown {cmd}") - except (ValueError, RuntimeError) as exc: - self.reply(f"ERR {cmd} {exc}") + command = parts[0].lower() + if command not in {"start", "stop", "read", "quit"}: + self.reply(f"ERR unknown {command}") + elif len(parts) != 1: + self.reply("ERR command start|stop|read|quit take no arguments") + elif command == "start": + self._streaming = True + self.reply("OK start") + elif command == "stop": + self._streaming = False + self.reply("OK stop") + elif command == "read": + self._read_round() + elif command == "quit": + self._streaming = False + self._should_quit = True + self.reply("OK quit") return not self._should_quit def serve(self): - self.reply("OK ready") - if self.autostart: - self.cmd_connect(["all"]) - if self.bank.runtimes: - self.cmd_run([]) - else: - raise RuntimeError("autostart could not connect any sensor") + if set(self.bank.runtimes) != set(self.bank.sources): + raise RuntimeError("connect all configured cameras before serving UART") + self.reply(f"OK ready {len(self.bank.sources)}") + self._streaming = True while not self._should_quit: - # Bounded command work keeps a noisy host from starving acquisition. for _ in range(16): line = self._poll_command_line(blocking=not self._streaming) if line is None: break if not self._dispatch(line): return - for sensor_id in sorted(self._streaming): - self._read_one(sensor_id, context="run") - if self._should_quit: - return - - + if self._streaming: + self._read_round() def open_serial(port, baud, timeout, write_timeout=1.0): try: import serial @@ -297,38 +187,35 @@ def main(argv=None): parser.add_argument("--timeout", type=float, default=0.1, help="idle read timeout in seconds") parser.add_argument("--write-timeout", type=float, default=1.0) parser.add_argument("--config", "-c", action="append", help="one shared config or repeat per sensor") - parser.add_argument("--camera-id", type=int, default=0) - parser.add_argument("--video", help="single-sensor default video file") - parser.add_argument("--sources", nargs="+", help="1-4 camera indices, device paths or video files") - parser.add_argument("--max-camera-index", type=int, default=10) - parser.add_argument("--loop-video", dest="loop_video", action="store_true", default=True) + parser.add_argument("--sources", required=True, nargs="+", help="1-4 camera indices, device paths or video files") parser.add_argument("--no-loop-video", dest="loop_video", action="store_false") parser.add_argument("--show", action="store_true", help="local desktop debug only") parser.add_argument("--verbose", action="store_true") - parser.add_argument("--autostart", action="store_true", help="connect and run configured sources on startup") - parser.add_argument("--read-only", action="store_true", help="autostart and ignore all UART commands") + parser.add_argument("--read-only", action="store_true", help="ignore all UART commands; exit from the login console") args = parser.parse_args(argv) - if args.baud <= 0 or args.max_camera_index < 0: - parser.error("baud must be positive; max camera index must be non-negative") + if args.baud <= 0: + parser.error("baud must be positive") if any(not math.isfinite(t) or t <= 0 for t in (args.timeout, args.write_timeout)): parser.error("timeouts must be finite and positive") - if args.sources and (args.video is not None or args.camera_id != 0): - parser.error("--sources cannot be combined with --video or --camera-id") configs = args.config or ["finger"] try: - sources = [parse_source(s) for s in args.sources] if args.sources else [ - str(Path(args.video).resolve()) if args.video else args.camera_id, - ] + sources = [parse_source(s) for s in args.sources] controller = UartFingerController( - None, sources=sources, configs=configs, max_camera_index=args.max_camera_index, + None, sources=sources, configs=configs, loop_video=args.loop_video, show=args.show, verbose=args.verbose, - autostart=args.autostart, read_only=args.read_only, + read_only=args.read_only, ) except ValueError as exc: parser.error(str(exc)) ser = None result = 0 try: + for sensor_id, source in controller.bank.sources.items(): + print( + f"Preparing sensor {sensor_id}: source={source} config={controller.bank.configs[sensor_id]}", + file=sys.stderr, flush=True, + ) + controller.bank.connect_all() ser = open_serial(args.port, args.baud, args.timeout, args.write_timeout) controller.ser = ser print(f"UART service on {args.port} @ {args.baud}; sources={sources}", file=sys.stderr, flush=True) diff --git a/serial/finger_api.py b/serial/finger_api.py index 5393843..3bb1a18 100644 --- a/serial/finger_api.py +++ b/serial/finger_api.py @@ -40,13 +40,13 @@ class ForceReading: class FingerSensorBank: - """Own 1–4 independent runtimes. read() returns None at EOF/capture failure. + """Own 1-4 independent runtimes; connect_all() precedes read_all(). Processing/connection failures raise exceptions; they never become zero force. A shared config can be reused, or each source can have its own config. """ - def __init__(self, sources=(0,), *, configs=("finger",), verbose=False): + def __init__(self, sources, *, configs=("finger",), verbose=False): sources = tuple(parse_source(str(source)) for source in sources) configs = tuple(configs) if not 1 <= len(sources) <= 4: @@ -57,45 +57,71 @@ class FingerSensorBank: raise ValueError("camera index must be non-negative") if len(configs) == 1: configs = configs * len(sources) + identities = [] + for source in sources: + if isinstance(source, int): + identity = f"camera:{source}" + elif source.startswith("/dev/video") and source[len("/dev/video"):].isdigit(): + identity = f"camera:{int(source[len('/dev/video'):])}" + else: + identity = str(Path(source).resolve()) + identities.append(identity) + if len(set(identities)) != len(identities): + raise ValueError("each sensor must use a different source") self.sources = dict(enumerate(sources, 1)) self.configs = dict(enumerate(configs, 1)) self.runtimes = {} self.verbose = verbose - def validate_id(self, sensor_id: int) -> None: - if sensor_id not in self.sources: - raise ValueError(f"sensor ID must be 1..{len(self.sources)}") - - def connect(self, sensor_id: int = 1, source: Source | None = None) -> Source: + def connect_all(self) -> None: + """Prepare every source, rolling back all runtimes on any failure.""" from orisys.finger import FingerRuntime - self.validate_id(sensor_id) - source = self.sources[sensor_id] if source is None else parse_source(str(source)) - if isinstance(source, int) and source < 0: - raise ValueError("camera index must be non-negative") - if is_video_file(source) and not Path(source).is_file(): - raise ValueError(f"file not found: {source}") - if source in (self.sources[i] for i in self.runtimes if i != sensor_id): - raise ValueError(f"source already connected: {source}") - self.disconnect(sensor_id) - runtime = FingerRuntime.open( - config=self.configs[sensor_id], video=source, verbose=self.verbose, - ) + if self.runtimes: + raise RuntimeError("already connected; close before reconnecting") + sensor_id = None try: - ok, _ = runtime.warmup(color=False) - if not ok: - raise RuntimeError("warmup could not read a frame") + for sensor_id, source in self.sources.items(): + if is_video_file(source) and not Path(source).is_file(): + raise ValueError(f"file not found: {source}") + runtime = FingerRuntime.open( + config=self.configs[sensor_id], video=source, verbose=self.verbose, + ) + self.runtimes[sensor_id] = runtime + ok, _ = runtime.warmup(color=False) + if not ok: + raise RuntimeError("warmup could not read a frame") + except Exception as exc: + try: + self.close() + except Exception: + logger.exception("Connection rollback failed") + raise RuntimeError(f"sensor {sensor_id}: {exc}") from exc except BaseException: - runtime.close() + self.close() raise - self.runtimes[sensor_id] = runtime - self.sources[sensor_id] = source - return source - def read(self, sensor_id: int = 1, *, loop_video=False) -> ForceReading | None: - self.validate_id(sensor_id) - if sensor_id not in self.runtimes: - raise ValueError("not connected") + def read_all(self, *, loop_video=False) -> tuple[ForceReading, ...]: + """Read a complete round. EOF/failure raises; no partial round is returned. + + The frames are read sequentially, not synchronized across sources. + """ + if set(self.runtimes) != set(self.sources): + raise RuntimeError("all configured sensors must be connected first") + readings = [] + for sensor_id in self.sources: + try: + reading = self._read(sensor_id, loop_video=loop_video) + except Exception as exc: + raise RuntimeError(f"sensor {sensor_id}: {exc}") from exc + if reading is None: + if is_video_file(self.sources[sensor_id]): + raise EOFError(f"sensor {sensor_id}: video ended") + raise RuntimeError(f"sensor {sensor_id}: capture failed") + readings.append(reading) + return tuple(readings) + + def _read(self, sensor_id: int, *, loop_video=False) -> ForceReading | None: runtime = self.runtimes[sensor_id] ok, image = runtime.grab_and_process(color=False) if not ok and loop_video and is_video_file(self.sources[sensor_id]): @@ -118,17 +144,11 @@ class FingerSensorBank: units = getattr(tactile, "force_units", ("unknown", "unknown", "unknown")) return ForceReading(sensor_id, *forces, tuple(units), image) - def disconnect(self, sensor_id: int = 1) -> None: - self.validate_id(sensor_id) - runtime = self.runtimes.pop(sensor_id, None) - if runtime is not None: - runtime.close() - def close(self) -> None: first_error = None for sensor_id in list(self.runtimes): try: - self.disconnect(sensor_id) + self.runtimes.pop(sensor_id).close() except Exception as exc: logger.exception("Sensor %s cleanup failed", sensor_id) first_error = first_error or exc diff --git a/tests/check_backend.py b/tests/check_backend.py new file mode 100644 index 0000000..3283f88 --- /dev/null +++ b/tests/check_backend.py @@ -0,0 +1,78 @@ +"""Exercise real SDK instances for 1-4 recorded inputs, without UART/hardware.""" + +import argparse +import itertools +import os +import sys +from functools import partial +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "serial")) +from finger_api import FingerSensorBank + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--recording", required=True, type=Path) + parser.add_argument("--scratch", required=True, type=Path) + args = parser.parse_args() + scratch = args.scratch.resolve() + scratch.mkdir(parents=True, exist_ok=True) + os.environ["NUMBA_CACHE_DIR"] = str(scratch / "numba") + + import cv2 + import numpy as np + import orisys + from orisys.finger import FingerRuntime + + print(f"SDK: {orisys.__file__}", flush=True) + capture = cv2.VideoCapture(str(args.recording.resolve())) + frames = [] + try: + for _ in range(12): + ok, frame = capture.read() + if not ok: + raise RuntimeError("recording must contain at least 12 frames") + frames.append(frame) + finally: + capture.release() + height, width = frames[0].shape[:2] + sources = [] + for sensor_id in range(1, 5): + path = scratch / f"sensor-{sensor_id}.avi" + writer = cv2.VideoWriter(str(path), cv2.VideoWriter_fourcc(*"MJPG"), 30, (width, height)) + try: + if not writer.isOpened(): + raise RuntimeError("cannot create test recording") + for frame in frames: + writer.write(np.roll(frame, sensor_id, axis=1)) + finally: + writer.release() + sources.append(str(path)) + + # Keep the SDK implementation intact; redirect only its generated UV cache. + open_runtime = partial(FingerRuntime.open, map_cache=str(scratch / "uv-map.npz")) + with patch.object(FingerRuntime, "open", side_effect=open_runtime): + for count in range(1, 5): + bank = FingerSensorBank(sources[:count]) + try: + bank.connect_all() + runtimes = list(bank.runtimes.values()) + for left, right in itertools.combinations(runtimes, 2): + assert left.capture is not right.capture + assert left.tactile.raw_sensor is not right.tactile.raw_sensor + assert not np.shares_memory(left.tactile.flow, right.tactile.flow) + for _ in range(3): + readings = bank.read_all() + assert [r.sensor_id for r in readings] == list(range(1, count + 1)) + assert all(np.isfinite([r.normal, r.shear_x, r.shear_y]).all() for r in readings) + print(f"PASS {count} sources: independent captures/processors, 3 complete rounds", flush=True) + finally: + bank.close() + assert all(runtime.capture is None for runtime in runtimes) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_uart.py b/tests/test_uart.py index 8a15804..823e59f 100644 --- a/tests/test_uart.py +++ b/tests/test_uart.py @@ -103,176 +103,126 @@ class HardwareFreeTest(unittest.TestCase): self.addCleanup(modules.stop) -class ExistingRegressionTests(HardwareFreeTest): - def test_partial_command_waits_for_terminator(self): - ser = ByteSerial(b"st") - controller = uart.UartFingerController(ser) - self.assertIsNone(controller._poll_command_line(blocking=False)) - ser.input.extend(b"op\n") - self.assertEqual(controller._poll_command_line(blocking=False), "stop") - - def test_cr_only_terminal_commands(self): - controller = uart.UartFingerController(ByteSerial(b"stop\rrun\r")) - self.assertEqual(controller._poll_command_line(blocking=False), "stop") - self.assertEqual(controller._poll_command_line(blocking=False), "run") - - def test_failed_warmup_releases_runtime(self): - runtime = FakeRuntime(warmup_error=True) - ser = ByteSerial() - controller = uart.UartFingerController(ser) - with patch("orisys.finger.FingerRuntime.open", return_value=runtime): - controller.cmd_connect([]) - self.assertTrue(runtime.closed) - self.assertTrue(ser.output[-1].startswith("ERR connect")) - - def test_unsuccessful_warmup_is_not_connected(self): - runtime = FakeRuntime(warmup_ok=False) - ser = ByteSerial() - controller = uart.UartFingerController(ser) - with patch("orisys.finger.FingerRuntime.open", return_value=runtime): - controller.cmd_connect([]) - self.assertTrue(ser.output[-1].startswith("ERR connect")) - self.assertTrue(runtime.closed) - - -class MultiSensorTests(HardwareFreeTest): - def test_autostart_service_streams_one_through_four_sensors_then_stops(self): - for count in range(1, 5): - with self.subTest(count=count): - ser = ByteSerial() - controller = uart.UartFingerController(ser, sources=list(range(count)), autostart=True) - original_write = ser.write - def write(data, original_write=original_write, ser=ser, count=count): - size = original_write(data) - if data.startswith(b"force,") and sum(line.startswith("force,") for line in ser.output) == count: - ser.input.extend(b"stop all\nquit\n") - return size - ser.write = write - controller.serve() - self.assertEqual(sum(line.startswith("force,") for line in ser.output), count) - self.assertEqual(ser.output[-1], "OK quit") - self.assertFalse(controller._streaming) - controller.close_runtime() - - def test_one_through_four_sensors_have_identifiable_readings(self): +class ConnectionFirstTests(HardwareFreeTest): + def test_connects_every_source_before_ready_or_force(self): for count in range(1, 5): with self.subTest(count=count): ser = ByteSerial() controller = uart.UartFingerController(ser, sources=list(range(count))) - controller.cmd_connect(["all"]) - controller._dispatch("read all") - forces = [line for line in ser.output if line.startswith("force,")] - expected = ["force,1.250000,-2.500000,3.750000"] if count == 1 else [ + events = [] + def open_runtime(events=events, **kwargs): + events.append(kwargs["video"]) + return FakeRuntime() + original_write = ser.write + def write(data, events=events, count=count, ser=ser, original_write=original_write): + self.assertEqual(events, list(range(count))) + size = original_write(data) + if data.startswith(b"force,") and sum(x.startswith("force,") for x in ser.output) == count: + ser.input.extend(b"quit\n") + return size + ser.write = write + with patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime): + controller.bank.connect_all() + controller.serve() + self.assertEqual(ser.output[0], f"OK ready {count}") + self.assertEqual([x for x in ser.output if x.startswith("force,")], [ f"force,{i},1.250000,-2.500000,3.750000" for i in range(1, count + 1) - ] - self.assertEqual(forces, expected) - self.assertFalse(controller._streaming) + ]) controller.close_runtime() - def test_selective_stop_and_reconnect_leave_other_sensor_running(self): - controller = uart.UartFingerController(ByteSerial(), sources=[0, 2, 4, 6]) - controller.cmd_connect(["all"]) - controller._dispatch("run all") - controller._dispatch("stop 2") - self.assertEqual(controller._streaming, {1, 3, 4}) - old = controller.bank.runtimes[3] - controller._dispatch("connect 3 8") - self.assertTrue(old.closed) - self.assertEqual(controller._streaming, {1, 4}) - self.assertEqual(controller.bank.sources[3], 8) + def test_one_failed_connection_rolls_back_every_opened_camera(self): + bank = FingerSensorBank([0, 2, 4]) + opened = [] + def open_runtime(**kwargs): + runtime = FakeRuntime(warmup_error=kwargs["video"] == 2) + opened.append(runtime) + return runtime + with (patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime), + self.assertRaisesRegex(RuntimeError, "sensor 2")): + bank.connect_all() + self.assertEqual(bank.runtimes, {}) + self.assertTrue(all(r.closed for r in opened)) - def test_failed_sensor_does_not_stop_healthy_sensor(self): + def test_no_uart_is_opened_if_camera_preparation_fails(self): + with (patch("orisys.finger.FingerRuntime.open", return_value=FakeRuntime(warmup_ok=False)), + patch.object(uart, "open_serial") as serial): + self.assertEqual(uart.main(["--sources", "0", "2"]), 1) + serial.assert_not_called() + + def test_legacy_commands_are_rejected(self): ser = ByteSerial() controller = uart.UartFingerController(ser, sources=[0, 2]) - controller.cmd_connect(["all"]) - controller.bank.runtimes[1].grab_error = True - controller.cmd_run([]) - for sensor_id in sorted(controller._streaming): - controller._read_one(sensor_id, context="run") - self.assertIn("ERR run 1 processing failed", ser.output) - self.assertIn("force,2,1.250000,-2.500000,3.750000", ser.output) - self.assertEqual(controller._streaming, {2}) - - def test_connect_all_retains_successes_when_one_source_fails(self): - ser = ByteSerial() - controller = uart.UartFingerController(ser, sources=[0, 2, 4]) - def open_runtime(**kwargs): - if kwargs["video"] == 2: - raise RuntimeError("camera unavailable") - return FakeRuntime() - with patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime): - controller.cmd_connect(["all"]) - self.assertEqual(set(controller.bank.runtimes), {1, 3}) - self.assertIn("ERR connect 2 camera unavailable", ser.output) - - def test_rejects_duplicate_live_source_without_closing_other_camera(self): - controller = uart.UartFingerController(ByteSerial(), sources=[0, 2]) - controller.cmd_connect(["all"]) - original = controller.bank.runtimes[2] - controller._dispatch("connect 2 0") - self.assertIs(controller.bank.runtimes[2], original) - self.assertFalse(original.closed) - - def test_config_is_selected_per_sensor(self): - bank = FingerSensorBank([0, 2], configs=["left.json", "right.json"]) - seen = [] - def open_runtime(**kwargs): - seen.append(kwargs["config"]) - return FakeRuntime() - with patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime): - bank.connect(1) - bank.connect(2) - self.assertEqual(seen, ["left.json", "right.json"]) - self.assertEqual(bank.read(1).units, ("raw", "raw", "raw")) - - def test_invalid_sensor_count_and_config_count(self): - for sources, configs in [([], ["finger"]), (range(5), ["finger"]), ([0, 2], ["a", "b", "c"])]: - with self.assertRaises(ValueError): - FingerSensorBank(sources, configs=configs) - - def test_invalid_command_id_and_unconnected_run_are_errors(self): - ser = ByteSerial() - controller = uart.UartFingerController(ser, sources=[0, 2]) - for command in ["run 0", "read 3", "stop -1", "connect 3 4", "run all", "run 2 extra"]: + for command in ["connect", "connect 2", "connect all", "run", "run all", "search", "exit"]: self.assertTrue(controller._dispatch(command)) - self.assertTrue(ser.output[-1].startswith("ERR ")) + self.assertTrue(ser.output[-1].startswith("ERR unknown")) - -class ProtocolTests(HardwareFreeTest): - def test_repeated_run_is_rejected_without_recursion(self): + def test_no_per_sensor_or_all_command_arguments(self): ser = ByteSerial() - controller = uart.UartFingerController(ser) - controller.cmd_connect([]) - controller._dispatch("run") - controller._dispatch("run") - self.assertEqual(ser.output[-1], "ERR run already running") - self.assertEqual(controller._streaming, {1}) + controller = uart.UartFingerController(ser, sources=[0, 2]) + for command in ["start 1", "stop all", "read 2", "quit extra"]: + self.assertTrue(controller._dispatch(command)) + self.assertTrue(ser.output[-1].startswith("ERR command")) - def test_camera_search_failure_is_reported_without_exiting(self): + def test_start_stop_and_read_control_entire_connected_bank(self): ser = ByteSerial() - controller = uart.UartFingerController(ser) - with patch("orisys.capture.list_cameras", side_effect=OSError("camera scan failed")): - self.assertTrue(controller._dispatch("search")) - self.assertEqual(ser.output[-1], "ERR search camera scan failed") - - def test_legacy_connect_and_batched_stop_quit(self): - ser = ByteSerial(b"connect 2\r\nrun\nstop\nquit\n") - controller = uart.UartFingerController(ser) - controller.serve() - self.assertEqual(ser.output, ["OK ready", "OK connect 2 (camera)", "OK run", "OK stop", "OK quit"]) + controller = uart.UartFingerController(ser, sources=[0, 2]) + controller.bank.connect_all() + controller._dispatch("stop") + controller._dispatch("read") + self.assertEqual(ser.output[-2:], [ + "force,1,1.250000,-2.500000,3.750000", + "force,2,1.250000,-2.500000,3.750000", + ]) self.assertFalse(controller._streaming) + controller._dispatch("start") + self.assertTrue(controller._streaming) + self.assertEqual(ser.output[-1], "OK start") - def test_stop_is_processed_before_next_frame(self): - ser = ByteSerial(b"stop\nquit\n") - controller = uart.UartFingerController(ser) - controller.cmd_connect([]) - controller.cmd_run([]) - controller.serve() - self.assertFalse(any(line.startswith("force,") for line in ser.output)) + def test_read_all_never_publishes_partial_round_on_processing_failure(self): + ser = ByteSerial() + controller = uart.UartFingerController(ser, sources=[0, 2]) + controller.bank.connect_all() + controller.bank.runtimes[2].grab_error = True + controller._streaming = True + controller._dispatch("read") + self.assertFalse(controller._streaming) + self.assertFalse(any(x.startswith("force,") for x in ser.output)) + self.assertIn("sensor 2", ser.output[-1]) - def test_read_only_autostarts_and_ignores_host_commands(self): - ser = ByteSerial(b"quit\nstop\nconnect 9\n" + b"a" * 1100 + b"\n") - controller = uart.UartFingerController(ser, read_only=True) + def test_serve_rejects_incomplete_bank_before_ready(self): + ser = ByteSerial() + controller = uart.UartFingerController(ser, sources=[0, 2]) + with self.assertRaises(RuntimeError): + controller.serve() + self.assertEqual(ser.output, []) + + +class FramingTests(HardwareFreeTest): + def test_partial_line_waits_for_terminator(self): + ser = ByteSerial(b"st") + controller = uart.UartFingerController(ser, sources=[0]) + self.assertIsNone(controller._poll_command_line(blocking=False)) + ser.input.extend(b"op\n") + self.assertEqual(controller._poll_command_line(blocking=False), "stop") + + def test_cr_lf_crlf_commands_are_separate(self): + controller = uart.UartFingerController(ByteSerial(b"stop\rstart\nread\r\n"), sources=[0]) + self.assertEqual(controller._poll_command_line(blocking=False), "stop") + self.assertEqual(controller._poll_command_line(blocking=False), "start") + self.assertEqual(controller._poll_command_line(blocking=False), "read") + + def test_overlong_command_is_discarded_until_terminator(self): + ser = ByteSerial(b"a" * 1025) + controller = uart.UartFingerController(ser, sources=[0]) + self.assertIsNone(controller._poll_command_line(blocking=False)) + ser.input.extend(b"quit\nstop\r") + self.assertEqual(controller._poll_command_line(blocking=False), "stop") + self.assertTrue(ser.output[-1].startswith("ERR command")) + + def test_read_only_ignores_every_command_and_still_streams(self): + ser = ByteSerial(b"stop\nquit\n" + b"a" * 1100 + b"\n") + controller = uart.UartFingerController(ser, sources=[0], read_only=True) + controller.bank.connect_all() original_write = ser.write def write(data): size = original_write(data) @@ -282,62 +232,69 @@ class ProtocolTests(HardwareFreeTest): ser.write = write with self.assertRaises(KeyboardInterrupt): controller.serve() - self.assertIn("OK run", ser.output) - self.assertIn("force,1.250000,-2.500000,3.750000", ser.output) - self.assertNotIn("OK quit", ser.output) - self.assertFalse(any(line.startswith("ERR ") for line in ser.output)) + self.assertEqual(ser.output, ["OK ready 1", "force,1,1.250000,-2.500000,3.750000"]) - def test_long_line_is_discarded_until_delimiter(self): - ser = ByteSerial(b"a" * 1025) - controller = uart.UartFingerController(ser) - self.assertIsNone(controller._poll_command_line(blocking=False)) - ser.input.extend(b"quit\nstop\r") - self.assertEqual(controller._poll_command_line(blocking=False), "stop") - self.assertEqual(ser.output, ["ERR command line too long (max 1024 bytes)"]) - self.assertFalse(controller._should_quit) - - def test_quoted_paths_and_malformed_quotes(self): - controller = uart.UartFingerController(ByteSerial(), sources=[0, 2]) - with patch.object(controller.bank, "connect", return_value="/tmp/a b.mp4") as connect: - controller._dispatch('connect 2 "/tmp/a b.mp4"') - connect.assert_called_once_with(2, "/tmp/a b.mp4") - controller._dispatch('connect "broken') - self.assertTrue(controller.ser.output[-1].startswith("ERR command")) - - def test_non_finite_measurements_are_not_published(self): - ser = ByteSerial() - controller = uart.UartFingerController(ser) - controller.cmd_connect([]) - controller.bank.runtimes[1].tactile.read_info = lambda *args: (float("nan"), 0, 0) - controller._dispatch("read") - self.assertEqual(ser.output[-1], "ERR read non-finite force measurement") - self.assertFalse(any(line.startswith("force,") for line in ser.output)) - - def test_serial_short_write_is_a_transport_failure(self): + def test_short_write_is_transport_failure(self): ser = ByteSerial() ser.write = lambda data: len(data) - 1 with self.assertRaises(OSError): - uart.UartFingerController(ser).reply("OK ready") + uart.UartFingerController(ser, sources=[0]).reply("OK ready 1") - def test_error_newlines_cannot_inject_protocol_records(self): + def test_error_newlines_do_not_inject_records(self): ser = ByteSerial() - uart.UartFingerController(ser).reply("ERR read bad\nOK quit\r\n") + uart.UartFingerController(ser, sources=[0]).reply("ERR read bad\nOK quit\r\n") self.assertEqual(ser.output, ["ERR read bad OK quit"]) - def test_empty_looping_video_retries_once_then_reports_eof(self): - ser = ByteSerial() - controller = uart.UartFingerController(ser) - runtime = FakeRuntime() - runtime.frames = 0 - controller.bank.sources[1] = "empty.mp4" - controller.bank.runtimes[1] = runtime - controller._dispatch("read") - self.assertEqual(ser.output, ["OK eof"]) - def test_close_attempts_every_sensor_even_when_one_close_fails(self): +class BankTests(HardwareFreeTest): + def test_invalid_count_config_count_and_duplicate_sources(self): + for sources, configs in [([], ["finger"]), (range(5), ["finger"]), + ([0, 2], ["a", "b", "c"]), ([0, "0"], ["finger"]), + ([0, "/dev/video0"], ["finger"]), ([-1], ["finger"])]: + with self.subTest(sources=sources), self.assertRaises(ValueError): + FingerSensorBank(sources, configs=configs) + + def test_configs_and_runtime_states_are_independent(self): + bank = FingerSensorBank([0, 2], configs=["left.json", "right.json"]) + configs = [] + def open_runtime(**kwargs): + configs.append(kwargs["config"]) + return FakeRuntime() + with patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime): + bank.connect_all() + self.assertEqual(configs, ["left.json", "right.json"]) + self.assertIsNot(bank.runtimes[1], bank.runtimes[2]) + readings = bank.read_all() + self.assertEqual([r.sensor_id for r in readings], [1, 2]) + self.assertEqual(readings[0].units, ("raw", "raw", "raw")) + + def test_no_frame_during_warmup_closes_runtime(self): + runtime = FakeRuntime(warmup_ok=False) + bank = FingerSensorBank([0]) + with (patch("orisys.finger.FingerRuntime.open", return_value=runtime), + self.assertRaisesRegex(RuntimeError, "warmup")): + bank.connect_all() + self.assertTrue(runtime.closed) + self.assertEqual(bank.runtimes, {}) + + def test_nonfinite_force_is_not_a_valid_round(self): + bank = FingerSensorBank([0]) + bank.connect_all() + bank.runtimes[1].tactile.read_info = lambda *args: (float("nan"), 0, 0) + with self.assertRaisesRegex(RuntimeError, "non-finite"): + bank.read_all() + + def test_eof_returns_no_partial_round(self): bank = FingerSensorBank([0, 2]) - bank.connect(1) - bank.connect(2) + bank.connect_all() + bank.runtimes[2].frames = 0 + bank.sources[2] = "ended.mp4" + with self.assertRaisesRegex(EOFError, "sensor 2"): + bank.read_all() + + def test_all_resources_are_attempted_even_when_close_fails(self): + bank = FingerSensorBank([0, 2]) + bank.connect_all() second = bank.runtimes[2] bank.runtimes[1].close = lambda: (_ for _ in ()).throw(RuntimeError("close failed")) with self.assertRaisesRegex(RuntimeError, "close failed"): @@ -347,38 +304,27 @@ class ProtocolTests(HardwareFreeTest): class CliTests(HardwareFreeTest): - def test_open_serial_has_finite_write_timeout_and_no_flow_control(self): - serial = types.ModuleType("serial") - serial.EIGHTBITS, serial.PARITY_NONE, serial.STOPBITS_ONE = 8, "N", 1 - serial.Serial = types.SimpleNamespace() - with patch.dict(sys.modules, {"serial": serial}), patch.object(serial, "Serial") as factory: - uart.open_serial("/dev/ttyHS1", 115200, 0.1, 0.5) - factory.assert_called_once_with( - port="/dev/ttyHS1", baudrate=115200, bytesize=8, parity="N", - stopbits=1, timeout=0.1, write_timeout=0.5, exclusive=True, - ) - - def test_invalid_options_do_not_open_uart(self): - for args in [["--timeout", "0"], ["--write-timeout", "nan"], ["--sources", "0", "1", "2", "3", "4"], - ["--sources", "0", "2", "--config", "a", "--config", "b", "--config", "c"]]: - with self.subTest(args=args), patch.object(uart, "open_serial") as open_serial: - with self.assertRaises(SystemExit) as exit_status: + def test_sources_are_required_and_legacy_options_are_removed(self): + for args in [[], ["--camera-id", "0"], ["--video", "a.mp4"], + ["--sources", "0", "--autostart"], ["--sources", "0", "--max-camera-index", "10"], + ["--sources", "0", "1", "2", "3", "4"], ["--sources", "0", "--timeout", "nan"]]: + with self.subTest(args=args), patch.object(uart, "open_serial") as serial: + with self.assertRaises(SystemExit) as status: uart.main(args) - self.assertEqual(exit_status.exception.code, 2) - open_serial.assert_not_called() + self.assertEqual(status.exception.code, 2) + serial.assert_not_called() - def test_transport_failure_returns_nonzero_and_closes_all_resources(self): - ser = ByteSerial() - ser.closed = False - ser.close = lambda: setattr(ser, "closed", True) - def serve(controller): - controller.cmd_connect(["all"]) - self.runtimes = list(controller.bank.runtimes.values()) - raise OSError("UART disconnected") - with patch.object(uart, "open_serial", return_value=ser), patch.object(uart.UartFingerController, "serve", serve): + def test_serial_failure_closes_every_connected_runtime(self): + opened = [] + def open_runtime(**kwargs): + runtime = FakeRuntime() + opened.append(runtime) + return runtime + with (patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime), + patch.object(uart, "open_serial", side_effect=OSError("UART unavailable"))): self.assertEqual(uart.main(["--sources", "0", "2"]), 1) - self.assertTrue(ser.closed) - self.assertTrue(all(runtime.closed for runtime in self.runtimes)) + self.assertEqual(len(opened), 2) + self.assertTrue(all(r.closed for r in opened)) if __name__ == "__main__":