ykcheungab
185 lines
9.0 KiB
Python
185 lines
9.0 KiB
Python
"""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
|