From 133a98e4187d12805650be178aa6af5ce79a6f35 Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Wed, 14 Oct 2026 15:49:58 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=AF=E6=8C=81=E9=80=9A=E8=BF=87=20UART=20?= =?UTF-8?q?=E6=89=8B=E5=8A=A8=E9=80=90=E5=8F=B0=E6=B7=BB=E5=8A=A0=E7=9B=B8?= =?UTF-8?q?=E6=9C=BA=E5=86=8D=E5=90=AF=E5=8A=A8=E8=BE=93=E5=87=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 保留已连接相机，添加失败不占用编号，并要求输出停止后才能新增。通过 23 项回归测试、真实 SDK 的 1–4 路逐台添加验证、Ruff 和文档检查。 --- docs/uart.md | 209 ++++++++++++------------ serial/example_uart.py | 77 ++++----- serial/finger_api.py | 98 ++++++----- tests/check_backend.py | 10 +- tests/test_uart.py | 358 +++++++++++++++++++++++++---------------- 5 files changed, 422 insertions(+), 330 deletions(-) diff --git a/docs/uart.md b/docs/uart.md index 4dc5f32..c1101fb 100644 --- a/docs/uart.md +++ b/docs/uart.md @@ -1,145 +1,154 @@ # UART finger service -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. +Start the service first, then manually add 1-4 camera sources through UART. +Each successful `add` opens and warms up one source. Send `start` when the set +is ready. There is no camera discovery, hot-plug monitoring or automatic connection. -## Connect the cameras and start once +## Start the service and add cameras -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`. - -Run from the board's `dragon-q6a` directory. These paths are examples, not detected -hardware identities: +In the Q6A TTL login shell, activate 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` directory: ```bash -# One camera. -python serial/example_uart.py --port /dev/ttyHS1 --sources /dev/video0 - -# Two cameras. -python serial/example_uart.py --port /dev/ttyHS1 --sources /dev/video0 /dev/video2 - -# Three cameras. -python serial/example_uart.py --port /dev/ttyHS1 --sources /dev/video0 /dev/video2 /dev/video4 - -# Four cameras. -python serial/example_uart.py --port /dev/ttyHS1 --sources /dev/video0 /dev/video2 /dev/video4 /dev/video6 +python serial/example_uart.py --port /dev/ttyHS1 --config finger ``` -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. - -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. - -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. - -## One protocol for every sensor count - -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: +The service opens only the data UART and replies `OK ready 0`. No camera is +opened and no measurement is sent until the host adds a source and starts output. +Send these lines through the data UART, waiting for each reply: ```text -OK ready 2 +add /dev/video0 +OK added 1 +add /dev/video2 +OK added 2 +start +OK start force,1,1.250000,-2.500000,3.750000 force,2,0.500000,0.100000,-0.200000 ``` -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. +Use actual capture nodes; the paths above are examples. One USB camera can expose +multiple `/dev/video*` nodes, so do not assume every node represents a different +camera. `v4l2-ctl --list-devices` shows device groups. Select one usable capture +node per camera. Camera indices such as `add 0` are also accepted. Sources must +be distinct; `0` and `/dev/video0` are treated as the same camera. -Control mode accepts only four commands, each acting on the entire camera set: +The first successful addition is sensor 1, the next sensor 2, up to sensor 4. +No sensor count is declared in advance. A failed addition replies `ERR add ...`, +closes its newly opened runtime if one exists, and leaves existing cameras and +IDs unchanged. It does not consume an ID. The service remains available for retry. -| Command | Result | +The TTL login console and data UART must use **different ports**. Data defaults +to `/dev/ttyHS1`, 115200 baud, 8N1 without flow control. It must not be occupied +by getty or another serial program. The script does not configure UART pins. +Logs stay in the login console; protocol records go to the data UART. +`--show` is optional local desktop debugging. + +## Commands and output + +| Command | Behavior | |---|---| -| `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. | +| `add SOURCE [CONFIG]` | Open and prepare one source while output is stopped; reply `OK added ID`. Optional config overrides the startup `--config` for this sensor only. | +| `start` | Begin continuous output for all added sensors; requires at least one sensor. Reply `OK start`. | +| `stop` | Pause continuous output for the whole set; keep all cameras open. Reply `OK stop`. | +| `read` | Publish one complete processing round; requires a sensor and preserves 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`. +To extend the camera set after streaming has started, pause before adding: -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. +```text +stop +OK stop +add /dev/video4 +OK added 3 +start +OK start +``` + +`add` while streaming replies `ERR add stop streaming first` and opens no source. +A fifth source or duplicate source is rejected. Removing/replacing cameras requires +quitting and starting a fresh service. `start`, `stop`, `read` and `quit` take no +arguments and always act on the full set. The older commands and automatic +startup options are removed. + +Commands are case-insensitive UTF-8 lines terminated by LF, CR or CRLF. Partial +lines survive read timeouts. Lines exceeding 1024 bytes are discarded through +their terminator. Quote source/config paths containing spaces: + +```text +add "/path/test clip.mp4" "/path/my config.json" +``` + +Every measurement has five CSV fields, even with only one sensor: +`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 units or +timestamp. The receiver can remain passive after sending setup commands and +`start`; it needs no further requests to receive continuous data. Filter `force,` +lines to distinguish measurements from replies and errors. + +Frames are processed sequentially, not synchronized across cameras. All sources +must succeed before a processing round is published. A capture/processing failure +produces `ERR read sensor ID: ...` and stops continuous output for the full set, +without fabricating zero values or transmitting a partial processing round. +A transport failure can still interrupt transmission between lines. `start` +retries processing; restart the service to reopen an unavailable camera. + +For recorded files, `--no-loop-video` replies `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 in the login shell ends the service and releases resources. ## Callable acquisition API `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: +Place its directory on the caller's module search path. Adding sources is explicit: ```python from finger_api import FingerSensorBank -bank = FingerSensorBank([0, 2], configs=["finger"]) +bank = FingerSensorBank(config="finger") try: - bank.connect_all() + first_id = bank.add(0) + second_id = bank.add(2, config="finger") for reading in bank.read_all(): print(reading.sensor_id, reading.normal, reading.shear_x, reading.shear_y, reading.units) finally: bank.close() ``` -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. +One thread owns the bank. `add()` returns a 1-based ID only after successful +preparation; it raises on failure and preserves previously added runtimes. +`read_all()` requires at least one source and returns a tuple only when all +sources succeed. File EOF raises `EOFError`; capture/processing failure raises +`RuntimeError`. Each reading contains its own unrolled image, three scalar forces +and SDK units; older SDKs without a units property report `unknown`. `close()` +attempts every runtime and clears the camera set. The API caller must pause its +own processing loop before adding; the UART service enforces this for commands. ## Backend evidence and remaining limits -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. +The SDK's `FingerRuntime.open()` creates separate capture, unwarping and sensor +state for each source. SDK code is unchanged. Verification on 2026-10-14 used +local SDK revision `3953747f1726b197a3f327a38c65d9878e333e12`: 1, 2, 3 and 4 +recorded sources were added incrementally, with complete readings after each +addition and three final rounds for each count. Existing runtime identities were +preserved; capture handles and sensor objects were distinct, and optical-flow +arrays did not share memory. Resources were released after each case. -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: +To rerun 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 tests/check_backend.py --recording /path/to/recording.avi --scratch .scratch/uart-manual-add/backend-run python -m unittest discover -s tests -v ``` -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). +Unit tests separately verify empty startup, explicit additions, failure isolation, +IDs/limits, stop-before-add, command framing, complete-round output and cleanup. +The shipped compiled ARM `orisys 0.5.0` wheel and Q6A physical camera/UART setup +remain unverified. Run the backend check inside the board's installed environment +to verify that SDK. USB bandwidth, sustained frame rates and command latency +require hardware validation. Camera opening/warmup and synchronous reads can delay +command handling. Serial writes have a configurable timeout, 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 67c5d35..f4aa781 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. Every configured camera -is prepared before UART output. All force records include a 1-based sensor ID. +The login console and data UART must be separate ports. Cameras are added +manually via UART before starting output. All force records include a 1-based sensor ID. Importing this module opens no devices; acquisition is owned by finger_api. """ @@ -12,6 +12,7 @@ import argparse import logging import math import os +import shlex import sys from collections import deque @@ -26,15 +27,14 @@ class UartFingerController: """Single-threaded command service; the caller owns serial and final cleanup.""" def __init__( - self, ser, *, sources, configs=("finger",), loop_video=True, - show=False, verbose=False, read_only=False, + self, ser, *, config="finger", loop_video=True, + show=False, verbose=False, ): self.ser = ser - self.bank = FingerSensorBank(sources, configs=configs, verbose=verbose) + self.bank = FingerSensorBank(config=config, verbose=verbose) self.loop_video = loop_video self.show = show self.verbose = verbose - self.read_only = read_only self._streaming = False self._should_quit = False self._partial = bytearray() @@ -68,8 +68,6 @@ class UartFingerController: return None # Never readline(): a timeout may return only half of a command. raw = self.ser.read(min(waiting, 4096) if waiting else 1) - if self.read_only: - return None for byte in raw: if byte in (10, 13): if self._discard_line: @@ -130,19 +128,39 @@ class UartFingerController: logger.exception("Sensor %s display failed", reading.sensor_id) def _dispatch(self, line): - if self.read_only: + 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 command = parts[0].lower() - if command not in {"start", "stop", "read", "quit"}: + if command == "add": + if len(parts) not in (2, 3): + self.reply("ERR command expected add SOURCE [CONFIG]") + elif self._streaming: + self.reply("ERR add stop streaming first") + else: + try: + sensor_id = self.bank.add( + parse_source(parts[1]), config=parts[2] if len(parts) == 3 else None, + ) + except Exception as exc: + logger.exception("Sensor addition failed") + self.reply(f"ERR add {exc}") + else: + self.reply(f"OK added {sensor_id}") + elif 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") + if not self.bank.runtimes: + self.reply("ERR start no sensors added") + else: + self._streaming = True + self.reply("OK start") elif command == "stop": self._streaming = False self.reply("OK stop") @@ -155,10 +173,8 @@ class UartFingerController: return not self._should_quit def serve(self): - 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 + self._streaming = False while not self._should_quit: for _ in range(16): line = self._poll_command_line(blocking=not self._streaming) @@ -168,6 +184,7 @@ class UartFingerController: return if self._streaming: self._read_round() + def open_serial(port, baud, timeout, write_timeout=1.0): try: import serial @@ -181,44 +198,30 @@ def open_serial(port, baud, timeout, write_timeout=1.0): def main(argv=None): - parser = argparse.ArgumentParser(description="UART finger control and read-only streaming") + parser = argparse.ArgumentParser(description="Manual UART finger service: add, start, stop, read, quit") parser.add_argument("--port", default="/dev/ttyHS1") parser.add_argument("--baud", type=int, default=115200) 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("--sources", required=True, nargs="+", help="1-4 camera indices, device paths or video files") + parser.add_argument("--config", "-c", default="finger", help="default config for add commands") 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("--read-only", action="store_true", help="ignore all UART commands; exit from the login console") args = parser.parse_args(argv) 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") - configs = args.config or ["finger"] - try: - sources = [parse_source(s) for s in args.sources] - controller = UartFingerController( - None, sources=sources, configs=configs, - loop_video=args.loop_video, show=args.show, verbose=args.verbose, - read_only=args.read_only, - ) - except ValueError as exc: - parser.error(str(exc)) + controller = UartFingerController( + None, config=args.config, loop_video=args.loop_video, + show=args.show, verbose=args.verbose, + ) 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) + print(f"UART service on {args.port} @ {args.baud}; waiting for add SOURCE, then start", file=sys.stderr, flush=True) controller.serve() except KeyboardInterrupt: print("Interrupted.", file=sys.stderr) diff --git a/serial/finger_api.py b/serial/finger_api.py index 3bb1a18..3d8276f 100644 --- a/serial/finger_api.py +++ b/serial/finger_api.py @@ -1,7 +1,7 @@ """Single-threaded finger acquisition API, independent of UART and the CLI. -Sensor IDs are 1-based, fixed by source order. Calls and runtime ownership belong -to one thread; the caller must close the bank in a finally block. Values use the +Sensor IDs are 1-based, assigned after each successful add(). Calls and runtime +ownership belong to one thread; close the bank in a finally block. Values use the SDK's FNORMAL/FSHEARX/FSHEARY axes and calibration units, not assumed Newtons. """ @@ -40,74 +40,68 @@ class ForceReading: class FingerSensorBank: - """Own 1-4 independent runtimes; connect_all() precedes read_all(). + """Own up to four independently added runtimes; add() 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, *, configs=("finger",), verbose=False): - sources = tuple(parse_source(str(source)) for source in sources) - configs = tuple(configs) - if not 1 <= len(sources) <= 4: - raise ValueError("provide 1 to 4 sources") - if len(configs) not in (1, len(sources)): - raise ValueError("provide one config or one config per source") - if any(isinstance(source, int) and source < 0 for source in sources): - 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)) + def __init__(self, *, config="finger", verbose=False): + self.default_config = config + self.sources = {} + self.configs = {} self.runtimes = {} self.verbose = verbose - def connect_all(self) -> None: - """Prepare every source, rolling back all runtimes on any failure.""" + @staticmethod + def _source_identity(source): + if isinstance(source, int): + return f"camera:{source}" + if source.startswith("/dev/video") and source[len("/dev/video"):].isdigit(): + return f"camera:{int(source[len('/dev/video'):])}" + return str(Path(source).resolve()) + + def add(self, source: Source, *, config: str | None = None) -> int: + """Open/prepare one source. Failure preserves existing sensors and IDs.""" + source = parse_source(str(source)) + if len(self.runtimes) >= 4: + raise ValueError("maximum 4 sensors") + if isinstance(source, int) and source < 0: + raise ValueError("camera index must be non-negative") + identity = self._source_identity(source) + if any(self._source_identity(existing) == identity for existing in self.sources.values()): + raise ValueError("source already added") + if is_video_file(source) and not Path(source).is_file(): + raise ValueError(f"file not found: {source}") from orisys.finger import FingerRuntime - if self.runtimes: - raise RuntimeError("already connected; close before reconnecting") - sensor_id = None + config = self.default_config if config is None else config + runtime = None try: - 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 + runtime = FingerRuntime.open(config=config, video=source, verbose=self.verbose) + ok, _ = runtime.warmup(color=False) + if not ok: + raise RuntimeError("warmup could not read a frame") except BaseException: - self.close() + if runtime is not None: + try: + runtime.close() + except Exception: + logger.exception("New sensor cleanup failed") raise + sensor_id = len(self.runtimes) + 1 + self.sources[sensor_id] = source + self.configs[sensor_id] = config + self.runtimes[sensor_id] = runtime + return sensor_id 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") + if not self.runtimes: + raise RuntimeError("no sensors added") readings = [] for sensor_id in self.sources: try: @@ -152,5 +146,7 @@ class FingerSensorBank: except Exception as exc: logger.exception("Sensor %s cleanup failed", sensor_id) first_error = first_error or exc + self.sources.clear() + self.configs.clear() if first_error is not None: raise first_error diff --git a/tests/check_backend.py b/tests/check_backend.py index 3283f88..9e764b6 100644 --- a/tests/check_backend.py +++ b/tests/check_backend.py @@ -55,9 +55,13 @@ def main(): 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]) + bank = FingerSensorBank() try: - bank.connect_all() + for source in sources[:count]: + existing = dict(bank.runtimes) + bank.add(source) + assert all(bank.runtimes[i] is runtime for i, runtime in existing.items()) + assert len(bank.read_all()) == len(bank.runtimes) runtimes = list(bank.runtimes.values()) for left, right in itertools.combinations(runtimes, 2): assert left.capture is not right.capture @@ -67,7 +71,7 @@ def main(): 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) + print(f"PASS {count} sources: incremental add/read, independent states, 3 final rounds", flush=True) finally: bank.close() assert all(runtime.capture is None for runtime in runtimes) diff --git a/tests/test_uart.py b/tests/test_uart.py index 823e59f..f44df11 100644 --- a/tests/test_uart.py +++ b/tests/test_uart.py @@ -90,98 +90,194 @@ class HardwareFreeTest(unittest.TestCase): stderr.__enter__() self.addCleanup(stderr.__exit__, None, None, None) finger = types.ModuleType("orisys.finger") - finger.FingerRuntime = types.SimpleNamespace(open=lambda **kwargs: FakeRuntime()) + finger.FingerRuntime = types.SimpleNamespace( + open=lambda **kwargs: FakeRuntime() + ) capture = types.ModuleType("orisys.capture") - capture.is_live_camera = lambda source: isinstance(source, int) or str(source).startswith("/dev/video") + capture.is_live_camera = lambda source: ( + isinstance(source, int) or str(source).startswith("/dev/video") + ) capture.list_cameras = lambda **kwargs: [] package = types.ModuleType("orisys") package.finger, package.capture = finger, capture - modules = patch.dict(sys.modules, { - "orisys": package, "orisys.finger": finger, "orisys.capture": capture, - }) + modules = patch.dict( + sys.modules, + { + "orisys": package, + "orisys.finger": finger, + "orisys.capture": capture, + }, + ) modules.start() self.addCleanup(modules.stop) -class ConnectionFirstTests(HardwareFreeTest): - def test_connects_every_source_before_ready_or_force(self): +class ManualAddTests(HardwareFreeTest): + def test_service_starts_empty_without_opening_a_camera(self): + ser = ByteSerial(b"quit\n") + controller = uart.UartFingerController(ser) + with patch("orisys.finger.FingerRuntime.open") as open_runtime: + controller.serve() + open_runtime.assert_not_called() + self.assertEqual(ser.output, ["OK ready 0", "OK quit"]) + + def test_add_one_through_four_then_start_produces_uniform_output(self): for count in range(1, 5): with self.subTest(count=count): - ser = ByteSerial() - controller = uart.UartFingerController(ser, sources=list(range(count))) - events = [] - def open_runtime(events=events, **kwargs): - events.append(kwargs["video"]) - return FakeRuntime() + commands = "".join(f"add {i * 2}\n" for i in range(count)) + "start\n" + ser = ByteSerial(commands.encode()) + controller = uart.UartFingerController(ser) original_write = ser.write - def write(data, events=events, count=count, ser=ser, original_write=original_write): - self.assertEqual(events, list(range(count))) + + def write(data, ser=ser, original_write=original_write, count=count): size = original_write(data) - if data.startswith(b"force,") and sum(x.startswith("force,") for x in ser.output) == count: + 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) - ]) + controller.serve() + self.assertEqual( + ser.output[: count + 2], + ["OK ready 0"] + + [f"OK added {i}" for i in range(1, count + 1)] + + ["OK start"], + ) + 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) + ], + ) controller.close_runtime() - 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() + def test_add_alone_does_not_start_output(self): + ser = ByteSerial(b"add 0\nquit\n") + uart.UartFingerController(ser).serve() + self.assertEqual(ser.output, ["OK ready 0", "OK added 1", "OK quit"]) + + def test_start_and_read_require_at_least_one_sensor(self): + ser = ByteSerial() + controller = uart.UartFingerController(ser) + for command in ("start", "read"): + controller._dispatch(command) + self.assertTrue(ser.output[-1].startswith(f"ERR {command}")) + self.assertFalse(controller._streaming) + + def test_add_requires_stop_and_never_interrupts_existing_stream(self): + ser = ByteSerial() + controller = uart.UartFingerController(ser) + controller._dispatch("add 0") + controller._dispatch("start") + first = controller.bank.runtimes[1] + with patch("orisys.finger.FingerRuntime.open") as open_runtime: + controller._dispatch("add 2") + open_runtime.assert_not_called() + self.assertEqual(ser.output[-1], "ERR add stop streaming first") + self.assertTrue(controller._streaming) + controller._dispatch("stop") + controller._dispatch("add 2") + self.assertEqual(ser.output[-1], "OK added 2") + self.assertIs(controller.bank.runtimes[1], first) + self.assertFalse(first.closed) + self.assertFalse(controller._streaming) + + def test_failed_add_preserves_other_sensor_and_reuses_available_id(self): + bank = FingerSensorBank() + self.assertEqual(bank.add(0), 1) + first = bank.runtimes[1] + failed = FakeRuntime(warmup_error=True) + with ( + patch("orisys.finger.FingerRuntime.open", return_value=failed), + self.assertRaises(RuntimeError), + ): + bank.add(2) + self.assertTrue(failed.closed) + self.assertFalse(first.closed) + self.assertEqual(bank.sources, {1: 0}) + self.assertEqual(bank.add(4), 2) + self.assertEqual(bank.sources, {1: 0, 2: 4}) + + def test_failed_warmup_without_exception_closes_new_runtime(self): + bank = FingerSensorBank() + failed = FakeRuntime(warmup_ok=False) + with ( + patch("orisys.finger.FingerRuntime.open", return_value=failed), + self.assertRaisesRegex(RuntimeError, "warmup"), + ): + bank.add(0) + self.assertTrue(failed.closed) self.assertEqual(bank.runtimes, {}) - self.assertTrue(all(r.closed for r in opened)) + self.assertEqual(bank.sources, {}) - 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_duplicate_source_and_fifth_camera_are_rejected_before_open(self): + bank = FingerSensorBank() + bank.add(0) + for source in ("0", "/dev/video0", -1): + with ( + self.subTest(source=source), + patch("orisys.finger.FingerRuntime.open") as open_runtime, + ): + with self.assertRaises(ValueError): + bank.add(source) + open_runtime.assert_not_called() + for source in (2, 4, 6): + bank.add(source) + with patch("orisys.finger.FingerRuntime.open") as open_runtime: + with self.assertRaisesRegex(ValueError, "4"): + bank.add(8) + open_runtime.assert_not_called() - def test_legacy_commands_are_rejected(self): + def test_each_sensor_can_use_its_own_config(self): + bank = FingerSensorBank(config="shared.json") + seen = [] + + def open_runtime(**kwargs): + seen.append(kwargs["config"]) + return FakeRuntime() + + with patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime): + bank.add(0) + bank.add(2, config="right.json") + self.assertEqual(seen, ["shared.json", "right.json"]) + self.assertIsNot(bank.runtimes[1], bank.runtimes[2]) + self.assertEqual([r.sensor_id for r in bank.read_all()], [1, 2]) + + def test_quoted_source_and_config_are_one_argument_each(self): ser = ByteSerial() - controller = uart.UartFingerController(ser, sources=[0, 2]) - for command in ["connect", "connect 2", "connect all", "run", "run all", "search", "exit"]: - self.assertTrue(controller._dispatch(command)) + controller = uart.UartFingerController(ser) + with patch.object(controller.bank, "add", return_value=1) as add: + controller._dispatch('add "/tmp/test clip.mp4" "my config.json"') + add.assert_called_once_with("/tmp/test clip.mp4", config="my config.json") + self.assertEqual(ser.output[-1], "OK added 1") + + def test_removed_commands_and_arguments_are_rejected(self): + ser = ByteSerial() + controller = uart.UartFingerController(ser) + for command in ("connect", "run", "search", "exit"): + controller._dispatch(command) self.assertTrue(ser.output[-1].startswith("ERR unknown")) - - def test_no_per_sensor_or_all_command_arguments(self): - ser = ByteSerial() - controller = uart.UartFingerController(ser, sources=[0, 2]) - for command in ["start 1", "stop all", "read 2", "quit extra"]: - self.assertTrue(controller._dispatch(command)) + for command in ( + "add", + "add 0 a extra", + "start 1", + "stop all", + "read 2", + "quit extra", + 'add "broken', + ): + controller._dispatch(command) self.assertTrue(ser.output[-1].startswith("ERR command")) - def test_start_stop_and_read_control_entire_connected_bank(self): + def test_processing_failure_stops_output_without_partial_round(self): ser = ByteSerial() - 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_read_all_never_publishes_partial_round_on_processing_failure(self): - ser = ByteSerial() - controller = uart.UartFingerController(ser, sources=[0, 2]) - controller.bank.connect_all() + controller = uart.UartFingerController(ser) + controller._dispatch("add 0") + controller._dispatch("add 2") controller.bank.runtimes[2].grab_error = True controller._streaming = True controller._dispatch("read") @@ -189,141 +285,125 @@ class ConnectionFirstTests(HardwareFreeTest): self.assertFalse(any(x.startswith("force,") for x in ser.output)) self.assertIn("sensor 2", ser.output[-1]) - 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]) + 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_lf_crlf_commands_are_separate(self): - controller = uart.UartFingerController(ByteSerial(b"stop\rstart\nread\r\n"), sources=[0]) + controller = uart.UartFingerController(ByteSerial(b"stop\rstart\nread\r\n")) 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]) + 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.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) - if data.startswith(b"force,"): - raise KeyboardInterrupt() - return size - ser.write = write - with self.assertRaises(KeyboardInterrupt): - controller.serve() - self.assertEqual(ser.output, ["OK ready 1", "force,1,1.250000,-2.500000,3.750000"]) - def test_short_write_is_transport_failure(self): ser = ByteSerial() ser.write = lambda data: len(data) - 1 with self.assertRaises(OSError): - uart.UartFingerController(ser, sources=[0]).reply("OK ready 1") + uart.UartFingerController(ser).reply("OK ready 0") def test_error_newlines_do_not_inject_records(self): ser = ByteSerial() - uart.UartFingerController(ser, sources=[0]).reply("ERR read bad\nOK quit\r\n") + uart.UartFingerController(ser).reply("ERR read bad\nOK quit\r\n") self.assertEqual(ser.output, ["ERR read bad OK quit"]) 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 = FingerSensorBank() + bank.add(0) 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_all() + bank = FingerSensorBank() + bank.add(0) + bank.add(2) 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() + bank = FingerSensorBank() + bank.add(0) + bank.add(2) second = bank.runtimes[2] - bank.runtimes[1].close = lambda: (_ for _ in ()).throw(RuntimeError("close failed")) + bank.runtimes[1].close = lambda: (_ for _ in ()).throw( + RuntimeError("close failed") + ) with self.assertRaisesRegex(RuntimeError, "close failed"): bank.close() self.assertTrue(second.closed) self.assertEqual(bank.runtimes, {}) + self.assertEqual(bank.sources, {}) class CliTests(HardwareFreeTest): - 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"]]: + def test_startup_opens_only_uart_and_waits_for_add(self): + ser = ByteSerial(b"quit\n") + ser.close = lambda: None + with ( + patch.object(uart, "open_serial", return_value=ser), + patch("orisys.finger.FingerRuntime.open") as runtime, + ): + self.assertEqual(uart.main([]), 0) + runtime.assert_not_called() + self.assertEqual(ser.output, ["OK ready 0", "OK quit"]) + + def test_automatic_setup_options_are_removed(self): + for args in ( + ["--sources", "0"], + ["--read-only"], + ["--autostart"], + ["--camera-id", "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(status.exception.code, 2) serial.assert_not_called() - def test_serial_failure_closes_every_connected_runtime(self): + def test_transport_failure_closes_cameras_added_by_commands(self): + ser = ByteSerial(b"add 0\nadd 2\nstart\n") + ser.closed = False + ser.close = lambda: setattr(ser, "closed", True) 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) + + original_write = ser.write + + def write(data): + if data.startswith(b"force,"): + raise OSError("UART disconnected") + return original_write(data) + + ser.write = write + with ( + patch.object(uart, "open_serial", return_value=ser), + patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime), + ): + self.assertEqual(uart.main([]), 1) self.assertEqual(len(opened), 2) + self.assertTrue(ser.closed) self.assertTrue(all(r.closed for r in opened))