"""Model validity, explicit unloaded references and portable schema 3 snapshots.""" import json import math from types import SimpleNamespace import numpy as np import pytest from orisys import calibration from orisys.finger import TactileProcessor from orisys_app_shared.acquisition.finger_worker import FingerAcquisitionWorker from orisys_app_shared.acquisition.sdk_worker import ( loaded_calibration_version, read_forces_from_sensor, ) from orisys_app_shared.force.units import select_display_forces from orisys_app_shared.record.calibration_snapshot import store_calibration_snapshot from orisys_app_shared.record.sample import to_record_sample def model_file(tmp_path): import hashlib attachment = tmp_path / "coordinates.npz" np.savez(attachment, uv_by_output_px=np.zeros((2, 2, 2)), valid_mask=np.ones((2, 2), bool)) digest = hashlib.sha256(attachment.read_bytes()).hexdigest() data = { "schema_version": 3, "unit": "N", "identity": {"sensor_type": "finger", "sn": "test"}, "pipeline": {"feature_signature_sha256": "a" * 64, "geometry_sha256": "b" * 64, "stitch_resolution": [2, 2], "dis_mode": 1, "hhd_downsample": 3, "source_scale": .42, "filters": {"measurement": {"flow_ema_alpha": 1.0}}, "coordinates": {"frame": "autocal_flat", "unit": "px", "origin_xy": [0, 0], "map_sha256": digest}}, "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]}, "artifacts": [{"role": "coordinate_map", "path": attachment.name, "sha256": digest}]}} path = tmp_path / "model.json" path.write_text(json.dumps(data)) return path def test_recorded_model_remains_loadable_without_original_resources(tmp_path): source = model_file(tmp_path) original = calibration.load(source, strict=True) recorded = tmp_path / "recording" relative = store_calibration_snapshot(recorded, source, "version") source.unlink() (tmp_path / "coordinates.npz").unlink() reloaded = calibration.load(recorded / relative, strict=True) assert reloaded.fingerprint == original.fingerprint prediction = reloaded.predict(1, 0, dict(delta_raw=1000, u=2.5, v=2.5, zero_valid=True, contact_found=True, mapping_valid=True), "z") assert prediction.values_N["fn"] == pytest.approx(1.75) def test_snapshot_keeps_its_axes_contact_and_status_after_model_swap(): from orisys.pipeline import FrameSnapshot from orisys_app_shared.acquisition.models import from_snapshot snapshot = FrameSnapshot(seq=0, timestamp=1., age_ms=0., fps=30., unit="N", normal_force=1., shear_x=20., shear_y=30., normal_force_raw=1000., shear_x_raw=20., shear_y_raw=30., torque=0., produce_ms=0., flow_ms=0., hhd_ms=0., force_units=("N", "raw", "raw"), calibration_status={"fn": "ok"}, contact_found=True, centroid=np.array([2., 3.])) next_sensor = SimpleNamespace(calibration=None, force_calibrated=False, centroid_found=False) packet = from_snapshot(snapshot, next_sensor, contact_analysed=True, contact_found=False) assert packet.calibrated_axes == ("fn",) assert packet.calibration_status == {"fn": "ok"} assert packet.contact.found def test_serial_read_unwraps_tactile_and_preserves_invalid_model_status(tmp_path): model = calibration.load(model_file(tmp_path), strict=True) info = SimpleNamespace(FNORMAL=0, FSHEARX=1, FSHEARY=2, FNORMAL_RAW=3, FSHEARX_RAW=4, FSHEARY_RAW=5) values = [math.nan, 2., 3., 2000., 2., 3.] sensor = SimpleNamespace(info=info, calibration=model, force_calibrated=True, calibration_status={"fn": "zero_unavailable"}, read_info=lambda *keys: [values[key] for key in keys]) tactile = TactileProcessor(sensor) reading = read_forces_from_sensor(tactile) assert reading["calibrated_axes"] == ("fn",) assert reading["calibration_status"] == {"fn": "zero_unavailable"} assert loaded_calibration_version(tactile) == model.fingerprint record = to_record_sample(SimpleNamespace(**reading)) assert record.fn_raw == 2000 and math.isnan(record.fn_n) assert record.calibration_status == {"fn": "zero_unavailable"} displayed = select_display_forces(units_pref="newton", sample=record) assert displayed.unit == "N" and not any(displayed.available) assert not displayed.using_raw_fallback def test_zero_is_explicit_complete_and_cancelled_on_contact(): applied = [] sensor = SimpleNamespace(calibration=SimpleNamespace(schema_version=3, channels={"fn": None}), centroid_found=False, set_calibration_reference=lambda raw, **kw: applied.append((raw, kw))) runtime = SimpleNamespace(tactile=SimpleNamespace(raw_sensor=sensor), set_calibration_reference=sensor.set_calibration_reference) worker = FingerAcquisitionWorker() worker.request_calibration_zero() assert applied == [] for index in range(29): worker._collect_zero(runtime, (index, 0, 0), index) assert applied == [({}, {"reference_id": "zero-measurement-pending"})] worker._collect_zero(runtime, (29, 0, 0), 29) assert applied[-1][0] == {"fn": 14.5} assert worker._zero_reference["samples"] == 30 worker.request_calibration_zero() sensor.centroid_found = True with pytest.raises(ValueError, match="cancelled"): worker._collect_zero(runtime, (1000, 0, 0), 30) assert applied[-1][0] == {} assert worker._zero_reference == {} def test_resume_cancels_partial_zero_because_tracker_is_rebased(): worker = FingerAcquisitionWorker() worker._zero_samples = [(1., 2., 3.)] worker._pipe = SimpleNamespace(resume=lambda **kwargs: None) worker.resume() assert worker._cancel_zero def test_hot_swap_uses_runtime_and_confirmed_serial_then_clears_zero(tmp_path): path = model_file(tmp_path) calls = [] runtime = SimpleNamespace(set_calibration=lambda path, **kw: calls.append((path, kw))) worker = FingerAcquisitionWorker() worker._zero_reference = {"raw": {"fn": 123}} worker.set_pending_calibration(path, device_id="test") assert calls == [] worker._consume_pending_calibration(runtime, 4, .2) assert calls == [(str(path), {"device_id": "test"})] assert worker._zero_reference == {} @pytest.mark.parametrize("frame_count,contact", [(31, False), (29, False), (31, True)]) def test_video_unloaded_reference_is_explicit_and_requires_the_whole_interval(monkeypatch, tmp_path, frame_count, contact): from orisys_app_shared.acquisition import sdk_worker from orisys_app_shared.analysis import service from orisys_app_shared.analysis.model import Provenance from orisys_app_shared.record.playback import RecordingFrame applied = [] sensor = SimpleNamespace(calibration=SimpleNamespace(schema_version=3, channels={"fn": None}), info=SimpleNamespace(FNORMAL_RAW=0, FSHEARX_RAW=1, FSHEARY_RAW=2), read_info=lambda *keys: (100., 2., 3.), centroid_found=contact, set_calibration_reference=lambda raw, **kw: applied.append(raw), flow=np.zeros((4, 4, 2)), valid_mask=None) class Chain: def __init__(self, *args): self.sensor = sensor self.open_seconds = 0. self.debug_files = [] self.index = 0 def __enter__(self): return self def __exit__(self, *args): return False def read(self): self.index += 1 return (True, np.zeros((4, 4))) if self.index <= frame_count else (False, None) def feed(self, frame): return frame monkeypatch.setattr(service, "_OpenChain", Chain) monkeypatch.setattr(sdk_worker, "process_chain", lambda sensor: (None, "")) processor = service.AnalysisService(tmp_path, chain=service.ChainSpec("finger"), store_force=False) reader = SimpleNamespace(session=lambda: {"synthetic": {"unloaded_reference_frames": 30}}, segment_path=lambda number: tmp_path / "clip.mp4") monkeypatch.setattr(processor, "reader", lambda: reader) monkeypatch.setattr(processor, "_segment_offset", lambda *args: (0, True)) rows = [RecordingFrame(i + 1, i / 30, "", 1, i + 1, True) for i in range(frame_count)] state = service._PassState(Provenance(), [1], {1: rows}, tmp_path) if frame_count < 30 or contact: with pytest.raises(ValueError): list(processor._pass_frames(state)) assert applied == [] else: assert len(list(processor._pass_frames(state))) == frame_count assert applied == [{"fn": 100.}] assert state.provenance.counts["unloaded_reference_frames"] == 30