diff --git a/docs/uart.md b/docs/uart.md index c1101fb..805d851 100644 --- a/docs/uart.md +++ b/docs/uart.md @@ -1,8 +1,8 @@ # UART finger service -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 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 @@ -40,17 +40,80 @@ 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]` | Open and prepare one source while output is stopped; reply `OK added ID`. Optional config overrides the startup `--config` for this sensor only. | +| `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. | @@ -70,8 +133,7 @@ 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. +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 @@ -109,7 +171,7 @@ from finger_api import FingerSensorBank bank = FingerSensorBank(config="finger") try: - first_id = bank.add(0) + 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) @@ -125,6 +187,8 @@ sources succeed. File EOF raises `EOFError`; capture/processing failure raises 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 @@ -135,12 +199,16 @@ 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 --scratch .scratch/uart-manual-add/backend-run +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 ``` diff --git a/serial/example_uart.py b/serial/example_uart.py index f4aa781..4ab79dd 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. Cameras are added -manually via UART before starting output. All force records include a 1-based sensor ID. +The login console and data UART must be separate ports. Cameras are selected +explicitly by UART commands or a startup setup file. All force records include a sensor ID. Importing this module opens no devices; acquisition is owned by finger_api. """ @@ -137,20 +137,40 @@ class UartFingerController: return True command = parts[0].lower() if command == "add": - if len(parts) not in (2, 3): - self.reply("ERR command expected add SOURCE [CONFIG]") - elif self._streaming: + if self._streaming: self.reply("ERR add stop streaming first") else: + parser = argparse.ArgumentParser(prog="add", add_help=False, exit_on_error=False) + parser.add_argument("source") + parser.add_argument("config", nargs="?") + parser.add_argument("--calibration") + try: + args = parser.parse_args(parts[1:]) + except (argparse.ArgumentError, SystemExit): + self.reply("ERR command expected add SOURCE [CONFIG] [--calibration PATH]") + return True try: sensor_id = self.bank.add( - parse_source(parts[1]), config=parts[2] if len(parts) == 3 else None, + 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: @@ -172,9 +192,11 @@ class UartFingerController: self.reply("OK quit") return not self._should_quit - def serve(self): + def serve(self, *, start=False): + if start and not self.bank.runtimes: + raise RuntimeError("no sensors added") self.reply(f"OK ready {len(self.bank.sources)}") - self._streaming = False + self._streaming = start while not self._should_quit: for _ in range(16): line = self._poll_command_line(blocking=not self._streaming) @@ -204,10 +226,14 @@ def main(argv=None): parser.add_argument("--timeout", type=float, default=0.1, help="idle read timeout in seconds") parser.add_argument("--write-timeout", type=float, default=1.0) parser.add_argument("--config", "-c", default="finger", help="default config for add commands") + parser.add_argument("--setup", help="JSON file listing sources, configs and calibrations to connect") + parser.add_argument("--start", action="store_true", help="start output after loading --setup") 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") args = parser.parse_args(argv) + if args.start and not args.setup: + parser.error("--start requires --setup") 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)): @@ -219,10 +245,12 @@ def main(argv=None): ser = None result = 0 try: + 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}; waiting for add SOURCE, then start", file=sys.stderr, flush=True) - controller.serve() + 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("Interrupted.", file=sys.stderr) except Exception: diff --git a/serial/finger_api.py b/serial/finger_api.py index 3d8276f..8a1ec63 100644 --- a/serial/finger_api.py +++ b/serial/finger_api.py @@ -7,6 +7,7 @@ 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 @@ -51,6 +52,7 @@ class FingerSensorBank: self.sources = {} self.configs = {} self.runtimes = {} + self.calibration_paths = {} self.verbose = verbose @staticmethod @@ -61,7 +63,7 @@ class FingerSensorBank: return f"camera:{int(source[len('/dev/video'):])}" return str(Path(source).resolve()) - def add(self, source: Source, *, config: str | None = None) -> int: + 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: @@ -73,12 +75,22 @@ class FingerSensorBank: 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: - runtime = FingerRuntime.open(config=config, video=source, verbose=self.verbose) + 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") @@ -93,8 +105,64 @@ class FingerSensorBank: 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. @@ -138,11 +206,18 @@ class FingerSensorBank: 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.runtimes.pop(sensor_id).close() + self._close_one(sensor_id) except Exception as exc: logger.exception("Sensor %s cleanup failed", sensor_id) first_error = first_error or exc diff --git a/serial/sensors.example.json b/serial/sensors.example.json new file mode 100644 index 0000000..d2ed05a --- /dev/null +++ b/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"} + ] +} diff --git a/tests/check_backend.py b/tests/check_backend.py index 9e764b6..106d380 100644 --- a/tests/check_backend.py +++ b/tests/check_backend.py @@ -2,6 +2,7 @@ import argparse import itertools +import json import os import sys from functools import partial @@ -16,6 +17,7 @@ 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) @@ -39,6 +41,7 @@ def main(): 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)) @@ -50,6 +53,10 @@ def main(): 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")) @@ -57,11 +64,17 @@ def main(): for count in range(1, 5): bank = FingerSensorBank() try: - for source in sources[:count]: + for index, source in enumerate(sources[:count]): existing = dict(bank.runtimes) - bank.add(source) + 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 @@ -75,6 +88,34 @@ def main(): 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 diff --git a/tests/test_uart.py b/tests/test_uart.py index f44df11..b7d5b29 100644 --- a/tests/test_uart.py +++ b/tests/test_uart.py @@ -252,7 +252,7 @@ class ManualAddTests(HardwareFreeTest): 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") + 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): diff --git a/tests/test_uart_setup.py b/tests/test_uart_setup.py new file mode 100644 index 0000000..150437a --- /dev/null +++ b/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()