ykcheungab
ee830d8505
把同帧测量、空载归零、模型推断与共用后处理收进一个 SDK 会话，Studio 显示与记录使用同一样本；旧 Sensor 读数接口保持兼容。以 force session 与受影响 SDK 套件、lint 与文档门禁验证，硬件与真实载荷精度未验证。 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CpCfJF6SvonMKmPZhzKZHb
621 lines
25 KiB
Python
621 lines
25 KiB
Python
"""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"]
|