接入力标定 schema 3 与多点空间增益模型
保留旧格式和 cal_path，校验 SN、特征身份及坐标附件，按帧返回逐轴状态。相关测试 263 项通过，已核对存量预测并实测 180 帧录像。
This commit is contained in:
ykcheungab
@@ -0,0 +1,59 @@
|
||||
"""Feature identities use effective measurement settings, never display metadata."""
|
||||
|
||||
from copy import deepcopy
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from orisys.calibration_context import build_context, domain_fingerprint
|
||||
|
||||
|
||||
def config():
|
||||
return {"camera": {"resolution_raw": [1280, 720], "resolution_stitch": [700, 750]},
|
||||
"tracker": {"dis_mode": 4}}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("raw,expected", [(None, 1), ("bad", 1), (0, 1), (-3, 1), ("3", 3), (3.7, 3)])
|
||||
def test_hhd_uses_actual_sensor_normalization(raw, expected):
|
||||
data = config()
|
||||
data["tracker"]["hhd_downsample"] = raw
|
||||
assert build_context(data)["hhd_downsample"] == expected
|
||||
|
||||
|
||||
def test_default_values_equal_explicit_effective_settings():
|
||||
implicit = build_context(config())
|
||||
data = config()
|
||||
data["tracker"]["hhd_downsample"] = 1
|
||||
explicit = build_context(data, {"flow_ema_alpha": 1.0})
|
||||
assert implicit["hhd_downsample"] == 1
|
||||
assert implicit["filters"]["measurement"] == {"flow_ema_alpha": 1.0}
|
||||
assert implicit["feature_signature_sha256"] == explicit["feature_signature_sha256"]
|
||||
|
||||
|
||||
def test_measurement_filters_change_identity_but_readout_and_paths_do_not():
|
||||
data = config()
|
||||
original = build_context(data)
|
||||
assert build_context(data, {"flow_ema_alpha": 0.5})["feature_signature_sha256"] != original["feature_signature_sha256"]
|
||||
data.update(name="other", video="D:/other.mp4", cal_path="E:/calibration.json")
|
||||
assert build_context(data, {"force_median_window": 7, "force_lowpass_cutoff_hz": 2}) == original
|
||||
|
||||
|
||||
def test_explicit_empty_filter_mapping_overrides_config_defaults():
|
||||
data = config()
|
||||
data["filters"] = {"flow_ema_alpha": 0.5}
|
||||
assert build_context(data, {})["filters"]["measurement"] == {"flow_ema_alpha": 1.0}
|
||||
|
||||
|
||||
def test_source_scale_numeric_types_have_identical_identity():
|
||||
assert build_context(config(), source_scale=1) == build_context(config(), source_scale=1.0)
|
||||
|
||||
|
||||
def test_domain_fingerprint_binds_map_content_not_only_dimensions():
|
||||
domain = {"map_x": np.zeros((2, 3)), "map_y": np.zeros((2, 3)),
|
||||
"valid_mask": np.ones((2, 3), dtype=bool), "world_xyz": np.zeros((2, 3, 3))}
|
||||
original = domain_fingerprint(domain)
|
||||
for key in domain:
|
||||
changed = deepcopy(domain)
|
||||
changed[key].flat[0] = not changed[key].flat[0] if key == "valid_mask" else 1
|
||||
assert domain_fingerprint(changed) != original
|
||||
assert domain_fingerprint({key: value.copy() for key, value in domain.items()}) == original
|
||||
@@ -0,0 +1,179 @@
|
||||
"""Schema 3 model validation and deterministic inference."""
|
||||
|
||||
import hashlib
|
||||
from copy import deepcopy
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from orisys.calibration import CalibrationError
|
||||
from orisys.calibration_models import parse_model
|
||||
|
||||
|
||||
def document():
|
||||
return {
|
||||
"schema_version": 3, "unit": "N",
|
||||
"identity": {"sensor_type": "finger", "sn": "DD02-GL"},
|
||||
"pipeline": {
|
||||
"stitch_resolution": [700, 750], "dis_mode": 4,
|
||||
"hhd_downsample": 3, "source_scale": 0.5,
|
||||
"filters": {"measurement": {"flow_ema_alpha": 1.0}},
|
||||
"feature_signature_sha256": "a" * 64,
|
||||
"geometry_sha256": "b" * 64,
|
||||
"coordinates": {"frame": "autocal_flat", "unit": "px",
|
||||
"origin_xy": [511, 1021], "map_sha256": "c" * 64},
|
||||
},
|
||||
"model": {
|
||||
"type": "spatial_gain/1", "input_profile": "fn_delta_autocal_uv/1",
|
||||
"output_frame": "sdk_force/1", "outputs": ["fn"],
|
||||
"params": {"points_uv": [[0, 0], [10, 0], [0, 10]],
|
||||
"gains_N_per_raw": [0.001, 0.002, 0.003],
|
||||
"triangles": [[0, 1, 2]], "valid_force_range_N": [0, 3]},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def features(**overrides):
|
||||
return dict(delta_raw=1000, u=2.5, v=2.5, contact_found=True,
|
||||
zero_valid=True, mapping_valid=True, **overrides)
|
||||
|
||||
|
||||
def test_interpolates_gain_not_sensitivity_and_roundtrips():
|
||||
data = document()
|
||||
model = parse_model(data)
|
||||
result = model.predict(1, 0.1, features(), "zero-1")
|
||||
assert result.values_N == {"fn": pytest.approx(1.75)}
|
||||
assert result.status == {"fn": "ok"}
|
||||
assert model.to_dict() == data
|
||||
assert "fx" not in model.channels
|
||||
data["model"]["params"]["gains_N_per_raw"][0] = 999
|
||||
assert model.predict(2, 0.2, features(), "zero-1").values_N["fn"] == 1.75
|
||||
|
||||
|
||||
def test_legacy_scalar_convert_explicitly_rejected():
|
||||
with pytest.raises(CalibrationError, match="schema 3 requires frame features"):
|
||||
parse_model(document()).convert(1, 2, 3)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("change,status", [
|
||||
({"u": 20}, "outside_coverage"), ({"contact_found": False}, "no_contact"),
|
||||
({"zero_valid": False}, "zero_unavailable"),
|
||||
({"mapping_valid": False}, "coordinate_unmapped"),
|
||||
({"delta_raw": float("nan")}, "invalid_input"),
|
||||
({"delta_raw": "1000"}, "invalid_input"),
|
||||
({"single_contact": False}, "invalid_input"),
|
||||
])
|
||||
def test_frame_invalidity(change, status):
|
||||
values = features()
|
||||
values.update(change)
|
||||
result = parse_model(document()).predict(1, 0.1, values, "zero")
|
||||
assert result.status == {"fn": status}
|
||||
assert result.values_N == {"fn": None}
|
||||
|
||||
|
||||
def test_out_of_range_retains_value():
|
||||
values = features()
|
||||
values["delta_raw"] = 3000
|
||||
result = parse_model(document()).predict(1, 0, values, "zero")
|
||||
assert result.values_N["fn"] == 5.25
|
||||
assert result.status["fn"] == "out_of_range"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mutate", [
|
||||
lambda d: d.update(schema_version=True),
|
||||
lambda d: d.update(extra="ignored"),
|
||||
lambda d: d["identity"].update(sn=""),
|
||||
lambda d: d["model"].update(outputs=["fn", "fx"]),
|
||||
lambda d: d["model"]["params"].update(gains_N_per_raw=[0.001, 0, 0.003]),
|
||||
lambda d: d["model"]["params"].update(triangles=[[0, 1, 1]]),
|
||||
lambda d: d["model"]["params"].update(triangles=[[0, 1, 2], [2, 1, 0]]),
|
||||
lambda d: d["pipeline"].pop("geometry_sha256"),
|
||||
lambda d: d["pipeline"]["coordinates"].update(map_sha256="bad"),
|
||||
])
|
||||
def test_rejects_bad_documents(mutate):
|
||||
data = document()
|
||||
mutate(data)
|
||||
with pytest.raises(ValueError):
|
||||
parse_model(data)
|
||||
|
||||
|
||||
def test_rejects_gap_and_overlapping_equal_area_meshes():
|
||||
data = document()
|
||||
params = data["model"]["params"]
|
||||
params.update(points_uv=[[0, 0], [10, 0], [10, 10], [0, 10]],
|
||||
gains_N_per_raw=[0.001] * 4, triangles=[[0, 1, 2]])
|
||||
with pytest.raises(ValueError, match="triang|mesh|hull"):
|
||||
parse_model(data)
|
||||
params["triangles"] = [[0, 1, 2], [0, 1, 3]]
|
||||
with pytest.raises(ValueError, match="overlap"):
|
||||
parse_model(data)
|
||||
params["triangles"] = [[0, 1, 2], [0, 2, 3]]
|
||||
assert parse_model(data).predict(1, 0, features(), "zero").status["fn"] == "ok"
|
||||
|
||||
|
||||
def test_global_gain_and_context():
|
||||
data = document()
|
||||
data["pipeline"].pop("coordinates")
|
||||
data["pipeline"].pop("geometry_sha256")
|
||||
data["model"].update(type="global_gain/1", input_profile="force_delta/1",
|
||||
outputs=["fn", "fx"], params={
|
||||
"gains_N_per_raw": {"fn": 0.002, "fx": 0.004},
|
||||
"valid_force_range_N": [-3, 3]})
|
||||
model = parse_model(data)
|
||||
result = model.predict(1, 0, {"delta_raw_by_axis": {"fn": 100, "fx": -100},
|
||||
"zero_valid": True, "contact_found": True}, "z")
|
||||
assert result.values_N == {"fn": 0.2, "fx": -0.4}
|
||||
model.validate_for_context(data["pipeline"])
|
||||
wrong = deepcopy(data["pipeline"])
|
||||
wrong["source_scale"] = 0.42
|
||||
with pytest.raises(ValueError, match="source_scale"):
|
||||
model.validate_for_context(wrong)
|
||||
model.reset("new-zero")
|
||||
data["pipeline"].pop("source_scale")
|
||||
assert parse_model(data).input_profile == "force_delta/1"
|
||||
assert model.predict(2, None, {"force_delta_raw": {"fn": 100, "fx": -100},
|
||||
"zero_valid": True}, "z").values_N == {"fn": 0.2, "fx": -0.4}
|
||||
assert model.predict(3, None, {"force_delta_raw": {"fn": 100, "fx": -100},
|
||||
"zero_valid": True, "contact_found": False}, "z").status == {
|
||||
"fn": "no_contact", "fx": "no_contact"}
|
||||
|
||||
|
||||
def test_unknown_backend_fails_explicitly():
|
||||
data = document()
|
||||
data["model"]["type"] = "onnx/1"
|
||||
with pytest.raises(ValueError, match="unsupported.*onnx/1"):
|
||||
parse_model(data)
|
||||
|
||||
|
||||
def test_coordinate_artifact_hash_and_paths(tmp_path):
|
||||
path = tmp_path / "map.npz"
|
||||
np.savez(path, uv_by_output_px=np.zeros((2, 3, 2)), valid_mask=np.ones((2, 3), dtype=bool),
|
||||
origin_xy=np.array([511, 1021]))
|
||||
digest = hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
data = document()
|
||||
data["pipeline"]["coordinates"]["map_sha256"] = digest
|
||||
data["model"]["artifacts"] = [{"role": "coordinate_map", "path": "map.npz", "sha256": digest}]
|
||||
model = parse_model(data, tmp_path / "calibration.json")
|
||||
assert model.coordinate_map.shape == (2, 3, 2)
|
||||
assert len(model.fingerprint) == 64
|
||||
assert not model.coordinate_map.flags.writeable
|
||||
for unsafe in ("../map.npz", "https://example.org/x.npz", "C:\\map.npz", "/map.npz"):
|
||||
data["model"]["artifacts"][0]["path"] = unsafe
|
||||
with pytest.raises(ValueError, match="relative"):
|
||||
parse_model(data, tmp_path / "calibration.json")
|
||||
data["model"]["artifacts"][0]["path"] = "map.npz"
|
||||
path.write_bytes(b"corrupt")
|
||||
with pytest.raises(ValueError, match="SHA-256"):
|
||||
parse_model(data, tmp_path / "calibration.json")
|
||||
|
||||
|
||||
def test_many_points_complete_mesh():
|
||||
from scipy.spatial import Delaunay
|
||||
data = document()
|
||||
points = np.array([[x, y] for x in range(5) for y in range(5)])
|
||||
data["model"]["params"].update(points_uv=points.tolist(), gains_N_per_raw=[0.001] * 25,
|
||||
triangles=Delaunay(points).simplices.tolist())
|
||||
model = parse_model(data)
|
||||
values = features()
|
||||
values.update(u=2, v=2)
|
||||
assert model.predict(1, 0, values, "z").values_N["fn"] == 1.0
|
||||
@@ -0,0 +1,120 @@
|
||||
"""File selection, SN isolation and legacy compatibility at the public loader boundary."""
|
||||
|
||||
import json
|
||||
import time
|
||||
from types import SimpleNamespace
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from orisys import calibration
|
||||
from orisys.calibration_context import build_context
|
||||
from orisys.sensors.calibration import resolve_calibration
|
||||
|
||||
|
||||
def model_doc():
|
||||
config = {"type": "single_camera", "camera": {"resolution_stitch": [64, 64]},
|
||||
"tracker": {"dis_mode": 1, "hhd_downsample": 3}}
|
||||
return {"schema_version": 3, "unit": "N", "identity": {"sensor_type": "finger", "sn": "A1"},
|
||||
"pipeline": build_context(config), "model": {
|
||||
"type": "global_gain/1", "input_profile": "force_delta/1", "output_frame": "sdk_force/1",
|
||||
"outputs": ["fn"], "params": {"gains_N_per_raw": {"fn": .001}, "valid_force_range_N": [0, 3]}}}
|
||||
|
||||
|
||||
def test_schema3_public_load_write_and_fn_only_sn(tmp_path):
|
||||
path = tmp_path / "finger" / "A1.json"
|
||||
path.parent.mkdir()
|
||||
path.write_text(json.dumps(model_doc()))
|
||||
model = calibration.load_for_device(tmp_path, "finger", "A1")
|
||||
assert tuple(model.channels) == ("fn",)
|
||||
assert calibration.check(model).ok
|
||||
target = tmp_path / "copy.json"
|
||||
calibration.write(model, target)
|
||||
assert json.loads(target.read_text())["schema_version"] == 3
|
||||
assert calibration.load(target).fingerprint == model.fingerprint
|
||||
with pytest.raises(calibration.CalibrationError, match="identity mismatch"):
|
||||
calibration.validate_for_device(model, "finger", "another-SN")
|
||||
|
||||
|
||||
def test_missing_sn_does_not_pick_any_neighbor_or_default(tmp_path):
|
||||
(tmp_path / "finger").mkdir()
|
||||
(tmp_path / "finger" / "A1.json").write_text(json.dumps(model_doc()))
|
||||
(tmp_path / "force_calibration.json").write_text(json.dumps(model_doc()))
|
||||
assert calibration.load_for_device(tmp_path, "finger", "B2") is None
|
||||
with pytest.raises(calibration.CalibrationError, match="directory"):
|
||||
calibration.load_for_device(tmp_path / "missing", "finger", "B2")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("change", ["unknown_type", "unknown_schema", "malformed"])
|
||||
def test_selected_bad_model_never_falls_back(tmp_path, change):
|
||||
data = model_doc()
|
||||
if change == "unknown_type":
|
||||
data["model"]["type"] = "onnx/1"
|
||||
elif change == "unknown_schema":
|
||||
data["schema_version"] = 99
|
||||
else:
|
||||
data["model"]["params"]["gains_N_per_raw"] = {}
|
||||
path = tmp_path / "selected.json"
|
||||
path.write_text(json.dumps(data))
|
||||
with pytest.raises(calibration.CalibrationError):
|
||||
calibration.load(path, sensor_type="box")
|
||||
|
||||
|
||||
def test_invalid_default_schema3_does_not_silently_become_raw(monkeypatch, tmp_path):
|
||||
from orisys import configs
|
||||
|
||||
data = model_doc()
|
||||
data["model"]["type"] = "onnx/1"
|
||||
path = tmp_path / "force_calibration.json"
|
||||
path.write_text(json.dumps(data))
|
||||
monkeypatch.setattr(configs, "resolve_sensor_bundle", lambda name: SimpleNamespace(force_calibration_json=path))
|
||||
with pytest.raises(calibration.CalibrationError, match="unsupported calibration model"):
|
||||
calibration.load(sensor_type="finger")
|
||||
|
||||
|
||||
def test_path_selection_keeps_explicit_and_sn_modes_separate(tmp_path):
|
||||
with pytest.raises(calibration.CalibrationError, match="mutually exclusive"):
|
||||
calibration.select_calibration_path(cal_path="file.json", calibration_dir=tmp_path)
|
||||
with pytest.raises(calibration.CalibrationError, match="requires"):
|
||||
calibration.select_calibration_path(device_id="A1")
|
||||
assert calibration.select_calibration_path(cal_path=tmp_path / "A1.json") == tmp_path / "A1.json"
|
||||
|
||||
|
||||
def test_profile_declares_only_available_axes_and_checks_family(tmp_path):
|
||||
path = tmp_path / "selected.json"
|
||||
path.write_text(json.dumps(model_doc()))
|
||||
options = dict(bundle_root=tmp_path, declared_default=None, profile_id="test", selection=path)
|
||||
resolved = resolve_calibration(product_family="finger", **options)
|
||||
assert resolved.axes == ("fn",) and resolved.document is not None
|
||||
assert resolved.fingerprint == resolved.document.fingerprint
|
||||
wrong = resolve_calibration(product_family="box", **options)
|
||||
assert wrong.document is None and wrong.issues[0].code == "calibration.model-family-mismatch"
|
||||
|
||||
|
||||
def test_processor_applies_resolved_selection_including_none(monkeypatch):
|
||||
from orisys.sensors import backends
|
||||
|
||||
monkeypatch.setattr(backends, "validate_profile", lambda profile: SimpleNamespace(usable=True))
|
||||
selected = calibration.parse(model_doc())
|
||||
calls = []
|
||||
sensor = SimpleNamespace(set_calibration=calls.append)
|
||||
for model in (selected, None):
|
||||
profile = SimpleNamespace(processing_backend="single_camera", config={"camera": {}},
|
||||
calibration=SimpleNamespace(document=model), profile_id="test")
|
||||
session = backends.create_processor(profile, sensor=sensor)
|
||||
assert session.sensor is sensor
|
||||
assert calls == [selected, None]
|
||||
|
||||
|
||||
def test_snapshot_keeps_per_frame_units_and_validity_copy():
|
||||
from orisys.pipeline import StreamPipeline
|
||||
|
||||
sensor = SimpleNamespace(info=object(), force_unit="N", force_units=("N", "raw", "raw"),
|
||||
calibration_status={"fn": "outside_coverage"}, fnormal=float("nan"),
|
||||
fshearx=2., fsheary=3., _force_raw=(1., 2., 3.))
|
||||
pipeline = StreamPipeline(sensor, lambda: None, include=())
|
||||
snapshot = pipeline._build(np.zeros((2, 2)), None, 0., 0., 0., time.perf_counter())
|
||||
assert snapshot.force_units == ("N", "raw", "raw")
|
||||
assert snapshot.calibration_status == {"fn": "outside_coverage"}
|
||||
sensor.calibration_status["fn"] = "ok"
|
||||
assert snapshot.calibration_status["fn"] == "outside_coverage"
|
||||
@@ -0,0 +1,308 @@
|
||||
"""Schema 3 lifecycle and readout tests without camera or hardware."""
|
||||
|
||||
from types import SimpleNamespace
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from orisys.calibration import CalibrationError
|
||||
from orisys.force_readout import ForceReadout
|
||||
from orisys.sensor import Sensor
|
||||
|
||||
|
||||
class RecordingModel:
|
||||
schema_version = 3
|
||||
channels = {"fn": None}
|
||||
input_profile = "fn_delta_autocal_uv/1"
|
||||
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
self.resets = []
|
||||
self.coordinate_map = np.array([[[0., 0.], [1., 0.]], [[0., 1.], [1., 1.]]])
|
||||
self.coordinate_valid_mask = np.array([[True, False], [True, True]])
|
||||
|
||||
def reset(self, reference_id):
|
||||
self.resets.append(reference_id)
|
||||
|
||||
def validate_for_context(self, context):
|
||||
if context != {"signature": "actual"}:
|
||||
raise CalibrationError("pipeline mismatch")
|
||||
|
||||
def predict(self, **kwargs):
|
||||
self.calls.append(kwargs)
|
||||
features = kwargs["features"]
|
||||
status = "ok"
|
||||
if not features["zero_valid"]:
|
||||
status = "zero_unavailable"
|
||||
elif not features["contact_found"]:
|
||||
status = "no_contact"
|
||||
elif not features["mapping_valid"]:
|
||||
status = "coordinate_unmapped"
|
||||
value = features["delta_raw"] * .1 if status == "ok" else None
|
||||
return SimpleNamespace(values_N={"fn": value}, status={"fn": status})
|
||||
|
||||
|
||||
def make_sensor(*, median=1):
|
||||
sensor = Sensor.__new__(Sensor)
|
||||
sensor.calibration = RecordingModel()
|
||||
sensor._config = {"filters": {"force_median_window": median}}
|
||||
sensor._force_readout = ForceReadout(sensor._config["filters"])
|
||||
sensor._force_raw = (10., 20., 30.)
|
||||
sensor._model_readouts = {}
|
||||
sensor._calibration_context = None
|
||||
sensor._reset_calibration_session()
|
||||
sensor.bind_calibration_context({"signature": "actual"})
|
||||
return sensor
|
||||
|
||||
|
||||
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.),
|
||||
contact_found=contact, mapping_valid=mapped)
|
||||
|
||||
|
||||
def test_once_per_frame_and_explicit_zero_subtracted_once():
|
||||
sensor = make_sensor()
|
||||
sensor.set_calibration_reference({"fn": 10.}, reference_id="zero-a")
|
||||
first = predict(sensor, 1, 25.)
|
||||
assert first.values_N["fn"] == pytest.approx(1.5)
|
||||
assert predict(sensor, 1, 99.) is first
|
||||
assert len(sensor.calibration.calls) == 1
|
||||
assert sensor.force_units == ("N", "raw", "raw")
|
||||
assert (sensor.fnormal, sensor.fshearx, sensor.fsheary) == (1.5, 20., 30.)
|
||||
assert sensor._force_raw_unfiltered == (25., 20., 30.)
|
||||
|
||||
|
||||
def test_unknown_zero_and_invalid_contact_never_become_zero_newtons():
|
||||
sensor = make_sensor()
|
||||
assert predict(sensor, 1, 20.).status["fn"] == "zero_unavailable"
|
||||
assert np.isnan(sensor.fnormal)
|
||||
sensor.set_calibration_reference({"fn": 10.}, reference_id="zero-b")
|
||||
assert predict(sensor, 2, 20., contact=False).status["fn"] == "no_contact"
|
||||
assert np.isnan(sensor.fnormal)
|
||||
assert predict(sensor, 3, 20., mapped=False).status["fn"] == "coordinate_unmapped"
|
||||
assert np.isnan(sensor.fnormal)
|
||||
|
||||
|
||||
def test_failed_frame_does_not_enter_newton_median_history():
|
||||
sensor = make_sensor(median=3)
|
||||
sensor.set_calibration_reference({"fn": 0.}, reference_id="zero")
|
||||
predict(sensor, 1, 10.)
|
||||
predict(sensor, 2, 1000., contact=False)
|
||||
predict(sensor, 3, 20.)
|
||||
predict(sensor, 4, 30.)
|
||||
assert sensor.fnormal == pytest.approx(2.)
|
||||
assert sensor._force_unfiltered[0] == 3.
|
||||
|
||||
|
||||
def test_new_reference_clears_cached_prediction_and_filter():
|
||||
sensor = make_sensor(median=3)
|
||||
sensor.set_calibration_reference({"fn": 0.}, reference_id="first")
|
||||
predict(sensor, 1, 10.)
|
||||
sensor.set_calibration_reference({"fn": 20.}, reference_id="second")
|
||||
assert sensor.calibration_result is None
|
||||
assert np.isnan(sensor.fnormal)
|
||||
predict(sensor, 1, 30.)
|
||||
assert sensor.fnormal == pytest.approx(1.)
|
||||
assert sensor.calibration.resets[-1] == "second"
|
||||
|
||||
|
||||
def test_requires_actual_pipeline_context():
|
||||
sensor = make_sensor()
|
||||
with pytest.raises(CalibrationError, match="pipeline mismatch"):
|
||||
sensor.bind_calibration_context({"signature": "different"})
|
||||
assert sensor._calibration_context == {"signature": "actual"}
|
||||
sensor._calibration_context = None
|
||||
with pytest.raises(CalibrationError, match="actual calibration context"):
|
||||
predict(sensor, 1, 10.)
|
||||
|
||||
|
||||
def test_contact_map_never_reuses_previous_coordinate():
|
||||
sensor = make_sensor()
|
||||
sensor.centroid_found = True
|
||||
sensor.centroid = (0, 1)
|
||||
assert sensor._calibration_contact_uv() == (0., 1.)
|
||||
sensor.centroid = (1, 0)
|
||||
assert sensor._calibration_contact_uv() is None
|
||||
sensor.centroid = (10, 10)
|
||||
assert sensor._calibration_contact_uv() is None
|
||||
sensor.centroid_found = False
|
||||
sensor.centroid = (0, 1)
|
||||
assert sensor._calibration_contact_uv() is None
|
||||
|
||||
|
||||
def test_hhd_infers_after_current_contact_only_once_and_clears_old_contact():
|
||||
sensor = make_sensor()
|
||||
sensor.set_calibration_reference({"fn": 10.}, reference_id="zero")
|
||||
field = np.zeros((2, 2, 2), dtype=np.float32)
|
||||
sensor._Sensor__hhd = SimpleNamespace(
|
||||
get_force=lambda *args, **kwargs: (None, (20., 0., 0., 0.)),
|
||||
get_normal=lambda: field, get_shear=lambda: field,
|
||||
)
|
||||
contact = {"centroid": (0, 1), "centroid_found": True,
|
||||
"divergence_map": np.zeros((2, 2))}
|
||||
sensor._Sensor__contact3d = SimpleNamespace(compute_foe=lambda *args, **kwargs: contact)
|
||||
sensor.cuda = sensor.verbose = False
|
||||
sensor._hhd_downsample = 1
|
||||
sensor._hhd_grid_shape = sensor._hhd_ref_shape = (2, 2)
|
||||
sensor.valid_mask = sensor.valid_mask_hhd = None
|
||||
sensor._model_frame_id = 0
|
||||
sensor.compute_hhd(field)
|
||||
assert sensor.calibration.calls[-1]["features"]["v"] == 1.
|
||||
assert sensor.fnormal == pytest.approx(1.)
|
||||
sensor.compute_contact()
|
||||
assert len(sensor.calibration.calls) == 1
|
||||
contact["centroid_found"] = False
|
||||
sensor.compute_hhd(field)
|
||||
assert len(sensor.calibration.calls) == 2
|
||||
assert sensor.calibration_status == {"fn": "no_contact"}
|
||||
assert sensor.calibration.calls[-1]["features"]["u"] is None
|
||||
assert np.isnan(sensor.fnormal)
|
||||
|
||||
|
||||
def test_real_model_adapter_uses_same_lifecycle_as_stateful_model():
|
||||
from orisys.calibration import parse
|
||||
|
||||
data = {
|
||||
"schema_version": 3, "unit": "N",
|
||||
"identity": {"sensor_type": "finger", "sn": "test"},
|
||||
"pipeline": {"stitch_resolution": [2, 2], "dis_mode": 4, "hhd_downsample": 1,
|
||||
"filters": {"measurement": {"flow_ema_alpha": 1.}},
|
||||
"feature_signature_sha256": "a" * 64},
|
||||
"model": {"type": "global_gain/1", "input_profile": "force_delta/1",
|
||||
"output_frame": "sdk_force/1", "outputs": ["fn"],
|
||||
"params": {"gains_N_per_raw": {"fn": .1}, "valid_force_range_N": [0., 3.]}},
|
||||
}
|
||||
sensor = make_sensor()
|
||||
sensor.calibration = parse(data)
|
||||
sensor.bind_calibration_context(data["pipeline"])
|
||||
sensor.set_calibration_reference({"fn": 10.}, reference_id="reference")
|
||||
result = sensor.apply_calibration_frame(frame_id=1, raw=(20., 30., 40.))
|
||||
assert result.values_N == {"fn": 1.}
|
||||
assert sensor.force_units == ("N", "raw", "raw")
|
||||
assert sensor.calibration_status == {"fn": "ok"}
|
||||
result = sensor.apply_calibration_frame(frame_id=2, raw=(20., 30., 40.), contact_found=False)
|
||||
assert result.status == {"fn": "no_contact"}
|
||||
|
||||
|
||||
def test_reversed_frame_requires_a_new_zero_reference():
|
||||
sensor = make_sensor()
|
||||
sensor.set_calibration_reference({"fn": 10.}, reference_id="reference")
|
||||
predict(sensor, 10, 20.)
|
||||
assert predict(sensor, 2, 20.).status == {"fn": "zero_unavailable"}
|
||||
assert sensor.calibration.resets[-1] is None
|
||||
|
||||
|
||||
def test_replacement_is_atomic_when_pipeline_cannot_be_validated():
|
||||
sensor = make_sensor()
|
||||
sensor.device_id = None
|
||||
original = sensor.calibration
|
||||
replacement = RecordingModel()
|
||||
with pytest.raises(CalibrationError, match="pipeline mismatch"):
|
||||
sensor.set_calibration(replacement, context={"signature": "different"})
|
||||
assert sensor.calibration is original
|
||||
sensor._calibration_context = None
|
||||
with pytest.raises(CalibrationError, match="actual pipeline context"):
|
||||
sensor.set_calibration(replacement)
|
||||
assert sensor.calibration is original
|
||||
sensor.set_calibration(replacement, context={"signature": "actual"})
|
||||
assert sensor.calibration is replacement
|
||||
assert sensor._calibration_reference == {}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("use_legacy", [False, True])
|
||||
def test_schema3_swap_cannot_relabel_cached_newtons_as_raw_or_legacy(use_legacy):
|
||||
from orisys.calibration import parse
|
||||
|
||||
sensor = make_sensor()
|
||||
sensor.device_id = None
|
||||
sensor.set_calibration_reference({"fn": 10.}, reference_id="reference")
|
||||
predict(sensor, 1, 25.)
|
||||
assert sensor.fnormal == 1.5
|
||||
legacy = parse({"schema_version": 2, "unit": "N", "channels": {
|
||||
"normal": {"method": "piecewise_linear", "points": [[0., 0.], [100., 2.]],
|
||||
"extrapolate": True},
|
||||
}}) if use_legacy else None
|
||||
sensor.set_calibration(legacy)
|
||||
expected = .5 if use_legacy else 25.
|
||||
assert sensor.fnormal == pytest.approx(expected)
|
||||
assert sensor.force_triplets()["unfiltered"]["newton"][0] == pytest.approx(expected)
|
||||
assert sensor.force_triplets()["filtered"]["raw"][0] == 25.
|
||||
assert sensor.calibration_status == {}
|
||||
|
||||
|
||||
def test_reset_invalidates_contact_cache_and_coordinate():
|
||||
sensor = make_sensor()
|
||||
sensor._Sensor__tracker = SimpleNamespace(reset=lambda: None)
|
||||
sensor.img = None
|
||||
sensor.compute_contact_valid = sensor.centroid_found = True
|
||||
sensor.centroid = (1, 1)
|
||||
sensor._model_contact_result = {"centroid": (1, 1)}
|
||||
sensor.compute_deformation_valid = True
|
||||
sensor.reset()
|
||||
assert sensor.compute_contact_valid is False
|
||||
assert sensor.compute_deformation_valid is False
|
||||
assert sensor.centroid_found is False
|
||||
assert sensor._model_contact_result is None
|
||||
assert sensor._calibration_contact_uv() is None
|
||||
with pytest.raises(CalibrationError, match="current deformation"):
|
||||
sensor.compute_contact()
|
||||
|
||||
|
||||
def test_schema3_rejects_measurement_filter_change_before_mutation():
|
||||
sensor = make_sensor()
|
||||
sensor.compute_deformation_valid = False
|
||||
with pytest.raises(CalibrationError, match="measurement identity"):
|
||||
sensor.configure(flow_ema_alpha=.5)
|
||||
assert sensor._force_readout.flow_ema_alpha == 1.
|
||||
assert sensor._calibration_context == {"signature": "actual"}
|
||||
|
||||
|
||||
def test_readout_configuration_applies_before_and_after_first_model_frame():
|
||||
sensor = make_sensor()
|
||||
sensor.compute_deformation_valid = False
|
||||
sensor.configure(force_median_window=3, lowpass_sample_rate_hz=20.)
|
||||
sensor.set_calibration_reference({"fn": 0.}, reference_id="reference")
|
||||
predict(sensor, 1, 10.)
|
||||
readout = sensor._model_readouts["fn"]
|
||||
assert readout.force_median_window == 3
|
||||
assert readout.lowpass_sample_rate_hz == 20.
|
||||
predict(sensor, 2, 20.)
|
||||
predict(sensor, 3, 30.)
|
||||
assert sensor.fnormal == 2.
|
||||
sensor.configure(force_median_window=1)
|
||||
assert readout.force_median_window == 1
|
||||
predict(sensor, 4, 40.)
|
||||
assert sensor.fnormal == 4.
|
||||
|
||||
|
||||
def test_generic_global_fallback_binds_actual_mirror_identity():
|
||||
from orisys.calibration import parse
|
||||
from orisys.calibration_context import build_context
|
||||
|
||||
sensor = make_sensor()
|
||||
sensor.device_id = None
|
||||
sensor.mirror = True
|
||||
sensor._config.update(camera={"resolution_stitch": [2, 2]}, tracker={"dis_mode": 1})
|
||||
context = build_context(sensor._config, sensor.filters_state(), preprocessing={"mirror": True})
|
||||
data = {"schema_version": 3, "unit": "N", "identity": {"sensor_type": "finger", "sn": "test"},
|
||||
"pipeline": context,
|
||||
"model": {"type": "global_gain/1", "input_profile": "force_delta/1",
|
||||
"output_frame": "sdk_force/1", "outputs": ["fn"],
|
||||
"params": {"gains_N_per_raw": {"fn": .1}, "valid_force_range_N": [0., 3.]}}}
|
||||
sensor._calibration_context = None
|
||||
sensor.set_calibration(parse(data))
|
||||
assert sensor._calibration_context == context
|
||||
sensor.mirror = False
|
||||
sensor._calibration_context = None
|
||||
with pytest.raises(CalibrationError, match="feature_signature"):
|
||||
sensor.set_calibration(parse(data))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("reference", [{"fn": float("nan")}, {"other": 0.}])
|
||||
def test_bad_zero_reference_is_rejected_without_reset(reference):
|
||||
sensor = make_sensor()
|
||||
sensor.set_calibration_reference({"fn": 10.}, reference_id="valid")
|
||||
with pytest.raises(ValueError):
|
||||
sensor.set_calibration_reference(reference, reference_id="bad")
|
||||
assert sensor._calibration_reference == {"fn": 10.}
|
||||
Reference in New Issue
Block a user