ykcheungab
180 lines
7.3 KiB
Python
180 lines
7.3 KiB
Python
"""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
|