支持传感器独立校准文件与批量配置启动

保留逐台添加，并通过显式配置一次连接全组；失败时撤销本批新增连接。通过 35 项回归测试、真实 SDK 校准与 1–4 路验证、Ruff 和文档检查。

README.md

@@ -0,0 +1,4 @@
+# Dragon Q6A deployment helpers
+
+See [UART setup, commands and API](docs/uart.md) for controlling or passively
+reading 1–4 sensors through a data UART separate from the TTL login console.

docs/uart.md

@@ -0,0 +1,222 @@
+# UART finger service
+
+Connect 1-4 sources individually through UART, or load an explicit setup file
+with their configurations and calibration paths. There is no camera discovery
+or hot-plug monitoring; every source is selected by the user.
+
+## Start the service and add cameras
+
+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
+python serial/example_uart.py --port /dev/ttyHS1 --config finger
+```
+
+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
+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
+```
+
+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.
+
+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.
+
+To select this sensor's force calibration, append `--calibration`:
+
+```text
+add /dev/video0 --calibration "/home/radxa/calibrations/finger/DD02-14.json"
+```
+
+An optional config still works: `add SOURCE CONFIG --calibration PATH`.
+Explicit calibration paths are passed to SDK `FingerRuntime.open(cal_path=...)`.
+A missing or invalid file fails the addition, without falling back to raw values.
+Without the option, normal SDK calibration selection applies. This does not
+replace calibration on a camera already added; restart to change that mapping.
+
+The Windows example `D:\orisys_ws\data\calibrations\finger\DD02-14.json` refers
+to a file on that computer. Copy it to the Q6A before using a board-local path.
+Quote Windows paths in UART commands, or use forward slashes. The user must
+match each camera to its physical sensor's calibration; no serial-number discovery
+is performed.
+
+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.
+
+## Batch setup and one-command startup
+
+Copy [sensors.example.json](../serial/sensors.example.json), then replace its
+camera paths and calibration filenames with the real mappings. `SENSOR-2` through
+`SENSOR-4` are placeholders. The JSON format is:
+
+```json
+{
+  "sensors": [
+    {"source": "/dev/video0", "config": "finger", "calibration": "calibrations/finger/DD02-14.json"},
+    {"source": "/dev/video2", "calibration": "calibrations/finger/SENSOR-2.json"}
+  ]
+}
+```
+
+List 1-4 entries. `source` is required; `config` defaults to startup `--config`.
+An omitted/null `calibration` uses normal SDK selection. Relative file paths
+resolve from the setup file's directory, independently of the working directory.
+Named configs such as `finger` remain names. Unknown fields are rejected.
+
+Once saved as `sensors.json`, one terminal command connects the whole set and
+starts continuous UART output:
+
+```bash
+python serial/example_uart.py --port /dev/ttyHS1 --setup sensors.json --start
+```
+
+Omit `--start` to connect and wait for UART commands. With neither option, empty
+manual startup is unchanged. `--start` requires `--setup`. Preparation failure
+rolls back the entire startup set and exits before opening the UART.
+
+An already-running service can also load the file:
+
+```text
+load "/home/radxa/sensors.json"
+OK loaded 1 2
+start
+```
+
+`load` only connects; it does not start output. It can extend manually added
+sensors, up to four total. Batch failure removes only that batch's new sensors
+and leaves previously added cameras unchanged. `add` and `load` require stopped
+output. Wait for each reply before proceeding.
+
+## Commands and output
+
+| Command | Behavior |
+|---|---|
+| `add SOURCE [CONFIG] [--calibration PATH]` | Prepare one source while output is stopped; reply `OK added ID`. Config/calibration apply only to this sensor. |
+| `load SETUP.json` | Add the sources and calibration mappings in a setup file while stopped; reply `OK loaded ID ...`. |
+| `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. |
+
+To extend the camera set after streaming has started, pause before adding:
+
+```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 protocol commands remain 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.
+Place its directory on the caller's module search path. Adding sources is explicit:
+
+```python
+from finger_api import FingerSensorBank
+
+bank = FingerSensorBank(config="finger")
+try:
+    first_id = bank.add(0, cal_path="/home/radxa/calibrations/finger/DD02-14.json")
+    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. `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.
+`bank.add_setup("sensors.json")` uses the same transactional batch loader as the
+CLI and UART. `bank.calibration_paths` records explicit calibration paths by ID.
+
+## Backend evidence and remaining limits
+
+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.
+The explicit `DD02-14.json` calibration was also tested on 1-4 recorded sources:
+each instance loaded its assigned file and reported three calibrated `N` axes.
+Four-source JSON loading passed, and invalid explicit calibration was rejected
+without altering an existing sensor. This validates software binding, not physical accuracy.
+
+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 --calibration /path/to/DD02-14.json --scratch .scratch/uart-calibration-setup/backend-run
+python -m unittest discover -s tests -v
+```
+
+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.

serial/example_uart.py

@@ -1,460 +1,271 @@
 #!/usr/bin/env python3
-"""UART finger controller for Radxa / Dragon Q6A (sensor-side receiver).
+"""Headless UART finger service for 1-4 sensors; see ../docs/uart.md.
 
-Combines ORY-FINGER processing (same path as ``examples/finger_pipeline.py``)
+The login console and data UART must be separate ports. Cameras are selected
-with a line-based UART command interface. A host (or ``uart_send.py``-style
+explicitly by UART commands or a startup setup file. All force records include a sensor ID.
-transmitter) sends commands; this script replies and streams forces.
+Importing this module opens no devices; acquisition is owned by finger_api.
-
-Dependencies (beyond orisys):
-  pip install pyserial
-
-Default UART (Dragon Q6A):
-  port=/dev/ttyHS1  baud=115200  8N1
-
-Commands (one line, case-insensitive):
-  search              List cameras (/dev/video*) via orisys.capture.list_cameras
-  connect [source]    Open FingerRuntime (camera index, /dev/videoN, or video file)
-  run                 Stream FNORMAL / FSHEARX / FSHEARY until stop (or EOF)
-  stop                Stop force streaming; keep source connected
-  quit | exit         Close runtime and serial, then exit
-
-Replies:
-  camera,<index>,<path>,<WxH>
-  OK search <N>
-  OK connect <source>
-  OK run
-  force,<fnormal>,<fshearx>,<fsheary>
-  OK eof              Video file ended (when --no-loop-video)
-  OK stop
-  OK quit
-  ERR <context> <message>
-
-Examples:
-  # Live camera over UART
-  python scripts/example_uart.py --port /dev/ttyHS1 --config finger
-
-  # Offline debug from a recorded file (no camera needed)
-  python scripts/example_uart.py --port /dev/ttyHS1 --video /path/to/clip.mp4
-  # then UART: connect   (uses --video)  or  connect /path/to/clip.mp4
-  # then UART: run
 """
 
 from __future__ import annotations
 
 import argparse
+import logging
+import math
 import os
 import shlex
 import sys
-from pathlib import Path
+from collections import deque
-from typing import Optional, Union
 
 os.environ.setdefault("OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS", "0")
 
-try:
+from finger_api import FingerSensorBank, parse_source
-    import serial
-except ImportError as exc:  # pragma: no cover
-    raise SystemExit(
-        "pyserial is required for this script.\n"
-        "  pip install pyserial\n"
-        f"Import error: {exc}"
-    ) from exc
 
-import cv2
+logger = logging.getLogger(__name__)
-from orisys.capture import is_live_camera, list_cameras
-from orisys.finger import FingerRuntime
-
-Source = Union[int, str]
-
-
-def _parse_source(token: str) -> Optional[Source]:
-    """Accept camera index int, numeric string, V4L path, or filesystem path."""
-    text = str(token).strip().strip('"').strip("'")
-    if not text:
-        return None
-    if text.isdigit() or (text.startswith("-") and text[1:].isdigit()):
-        return int(text)
-    return text
-
-
-def _is_video_file(source: Source) -> bool:
-    """True when source is a filesystem video path (not a live camera)."""
-    if is_live_camera(source):
-        return False
-    if isinstance(source, int):
-        return False
-    path = Path(str(source))
-    return path.is_file() or path.suffix.lower() in {
-        ".mp4",
-        ".avi",
-        ".mkv",
-        ".mov",
-        ".m4v",
-        ".wmv",
-        ".webm",
-    }
 
 
 class UartFingerController:
-    """Serial command dispatcher around a single FingerRuntime."""
+    """Single-threaded command service; the caller owns serial and final cleanup."""
 
     def __init__(
-        self,
+        self, ser, *, config="finger", loop_video=True,
-        ser: serial.Serial,
+        show=False, verbose=False,
-        *,
-        config: str = "finger",
-        default_source: Source = 0,
-        max_camera_index: int = 10,
-        loop_video: bool = True,
-        show: bool = False,
-        verbose: bool = False,
     ):
         self.ser = ser
-        self.config = config
+        self.bank = FingerSensorBank(config=config, verbose=verbose)
-        self.default_source = default_source
-        self.max_camera_index = max_camera_index
         self.loop_video = loop_video
         self.show = show
         self.verbose = verbose
-
-        self.runtime: Optional[FingerRuntime] = None
-        self._connected_source: Optional[Source] = None
-        self._last_search_source: Optional[Source] = None
         self._streaming = False
         self._should_quit = False
+        self._partial = bytearray()
+        self._lines = deque()
+        self._discard_line = False
 
-    def reply(self, line: str) -> None:
+    def reply(self, line):
-        text = line.rstrip("\r\n") + "\n"
+        # SDK exceptions can contain newlines; every reply must be one record.
-        self.ser.write(text.encode("utf-8"))
+        text = " ".join(str(line).splitlines()) + "\n"
-        self.ser.flush()
+        data = text.encode("utf-8")
+        if self.ser.write(data) != len(data):
+            raise OSError("incomplete UART write")
         if self.verbose:
-            print(f"<< {text.rstrip()}", flush=True)
+            print(f"<< {text.rstrip()}", file=sys.stderr, flush=True)
 
-    def close_runtime(self) -> None:
+    def close_runtime(self):
         self._streaming = False
-        if self.runtime is not None:
-            try:
-                self.runtime.close()
-            except Exception:  # noqa: BLE001
-                pass
-            self.runtime = None
-            self._connected_source = None
-        if self.show:
-            try:
-                cv2.destroyAllWindows()
-            except Exception:  # noqa: BLE001
-                pass
-
-    def _resolve_connect_source(self, args: list[str]) -> Source:
-        if args:
-            # Join remaining tokens so unquoted paths with spaces still work.
-            joined = " ".join(args).strip()
-            parsed = _parse_source(joined)
-            if parsed is None:
-                raise ValueError("empty connect source")
-            return parsed
-        # Prefer --video default over last search so offline debug stays on file.
-        if isinstance(self.default_source, str) and not is_live_camera(self.default_source):
-            return self.default_source
-        if self._last_search_source is not None:
-            return self._last_search_source
-        return self.default_source
-
-    def cmd_search(self, _args: list[str]) -> None:
         try:
-            cameras = list_cameras(max_index=self.max_camera_index)
+            self.bank.close()
-        except Exception as exc:  # noqa: BLE001
+        finally:
-            self.reply(f"ERR search {exc}")
-            return
-
-        self._last_search_source = None
-        for cam in cameras:
-            path = (cam.path or f"/dev/video{cam.index}").strip() or f"/dev/video{cam.index}"
-            size = f"{cam.width}x{cam.height}"
-            self.reply(f"camera,{cam.index},{path},{size}")
-            if self._last_search_source is None:
-                self._last_search_source = cam.index
-        self.reply(f"OK search {len(cameras)}")
-
-    def cmd_connect(self, args: list[str]) -> None:
-        try:
-            source = self._resolve_connect_source(args)
-        except ValueError as exc:
-            self.reply(f"ERR connect {exc}")
-            return
-
-        if isinstance(source, str) and not is_live_camera(source):
-            path = Path(source)
-            if not path.is_file():
-                self.reply(f"ERR connect file not found: {source}")
-                return
-
-        self._streaming = False
-        self.close_runtime()
-        try:
-            runtime = FingerRuntime.open(
-                config=self.config,
-                video=source,
-                verbose=self.verbose,
-            )
-            runtime.warmup(color=False)
-        except Exception as exc:  # noqa: BLE001
-            self.reply(f"ERR connect {exc}")
-            return
-
-        self.runtime = runtime
-        self._connected_source = source
-        kind = "file" if _is_video_file(source) else "camera"
-        self.reply(f"OK connect {source} ({kind})")
-
-    def _rewind_video(self) -> bool:
-        """Seek connected file source back to frame 0. Return True on success."""
-        if self.runtime is None or self._connected_source is None:
-            return False
-        if not _is_video_file(self._connected_source):
-            return False
-        cap = self.runtime.capture
-        if cap is None:
-            return False
-        try:
-            return bool(cap.set(cv2.CAP_PROP_POS_FRAMES, 0))
-        except Exception:  # noqa: BLE001
-            return False
-
-    def _poll_command_line(self, *, blocking: bool = True) -> Optional[str]:
-        """Read one command line if available; return None on timeout/empty."""
-        if not blocking and self.ser.in_waiting <= 0:
-            return None
-        raw = self.ser.readline()
-        if not raw:
-            return None
-        try:
-            line = raw.decode("utf-8", errors="replace").strip()
-        except Exception:  # noqa: BLE001
-            return None
-        return line or None
-
-    def cmd_run(self, _args: list[str]) -> None:
-        if self.runtime is None:
-            self.reply("ERR run not connected")
-            return
-
-        self._streaming = True
-        self.reply("OK run")
-        tactile = self.runtime.tactile
-        is_file = self._connected_source is not None and _is_video_file(self._connected_source)
-
-        while self._streaming and not self._should_quit:
-            ok, unrolled = self.runtime.grab_and_process(color=False)
-            if not ok:
-                if is_file and self.loop_video and self._rewind_video():
-                    if self.verbose:
-                        print("video EOF: looping", flush=True)
-                    continue
-                if is_file:
-                    self.reply("OK eof")
-                else:
-                    self.reply("ERR run grab failed")
-                self._streaming = False
-                break
-
-            tactile.compute_deformation()
-            tactile.compute_contact()
-            fnormal, fshearx, fsheary = tactile.read_info(
-                tactile.info.FNORMAL,
-                tactile.info.FSHEARX,
-                tactile.info.FSHEARY,
-            )
-            self.reply(f"force,{float(fnormal):.6f},{float(fshearx):.6f},{float(fsheary):.6f}")
-
             if self.show:
-                try:
+                import cv2
-                    import orisys
 
-                    arrows = orisys.util.draw_arrows(
+                cv2.destroyAllWindows()
-                        unrolled,
-                        tactile.flow,
-                        threshold=0.4,
-                        grid_spacing=10,
-                        arrow_scale=1.0,
-                        valid_mask=tactile.valid_mask,
-                    )
-                    cv2.imshow("UV Unrolled", arrows)
-                    key = cv2.waitKey(1) & 0xFF
-                    if key == ord("q"):
-                        self._should_quit = True
-                        break
-                    if key == ord("r"):
-                        tactile.reset()
-                except Exception as exc:  # noqa: BLE001
-                    if self.verbose:
-                        print(f"show error: {exc}", file=sys.stderr)
 
-            # Non-blocking command poll so force streaming is not delayed by UART timeout.
+    def _poll_command_line(self, *, blocking=True):
-            pending = self._poll_command_line(blocking=False)
+        if self._lines:
-            if pending is not None:
+            return self._lines.popleft()
-                if self.verbose:
+        waiting = self.ser.in_waiting
-                    print(f">> {pending}", flush=True)
+        if not waiting and not blocking:
-                if not self._dispatch(pending, allow_run=False):
+            return None
-                    break
+        # Never readline(): a timeout may return only half of a command.
+        raw = self.ser.read(min(waiting, 4096) if waiting else 1)
+        for byte in raw:
+            if byte in (10, 13):
+                if self._discard_line:
+                    self._discard_line = False
+                    self._partial.clear()
+                    self.reply("ERR command line too long (max 1024 bytes)")
+                elif self._partial:
+                    self._lines.append(self._partial.decode("utf-8", errors="replace").strip())
+                    self._partial.clear()
+            elif not self._discard_line:
+                self._partial.append(byte)
+                if len(self._partial) > 1024:
+                    self._partial.clear()
+                    self._discard_line = True
+        return self._lines.popleft() if self._lines else None
 
-    def cmd_stop(self, _args: list[str]) -> None:
+    def _read_round(self):
-        self._streaming = False
+        try:
-        self.reply("OK stop")
+            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("Acquisition round failed")
+            self._streaming = False
+            self.reply(f"ERR read {exc}")
+            return
+        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 cmd_quit(self, _args: list[str]) -> None:
+    def _show(self, reading):
-        self._streaming = False
+        import cv2
-        self._should_quit = True
+        import orisys
-        self.reply("OK quit")
 
-    def _dispatch(self, line: str, *, allow_run: bool = True) -> bool:
+        tactile = self.bank.runtimes[reading.sensor_id].tactile
-        """Dispatch one command line. Return False if the main loop should exit."""
+        try:
+            arrows = orisys.util.draw_arrows(
+                reading.image, tactile.flow, threshold=0.4, grid_spacing=10,
+                arrow_scale=1.0, valid_mask=tactile.valid_mask,
+            )
+            cv2.imshow(f"UV Unrolled {reading.sensor_id}", arrows)
+            key = cv2.waitKey(1) & 0xFF
+            if key == ord("q"):
+                self._streaming = False
+                self._should_quit = True
+                self.reply("OK quit")
+            elif key == ord("r"):
+                tactile.reset()
+        except Exception:
+            logger.exception("Sensor %s display failed", reading.sensor_id)
+
+    def _dispatch(self, line):
         try:
             parts = shlex.split(line)
-        except ValueError:
+        except ValueError as exc:
-            parts = line.split()
+            self.reply(f"ERR command {exc}")
+            return True
         if not parts:
             return True
-        cmd = parts[0].lower()
+        command = parts[0].lower()
-        args = parts[1:]
+        if command == "add":
-
+            if self._streaming:
-        if cmd == "search":
+                self.reply("ERR add stop streaming first")
-            self.cmd_search(args)
-        elif cmd == "connect":
-            self.cmd_connect(args)
-        elif cmd == "run":
-            if not allow_run:
-                self.reply("ERR run already running")
             else:
-                self.cmd_run(args)
+                parser = argparse.ArgumentParser(prog="add", add_help=False, exit_on_error=False)
-        elif cmd == "stop":
+                parser.add_argument("source")
-            self.cmd_stop(args)
+                parser.add_argument("config", nargs="?")
-        elif cmd in ("quit", "exit"):
+                parser.add_argument("--calibration")
-            self.cmd_quit(args)
+                try:
-            return False
+                    args = parser.parse_args(parts[1:])
-        else:
+                except (argparse.ArgumentError, SystemExit):
-            self.reply(f"ERR unknown {cmd}")
+                    self.reply("ERR command expected add SOURCE [CONFIG] [--calibration PATH]")
+                    return True
+                try:
+                    sensor_id = self.bank.add(
+                        parse_source(args.source), config=args.config, cal_path=args.calibration,
+                    )
+                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 == "load":
+            if len(parts) != 2:
+                self.reply("ERR command expected load SETUP.json")
+            elif self._streaming:
+                self.reply("ERR load stop streaming first")
+            else:
+                try:
+                    sensor_ids = self.bank.add_setup(parts[1])
+                except Exception as exc:
+                    logger.exception("Setup loading failed")
+                    self.reply(f"ERR load {exc}")
+                else:
+                    self.reply("OK loaded " + " ".join(map(str, sensor_ids)))
+        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":
+            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")
+        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) -> None:
+    def serve(self, *, start=False):
-        self.reply("OK ready")
+        if start and not self.bank.runtimes:
+            raise RuntimeError("no sensors added")
+        self.reply(f"OK ready {len(self.bank.sources)}")
+        self._streaming = start
         while not self._should_quit:
-            line = self._poll_command_line()
+            for _ in range(16):
-            if line is None:
+                line = self._poll_command_line(blocking=not self._streaming)
-                continue
+                if line is None:
-            if self.verbose:
+                    break
-                print(f">> {line}", flush=True)
+                if not self._dispatch(line):
-            if not self._dispatch(line):
+                    return
-                break
+            if self._streaming:
+                self._read_round()
 
-
+def open_serial(port, baud, timeout, write_timeout=1.0):
-def open_serial(port: str, baud: int, timeout: float) -> serial.Serial:
     try:
-        return serial.Serial(
+        import serial
-            port=port,
+    except ImportError as exc:
-            baudrate=baud,
+        raise RuntimeError("pyserial is required: python -m pip install pyserial") from exc
-            bytesize=serial.EIGHTBITS,
+    return serial.Serial(
-            parity=serial.PARITY_NONE,
+        port=port, baudrate=baud, bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE,
-            stopbits=serial.STOPBITS_ONE,
+        stopbits=serial.STOPBITS_ONE, timeout=timeout, write_timeout=write_timeout,
-            timeout=timeout,
+        exclusive=True,
-        )
+    )
-    except serial.SerialException as exc:
-        print(f"Serial error: {exc}", file=sys.stderr)
-        print("Check:", file=sys.stderr)
-        print(f"  1. Device node {port} exists", file=sys.stderr)
-        print("  2. User has access (sudo or dialout group)", file=sys.stderr)
-        print("  3. Wiring / USB-TTL connection", file=sys.stderr)
-        raise SystemExit(1) from exc
 
 
-def main(argv: list[str] | None = None) -> int:
+def main(argv=None):
-    parser = argparse.ArgumentParser(
+    parser = argparse.ArgumentParser(description="Manual UART finger service: add, start, stop, read, quit")
-        description="UART finger controller (search/connect/run/stop force stream)"
+    parser.add_argument("--port", default="/dev/ttyHS1")
-    )
+    parser.add_argument("--baud", type=int, default=115200)
-    parser.add_argument("--port", default="/dev/ttyHS1", help="Serial device (default: /dev/ttyHS1)")
+    parser.add_argument("--timeout", type=float, default=0.1, help="idle read timeout in seconds")
-    parser.add_argument("--baud", type=int, default=115200, help="Baud rate (default: 115200)")
+    parser.add_argument("--write-timeout", type=float, default=1.0)
-    parser.add_argument(
+    parser.add_argument("--config", "-c", default="finger", help="default config for add commands")
-        "--timeout",
+    parser.add_argument("--setup", help="JSON file listing sources, configs and calibrations to connect")
-        type=float,
+    parser.add_argument("--start", action="store_true", help="start output after loading --setup")
-        default=0.1,
+    parser.add_argument("--no-loop-video", dest="loop_video", action="store_false")
-        help="Serial readline timeout seconds (default: 0.1)",
+    parser.add_argument("--show", action="store_true", help="local desktop debug only")
-    )
+    parser.add_argument("--verbose", action="store_true")
-    parser.add_argument("--config", "-c", default="finger", help="Orisys config name or path")
-    parser.add_argument(
-        "--camera-id",
-        type=int,
-        default=0,
-        help="Default camera index when connect has no argument (ignored if --video is set)",
-    )
-    parser.add_argument(
-        "--video",
-        default=None,
-        help="Default video file for connect (offline debug without a camera)",
-    )
-    parser.add_argument(
-        "--max-camera-index",
-        type=int,
-        default=10,
-        help="Highest index probed by search",
-    )
-    parser.add_argument(
-        "--loop-video",
-        dest="loop_video",
-        action="store_true",
-        default=True,
-        help="Loop video file on EOF during run (default: on)",
-    )
-    parser.add_argument(
-        "--no-loop-video",
-        dest="loop_video",
-        action="store_false",
-        help="Stop run with OK eof when video file ends",
-    )
-    parser.add_argument(
-        "--show",
-        action="store_true",
-        help="Show UV unrolled OpenCV window (local debug only)",
-    )
-    parser.add_argument("--verbose", action="store_true", help="Echo UART traffic and SDK logs")
     args = parser.parse_args(argv)
-
+    if args.start and not args.setup:
-    if args.video:
+        parser.error("--start requires --setup")
-        video_path = Path(args.video)
+    if args.baud <= 0:
-        if not video_path.is_file():
+        parser.error("baud must be positive")
-            print(f"Video file not found: {args.video}", file=sys.stderr)
+    if any(not math.isfinite(t) or t <= 0 for t in (args.timeout, args.write_timeout)):
-            return 1
+        parser.error("timeouts must be finite and positive")
-        default_source: Source = str(video_path.resolve())
-    else:
-        default_source = args.camera_id
-
-    ser = open_serial(args.port, args.baud, args.timeout)
     controller = UartFingerController(
-        ser,
+        None, config=args.config, loop_video=args.loop_video,
-        config=args.config,
+        show=args.show, verbose=args.verbose,
-        default_source=default_source,
-        max_camera_index=args.max_camera_index,
-        loop_video=args.loop_video,
-        show=args.show,
-        verbose=args.verbose,
     )
-
+    ser = None
-    print(f"UART finger controller on {args.port} @ {args.baud} bps", flush=True)
+    result = 0
-    print(f"Default source: {default_source}", flush=True)
-    print("Commands: search | connect [source|file] | run | stop | quit", flush=True)
-
     try:
-        controller.serve()
+        if args.setup:
+            controller.bank.add_setup(args.setup)
+        ser = open_serial(args.port, args.baud, args.timeout, args.write_timeout)
+        controller.ser = ser
+        print(f"UART service on {args.port} @ {args.baud}; {len(controller.bank.runtimes)} sensors ready", file=sys.stderr, flush=True)
+        controller.serve(start=args.start)
     except KeyboardInterrupt:
-        print("\nInterrupted.", flush=True)
+        print("Interrupted.", file=sys.stderr)
-        try:
+    except Exception:
-            controller.reply("OK quit")
+        logger.exception("UART service failed on %s", args.port)
-        except Exception:  # noqa: BLE001
+        result = 1
-            pass
     finally:
-        controller.close_runtime()
+        try:
-        if ser.is_open:
+            controller.close_runtime()
-            ser.close()
+        except Exception:
-            print("Serial closed.", flush=True)
+            logger.exception("Runtime cleanup failed")
-    return 0
+            result = 1
+        finally:
+            if ser is not None:
+                ser.close()
+    return result
 
 
 if __name__ == "__main__":

serial/finger_api.py

@@ -0,0 +1,227 @@
+"""Single-threaded finger acquisition API, independent of UART and the CLI.
+
+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.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import math
+from dataclasses import dataclass
+from pathlib import Path
+
+Source = int | str
+logger = logging.getLogger(__name__)
+
+
+def parse_source(token: str) -> Source:
+    text = token.strip()
+    if not text:
+        raise ValueError("empty source")
+    return int(text) if text.isdigit() or (text.startswith("-") and text[1:].isdigit()) else text
+
+
+def is_video_file(source: Source) -> bool:
+    from orisys.capture import is_live_camera
+
+    return not is_live_camera(source) and not isinstance(source, int)
+
+
+@dataclass(frozen=True)
+class ForceReading:
+    sensor_id: int
+    normal: float
+    shear_x: float
+    shear_y: float
+    units: tuple[str, str, str]
+    image: object
+
+
+class FingerSensorBank:
+    """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, *, config="finger", verbose=False):
+        self.default_config = config
+        self.sources = {}
+        self.configs = {}
+        self.runtimes = {}
+        self.calibration_paths = {}
+        self.verbose = verbose
+
+    @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, cal_path: 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}")
+        if cal_path is not None:
+            if not str(cal_path).strip():
+                raise ValueError("empty calibration path")
+            calibration = Path(cal_path).expanduser().resolve()
+            if not calibration.is_file():
+                raise ValueError(f"calibration file not found: {calibration}")
+            cal_path = str(calibration)
+        from orisys.finger import FingerRuntime
+
+        config = self.default_config if config is None else config
+        runtime = None
+        try:
+            calibration_options = {"cal_path": cal_path} if cal_path is not None else {}
+            runtime = FingerRuntime.open(
+                config=config, video=source, verbose=self.verbose, **calibration_options,
+            )
+            ok, _ = runtime.warmup(color=False)
+            if not ok:
+                raise RuntimeError("warmup could not read a frame")
+        except BaseException:
+            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
+        self.calibration_paths[sensor_id] = cal_path
+        return sensor_id
+
+    def add_setup(self, path: str | Path) -> tuple[int, ...]:
+        """Add an explicit JSON setup; rollback only this batch on failure.
+
+        Relative file paths are based on the setup file, not the process cwd.
+        Schema/identity validation precedes opening any camera.
+        """
+        path = Path(path).expanduser().resolve()
+        document = json.loads(path.read_text(encoding="utf-8"))
+        if not isinstance(document, dict) or set(document) != {"sensors"}:
+            raise ValueError("setup must contain only a sensors list")
+        sensors = document["sensors"]
+        if not isinstance(sensors, list) or not 1 <= len(sensors) <= 4:
+            raise ValueError("setup must contain 1 to 4 sensors")
+        if len(self.runtimes) + len(sensors) > 4:
+            raise ValueError("maximum 4 sensors including those already added")
+        prepared = []
+        identities = {self._source_identity(source) for source in self.sources.values()}
+        for sensor in sensors:
+            if (not isinstance(sensor, dict) or "source" not in sensor
+                    or set(sensor) - {"source", "config", "calibration"}):
+                raise ValueError("sensor fields: source, optional config and calibration")
+            if type(sensor["source"]) not in (int, str):
+                raise ValueError("source must be a camera index or path")
+            source = parse_source(str(sensor["source"]))
+            if isinstance(source, int) and source < 0:
+                raise ValueError("camera index must be non-negative")
+            if is_video_file(source):
+                source = str((path.parent / Path(source).expanduser()).resolve())
+            identity = self._source_identity(source)
+            if identity in identities:
+                raise ValueError("source already added or repeated in setup")
+            identities.add(identity)
+            config = sensor.get("config")
+            calibration = sensor.get("calibration")
+            for name, value in (("config", config), ("calibration", calibration)):
+                if value is not None and (not isinstance(value, str) or not value.strip()):
+                    raise ValueError(f"{name} must be a non-empty string or null")
+            if config is not None and (Path(config).suffix.lower() == ".json" or "/" in config or "\\" in config):
+                config = str((path.parent / Path(config).expanduser()).resolve())
+            if calibration is not None:
+                calibration = str((path.parent / Path(calibration).expanduser()).resolve())
+            prepared.append({"source": source, "config": config, "cal_path": calibration})
+        added = []
+        try:
+            for sensor in prepared:
+                added.append(self.add(**sensor))
+        except BaseException:
+            for sensor_id in added:
+                try:
+                    self._close_one(sensor_id)
+                except Exception:
+                    logger.exception("Setup rollback failed for sensor %s", sensor_id)
+            raise
+        return tuple(added)
+
+    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 not self.runtimes:
+            raise RuntimeError("no sensors added")
+        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]):
+            import cv2
+
+            # One rewind attempt per read prevents an empty file from spinning forever.
+            if runtime.capture is not None and runtime.capture.set(cv2.CAP_PROP_POS_FRAMES, 0):
+                runtime.tactile.reset()
+                ok, image = runtime.grab_and_process(color=False)
+        if not ok:
+            return None
+        tactile = runtime.tactile
+        tactile.compute_deformation()
+        tactile.compute_contact()
+        forces = tuple(float(value) for value in tactile.read_info(
+            tactile.info.FNORMAL, tactile.info.FSHEARX, tactile.info.FSHEARY,
+        ))
+        if not all(math.isfinite(value) for value in forces):
+            raise RuntimeError("non-finite force measurement")
+        units = getattr(tactile, "force_units", ("unknown", "unknown", "unknown"))
+        return ForceReading(sensor_id, *forces, tuple(units), image)
+
+    def _close_one(self, sensor_id):
+        runtime = self.runtimes.pop(sensor_id)
+        self.sources.pop(sensor_id)
+        self.configs.pop(sensor_id)
+        self.calibration_paths.pop(sensor_id)
+        runtime.close()
+
+    def close(self) -> None:
+        first_error = None
+        for sensor_id in list(self.runtimes):
+            try:
+                self._close_one(sensor_id)
+            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

serial/sensors.example.json

@@ -0,0 +1,8 @@
+{
+  "sensors": [
+    {"source": "/dev/video0", "calibration": "../calibrations/finger/DD02-14.json"},
+    {"source": "/dev/video2", "calibration": "../calibrations/finger/SENSOR-2.json"},
+    {"source": "/dev/video4", "calibration": "../calibrations/finger/SENSOR-3.json"},
+    {"source": "/dev/video6", "calibration": "../calibrations/finger/SENSOR-4.json"}
+  ]
+}

tests/check_backend.py

@@ -0,0 +1,123 @@
+"""Exercise real SDK instances for 1-4 recorded inputs, without UART/hardware."""
+
+import argparse
+import itertools
+import json
+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)
+    parser.add_argument("--calibration", type=Path, help="explicit calibration used for all offline test sources")
+    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 = []
+    calibrations = []
+    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))
+        if args.calibration:
+            calibration = scratch / f"sensor-{sensor_id}-calibration.json"
+            calibration.write_bytes(args.calibration.read_bytes())
+            calibrations.append(str(calibration))
+
+    # 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()
+            try:
+                for index, source in enumerate(sources[:count]):
+                    existing = dict(bank.runtimes)
+                    bank.add(source, cal_path=calibrations[index] if calibrations else None)
+                    assert all(bank.runtimes[i] is runtime for i, runtime in existing.items())
+                    assert len(bank.read_all()) == len(bank.runtimes)
+                if calibrations:
+                    for sensor_id, runtime in bank.runtimes.items():
+                        calibration = runtime.tactile.raw_sensor.calibration
+                        assert calibration is not None
+                        assert calibration.source_path.resolve() == Path(calibrations[sensor_id - 1])
+                        assert runtime.tactile.force_units == ("N", "N", "N")
+                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: incremental add/read, independent states, 3 final rounds", flush=True)
+            finally:
+                bank.close()
+            assert all(runtime.capture is None for runtime in runtimes)
+        setup = scratch / "sensors.json"
+        setup.write_text(json.dumps({"sensors": [
+            {"source": source, **({"calibration": calibrations[i]} if calibrations else {})}
+            for i, source in enumerate(sources)
+        ]}), encoding="utf-8")
+        bank = FingerSensorBank()
+        try:
+            assert bank.add_setup(setup) == (1, 2, 3, 4)
+            assert len(bank.read_all()) == 4
+            bad_calibration = scratch / "invalid-calibration.json"
+            bad_calibration.write_text("{}", encoding="utf-8")
+        finally:
+            bank.close()
+        bank = FingerSensorBank()
+        try:
+            bank.add(sources[0])
+            first = bank.runtimes[1]
+            try:
+                bank.add(sources[1], cal_path=str(bad_calibration))
+            except ValueError:
+                pass
+            else:
+                raise AssertionError("invalid explicit calibration was accepted")
+            assert bank.runtimes == {1: first}
+            assert len(bank.read_all()) == 1
+        finally:
+            bank.close()
+        print("PASS 4-source JSON setup; invalid calibration rejected without fallback", flush=True)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_uart.py

@@ -0,0 +1,411 @@
+"""Hardware-free protocol and sensor lifecycle regression tests."""
+
+import io
+import sys
+import types
+import unittest
+from contextlib import redirect_stderr
+from pathlib import Path
+from unittest.mock import patch
+
+SERIAL_DIR = Path(__file__).resolve().parents[1] / "serial"
+sys.path.insert(0, str(SERIAL_DIR))
+import example_uart as uart
+from finger_api import FingerSensorBank
+
+
+class ByteSerial:
+    def __init__(self, data=b""):
+        self.input = bytearray(data)
+        self.output = []
+
+    @property
+    def in_waiting(self):
+        return len(self.input)
+
+    def read(self, size=1):
+        result = bytes(self.input[:size])
+        del self.input[:size]
+        return result
+
+    def readline(self):
+        end = self.input.find(b"\n")
+        return self.read(end + 1 if end >= 0 else len(self.input))
+
+    def write(self, data):
+        self.output.append(data.decode().rstrip("\n"))
+        return len(data)
+
+    def flush(self):
+        pass
+
+
+class FakeTactile:
+    info = types.SimpleNamespace(FNORMAL=1, FSHEARX=2, FSHEARY=3)
+    force_units = ("raw", "raw", "raw")
+
+    def compute_deformation(self):
+        pass
+
+    def compute_contact(self):
+        pass
+
+    def read_info(self, *args):
+        return 1.25, -2.5, 3.75
+
+    def reset(self):
+        pass
+
+
+class FakeRuntime:
+    def __init__(self, *, warmup_error=False, warmup_ok=True, grab_error=False):
+        self.tactile = FakeTactile()
+        self.closed = False
+        self.warmup_error = warmup_error
+        self.warmup_ok = warmup_ok
+        self.grab_error = grab_error
+        self.capture = types.SimpleNamespace(set=lambda *args: True)
+        self.frames = 1
+
+    def warmup(self, **kwargs):
+        if self.warmup_error:
+            raise RuntimeError("warmup failed")
+        return self.warmup_ok, None
+
+    def grab_and_process(self, **kwargs):
+        if self.grab_error:
+            raise RuntimeError("processing failed")
+        if self.frames:
+            self.frames -= 1
+            return True, None
+        return False, None
+
+    def close(self):
+        self.closed = True
+
+
+class HardwareFreeTest(unittest.TestCase):
+    def setUp(self):
+        stderr = redirect_stderr(io.StringIO())
+        stderr.__enter__()
+        self.addCleanup(stderr.__exit__, None, None, None)
+        finger = types.ModuleType("orisys.finger")
+        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.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.start()
+        self.addCleanup(modules.stop)
+
+
+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):
+                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, 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
+                    ):
+                        ser.input.extend(b"quit\n")
+                    return size
+
+                ser.write = write
+                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_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.assertEqual(bank.sources, {})
+
+    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_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)
+        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", cal_path=None)
+        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"))
+        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_processing_failure_stops_output_without_partial_round(self):
+        ser = ByteSerial()
+        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")
+        self.assertFalse(controller._streaming)
+        self.assertFalse(any(x.startswith("force,") for x in ser.output))
+        self.assertIn("sensor 2", ser.output[-1])
+
+
+class FramingTests(HardwareFreeTest):
+    def test_partial_line_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_lf_crlf_commands_are_separate(self):
+        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)
+        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_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 0")
+
+    def test_error_newlines_do_not_inject_records(self):
+        ser = ByteSerial()
+        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_nonfinite_force_is_not_a_valid_round(self):
+        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()
+        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()
+        bank.add(0)
+        bank.add(2)
+        second = bank.runtimes[2]
+        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_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_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
+
+        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))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_uart_setup.py

@@ -0,0 +1,216 @@
+"""Calibration binding and transactional setup regression tests."""
+
+import json
+import sys
+import uuid
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "serial"))
+import example_uart as uart
+from finger_api import FingerSensorBank
+from test_uart import ByteSerial, FakeRuntime, HardwareFreeTest
+
+
+class SetupTests(HardwareFreeTest):
+    def setUp(self):
+        super().setUp()
+        self.folder = (
+            Path(__file__).resolve().parents[1]
+            / ".scratch/uart-calibration-setup/tests"
+            / uuid.uuid4().hex
+        )
+        self.folder.mkdir(parents=True)
+        self.calibration = self.folder / "sensor calibration.json"
+        self.calibration.write_text("{}", encoding="utf-8")
+
+    def setup_file(self, sensors):
+        path = self.folder / "sensors.json"
+        path.write_text(json.dumps({"sensors": sensors}), encoding="utf-8")
+        return path
+
+    def test_explicit_calibration_is_bound_to_correct_runtime(self):
+        bank = FingerSensorBank()
+        with patch(
+            "orisys.finger.FingerRuntime.open", return_value=FakeRuntime()
+        ) as factory:
+            bank.add(0, cal_path=str(self.calibration))
+        self.assertEqual(
+            factory.call_args.kwargs["cal_path"], str(self.calibration.resolve())
+        )
+        self.assertEqual(bank.calibration_paths[1], str(self.calibration.resolve()))
+
+    def test_missing_and_empty_calibration_do_not_fall_back(self):
+        for calibration in ("", str(self.folder / "missing.json")):
+            with (
+                self.subTest(calibration=calibration),
+                patch("orisys.finger.FingerRuntime.open") as factory,
+            ):
+                with self.assertRaises(ValueError):
+                    FingerSensorBank().add(0, cal_path=calibration)
+                factory.assert_not_called()
+
+    def test_uart_add_accepts_calibration_and_optional_config(self):
+        controller = uart.UartFingerController(ByteSerial())
+        command = f'add 0 custom.json --calibration "{self.calibration}"'
+        with patch(
+            "orisys.finger.FingerRuntime.open", return_value=FakeRuntime()
+        ) as factory:
+            controller._dispatch(command)
+        self.assertEqual(controller.ser.output, ["OK added 1"])
+        self.assertEqual(factory.call_args.kwargs["config"], "custom.json")
+        self.assertEqual(
+            factory.call_args.kwargs["cal_path"], str(self.calibration.resolve())
+        )
+
+    def test_setup_paths_are_relative_to_setup_file(self):
+        clip = self.folder / "recording.avi"
+        clip.touch()
+        config = self.folder / "sensor.json"
+        config.write_text("{}", encoding="utf-8")
+        setup = self.setup_file(
+            [
+                {
+                    "source": "recording.avi",
+                    "config": "sensor.json",
+                    "calibration": self.calibration.name,
+                },
+                {"source": 2},
+            ]
+        )
+        seen = []
+
+        def open_runtime(**kwargs):
+            seen.append(kwargs)
+            return FakeRuntime()
+
+        bank = FingerSensorBank()
+        with patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime):
+            self.assertEqual(bank.add_setup(setup), (1, 2))
+        self.assertEqual(seen[0]["video"], str(clip.resolve()))
+        self.assertEqual(seen[0]["config"], str(config.resolve()))
+        self.assertEqual(seen[0]["cal_path"], str(self.calibration.resolve()))
+        self.assertNotIn("cal_path", seen[1])
+
+    def test_distinct_calibration_files_are_selected_in_source_order(self):
+        second = self.folder / "second.json"
+        second.write_text("{}", encoding="utf-8")
+        setup = self.setup_file(
+            [
+                {"source": 0, "calibration": self.calibration.name},
+                {"source": 2, "calibration": second.name},
+            ]
+        )
+        seen = []
+
+        def open_runtime(**kwargs):
+            seen.append(kwargs["cal_path"])
+            return FakeRuntime()
+
+        with patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime):
+            FingerSensorBank().add_setup(setup)
+        self.assertEqual(seen, [str(self.calibration.resolve()), str(second.resolve())])
+
+    def test_batch_failure_rolls_back_only_new_sensors(self):
+        bank = FingerSensorBank()
+        bank.add(0)
+        original = bank.runtimes[1]
+        opened = []
+
+        def open_runtime(**kwargs):
+            runtime = FakeRuntime(warmup_error=kwargs["video"] == 4)
+            opened.append(runtime)
+            return runtime
+
+        setup = self.setup_file([{"source": 2}, {"source": 4}])
+        with (
+            patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime),
+            self.assertRaises(RuntimeError),
+        ):
+            bank.add_setup(setup)
+        self.assertEqual(bank.sources, {1: 0})
+        self.assertEqual(bank.calibration_paths, {1: None})
+        self.assertIs(bank.runtimes[1], original)
+        self.assertFalse(original.closed)
+        self.assertTrue(all(runtime.closed for runtime in opened))
+        self.assertEqual(bank.add(6), 2)
+
+    def test_setup_rejects_typos_bad_types_duplicates_and_excess_count_before_open(
+        self,
+    ):
+        for sensors in (
+            [],
+            [{"source": True}],
+            [{"source": 0, "calibration_file": "oops"}],
+            [{"source": 0}, {"source": "/dev/video0"}],
+            [{"source": i} for i in range(5)],
+            [{"source": 0, "calibration": 42}],
+            [{"source": 0, "config": ""}],
+        ):
+            with (
+                self.subTest(sensors=sensors),
+                patch("orisys.finger.FingerRuntime.open") as factory,
+            ):
+                with self.assertRaises(ValueError):
+                    FingerSensorBank().add_setup(self.setup_file(sensors))
+                factory.assert_not_called()
+
+    def test_uart_load_can_extend_manual_setup_but_requires_stop(self):
+        setup = self.setup_file([{"source": 2}, {"source": 4}, {"source": 6}])
+        controller = uart.UartFingerController(ByteSerial())
+        controller._dispatch("add 0")
+        controller._dispatch("start")
+        controller._dispatch(f'load "{setup}"')
+        self.assertEqual(controller.ser.output[-1], "ERR load stop streaming first")
+        controller._dispatch("stop")
+        controller._dispatch(f'load "{setup}"')
+        self.assertEqual(controller.ser.output[-1], "OK loaded 2 3 4")
+        self.assertFalse(controller._streaming)
+
+    def test_one_cli_command_loads_four_sensors_and_starts_output(self):
+        setup = self.setup_file(
+            [{"source": i * 2, "calibration": self.calibration.name} for i in range(4)]
+        )
+        ser = ByteSerial()
+        ser.closed = False
+        ser.close = lambda: setattr(ser, "closed", True)
+        original_write = ser.write
+
+        def write(data):
+            size = original_write(data)
+            if (
+                data.startswith(b"force,")
+                and sum(line.startswith("force,") for line in ser.output) == 4
+            ):
+                ser.input.extend(b"quit\n")
+            return size
+
+        ser.write = write
+        with patch.object(uart, "open_serial", return_value=ser):
+            self.assertEqual(uart.main(["--setup", str(setup), "--start"]), 0)
+        self.assertEqual(ser.output[0], "OK ready 4")
+        self.assertEqual(sum(line.startswith("force,") for line in ser.output), 4)
+        self.assertTrue(ser.closed)
+
+    def test_start_option_requires_setup(self):
+        with (
+            patch.object(uart, "open_serial") as serial,
+            self.assertRaises(SystemExit) as status,
+        ):
+            uart.main(["--start"])
+        self.assertEqual(status.exception.code, 2)
+        serial.assert_not_called()
+
+    def test_setup_without_start_waits_for_uart_commands(self):
+        setup = self.setup_file([{"source": 0}, {"source": 2}])
+        ser = ByteSerial(b"quit\n")
+        ser.close = lambda: None
+        with patch.object(uart, "open_serial", return_value=ser):
+            self.assertEqual(uart.main(["--setup", str(setup)]), 0)
+        self.assertEqual(ser.output, ["OK ready 2", "OK quit"])
+
+    def test_bad_startup_setup_never_opens_uart(self):
+        setup = self.setup_file([{"source": 0, "calibration": "missing.json"}])
+        with patch.object(uart, "open_serial") as serial:
+            self.assertEqual(uart.main(["--setup", str(setup)]), 1)
+        serial.assert_not_called()