补充 Finger 模型切换与逐帧接触状态
按实际几何和处理参数校验切换模型,并保留帧内接触状态。88 项相关测试通过。
This commit is contained in:
@@ -3,6 +3,8 @@
|
||||
Purpose: reference. Scope: SDK 标定文件选择、SN 绑定和推断接口。
|
||||
状态:2026-10-18 已实现 schema 3 的空间/全局增益;ML 后端仅保留契约,尚未实现。
|
||||
保持 `cal_path` 与现有单点文件;新增空间模型和机器学习模型使用同一入口。
|
||||
运行中的 Finger 可调用 `runtime.set_calibration(file, device_id=sn)`;调用者须在处理线程的帧边界执行。
|
||||
该接口按实际几何、展开域及处理参数重新校验,成功切换后清除会话零点;校验失败保留原模型。
|
||||
|
||||
## 现状与边界
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
@@ -386,6 +387,31 @@ class FingerRuntime:
|
||||
"""Set an explicitly measured unloaded reference for this finger stream."""
|
||||
self.tactile.raw_sensor.set_calibration_reference(raw, reference_id=reference_id)
|
||||
|
||||
def set_calibration(self, calibration, *, device_id=None):
|
||||
"""Replace the model at a caller-owned frame boundary, validating actual geometry."""
|
||||
from .. import calibration as cal_mod
|
||||
from ..calibration_context import build_context, domain_fingerprint, sha256_file
|
||||
|
||||
sensor = self.tactile.raw_sensor
|
||||
selected = cal_mod.load(calibration, strict=True) if isinstance(calibration, (str, Path)) else calibration
|
||||
serial = device_id or sensor.device_id
|
||||
if selected is not None and serial:
|
||||
selected = cal_mod.validate_for_device(selected, "finger", serial)
|
||||
context = None
|
||||
if getattr(selected, "schema_version", None) == 3:
|
||||
if selected.input_profile == "fn_delta_autocal_uv/1":
|
||||
uv_map = selected.coordinate_map
|
||||
if uv_map is None or uv_map.shape[:2] != (self.geometry.out_height, self.geometry.out_width):
|
||||
raise ValueError("calibration coordinate map does not match finger output shape")
|
||||
context = build_context(
|
||||
self.geometry.config, sensor.filters_state(),
|
||||
geometry_sha256=sha256_file(self.geometry.export_paths.mesh),
|
||||
source_scale=self.geometry.source_scale, coordinates=selected.pipeline.get("coordinates"),
|
||||
output_domain_sha256=domain_fingerprint(self.domain), processing_backend="uv_unroll",
|
||||
preprocessing={"flip_source_vertical": self.frame_processor.flip_source_vertical,
|
||||
"mirror": sensor.mirror})
|
||||
sensor.set_calibration(selected, context=context)
|
||||
|
||||
@property
|
||||
def calibration_status(self):
|
||||
"""Copy the current per-axis calibration validity state."""
|
||||
|
||||
@@ -156,6 +156,7 @@ class FrameSnapshot:
|
||||
source_metadata: Any = None
|
||||
force_units: tuple[str, str, str] | None = None
|
||||
calibration_status: dict[str, str] | None = None
|
||||
contact_found: bool | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -658,6 +659,7 @@ class StreamPipeline:
|
||||
torque=float(getattr(sensor, "ftor", 0.0)),
|
||||
force_units=tuple(getattr(sensor, "force_units", (getattr(sensor, "force_unit", "raw"),) * 3)),
|
||||
calibration_status=dict(getattr(sensor, "calibration_status", {})),
|
||||
contact_found=bool(getattr(sensor, "centroid_found", False)),
|
||||
# Both already private per-frame copies owned by this snapshot.
|
||||
image=frame if "image" in include else None,
|
||||
flow=flow if "flow" in include else None,
|
||||
|
||||
@@ -55,6 +55,45 @@ def make_sensor(*, median=1):
|
||||
return sensor
|
||||
|
||||
|
||||
def test_finger_model_swap_validates_actual_runtime_and_invalidates_zero(monkeypatch, tmp_path):
|
||||
import orisys.calibration_context as context_mod
|
||||
from orisys.finger.runtime import FingerRuntime
|
||||
|
||||
sensor = make_sensor()
|
||||
sensor.device_id = None
|
||||
sensor.mirror = False
|
||||
sensor.set_calibration_reference({"fn": 10}, reference_id="old-zero")
|
||||
runtime = FingerRuntime.__new__(FingerRuntime)
|
||||
runtime.tactile = SimpleNamespace(raw_sensor=sensor)
|
||||
mesh = tmp_path / "mesh.npz"
|
||||
mesh.write_bytes(b"actual geometry")
|
||||
runtime.geometry = SimpleNamespace(config={}, export_paths=SimpleNamespace(mesh=mesh),
|
||||
source_scale=.42, out_height=2, out_width=2)
|
||||
runtime.domain = object()
|
||||
runtime.frame_processor = SimpleNamespace(flip_source_vertical=True)
|
||||
monkeypatch.setattr(context_mod, "domain_fingerprint", lambda domain: "domain")
|
||||
contexts = []
|
||||
|
||||
def context(config, filters, **kwargs):
|
||||
contexts.append(kwargs)
|
||||
return {"signature": "actual"}
|
||||
|
||||
monkeypatch.setattr(context_mod, "build_context", context)
|
||||
selected = RecordingModel()
|
||||
selected.pipeline = {"coordinates": {"frame": "autocal_flat"}}
|
||||
runtime.set_calibration(selected)
|
||||
assert sensor.calibration is selected
|
||||
assert sensor.calibration_status == {"fn": "zero_unavailable"}
|
||||
assert contexts[0]["geometry_sha256"] == context_mod.sha256_file(mesh)
|
||||
assert contexts[0]["output_domain_sha256"] == "domain"
|
||||
wrong_shape = RecordingModel()
|
||||
wrong_shape.pipeline = selected.pipeline
|
||||
wrong_shape.coordinate_map = np.zeros((1, 1, 2))
|
||||
with pytest.raises(ValueError, match="shape"):
|
||||
runtime.set_calibration(wrong_shape)
|
||||
assert sensor.calibration is selected
|
||||
|
||||
|
||||
def predict(sensor, frame, fn, *, contact=True, mapped=True):
|
||||
return sensor.apply_calibration_frame(frame_id=frame, timestamp_s=frame / 30.,
|
||||
raw=(fn, 20., 30.), uv=(0., 0.),
|
||||
|
||||
Reference in New Issue
Block a user