ykcheungab
431 lines
14 KiB
Python
431 lines
14 KiB
Python
"""Tests for orisys.calibration load/parse/write and Sensor apply."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from orisys import Sensor
|
|
from orisys.calibration import (
|
|
CalSample,
|
|
CalibrationError,
|
|
build,
|
|
check,
|
|
filter_mismatches,
|
|
filters_fingerprint,
|
|
load,
|
|
parse,
|
|
pipeline_from_sensor_json,
|
|
write,
|
|
)
|
|
|
|
|
|
def _box_cal_path() -> Path:
|
|
# Editable checkout: repo_root/sensors/box/force_calibration.json
|
|
here = Path(__file__).resolve()
|
|
candidate = here.parents[1] / "sensors" / "box" / "force_calibration.json"
|
|
assert candidate.is_file(), f"missing shipped cal: {candidate}"
|
|
return candidate
|
|
|
|
|
|
def test_parse_box_v1_piecewise():
|
|
cal = load(sensor_type="box")
|
|
assert cal is not None
|
|
assert cal.get("FN") is not None
|
|
assert cal.get("FN").convert(10000.0) == pytest.approx(0.377)
|
|
assert cal.get("FX").convert_component(36000.0) == pytest.approx(1.19)
|
|
assert cal.get("FX").convert_component(-36000.0) == pytest.approx(-1.19)
|
|
assert cal.get("FY").convert_component(36000.0) == pytest.approx(1.19)
|
|
|
|
|
|
def test_parse_v2_tmp_file(tmp_path: Path):
|
|
doc = {
|
|
"schema_version": 2,
|
|
"unit": "N",
|
|
"channels": {
|
|
"normal": {
|
|
"method": "piecewise_linear",
|
|
"points": [[0.0, 0.0], [1000.0, 1.0]],
|
|
"extrapolate": True,
|
|
"clamp_output_min": 0.0,
|
|
},
|
|
"shear_x": {
|
|
"method": "piecewise_linear",
|
|
"signed": True,
|
|
"points": [[0.0, 0.0], [500.0, 2.0]],
|
|
"extrapolate": True,
|
|
"clamp_output_min": 0.0,
|
|
},
|
|
"shear_y": {
|
|
"method": "piecewise_linear",
|
|
"signed": True,
|
|
"points": [[0.0, 0.0], [500.0, 3.0]],
|
|
"extrapolate": True,
|
|
"clamp_output_min": 0.0,
|
|
},
|
|
},
|
|
"meta": {"sensor_type": "box", "sn": "TEST-001"},
|
|
}
|
|
path = tmp_path / "force_calibration_v2.json"
|
|
path.write_text(json.dumps(doc), encoding="utf-8")
|
|
cal = load(path)
|
|
assert cal is not None
|
|
assert check(cal).ok
|
|
assert cal.convert(1000.0, 500.0, -500.0) == pytest.approx((1.0, 2.0, -3.0))
|
|
|
|
|
|
def test_build_write_load_roundtrip(tmp_path: Path):
|
|
samples = {
|
|
"fn": [
|
|
CalSample(weight_n=0.0, raw_mean=0.0, n=10),
|
|
CalSample(weight_n=1.0, raw_mean=1000.0, n=10),
|
|
],
|
|
"fx": [
|
|
CalSample(weight_n=0.0, raw_mean=0.0, n=10),
|
|
CalSample(weight_n=2.0, raw_mean=500.0, n=10),
|
|
],
|
|
"fy": [
|
|
CalSample(weight_n=0.0, raw_mean=0.0, n=10),
|
|
CalSample(weight_n=3.0, raw_mean=500.0, n=10),
|
|
],
|
|
}
|
|
cal = build(
|
|
samples,
|
|
identity={"sensor_type": "box", "sn": "RT-1"},
|
|
pipeline={"dis_mode": 1},
|
|
origin="user",
|
|
)
|
|
path = tmp_path / "out.json"
|
|
write(cal, path, overwrite=True)
|
|
loaded = load(path)
|
|
assert loaded is not None
|
|
assert loaded.convert(1000.0, 500.0, -500.0) == pytest.approx((1.0, 2.0, -3.0))
|
|
|
|
|
|
def test_check_rejects_bad_schema_and_short_points():
|
|
bad_schema = {"schema_version": 99, "channels": {"normal": {"method": "linear", "slope": 1, "offset": 0}}}
|
|
result = check(bad_schema)
|
|
assert not result.ok
|
|
assert any("schema_version" in e for e in result.errors)
|
|
|
|
short = {
|
|
"schema_version": 2,
|
|
"channels": {
|
|
"normal": {
|
|
"method": "piecewise_linear",
|
|
"points": [[0.0, 0.0]],
|
|
}
|
|
},
|
|
}
|
|
result2 = check(short)
|
|
assert not result2.ok
|
|
|
|
|
|
def test_sensor_cal_path_v2_applies_to_read_info(tmp_path: Path):
|
|
doc = {
|
|
"schema_version": 2,
|
|
"unit": "N",
|
|
"channels": {
|
|
"normal": {
|
|
"method": "piecewise_linear",
|
|
"points": [[0.0, 0.0], [100.0, 5.0]],
|
|
"extrapolate": True,
|
|
"clamp_output_min": 0.0,
|
|
},
|
|
"shear_x": {
|
|
"method": "piecewise_linear",
|
|
"signed": True,
|
|
"points": [[0.0, 0.0], [100.0, 7.0]],
|
|
"extrapolate": True,
|
|
"clamp_output_min": 0.0,
|
|
},
|
|
"shear_y": {
|
|
"method": "piecewise_linear",
|
|
"signed": True,
|
|
"points": [[0.0, 0.0], [100.0, 9.0]],
|
|
"extrapolate": True,
|
|
"clamp_output_min": 0.0,
|
|
},
|
|
},
|
|
}
|
|
path = tmp_path / "v2.json"
|
|
path.write_text(json.dumps(doc), encoding="utf-8")
|
|
sensor = Sensor(
|
|
"external",
|
|
config_name="box",
|
|
cal_path=str(path),
|
|
verbose=False,
|
|
backend="cpu",
|
|
)
|
|
try:
|
|
assert sensor.force_calibrated
|
|
assert sensor.force_unit == "N"
|
|
sensor._force_raw = (100.0, 100.0, -100.0)
|
|
sensor._apply_force_filter()
|
|
fn, fx, fy = sensor.read_info(
|
|
sensor.info.FNORMAL,
|
|
sensor.info.FSHEARX,
|
|
sensor.info.FSHEARY,
|
|
)
|
|
fn_raw, fx_raw, fy_raw = sensor.read_info(
|
|
sensor.info.FNORMAL_RAW,
|
|
sensor.info.FSHEARX_RAW,
|
|
sensor.info.FSHEARY_RAW,
|
|
)
|
|
assert (fn_raw, fx_raw, fy_raw) == pytest.approx((100.0, 100.0, -100.0))
|
|
assert (fn, fx, fy) == pytest.approx((5.0, 7.0, -9.0))
|
|
finally:
|
|
sensor.disconnect()
|
|
|
|
|
|
def test_set_calibration_swaps_curve(tmp_path: Path):
|
|
path_a = tmp_path / "a.json"
|
|
path_b = tmp_path / "b.json"
|
|
path_a.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 2,
|
|
"channels": {
|
|
"normal": {
|
|
"method": "piecewise_linear",
|
|
"points": [[0.0, 0.0], [10.0, 1.0]],
|
|
}
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
path_b.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 2,
|
|
"channels": {
|
|
"normal": {
|
|
"method": "piecewise_linear",
|
|
"points": [[0.0, 0.0], [10.0, 2.0]],
|
|
}
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
sensor = Sensor(
|
|
"external",
|
|
config_name="box",
|
|
cal_path=str(path_a),
|
|
verbose=False,
|
|
backend="cpu",
|
|
)
|
|
try:
|
|
sensor._force_raw = (10.0, 0.0, 0.0)
|
|
sensor._apply_force_filter()
|
|
assert sensor.fnormal == pytest.approx(1.0)
|
|
sensor.set_calibration(path_b)
|
|
sensor._apply_force_filter()
|
|
assert sensor.fnormal == pytest.approx(2.0)
|
|
finally:
|
|
sensor.disconnect()
|
|
|
|
|
|
def test_missing_cal_path_raises():
|
|
with pytest.raises(CalibrationError):
|
|
Sensor(
|
|
"external",
|
|
config_name="box",
|
|
cal_path="D:/no/such/force_calibration.json",
|
|
verbose=False,
|
|
backend="cpu",
|
|
)
|
|
|
|
|
|
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 == 1
|
|
assert cal.convert(10000.0, 36000.0, -36000.0) == pytest.approx((0.377, 1.19, -1.19))
|
|
|
|
|
|
# ------------------------------------------------------- filter fingerprint
|
|
ALL_OFF_FILTERS = {
|
|
"flow_ema_alpha": 1.0,
|
|
"force_median_window": 1,
|
|
"force_lowpass_cutoff_hz": 0.0,
|
|
"force_lowpass_order": 2,
|
|
"force_lowpass_sample_rate_hz": "auto",
|
|
}
|
|
|
|
|
|
def _sensor_with_filters(tmp_path: Path, filters: dict) -> Sensor:
|
|
"""An external-frame Sensor whose bundle carries ``filters``."""
|
|
bundle = _box_cal_path().parent
|
|
copy = tmp_path / "box"
|
|
shutil.copytree(bundle, copy)
|
|
path = copy / "sensor.json"
|
|
cfg = json.loads(path.read_text(encoding="utf-8"))
|
|
cfg["filters"] = filters
|
|
path.write_text(json.dumps(cfg, indent=2), encoding="utf-8")
|
|
return Sensor(
|
|
"external",
|
|
config_name=str(path),
|
|
cal_path=str(copy / "force_calibration.json"),
|
|
verbose=False,
|
|
backend="cpu",
|
|
)
|
|
|
|
|
|
def test_the_fingerprint_keeps_measurement_and_readout_apart():
|
|
"""A change to one group must not be confused with a change to the other."""
|
|
fp = filters_fingerprint(ALL_OFF_FILTERS)
|
|
assert fp["measurement"] == {"flow_ema_alpha": 1.0}
|
|
assert fp["readout"] == {
|
|
"force_median_window": 1,
|
|
"force_lowpass_cutoff_hz": 0.0,
|
|
"force_lowpass_order": 2,
|
|
"force_lowpass_rate_auto": True,
|
|
}
|
|
|
|
|
|
def test_a_preset_expands_without_inventing_anything():
|
|
fp = filters_fingerprint({"preset": "low_noise"})
|
|
assert fp["preset"] == "low_noise"
|
|
assert fp["measurement"] == {"flow_ema_alpha": 0.3}
|
|
assert fp["readout"] == {"force_median_window": 3, "force_lowpass_cutoff_hz": 10.0}
|
|
# An explicit key wins over the preset.
|
|
explicit = filters_fingerprint({"preset": "low_noise", "force_lowpass_cutoff_hz": 0.0})
|
|
assert explicit["readout"]["force_lowpass_cutoff_hz"] == 0.0
|
|
# Keys no source states are absent, never filled in from a default.
|
|
assert filters_fingerprint({}) == {}
|
|
assert filters_fingerprint(None) == {}
|
|
assert filters_fingerprint({"flow_ema_alpha": "not-a-number"}) == {}
|
|
|
|
|
|
def test_the_deprecated_master_switch_still_means_all_off():
|
|
fp = filters_fingerprint(dict(ALL_OFF_FILTERS, enabled=False, flow_ema_alpha=0.3))
|
|
assert fp["measurement"] == {"flow_ema_alpha": 1.0}
|
|
assert fp["readout"]["force_median_window"] == 1
|
|
assert fp["readout"]["force_lowpass_cutoff_hz"] == 0.0
|
|
|
|
|
|
def test_a_measurement_change_warns_and_a_readout_change_only_notes():
|
|
recorded = filters_fingerprint({"preset": "balanced"})
|
|
expected = filters_fingerprint({"preset": "low_noise"})
|
|
warns, notes = filter_mismatches(recorded, expected)
|
|
assert any("not comparable" in w and "flow_ema_alpha" in w for w in warns)
|
|
assert not any("flow_ema_alpha" in n for n in notes)
|
|
assert any("force_lowpass_cutoff_hz" in n for n in notes)
|
|
assert filter_mismatches(recorded, recorded) == ([], [])
|
|
|
|
|
|
def test_check_reports_the_two_groups_apart():
|
|
samples = {
|
|
key: [
|
|
CalSample(weight_n=0.0, raw_mean=0.0, n=10),
|
|
CalSample(weight_n=1.0, raw_mean=1000.0, n=10),
|
|
]
|
|
for key in ("fn", "fx", "fy")
|
|
}
|
|
doc = build(
|
|
samples,
|
|
identity={"sensor_type": "box", "sn": "FP-1"},
|
|
pipeline={"filters": filters_fingerprint({"preset": "balanced"})},
|
|
)
|
|
result = check(doc, pipeline={"filters": filters_fingerprint({"preset": "low_noise"})})
|
|
assert result.ok, "a filter difference must not refuse a calibration"
|
|
assert any("not comparable" in w for w in result.warnings)
|
|
assert any("readout" in n for n in result.notes)
|
|
assert not any("readout" in w for w in result.warnings)
|
|
|
|
|
|
def test_the_fingerprint_survives_a_write_and_read(tmp_path: Path):
|
|
samples = {
|
|
key: [
|
|
CalSample(weight_n=0.0, raw_mean=0.0, n=10),
|
|
CalSample(weight_n=1.0, raw_mean=1000.0, n=10),
|
|
]
|
|
for key in ("fn", "fx", "fy")
|
|
}
|
|
doc = build(
|
|
samples,
|
|
identity={"sensor_type": "box", "sn": "FP-2"},
|
|
pipeline={"filters": filters_fingerprint({"preset": "balanced"})},
|
|
)
|
|
path = tmp_path / "fp.json"
|
|
write(doc, path, overwrite=True)
|
|
raw = json.loads(path.read_text(encoding="utf-8"))
|
|
assert raw["meta"]["filters"]["measurement"] == {"flow_ema_alpha": 0.5}
|
|
# The flat keys stay for readers that look them up by name.
|
|
assert raw["meta"]["flow_ema_alpha"] == 0.5
|
|
|
|
loaded = load(path)
|
|
assert loaded.pipeline["filters"]["measurement"] == {"flow_ema_alpha": 0.5}
|
|
assert loaded.pipeline["filters"]["readout"]["force_median_window"] == 3
|
|
assert loaded.pipeline["filters"]["preset"] == "balanced"
|
|
assert loaded.to_dict()["meta"]["filters"] == doc.to_dict()["meta"]["filters"]
|
|
|
|
|
|
def test_a_legacy_flat_fingerprint_is_still_read_back(tmp_path: Path):
|
|
"""Documents written before the groups existed keep being comparable."""
|
|
samples = {
|
|
key: [
|
|
CalSample(weight_n=0.0, raw_mean=0.0, n=10),
|
|
CalSample(weight_n=1.0, raw_mean=1000.0, n=10),
|
|
]
|
|
for key in ("fn", "fx", "fy")
|
|
}
|
|
doc = build(samples, identity={"sensor_type": "box", "sn": "FP-3"})
|
|
legacy = json.loads(json.dumps(doc.to_dict()))
|
|
legacy["meta"].pop("filters", None)
|
|
legacy["meta"].update(
|
|
{"flow_ema_alpha": 0.3, "force_median_window": 3, "force_lowpass_cutoff_hz": 10.0}
|
|
)
|
|
path = tmp_path / "legacy.json"
|
|
path.write_text(json.dumps(legacy), encoding="utf-8")
|
|
loaded = load(path)
|
|
assert loaded.pipeline["filters"]["measurement"] == {"flow_ema_alpha": 0.3}
|
|
assert loaded.pipeline["filters"]["readout"]["force_median_window"] == 3
|
|
assert loaded.pipeline["filters"]["readout"]["force_lowpass_cutoff_hz"] == 10.0
|
|
|
|
|
|
def test_the_effective_state_beats_the_config_file(tmp_path: Path):
|
|
"""A file states an intent; the device states what it is doing."""
|
|
bundle_json = _box_cal_path().parent / "sensor.json"
|
|
from_file = pipeline_from_sensor_json(bundle_json)
|
|
assert from_file["filters"]["measurement"] == {"flow_ema_alpha": 1.0}
|
|
assert "flow_ema_alpha" not in from_file, "loose top-level filter keys are gone"
|
|
|
|
sensor = _sensor_with_filters(tmp_path, {"preset": "balanced"})
|
|
try:
|
|
from_state = pipeline_from_sensor_json(
|
|
bundle_json, filters_state=sensor.filters_state()
|
|
)
|
|
finally:
|
|
sensor.disconnect()
|
|
assert from_state["filters"]["measurement"] == {"flow_ema_alpha": 0.5}
|
|
assert from_state["filters"]["readout"]["force_median_window"] == 3
|
|
assert from_state["filters"]["source"] == "sensor.filters_state"
|
|
assert from_state["config_ref"] == str(bundle_json)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"preset, alpha, median, cutoff",
|
|
[
|
|
("off", 1.0, 1, 0.0),
|
|
("low_latency", 1.0, 3, 0.0),
|
|
("balanced", 0.5, 3, 0.0),
|
|
("low_noise", 0.3, 3, 10.0),
|
|
],
|
|
)
|
|
def test_every_preset_reports_what_it_does(
|
|
tmp_path: Path, preset: str, alpha: float, median: int, cutoff: float
|
|
):
|
|
sensor = _sensor_with_filters(tmp_path, {"preset": preset})
|
|
try:
|
|
fp = filters_fingerprint(sensor.filters_state())
|
|
finally:
|
|
sensor.disconnect()
|
|
assert fp["measurement"]["flow_ema_alpha"] == pytest.approx(alpha)
|
|
assert fp["readout"]["force_median_window"] == median
|
|
assert fp["readout"]["force_lowpass_cutoff_hz"] == pytest.approx(cutoff)
|