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

保留逐台添加,并通过显式配置一次连接全组;失败时撤销本批新增连接。通过 35 项回归测试、真实 SDK 校准与 1–4 路验证、Ruff 和文档检查。
This commit is contained in:
ykcheungab
2026-10-14 16:11:15 +08:00
parent 133a98e418
commit a723e96dcc
7 changed files with 460 additions and 24 deletions
+76 -8
View File
@@ -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
```
+38 -10
View File
@@ -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:
+78 -3
View File
@@ -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
+8
View File
@@ -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"}
]
}
+43 -2
View File
@@ -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
+1 -1
View File
@@ -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):
+216
View File
@@ -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()