Compare commits
5 Commits
14d5b1b086
..
main
| Author | SHA1 | Date | |
|---|---|---|---|
 zhangyijian
|
653ee60ef7 | ||
 ykcheungab
|
a723e96dcc | ||
|
ykcheungab
|
133a98e418 | ||
|
ykcheungab
|
07339c3374 | ||
|
ykcheungab
|
c970272361 |
@@ -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.
|
||||
|
||||
+222
@@ -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.
|
||||
+358
-363
@@ -1,460 +1,455 @@
|
||||
#!/usr/bin/env python3
|
||||
"""UART finger controller for Radxa / Dragon Q6A (sensor-side receiver).
|
||||
"""Headless UART finger service for 1-4 sensors.
|
||||
|
||||
Combines ORY-FINGER processing (same path as ``examples/finger_pipeline.py``)
|
||||
with a line-based UART command interface. A host (or ``uart_send.py``-style
|
||||
transmitter) sends commands; this script replies and streams forces.
|
||||
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.
|
||||
|
||||
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
|
||||
Commands (one line, case-insensitive, CR or LF terminated):
|
||||
add SOURCE [CONFIG] [--calibration PATH]
|
||||
load SETUP.json
|
||||
remove SENSOR_ID
|
||||
start | stop | read | quit | exit
|
||||
list | status | help
|
||||
|
||||
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
|
||||
OK ready <n>
|
||||
OK added <id> <kind> <source>
|
||||
OK removed <id>
|
||||
OK loaded <id> [<id> ...]
|
||||
OK start | OK stop | OK quit
|
||||
OK list <n>
|
||||
sensor,<id>,<kind>,<source>,<config>,<calibration>
|
||||
OK status sensors=<n> streaming=<0|1> loop_video=<0|1>
|
||||
force,<id>,<fnormal>,<fshearx>,<fsheary>
|
||||
OK eof <context>
|
||||
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 difflib
|
||||
import logging
|
||||
import math
|
||||
import os
|
||||
import shlex
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Optional, Union
|
||||
import time
|
||||
from collections import deque
|
||||
|
||||
os.environ.setdefault("OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS", "0")
|
||||
|
||||
try:
|
||||
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
|
||||
from finger_api import FingerSensorBank, is_video_file, parse_source
|
||||
|
||||
import cv2
|
||||
from orisys.capture import is_live_camera, list_cameras
|
||||
from orisys.finger import FingerRuntime
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
Source = Union[int, str]
|
||||
_KNOWN_COMMANDS = (
|
||||
"add", "load", "remove", "start", "stop", "read", "quit", "exit",
|
||||
"list", "status", "help",
|
||||
)
|
||||
|
||||
|
||||
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",
|
||||
}
|
||||
_HELP_LINES = (
|
||||
"add SOURCE [CONFIG] [--calibration PATH]",
|
||||
"load SETUP.json",
|
||||
"remove SENSOR_ID",
|
||||
"start | stop | read | quit",
|
||||
"list | status | help",
|
||||
)
|
||||
|
||||
|
||||
class UartFingerController:
|
||||
"""Serial command dispatcher around a single FingerRuntime."""
|
||||
"""Single-threaded command service; the caller owns serial and final cleanup."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
ser: serial.Serial,
|
||||
*,
|
||||
config: str = "finger",
|
||||
default_source: Source = 0,
|
||||
max_camera_index: int = 10,
|
||||
loop_video: bool = True,
|
||||
show: bool = False,
|
||||
verbose: bool = False,
|
||||
self, ser, *, config="finger", loop_video=True,
|
||||
show=False, verbose=False, max_hz=0.0, prompt=False,
|
||||
):
|
||||
self.ser = ser
|
||||
self.config = config
|
||||
self.default_source = default_source
|
||||
self.max_camera_index = max_camera_index
|
||||
self.bank = FingerSensorBank(config=config, verbose=verbose)
|
||||
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.max_hz = max_hz
|
||||
self.prompt = prompt
|
||||
self._streaming = False
|
||||
self._should_quit = False
|
||||
self._partial = bytearray()
|
||||
self._lines = deque()
|
||||
self._discard_line = False
|
||||
|
||||
def reply(self, line: str) -> None:
|
||||
text = line.rstrip("\r\n") + "\n"
|
||||
self.ser.write(text.encode("utf-8"))
|
||||
# ------------------------------------------------------------------ io
|
||||
|
||||
def reply(self, line):
|
||||
# SDK exceptions can contain newlines; every reply must be one record.
|
||||
text = " ".join(str(line).splitlines()) + "\n"
|
||||
data = text.encode("utf-8")
|
||||
if self.ser.write(data) != len(data):
|
||||
raise OSError("incomplete UART write")
|
||||
self.ser.flush()
|
||||
if self.verbose:
|
||||
print(f"<< {text.rstrip()}", flush=True)
|
||||
print(f"<< {text.rstrip()}", file=sys.stderr, flush=True)
|
||||
|
||||
def close_runtime(self) -> None:
|
||||
def _maybe_prompt(self):
|
||||
if self.prompt and not self._streaming:
|
||||
self.reply(">")
|
||||
|
||||
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
|
||||
self.bank.close()
|
||||
finally:
|
||||
if self.show:
|
||||
try:
|
||||
import cv2
|
||||
|
||||
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)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
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():
|
||||
def _poll_command_line(self, *, blocking=True):
|
||||
if self._lines:
|
||||
line = self._lines.popleft()
|
||||
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
|
||||
print(f">> {line}", file=sys.stderr, flush=True)
|
||||
return line
|
||||
waiting = self.ser.in_waiting
|
||||
if not waiting and not blocking:
|
||||
return None
|
||||
# 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
|
||||
if self._lines:
|
||||
line = self._lines.popleft()
|
||||
if self.verbose:
|
||||
print(f">> {line}", file=sys.stderr, flush=True)
|
||||
return line
|
||||
return None
|
||||
|
||||
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}")
|
||||
# ------------------------------------------------------------- streaming
|
||||
|
||||
if self.show:
|
||||
def _read_round(self):
|
||||
try:
|
||||
readings = self.bank.read_all(loop_video=self.loop_video)
|
||||
except EOFError as exc:
|
||||
self._streaming = False
|
||||
self.reply(f"OK eof {exc}")
|
||||
self._maybe_prompt()
|
||||
return
|
||||
except Exception as exc:
|
||||
logger.exception("Acquisition round failed")
|
||||
self._streaming = False
|
||||
self.reply(f"ERR read {exc}")
|
||||
self._maybe_prompt()
|
||||
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 _show(self, reading):
|
||||
import cv2
|
||||
import orisys
|
||||
|
||||
tactile = self.bank.runtimes[reading.sensor_id].tactile
|
||||
try:
|
||||
arrows = orisys.util.draw_arrows(
|
||||
unrolled,
|
||||
tactile.flow,
|
||||
threshold=0.4,
|
||||
grid_spacing=10,
|
||||
arrow_scale=1.0,
|
||||
valid_mask=tactile.valid_mask,
|
||||
reading.image, tactile.flow, threshold=0.4, grid_spacing=10,
|
||||
arrow_scale=1.0, valid_mask=tactile.valid_mask,
|
||||
)
|
||||
cv2.imshow("UV Unrolled", arrows)
|
||||
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
|
||||
break
|
||||
if key == ord("r"):
|
||||
self.reply("OK quit")
|
||||
elif key == ord("r"):
|
||||
tactile.reset()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
if self.verbose:
|
||||
print(f"show error: {exc}", file=sys.stderr)
|
||||
except Exception:
|
||||
logger.exception("Sensor %s display failed", reading.sensor_id)
|
||||
|
||||
# Non-blocking command poll so force streaming is not delayed by UART timeout.
|
||||
pending = self._poll_command_line(blocking=False)
|
||||
if pending is not None:
|
||||
if self.verbose:
|
||||
print(f">> {pending}", flush=True)
|
||||
if not self._dispatch(pending, allow_run=False):
|
||||
break
|
||||
# ------------------------------------------------------------- dispatch
|
||||
|
||||
def cmd_stop(self, _args: list[str]) -> None:
|
||||
def _reply_hint(self, command: str):
|
||||
close = difflib.get_close_matches(command, _KNOWN_COMMANDS, n=1)
|
||||
suffix = f" (did you mean '{close[0]}'?)" if close else ""
|
||||
self.reply(f"ERR unknown {command}{suffix}")
|
||||
|
||||
def _cmd_add(self, parts):
|
||||
if self._streaming:
|
||||
self.reply("ERR add stop streaming first")
|
||||
return
|
||||
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)
|
||||
except (argparse.ArgumentError, SystemExit):
|
||||
self.reply("ERR command expected add SOURCE [CONFIG] [--calibration PATH]")
|
||||
return
|
||||
try:
|
||||
sensor_id = self.bank.add(
|
||||
parse_source(args.source), config=args.config, cal_path=args.calibration,
|
||||
)
|
||||
except FileNotFoundError:
|
||||
self.reply(f"ERR add source not found: {args.source}")
|
||||
except ValueError as exc:
|
||||
self.reply(f"ERR add {exc}")
|
||||
except Exception as exc:
|
||||
logger.exception("Sensor addition failed")
|
||||
self.reply(
|
||||
f"ERR add failed for {args.source}: {type(exc).__name__}: {exc}"
|
||||
)
|
||||
else:
|
||||
src = self.bank.sources[sensor_id]
|
||||
kind = "file" if is_video_file(src) else "camera"
|
||||
self.reply(f"OK added {sensor_id} {kind} {src}")
|
||||
|
||||
def _cmd_load(self, parts):
|
||||
if len(parts) != 1:
|
||||
self.reply("ERR command expected load SETUP.json")
|
||||
return
|
||||
if self._streaming:
|
||||
self.reply("ERR load stop streaming first")
|
||||
return
|
||||
try:
|
||||
sensor_ids = self.bank.add_setup(parts[0])
|
||||
except FileNotFoundError:
|
||||
self.reply(f"ERR load setup not found: {parts[0]}")
|
||||
except ValueError as exc:
|
||||
self.reply(f"ERR load {exc}")
|
||||
except Exception as exc:
|
||||
logger.exception("Setup loading failed")
|
||||
self.reply(f"ERR load failed: {type(exc).__name__}: {exc}")
|
||||
else:
|
||||
self.reply("OK loaded " + " ".join(map(str, sensor_ids)))
|
||||
|
||||
def _cmd_remove(self, parts):
|
||||
if self._streaming:
|
||||
self.reply("ERR remove stop streaming first")
|
||||
return
|
||||
if len(parts) != 1 or not parts[0].isdigit():
|
||||
self.reply("ERR command expected remove SENSOR_ID")
|
||||
return
|
||||
sensor_id = int(parts[0])
|
||||
try:
|
||||
self.bank.remove_sensor(sensor_id)
|
||||
except ValueError as exc:
|
||||
self.reply(f"ERR remove {exc}")
|
||||
except Exception as exc:
|
||||
logger.exception("Sensor removal failed")
|
||||
self.reply(f"ERR remove failed: {type(exc).__name__}: {exc}")
|
||||
else:
|
||||
self.reply(f"OK removed {sensor_id}")
|
||||
|
||||
def _cmd_list(self):
|
||||
for info in self.bank.list_sensors():
|
||||
cal = info.calibration or "-"
|
||||
self.reply(
|
||||
f"sensor,{info.sensor_id},{info.kind},{info.source},{info.config},{cal}"
|
||||
)
|
||||
self.reply(f"OK list {len(self.bank.sources)}")
|
||||
|
||||
def _cmd_status(self):
|
||||
self.reply(
|
||||
f"OK status sensors={len(self.bank.sources)} "
|
||||
f"streaming={int(self._streaming)} "
|
||||
f"loop_video={int(self.loop_video)}"
|
||||
)
|
||||
|
||||
def _cmd_help(self):
|
||||
self.reply("OK help " + " | ".join(_HELP_LINES))
|
||||
|
||||
def _cmd_start(self, parts):
|
||||
if parts:
|
||||
self.reply("ERR command start takes no arguments")
|
||||
return
|
||||
if not self.bank.runtimes:
|
||||
self.reply("ERR start no sensors added")
|
||||
return
|
||||
self._streaming = True
|
||||
self.reply("OK start")
|
||||
|
||||
def _cmd_stop(self, parts):
|
||||
if parts:
|
||||
self.reply("ERR command stop takes no arguments")
|
||||
return
|
||||
self._streaming = False
|
||||
self.reply("OK stop")
|
||||
self._maybe_prompt()
|
||||
|
||||
def cmd_quit(self, _args: list[str]) -> None:
|
||||
def _cmd_read(self, parts):
|
||||
if parts:
|
||||
self.reply("ERR command read takes no arguments")
|
||||
return
|
||||
if not self.bank.runtimes:
|
||||
self.reply("ERR read no sensors added")
|
||||
return
|
||||
self._read_round()
|
||||
|
||||
def _cmd_quit(self, parts):
|
||||
if parts:
|
||||
self.reply("ERR command quit takes no arguments")
|
||||
return
|
||||
self._streaming = False
|
||||
self._should_quit = True
|
||||
self.reply("OK quit")
|
||||
|
||||
def _dispatch(self, line: str, *, allow_run: bool = True) -> bool:
|
||||
"""Dispatch one command line. Return False if the main loop should exit."""
|
||||
def _dispatch(self, line):
|
||||
try:
|
||||
parts = shlex.split(line)
|
||||
except ValueError:
|
||||
parts = line.split()
|
||||
if not parts:
|
||||
except ValueError as exc:
|
||||
self.reply(f"ERR command {exc}")
|
||||
self._maybe_prompt()
|
||||
return True
|
||||
cmd = parts[0].lower()
|
||||
if not parts:
|
||||
self._maybe_prompt()
|
||||
return True
|
||||
command = parts[0].lower()
|
||||
args = parts[1:]
|
||||
|
||||
if cmd == "search":
|
||||
self.cmd_search(args)
|
||||
elif cmd == "connect":
|
||||
self.cmd_connect(args)
|
||||
elif cmd == "run":
|
||||
if not allow_run:
|
||||
self.reply("ERR run already running")
|
||||
if command == "add":
|
||||
self._cmd_add(args)
|
||||
elif command == "load":
|
||||
self._cmd_load(args)
|
||||
elif command == "remove":
|
||||
self._cmd_remove(args)
|
||||
elif command == "start":
|
||||
self._cmd_start(args)
|
||||
elif command == "stop":
|
||||
self._cmd_stop(args)
|
||||
elif command == "read":
|
||||
self._cmd_read(args)
|
||||
elif command in ("quit", "exit"):
|
||||
self._cmd_quit(args)
|
||||
elif command == "list":
|
||||
self._cmd_list()
|
||||
elif command == "status":
|
||||
self._cmd_status()
|
||||
elif command in ("help", "?"):
|
||||
self._cmd_help()
|
||||
else:
|
||||
self.cmd_run(args)
|
||||
elif cmd == "stop":
|
||||
self.cmd_stop(args)
|
||||
elif cmd in ("quit", "exit"):
|
||||
self.cmd_quit(args)
|
||||
return False
|
||||
else:
|
||||
self.reply(f"ERR unknown {cmd}")
|
||||
self._reply_hint(command)
|
||||
|
||||
self._maybe_prompt()
|
||||
return not self._should_quit
|
||||
|
||||
def serve(self) -> None:
|
||||
self.reply("OK ready")
|
||||
# ---------------------------------------------------------------- serve
|
||||
|
||||
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._maybe_prompt()
|
||||
self._streaming = start
|
||||
while not self._should_quit:
|
||||
line = self._poll_command_line()
|
||||
for _ in range(16):
|
||||
line = self._poll_command_line(blocking=not self._streaming)
|
||||
if line is None:
|
||||
continue
|
||||
if self.verbose:
|
||||
print(f">> {line}", flush=True)
|
||||
if not self._dispatch(line):
|
||||
break
|
||||
if not self._dispatch(line):
|
||||
return
|
||||
if self._streaming:
|
||||
t0 = time.monotonic()
|
||||
self._read_round()
|
||||
if self.max_hz > 0 and self._streaming:
|
||||
budget = 1.0 / self.max_hz - (time.monotonic() - t0)
|
||||
if budget > 0:
|
||||
time.sleep(budget)
|
||||
|
||||
|
||||
def open_serial(port: str, baud: int, timeout: float) -> serial.Serial:
|
||||
def open_serial(port, baud, timeout, write_timeout=1.0):
|
||||
try:
|
||||
import serial
|
||||
except ImportError as exc:
|
||||
raise RuntimeError("pyserial is required: python -m pip install pyserial") from exc
|
||||
return serial.Serial(
|
||||
port=port,
|
||||
baudrate=baud,
|
||||
bytesize=serial.EIGHTBITS,
|
||||
parity=serial.PARITY_NONE,
|
||||
stopbits=serial.STOPBITS_ONE,
|
||||
timeout=timeout,
|
||||
port=port, baudrate=baud, bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE,
|
||||
stopbits=serial.STOPBITS_ONE, timeout=timeout, write_timeout=write_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(
|
||||
description="UART finger controller (search/connect/run/stop force stream)"
|
||||
description="Manual UART finger service: add, load, remove, start, stop, read, list, status, quit"
|
||||
)
|
||||
parser.add_argument("--port", default="/dev/ttyHS1", help="Serial device (default: /dev/ttyHS1)")
|
||||
parser.add_argument("--baud", type=int, default=115200, help="Baud rate (default: 115200)")
|
||||
parser.add_argument("--port", default="/dev/ttyHS1")
|
||||
parser.add_argument("--baud", type=int, default=115200)
|
||||
parser.add_argument("--timeout", type=float, default=0.1, help="idle read timeout in seconds")
|
||||
parser.add_argument("--write-timeout", type=float, default=1.0)
|
||||
parser.add_argument("--config", "-c", 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")
|
||||
parser.add_argument(
|
||||
"--timeout",
|
||||
type=float,
|
||||
default=0.1,
|
||||
help="Serial readline timeout seconds (default: 0.1)",
|
||||
)
|
||||
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)",
|
||||
"--max-hz", type=float, default=0.0,
|
||||
help="cap force rounds per second while streaming (0 = unlimited)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--video",
|
||||
default=None,
|
||||
help="Default video file for connect (offline debug without a camera)",
|
||||
"--prompt", action="store_true",
|
||||
help="emit a '>' line before each idle command wait (off by default)",
|
||||
)
|
||||
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.video:
|
||||
video_path = Path(args.video)
|
||||
if not video_path.is_file():
|
||||
print(f"Video file not found: {args.video}", file=sys.stderr)
|
||||
return 1
|
||||
default_source: Source = str(video_path.resolve())
|
||||
else:
|
||||
default_source = args.camera_id
|
||||
|
||||
ser = open_serial(args.port, args.baud, args.timeout)
|
||||
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)):
|
||||
parser.error("timeouts must be finite and positive")
|
||||
if not math.isfinite(args.max_hz) or args.max_hz < 0:
|
||||
parser.error("--max-hz must be finite and non-negative")
|
||||
controller = UartFingerController(
|
||||
ser,
|
||||
config=args.config,
|
||||
default_source=default_source,
|
||||
max_camera_index=args.max_camera_index,
|
||||
loop_video=args.loop_video,
|
||||
show=args.show,
|
||||
verbose=args.verbose,
|
||||
None, config=args.config, loop_video=args.loop_video,
|
||||
show=args.show, verbose=args.verbose, max_hz=args.max_hz, prompt=args.prompt,
|
||||
)
|
||||
|
||||
print(f"UART finger controller on {args.port} @ {args.baud} bps", flush=True)
|
||||
print(f"Default source: {default_source}", flush=True)
|
||||
print("Commands: search | connect [source|file] | run | stop | quit", flush=True)
|
||||
|
||||
ser = None
|
||||
result = 0
|
||||
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}; "
|
||||
f"{len(controller.bank.runtimes)} sensors ready",
|
||||
file=sys.stderr, flush=True,
|
||||
)
|
||||
controller.serve(start=args.start)
|
||||
except KeyboardInterrupt:
|
||||
print("\nInterrupted.", flush=True)
|
||||
try:
|
||||
controller.reply("OK quit")
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
print("Interrupted.", file=sys.stderr)
|
||||
except Exception:
|
||||
logger.exception("UART service failed on %s", args.port)
|
||||
result = 1
|
||||
finally:
|
||||
try:
|
||||
controller.close_runtime()
|
||||
if ser.is_open:
|
||||
except Exception:
|
||||
logger.exception("Runtime cleanup failed")
|
||||
result = 1
|
||||
finally:
|
||||
if ser is not None:
|
||||
ser.close()
|
||||
print("Serial closed.", flush=True)
|
||||
return 0
|
||||
return result
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -0,0 +1,258 @@
|
||||
"""Single-threaded finger acquisition API, independent of UART and the CLI.
|
||||
|
||||
Sensor IDs are 1-based, assigned once per successful add() and never reused
|
||||
within a process, so a host that caches IDs stays correct across removals.
|
||||
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
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SensorInfo:
|
||||
sensor_id: int
|
||||
source: Source
|
||||
kind: str # "camera" or "file"
|
||||
config: str
|
||||
calibration: str | None
|
||||
|
||||
|
||||
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: dict[int, Source] = {}
|
||||
self.configs: dict[int, str] = {}
|
||||
self.runtimes: dict[int, object] = {}
|
||||
self.calibration_paths: dict[int, str | None] = {}
|
||||
self.verbose = verbose
|
||||
self._next_id = 1
|
||||
|
||||
@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 = self._next_id
|
||||
self._next_id += 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.remove_sensor(sensor_id)
|
||||
except Exception:
|
||||
logger.exception("Setup rollback failed for sensor %s", sensor_id)
|
||||
raise
|
||||
return tuple(added)
|
||||
|
||||
def remove_sensor(self, sensor_id: int) -> None:
|
||||
"""Close and forget one sensor. Its ID is never reused."""
|
||||
if sensor_id not in self.runtimes:
|
||||
raise ValueError(f"no sensor {sensor_id}")
|
||||
runtime = self.runtimes.pop(sensor_id)
|
||||
self.sources.pop(sensor_id, None)
|
||||
self.configs.pop(sensor_id, None)
|
||||
self.calibration_paths.pop(sensor_id, None)
|
||||
runtime.close()
|
||||
|
||||
def list_sensors(self) -> tuple[SensorInfo, ...]:
|
||||
"""Snapshot of currently registered sensors, in ID order."""
|
||||
return tuple(
|
||||
SensorInfo(
|
||||
sensor_id=sid,
|
||||
source=self.sources[sid],
|
||||
kind="file" if is_video_file(self.sources[sid]) else "camera",
|
||||
config=self.configs.get(sid, "-"),
|
||||
calibration=self.calibration_paths.get(sid),
|
||||
)
|
||||
for sid in sorted(self.runtimes)
|
||||
)
|
||||
|
||||
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 sorted(self.runtimes):
|
||||
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(self) -> None:
|
||||
first_error = None
|
||||
for sensor_id in list(self.runtimes):
|
||||
try:
|
||||
self.remove_sensor(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()
|
||||
self.calibration_paths.clear()
|
||||
if first_error is not None:
|
||||
raise first_error
|
||||
@@ -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"}
|
||||
]
|
||||
}
|
||||
@@ -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())
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user