"""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 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.), 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.} class FieldRecordingModel: schema_version = 3 channels = {"fn": "N"} input_profile = "hhd_normal_l1_field/1" def __init__(self): self.calls = [] self.resets = [] self.begin_calls = [] self.coordinate_map = np.zeros((2, 2, 2), dtype=np.float32) self.coordinate_valid_mask = np.ones((2, 2), dtype=bool) def reset(self, reference_id): self.resets.append(reference_id) def begin_reference(self, reference_id): self.begin_calls.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"] if "field" not in features: return SimpleNamespace(values_N={"fn": None}, status={"fn": "invalid_input"}, diagnostics={}) return SimpleNamespace(values_N={"fn": 1.0}, status={"fn": "ok"}, diagnostics={}) def test_hhd_field_model_infers_from_same_frame_field_without_contact(): sensor = make_sensor() sensor.calibration = FieldRecordingModel() sensor._calibration_context = {"signature": "actual"} sensor.cuda = sensor.verbose = False sensor.valid_mask = sensor.valid_mask_hhd = None sensor._hhd_downsample = 1 sensor._hhd_grid_shape = sensor._hhd_ref_shape = (2, 2) sensor._model_frame_id = 0 field = np.zeros((2, 2, 2), dtype=np.float32) calls = [] def get_field(valid_mask=None): calls.append(valid_mask) return {"d": field, "valid_mask": np.ones((2, 2), dtype=bool), "area_scale": 1.0, "grid_shape": (2, 2), "reference_shape": (2, 2)} sensor._Sensor__hhd = SimpleNamespace( get_force=lambda *args, **kwargs: (None, (10., 0., 0., 0.)), get_normal=lambda: field, get_shear=lambda: field, get_field=get_field, ) sensor.compute_hhd(np.zeros((2, 2, 2), dtype=np.float32)) features = sensor.calibration.calls[-1]["features"] assert features["field"]["grid_shape"] == (2, 2) assert features["output_valid_mask"].shape == (2, 2) assert "contact_found" not in features assert sensor.calibration_status == {"fn": "ok"} assert sensor.fnormal == pytest.approx(1.0) assert calls == [None] def test_weighted_unloaded_reference_is_declared_not_a_scalar_zero(): sensor = make_sensor() sensor.calibration = FieldRecordingModel() sensor._calibration_context = {"signature": "actual"} sensor.declare_unloaded_reference(reference_id="video-unloaded-100") assert sensor.calibration.begin_calls == ["video-unloaded-100"] assert sensor._calibration_reference_id == "video-unloaded-100" with pytest.raises(CalibrationError, match="scalar zero"): sensor.set_calibration_reference({"fn": 0.}, reference_id="bad") with pytest.raises(CalibrationError, match="hhd_normal_l1_field"): make_sensor().declare_unloaded_reference(reference_id="scalar-model") def test_finger_runtime_binds_field_identity_and_forwards_the_window(monkeypatch, tmp_path): import orisys.calibration_context as context_mod from orisys.field_calibration import FIELD_IDENTITY from orisys.finger.runtime import FingerRuntime sensor = make_sensor() sensor.calibration = FieldRecordingModel() sensor._calibration_context = {"signature": "actual"} sensor.device_id = None sensor.mirror = False 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 = FieldRecordingModel() selected.pipeline = {"coordinates": {"frame": "autocal_flat"}} runtime.set_calibration(selected) assert contexts[0]["field"] == FIELD_IDENTITY runtime.declare_unloaded_reference("runtime-window") assert sensor.calibration.begin_calls == ["runtime-window"] # ------------------------------------------------- session measurement seam def test_measurement_session_skips_old_inference_and_readout(): """The new entry publishes same-frame raw only: no model call, no readout.""" sensor = make_sensor() sensor.cuda = sensor.verbose = False sensor.valid_mask = sensor.valid_mask_hhd = None sensor._hhd_downsample = 1 sensor._hhd_grid_shape = sensor._hhd_ref_shape = (2, 2) sensor._model_frame_id = 0 field = np.zeros((2, 2, 2), dtype=np.float32) sensor._Sensor__hhd = SimpleNamespace( get_force=lambda *args, **kwargs: (None, (10., 0., 0., 0.)), get_normal=lambda: field, get_shear=lambda: field, ) sensor.enable_measurement_session() assert sensor.measurement_session is True sensor.compute_hhd(field) assert sensor.calibration.calls == [], "the session owns the one model call" assert sensor._force_readout.filtered_raw is None assert sensor._force_raw == (10., 0., 0.) assert np.isnan(sensor.fnormal) assert sensor.calibration_status == {} # Contact statistics stay available, still without a model inference. sensor._Sensor__contact3d = SimpleNamespace( compute_foe=lambda *args, **kwargs: { "centroid": (0, 1), "centroid_found": True, "divergence_map": np.zeros((2, 2)), } ) sensor.compute_contact() assert sensor.calibration.calls == [] assert sensor.centroid_found is True sensor.disable_measurement_session() assert sensor.measurement_session is False def test_reset_tracker_confirms_without_touching_the_calibration_reference(): sensor = make_sensor() resets = [] sensor._Sensor__tracker = SimpleNamespace(reset=lambda: resets.append("tracker")) sensor.set_calibration_reference({"fn": 10.}, reference_id="reference") sensor._force_unfiltered = (1., 2., 3.) sensor.compute_deformation_valid = True sensor.reset_tracker() assert resets == ["tracker"] assert sensor._force_unfiltered is None assert sensor._calibration_reference == {"fn": 10.} assert sensor._calibration_reference_id == "reference" assert sensor.compute_deformation_valid is False sensor.enable_measurement_session() sensor.reset_tracker() assert sensor.measurement_session is True, "the seam survives a frame-boundary reset" def test_measure_current_frame_returns_this_frames_features_once(): sensor = make_sensor() sensor.cuda = sensor.verbose = False sensor.valid_mask = np.ones((2, 2), dtype=bool) sensor.valid_mask_hhd = np.ones((2, 2), dtype=bool) sensor._hhd_mask_f32 = None sensor._hhd_downsample = 1 sensor._hhd_grid_shape = sensor._hhd_ref_shape = (2, 2) sensor._model_frame_id = 0 field = np.zeros((2, 2, 2), dtype=np.float32) sensor._Sensor__hhd = SimpleNamespace( get_force=lambda *args, **kwargs: (None, (10., 20., 30., 0.)), get_normal=lambda: field, get_shear=lambda: field, get_field=lambda valid_mask=None: { "d": field, "valid_mask": np.ones((2, 2), dtype=bool), "area_scale": 4.0, "grid_shape": (2, 2), "reference_shape": (2, 2), }, ) sensor._Sensor__contact3d = SimpleNamespace( compute_foe=lambda *args, **kwargs: { "centroid": (0.4, 1.2), "centroid_found": True, "divergence_map": np.zeros((2, 2)), } ) flow_calls = [] sensor.compute_flow = lambda image=None: ( flow_calls.append(image), setattr(sensor, "flow", field) ) sensor.enable_measurement_session() sensor.img = np.zeros((2, 2), dtype=np.uint8) measurement = sensor.measure_current_frame( frame_id=5, want_field=True, want_contact=True ) assert flow_calls == [sensor.img] assert measurement.raw == (10., 20., 30.) assert measurement.raw_valid == (True, True, True) assert measurement.field["grid_shape"] == (2, 2) assert measurement.output_valid_mask.shape == (2, 2) assert measurement.contact_found is True assert measurement.centroid == (0.4, 1.2) assert measurement.single_contact is True def test_read_measurement_frame_prepares_the_source_frame_once(): sensor = make_sensor() sensor.cuda = sensor.verbose = False sensor.valid_mask = sensor.valid_mask_hhd = None sensor._hhd_downsample = 1 sensor._hhd_grid_shape = sensor._hhd_ref_shape = (2, 2) sensor._model_frame_id = 0 field = np.zeros((2, 2, 2), dtype=np.float32) sensor._Sensor__hhd = SimpleNamespace( get_force=lambda *args, **kwargs: (None, (1., 2., 3., 0.)), get_normal=lambda: field, get_shear=lambda: field, ) prepared = [] sensor.get_img = lambda frame=None: (prepared.append(frame), setattr(sensor, "img", frame)) sensor.compute_flow = lambda image=None: setattr(sensor, "flow", field) sensor.enable_measurement_session() frame = np.ones((2, 2, 3), dtype=np.uint8) measurement = sensor.read_measurement_frame(frame, frame_id=7) assert prepared == [frame] assert measurement.raw == (1., 2., 3.) assert measurement.field is None, "no field unless the model asked for it" assert measurement.contact_found is None, "no contact unless it was requested" def test_finger_runtime_session_seam_delegates_to_its_sensor(): from orisys.finger.runtime import FingerRuntime events = [] class FakeSensor: def enable_measurement_session(self): events.append("enable") def disable_measurement_session(self): events.append("disable") def reset_tracker(self): events.append("reset") def measure_current_frame(self, *, frame_id, want_field=False, want_contact=False): events.append(("measure", frame_id, want_field, want_contact)) return "measurement" runtime = object.__new__(FingerRuntime) runtime.tactile = SimpleNamespace(raw_sensor=FakeSensor()) processed = [] runtime.process_frame = lambda frame, color=False: processed.append((frame, color)) frame = np.zeros((2, 2, 3), dtype=np.uint8) result = runtime.read_measurement_frame( frame, frame_id=9, want_field=True, want_contact=False ) assert processed == [(frame, False)] assert result == "measurement" assert events[-1] == ("measure", 9, True, False) runtime.enable_measurement_session() runtime.reset_tracker() runtime.disable_measurement_session() assert events[1:] == ["enable", "reset", "disable"]