ykcheungab
246 lines
9.0 KiB
Python
246 lines
9.0 KiB
Python
"""Calibration diagnostics must describe the model that is exported.
|
|
|
|
The review found the panel reporting a private least-squares fit while the
|
|
exported file carried a piecewise-linear curve, validation points leaking into
|
|
the build, and repeated loads being overwritten so only the last one reached the
|
|
model.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from orisys_app_shared.force.calibration_io import (
|
|
CalSample,
|
|
CalSessionStore,
|
|
build_export_model,
|
|
build_force_calibration_v2,
|
|
merge_samples,
|
|
predict_channel,
|
|
write_force_calibration,
|
|
)
|
|
from orisys_app_shared.force.calibration_recorder import CalibrationRecorder
|
|
from orisys_app_shared.force.session import SessionSample
|
|
|
|
SCRATCH = Path(__file__).resolve().parent / "_scratch"
|
|
|
|
|
|
def _root() -> Path:
|
|
shutil.rmtree(SCRATCH, ignore_errors=True)
|
|
SCRATCH.mkdir(parents=True, exist_ok=True)
|
|
return SCRATCH
|
|
|
|
|
|
def _point(recorder, load: float, raw: float, *, channel: str = "fn"):
|
|
recorder.start(channel=channel, reference_load_n=load, duration_s=0.05)
|
|
for index in range(8):
|
|
recorder.feed(
|
|
SessionSample(
|
|
index,
|
|
index * 0.01,
|
|
fn_raw=raw,
|
|
fx_raw=raw,
|
|
fy_raw=raw,
|
|
fn_n=raw / 10.0,
|
|
fx_n=raw / 10.0,
|
|
fy_n=raw / 10.0,
|
|
)
|
|
)
|
|
return recorder.finish_now()
|
|
|
|
|
|
def test_the_diagnosed_model_is_the_exported_model():
|
|
recorder = CalibrationRecorder()
|
|
for load, raw in ((1.0, 100.0), (2.0, 400.0), (3.0, 300.6), (5.0, 500.9)):
|
|
_point(recorder, load, raw)
|
|
|
|
fit = recorder.fit("fn")
|
|
assert fit.usable
|
|
assert fit.method == "piecewise_linear"
|
|
assert fit.breakpoints == 4
|
|
# A piecewise curve passes through the points it was built from.
|
|
assert fit.max_residual_n == pytest.approx(0.0, abs=1e-9)
|
|
|
|
|
|
def test_validation_points_are_held_out_and_measured():
|
|
recorder = CalibrationRecorder()
|
|
points = [
|
|
_point(recorder, 1.0, 100.0),
|
|
_point(recorder, 2.0, 400.0),
|
|
_point(recorder, 3.0, 300.6),
|
|
_point(recorder, 5.0, 500.9),
|
|
]
|
|
recorder.set_validation(points[1], True)
|
|
fit = recorder.fit("fn")
|
|
assert fit.n_points == 3, "a validation point entered the build"
|
|
assert fit.validation_points == 1
|
|
assert fit.validation_error_n > 0.05, "the off-trend point was not measured"
|
|
assert fit.breakpoints == 3
|
|
assert points[1].residual_n != points[1].residual_n # NaN: not in the fit
|
|
|
|
|
|
def test_excluded_points_leave_the_build():
|
|
recorder = CalibrationRecorder()
|
|
points = [_point(recorder, 1.0, 100.0), _point(recorder, 5.0, 500.0)]
|
|
outlier = _point(recorder, 3.0, 900.0)
|
|
recorder.set_excluded(outlier, True)
|
|
|
|
fit = recorder.fit("fn")
|
|
assert fit.n_points == 2
|
|
store = CalSessionStore()
|
|
for point in recorder.all_points():
|
|
if point.excluded or point.is_validation:
|
|
continue
|
|
store.aggregate(
|
|
point.channel,
|
|
CalSample(
|
|
weight_n=float(point.reference_load_n),
|
|
raw_mean=float(point.loaded_stats().mean),
|
|
n=int(point.loaded_stats().n),
|
|
),
|
|
)
|
|
doc, _warnings = build_force_calibration_v2(store)
|
|
points_in_doc = doc["channels"]["normal"]["points"]
|
|
assert [point[0] for point in points_in_doc] == [100.0, 500.0], (
|
|
"the excluded point reached the exported coefficients"
|
|
)
|
|
assert points # silence linters about the unused list
|
|
|
|
|
|
def test_validation_points_never_reach_the_exported_coefficients():
|
|
recorder = CalibrationRecorder()
|
|
_point(recorder, 1.0, 100.0)
|
|
_point(recorder, 5.0, 500.0)
|
|
held_out = _point(recorder, 3.0, 300.0)
|
|
recorder.set_validation(held_out, True)
|
|
|
|
store = CalSessionStore()
|
|
for point in recorder.all_points():
|
|
if point.excluded or point.is_validation:
|
|
continue
|
|
store.aggregate(
|
|
point.channel,
|
|
CalSample(
|
|
weight_n=float(point.reference_load_n),
|
|
raw_mean=float(point.loaded_stats().mean),
|
|
n=int(point.loaded_stats().n),
|
|
),
|
|
)
|
|
doc, _ = build_force_calibration_v2(store)
|
|
raws = [point[0] for point in doc["channels"]["normal"]["points"]]
|
|
assert 300.0 not in raws, "a validation point was used to build the model"
|
|
|
|
|
|
def test_repeated_loads_are_merged_not_overwritten():
|
|
"""Two samples of one load carry scatter the model should see."""
|
|
first = CalSample(weight_n=2.0, raw_mean=100.0, n=10, std=1.0)
|
|
second = CalSample(weight_n=2.0, raw_mean=110.0, n=30, std=2.0)
|
|
merged = merge_samples(first, second)
|
|
assert merged.n == 40
|
|
assert merged.raw_mean == pytest.approx((100.0 * 10 + 110.0 * 30) / 40)
|
|
assert merged.std > 2.0, "the scatter between the two samples was lost"
|
|
|
|
store = CalSessionStore()
|
|
store.aggregate("fn", first)
|
|
store.aggregate("fn", second)
|
|
assert len(store.channel("fn")) == 1
|
|
assert store.channel("fn")[0].n == 40
|
|
|
|
|
|
def test_predict_channel_applies_the_exported_curve():
|
|
store = CalSessionStore()
|
|
store.aggregate("fn", CalSample(weight_n=0.0, raw_mean=0.0, n=10))
|
|
store.aggregate("fn", CalSample(weight_n=10.0, raw_mean=100.0, n=10))
|
|
model = build_export_model({key: store.channel(key) for key in ("fn", "fx", "fy")})
|
|
assert predict_channel(model, "fn", 50.0) == pytest.approx(5.0, abs=1e-6)
|
|
|
|
|
|
def test_a_channel_with_one_point_is_not_exported():
|
|
store = CalSessionStore()
|
|
store.aggregate("fn", CalSample(weight_n=1.0, raw_mean=10.0, n=10))
|
|
with pytest.raises(ValueError):
|
|
build_force_calibration_v2(store)
|
|
|
|
|
|
# ---------------------------------------------------------------- one preparation
|
|
|
|
|
|
def _session_with_duplicates_and_validation():
|
|
recorder = CalibrationRecorder()
|
|
points = [
|
|
_point(recorder, 1.0, 100.0),
|
|
_point(recorder, 2.0, 200.0),
|
|
# A deliberate repeat of load 2.0 at a different raw level.
|
|
_point(recorder, 2.0, 240.0),
|
|
_point(recorder, 4.0, 400.0),
|
|
_point(recorder, 3.0, 290.0),
|
|
]
|
|
recorder.set_validation(points[4], True)
|
|
return recorder, points
|
|
|
|
|
|
def test_duplicate_loads_are_merged_before_the_model_is_built():
|
|
_recorder, _points = _session_with_duplicates_and_validation()
|
|
prepared = _recorder.prepared()
|
|
assert prepared.samples["fn"], "no training samples"
|
|
loads = [sample.weight_n for sample in prepared.samples["fn"]]
|
|
assert loads == sorted(set(loads)), f"duplicate loads were not merged: {loads}"
|
|
assert len(loads) == 3
|
|
merged = next(sample for sample in prepared.samples["fn"] if sample.weight_n == 2.0)
|
|
assert merged.n == 16, "the repeat's samples were not combined"
|
|
assert merged.raw_mean == pytest.approx((200.0 + 240.0) / 2.0)
|
|
assert prepared.points_in_fit("fn") == 3
|
|
|
|
|
|
def test_the_exported_file_predicts_exactly_what_the_panel_shows():
|
|
"""Read the written document back and compare it with the diagnostics.
|
|
|
|
The panel's numbers must describe the file that is actually written, so the
|
|
test loads the file with the SDK (not with the prepared object) and predicts
|
|
every training and validation point from it.
|
|
"""
|
|
import orisys.calibration as sdk_calibration
|
|
|
|
recorder, points = _session_with_duplicates_and_validation()
|
|
prepared = recorder.prepared(meta={"sensor_type": "box", "sn": "SN9"})
|
|
assert prepared.ok
|
|
|
|
document = prepared.to_document()
|
|
path = _root() / "force_calibration.json"
|
|
write_force_calibration(document, path)
|
|
from_file = sdk_calibration.load(path, strict=True)
|
|
|
|
for point in points:
|
|
raw = float(point.loaded_stats().mean)
|
|
predicted = float(from_file.get("fn").convert(raw))
|
|
value = prepared.predict("fn", raw)
|
|
assert predicted == pytest.approx(value, abs=1e-9)
|
|
if point.is_validation:
|
|
assert prepared.validation_errors[id(point)] == pytest.approx(
|
|
abs(predicted - point.reference_load_n), abs=1e-9
|
|
)
|
|
else:
|
|
assert prepared.residuals[id(point)] == pytest.approx(
|
|
predicted - point.reference_load_n, abs=1e-9
|
|
)
|
|
|
|
# The validation point is measured, never built in.
|
|
assert prepared.validation_points("fn") == 1
|
|
assert prepared.max_validation_error("fn") > 0.0
|
|
exported_raws = sorted(point[0] for point in document["channels"]["normal"]["points"])
|
|
assert exported_raws == [100.0, 220.0, 400.0], exported_raws
|
|
|
|
|
|
def test_the_recorder_fit_and_the_preparation_agree():
|
|
recorder, _points = _session_with_duplicates_and_validation()
|
|
prepared = recorder.prepared()
|
|
fit = recorder.fit("fn", prepared=prepared)
|
|
assert fit.n_points == prepared.points_in_fit("fn")
|
|
assert fit.breakpoints == prepared.breakpoints("fn")
|
|
assert fit.max_residual_n == pytest.approx(prepared.max_residual("fn"))
|
|
assert fit.validation_error_n == pytest.approx(prepared.max_validation_error("fn"))
|
|
# ...and a fit that prepares on its own gets the same numbers.
|
|
assert recorder.fit("fn").as_dict() == fit.as_dict()
|