From 80f7da5a209a92683f066ea05ff195ba146841e0 Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Mon, 5 Oct 2026 12:21:04 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E6=8C=89=E8=AE=BE=E5=A4=87?= =?UTF-8?q?=20ID=20=E9=80=89=E6=8B=A9=E5=92=8C=E6=A0=A1=E9=AA=8C=E6=A0=87?= =?UTF-8?q?=E5=AE=9A=E6=96=87=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 避免多台传感器误用默认标定，缺文件时保持 raw。已通过 SDK 回归测试和文档检查。 --- README.md | 29 ++++++++++++-- src/orisys/calibration.py | 55 +++++++++++++++++++++++++ src/orisys/finger/runtime.py | 10 ++++- src/orisys/sensor.py | 68 ++++++++++++++++++++++++++++--- tests/test_calibration.py | 78 ++++++++++++++++++++++++++++++++++++ tests/test_finger_runtime.py | 31 ++++++++++++++ 6 files changed, 261 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 4875b4a..be9db54 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,8 @@ PyPI: [https://pypi.org/project/orisys/](https://pypi.org/project/orisys/) ## Repository Layout - `src/orisys/`: installable SDK package -- `config/`: default and lab/runtime JSON configurations +- `sensors/`: product bundles containing the current `sensor.json` configurations and assets +- `config/`: legacy JSON configurations retained for compatibility - `examples/`: runnable demos (`basic_pipeline.py`, `mini_pipeline.py`, `finger_pipeline*.py`) plus shared `gui/`, `pipeline/`, `viz/` - `scripts/`: maintenance, calibration, packaging, and demo scripts - `tools/`: non-package tooling such as patch-generation utilities @@ -73,12 +74,13 @@ Run commands from the repository root. ## Configuration -The canonical configuration files now live in the repo-level `config/` folder. +Product configurations live in `sensors//sensor.json`. The repo-level `config/` +folder remains available for older configurations and aliases. The package loader `orisys.configs` resolves configs from: 1. an explicit JSON file path -2. a config file stem under `config/` (for example `box`, `mini`, `finger`) -3. a product alias such as `dd02` -> `dd02-ov` +2. `sensors//sensor.json` (for example `box`, `mini`, `finger`) +3. a file stem under the legacy `config/` folder, including aliases such as `dd02` -> `dd02-ov` 4. a backward-compatible JSON `"name"` match, such as `"prototype"` Examples: @@ -90,6 +92,25 @@ sensor = orisys.Sensor(0, config_name="box") cfg = orisys.export_config("box", "./config/exported_config.json") ``` +For multiple physical sensors, keep calibration files in a user-selected directory: +`//.json` (for example, +`./calibrations/box/BOX-001.json`). Each file uses the existing calibration JSON +format and must contain matching `identity.sensor_type` and `identity.sn`, Newton +curves for FN/FX/FY, and a compatible measurement pipeline. Enter the device ID +manually until hardware identification is available: + +```python +sensor = orisys.Sensor(0, config_name="box", calibration_dir="./calibrations", device_id="BOX-001") +``` + +If that ID has no file, the sensor reports raw force; it never falls back to the +product's default. An invalid or mismatched file raises `CalibrationError`. +`FingerRuntime.open` accepts the same `calibration_dir` and `device_id` arguments. +Existing calls without a device ID still use the product bundle default, and +explicit `cal_path` remains available for one-off use. `cal_path` and +`calibration_dir` cannot be combined. Pass `device_id` with `cal_path` to validate +that explicit file against the selected device. + ## Examples and Tools - Example entry points: `examples/README.md` diff --git a/src/orisys/calibration.py b/src/orisys/calibration.py index 3442047..6291373 100644 --- a/src/orisys/calibration.py +++ b/src/orisys/calibration.py @@ -7,6 +7,7 @@ curves in ``_apply_force_filter``; ``read_info(FNORMAL, …)`` returns Newtons. from __future__ import annotations import json +import re import warnings from collections.abc import Iterable, Mapping from copy import deepcopy @@ -805,6 +806,60 @@ def load( return None +def load_for_device( + calibration_dir: str | Path, + sensor_type: str, + device_id: str, + *, + pipeline: Mapping[str, Any] | None = None, +) -> Calibration | None: + """Load ``//.json``; missing file means raw.""" + root = Path(calibration_dir).expanduser() + if not root.is_dir(): + raise CalibrationError(f"calibration directory not found: {root}") + for label, value in (("sensor_type", sensor_type), ("device_id", device_id)): + if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", value): + raise CalibrationError(f"invalid {label}: {value!r}") + path = root / sensor_type / f"{device_id}.json" + if not path.exists(): + if path.is_symlink(): + raise CalibrationError(f"calibration link is broken: {path}") + return None + if not path.is_file(): + raise CalibrationError(f"calibration path is not a file: {path}") + cal = load(path, strict=True) + assert cal is not None + return validate_for_device(cal, sensor_type, device_id, pipeline=pipeline) + + +def validate_for_device( + cal: Calibration, + sensor_type: str, + device_id: str, + *, + pipeline: Mapping[str, Any] | None = None, +) -> Calibration: + """Reject a calibration for another device or measurement pipeline.""" + path = cal.source_path or "" + if cal.identity.get("sensor_type") != sensor_type or cal.identity.get("sn") != device_id: + raise CalibrationError(f"calibration identity mismatch: {path}") + if cal.unit != "N" or any(cal.channels[axis].unit != "N" for axis in cal.channels): + raise CalibrationError(f"calibration unit must be N: {path}") + if set(CHANNEL_IDS) - cal.channels.keys(): + raise CalibrationError(f"calibration requires FN, FX and FY: {path}") + if pipeline is not None: + for key in ("stitch_resolution", "dis_mode", "hhd_downsample"): + recorded, current = cal.pipeline.get(key), pipeline.get(key) + if recorded is not None and current is not None and recorded != current: + raise CalibrationError(f"calibration pipeline mismatch ({key}): {path}") + measurement, _ = filter_mismatches( + cal.pipeline.get("filters"), pipeline.get("filters") + ) + if measurement: + raise CalibrationError(f"calibration pipeline mismatch ({'; '.join(measurement)}): {path}") + return cal + + def pipeline_from_sensor_json( path: str | Path | None, *, diff --git a/src/orisys/finger/runtime.py b/src/orisys/finger/runtime.py index eaf4478..dded85f 100644 --- a/src/orisys/finger/runtime.py +++ b/src/orisys/finger/runtime.py @@ -254,6 +254,9 @@ class FingerRuntime: map_cache=DEFAULT_FINGER_MAP_CACHE, source_scale=None, calib=None, + cal_path=None, + calibration_dir=None, + device_id=None, video=0, camera_id=None, flip_source_vertical=True, @@ -293,6 +296,12 @@ class FingerRuntime: ) domain.remap.ensure_fast_maps() + sensor = Sensor( + "external", config_name=config, cal_path=cal_path, + calibration_dir=calibration_dir, device_id=device_id, + verbose=verbose, mirror=mirror, + ) + capture_config = CaptureConfig.from_camera_dict( video, geometry.config.get("camera", {}), @@ -307,7 +316,6 @@ class FingerRuntime: capture_source.start_background() capture = capture_source.cap - sensor = Sensor("external", config_name=config, verbose=verbose, mirror=mirror) if not is_live_camera(video): # File playback: wall-clock cadence is not signal time, so pin the # force-filter design rate to the configured camera fps. diff --git a/src/orisys/sensor.py b/src/orisys/sensor.py index a4ec2f4..e72c17f 100644 --- a/src/orisys/sensor.py +++ b/src/orisys/sensor.py @@ -143,6 +143,7 @@ class Sensor: backend='auto', capture_config: CaptureConfig | None = None, auto_reconnect: bool = False, capture_mode: str = "latest", capture_queue_size: int = 1, capture_overflow: str = "drop_oldest", + calibration_dir: str | Path | None = None, device_id: str | None = None, **_deprecated_capture_kwargs): """ Initialize Sensor with video source and configuration. @@ -154,6 +155,8 @@ class Sensor: verbose: Print initialization logs (default: True) config_name: Configuration name (default: "prototype") cal_path: Explicit force_calibration.json path (overrides bundle default) + calibration_dir: User-managed root containing /.json. + device_id: Manually entered or hardware-reported device identity. override: Skip auto-detection, use provided box (default: False) box: Manual bounding boxes when override=True (default: None) backend: Poisson backend (default: "auto") @@ -184,6 +187,15 @@ class Sensor: self.debug = debug self.config_name = config_name self.cal_path = cal_path + if calibration_dir is not None and device_id is None: + raise cal_mod.CalibrationError("device_id is required with calibration_dir") + if device_id is not None and calibration_dir is None and cal_path is None: + raise cal_mod.CalibrationError("device_id requires calibration_dir or cal_path") + if calibration_dir is not None and cal_path is not None: + raise cal_mod.CalibrationError("cal_path and calibration_dir are mutually exclusive") + self.calibration_dir = calibration_dir + self.device_id = device_id + self.device_calibration_missing = False self.mirror = mirror self._capture_config_override = capture_config self._auto_reconnect = bool(auto_reconnect) @@ -1273,6 +1285,11 @@ class Sensor: bundle from another root and stopped there without ever asking whether that root had coefficients. """ + if getattr(self, "device_calibration_missing", False): + return ( + f"Force calibration: raw (no file for {self.device_id!r} " + f"under {self.calibration_dir})" + ) attempts = self._force_cal_attempts or [] # ``found_root`` is keyed even when nothing was found (valued None), so # test its value, not the key. @@ -1306,14 +1323,30 @@ class Sensor: return "\n".join(lines) def _init_force_calibration(self) -> None: - """Load force calibration from ``cal_path`` or the product bundle.""" + """Load the selected force calibration before processing frames.""" self.calibration = None self._force_cal_sensor_name = None self._force_cal_attempts = None + family = None + pipeline = None + if self.device_id is not None: + family, pipeline = self._device_calibration_expectations() + if self.cal_path is not None and str(self.cal_path).strip(): # Explicit path: missing/invalid raises (no silent fallback). self.calibration = cal_mod.load(self.cal_path, strict=True) + if self.device_id is not None: + self.calibration = cal_mod.validate_for_device( + self.calibration, family, self.device_id, pipeline=pipeline + ) + return + + if self.calibration_dir is not None: + self.calibration = cal_mod.load_for_device( + self.calibration_dir, family, self.device_id, pipeline=pipeline + ) + self.device_calibration_missing = self.calibration is None return attempts: list[dict] = [] @@ -1333,6 +1366,24 @@ class Sensor: # without coefficients is not that. self._force_cal_attempts = attempts or None + def _device_calibration_expectations(self): + family = next( + (name for name in self._bundle_name_candidates() if name in configs.PRODUCT_BUNDLES), + None, + ) + if family is None: + raise cal_mod.CalibrationError( + f"cannot determine product family from config_name={self.config_name!r}" + ) + camera = self._config.get("camera", {}) + tracker = self._config.get("tracker", {}) + return family, { + "stitch_resolution": camera.get("resolution_stitch"), + "dis_mode": tracker.get("dis_mode"), + "hhd_downsample": tracker.get("hhd_downsample"), + "filters": cal_mod.filters_fingerprint(self._config.get("filters")), + } + @property def force_calibrated(self) -> bool: """True when at least one channel curve is loaded.""" @@ -1349,14 +1400,21 @@ class Sensor: """Attach or replace the active force calibration (live swap after save).""" if cal is None: self.calibration = None + self.device_calibration_missing = False self._force_cal_sensor_name = None self._force_cal_attempts = None return if isinstance(cal, cal_mod.Calibration): - self.calibration = cal - self._force_cal_attempts = None - return - self.calibration = cal_mod.load(cal, strict=True) + selected = cal + else: + selected = cal_mod.load(cal, strict=True) + if self.device_id is not None: + family, pipeline = self._device_calibration_expectations() + selected = cal_mod.validate_for_device( + selected, family, self.device_id, pipeline=pipeline + ) + self.calibration = selected + self.device_calibration_missing = False self._force_cal_attempts = None def _apply_force_filter(self): diff --git a/tests/test_calibration.py b/tests/test_calibration.py index 62d1991..f19e5ed 100644 --- a/tests/test_calibration.py +++ b/tests/test_calibration.py @@ -17,6 +17,7 @@ from orisys.calibration import ( filter_mismatches, filters_fingerprint, load, + load_for_device, parse, pipeline_from_sensor_json, write, @@ -241,6 +242,83 @@ def test_missing_cal_path_raises(): ) +def _device_file(root: Path, device_id: str, *, family: str = "box") -> Path: + path = root / family / f"{device_id}.json" + path.parent.mkdir(parents=True, exist_ok=True) + doc = json.loads(_box_cal_path().read_text(encoding="utf-8")) + doc["identity"] = {"sensor_type": family, "sn": device_id} + path.write_text(json.dumps(doc), encoding="utf-8") + return path + + +def test_device_calibration_selection_and_missing_raw(tmp_path: Path): + root = tmp_path / "calibrations" + root.mkdir() + selected = _device_file(root, "BOX-001") + assert load_for_device(root, "box", "BOX-001").source_path == selected.resolve() + assert load_for_device(root, "box", "BOX-002") is None + + sensor = Sensor( + "external", config_name="box", calibration_dir=root, + device_id="BOX-001", verbose=False, backend="cpu", + ) + try: + assert sensor.force_unit == "N" + assert sensor.calibration.source_path == selected.resolve() + other = _device_file(root, "BOX-003") + with pytest.raises(CalibrationError, match="identity mismatch"): + sensor.set_calibration(other) + assert sensor.calibration.source_path == selected.resolve() + finally: + sensor.disconnect() + + sensor = Sensor( + "external", config_name="box", calibration_dir=root, + device_id="BOX-002", verbose=False, backend="cpu", + ) + try: + assert sensor.force_unit == "raw" + assert sensor.device_calibration_missing + assert sensor.calibration is None + finally: + sensor.disconnect() + + +def test_device_calibration_rejects_wrong_identity_and_pipeline(tmp_path: Path): + root = tmp_path / "calibrations" + root.mkdir() + path = _device_file(root, "BOX-001") + doc = json.loads(path.read_text(encoding="utf-8")) + doc["identity"]["sn"] = "BOX-002" + path.write_text(json.dumps(doc), encoding="utf-8") + with pytest.raises(CalibrationError, match="identity mismatch"): + load_for_device(root, "box", "BOX-001") + with pytest.raises(CalibrationError, match="identity mismatch"): + Sensor( + "external", config_name="box", cal_path=path, + device_id="BOX-001", verbose=False, backend="cpu", + ) + + doc["identity"]["sn"] = "BOX-001" + doc["pipeline"] = {"dis_mode": 999} + path.write_text(json.dumps(doc), encoding="utf-8") + with pytest.raises(CalibrationError, match="pipeline mismatch"): + load_for_device(root, "box", "BOX-001", pipeline={"dis_mode": 1}) + + +def test_device_calibration_rejects_partial_axes_and_unsafe_id(tmp_path: Path): + root = tmp_path / "calibrations" + root.mkdir() + path = _device_file(root, "BOX-001") + doc = json.loads(path.read_text(encoding="utf-8")) + del doc["channels"]["shear_y"] + path.write_text(json.dumps(doc), encoding="utf-8") + with pytest.raises(CalibrationError, match="requires FN, FX and FY"): + load_for_device(root, "box", "BOX-001") + with pytest.raises(CalibrationError, match="invalid device_id"): + load_for_device(root, "box", "../BOX-001") + + def test_parse_shipped_box_file_directly(): cal = parse(json.loads(_box_cal_path().read_text(encoding="utf-8")), source_path=_box_cal_path()) assert cal.schema_version == 2 diff --git a/tests/test_finger_runtime.py b/tests/test_finger_runtime.py index 733056b..b1d7700 100644 --- a/tests/test_finger_runtime.py +++ b/tests/test_finger_runtime.py @@ -1,4 +1,5 @@ from pathlib import Path +from types import SimpleNamespace import numpy as np import pytest @@ -159,6 +160,36 @@ def test_finger_runtime_get_img_returns_none_on_capture_failure(): assert runtime.get_img() is None +def test_finger_open_selects_device_calibration_before_capture(monkeypatch, tmp_path): + import orisys.finger.runtime as runtime_mod + + selected = {} + geometry = SimpleNamespace( + export_paths=SimpleNamespace(as_tuple=lambda: ("mesh", "intrinsics", "pose")), + map_cache="cache", source_K=np.eye(3), source_width=2, source_height=2, + out_width=2, out_height=2, edge_expand=1.0, source_scale=1.0, + config={"camera": {}}, + ) + monkeypatch.setattr(runtime_mod, "resolve_finger_geometry", lambda **_: geometry) + monkeypatch.setattr( + runtime_mod, "build_or_load_domain", + lambda *args, **kwargs: SimpleNamespace( + remap=SimpleNamespace(ensure_fast_maps=lambda: None) + ), + ) + monkeypatch.setattr(runtime_mod, "Sensor", lambda *args, **kwargs: selected.update(kwargs)) + monkeypatch.setattr(runtime_mod.CaptureConfig, "from_camera_dict", lambda *args, **kwargs: None) + + def stop_before_capture(*args, **kwargs): + assert selected["calibration_dir"] == tmp_path + assert selected["device_id"] == "FINGER-001" + raise RuntimeError("capture reached") + + monkeypatch.setattr(runtime_mod, "open_capture", stop_before_capture) + with pytest.raises(RuntimeError, match="capture reached"): + runtime_mod.FingerRuntime.open(calibration_dir=tmp_path, device_id="FINGER-001") + + def test_finger_runtime_get_img_returns_unrolled_frame_on_success(): from orisys.finger.runtime import FingerRuntime