ykcheungab
c1e6d21030
回放按声明的 30 帧加权空载窗口取基线并逐帧保存外推质量，field 模型单独进缓存键，实时入口拒绝。 新增测试覆盖窗口、质量与拒绝路径。全量门禁的 scene3d 两项失败与 studio 访问冲突为既有问题， 已归因并另立 studio-gate-hang。
340 lines
14 KiB
Python
340 lines
14 KiB
Python
"""S4: the field-weighted model in the replay pass, result payload and live entries.
|
|
|
|
The field model is an *experiment*: the SDK computes ``P = A * sum(g * q)`` on the
|
|
HHD grid and subtracts a baseline measured from a declared 30-frame unloaded
|
|
window. The app's part is to declare that window exactly once, keep the numeric
|
|
quality (extrapolated fraction) out of the validity decision, store it per frame,
|
|
and refuse the model at the live entries it was not wired into.
|
|
|
|
The SDK is faked in ``_field_fakes``: these tests describe the app-side contract
|
|
(which calls it makes, in which order, and what it stores), not the SDK's own
|
|
arithmetic, which ``orisys-sdk/tests/test_field_calibration.py`` owns.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
from types import SimpleNamespace
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from _field_fakes import (
|
|
FakeChain,
|
|
FakeFieldSensor,
|
|
FakeReader,
|
|
field_rows,
|
|
install_pass,
|
|
minimal_recording,
|
|
new_pass_state,
|
|
write_model_file,
|
|
)
|
|
from orisys_app_shared.acquisition.finger_worker import FingerAcquisitionWorker
|
|
from orisys_app_shared.acquisition.sdk_worker import (
|
|
SdkAcquisitionWorker,
|
|
live_field_model_rejection,
|
|
read_forces_from_sensor,
|
|
)
|
|
from orisys_app_shared.analysis import model as model_module
|
|
from orisys_app_shared.analysis.cache import AnalysisWriter, ResultCache
|
|
from orisys_app_shared.analysis.model import (
|
|
FIELD_KEY_SUFFIX,
|
|
FIELD_PAYLOAD,
|
|
FIELD_PROFILE,
|
|
FRAME_FLAG_FIELD,
|
|
ForceSample,
|
|
Provenance,
|
|
decode_frame,
|
|
encode_frame,
|
|
frame_payload_bytes,
|
|
result_key,
|
|
)
|
|
from orisys_app_shared.analysis.reader import AnalysisReader
|
|
from orisys_app_shared.analysis.service import (
|
|
AnalysisService,
|
|
ChainSpec,
|
|
calibration_profile,
|
|
)
|
|
from orisys_app_shared.record.playback import RecordingFrame
|
|
|
|
|
|
# ------------------------------------------------------------------- payload
|
|
def test_field_quality_round_trips_in_the_frame_record():
|
|
payload = {
|
|
"frame_id": 7,
|
|
"t_rel_s": 0.25,
|
|
"grid_id": 0,
|
|
"dx": np.zeros(0, dtype=np.float32),
|
|
"dy": np.zeros(0, dtype=np.float32),
|
|
"magnitude": np.zeros(0, dtype=np.float32),
|
|
"active": np.zeros(0, dtype=bool),
|
|
"field_quality": (0.25, 0.9),
|
|
}
|
|
decoded = decode_frame(encode_frame(**payload))
|
|
assert decoded.field_quality == pytest.approx((0.25, 0.9))
|
|
assert decoded.frame_id == 7 and decoded.t_rel_s == pytest.approx(0.25)
|
|
|
|
legacy = decode_frame(encode_frame(**{**payload, "field_quality": None}))
|
|
assert legacy.field_quality is None, "a frame without the flag must read as unknown"
|
|
|
|
assert frame_payload_bytes(0) == frame_payload_bytes(0, with_field_quality=False)
|
|
assert (
|
|
frame_payload_bytes(0, with_field_quality=True)
|
|
== frame_payload_bytes(0) + FIELD_PAYLOAD.size
|
|
)
|
|
assert FRAME_FLAG_FIELD != 0 and FIELD_PROFILE == "hhd_normal_l1_field/1"
|
|
|
|
|
|
def test_a_frame_with_force_and_field_quality_decodes_both():
|
|
sample = ForceSample.from_reading(
|
|
{
|
|
"normal_force": 1.5,
|
|
"normal_force_raw": 100.0,
|
|
"valid": True,
|
|
"force_calibrated": True,
|
|
"calibrated_axes": ("fn",),
|
|
"calibration_status": {"fn": "ok"},
|
|
"calibration_diagnostics": {
|
|
"extrapolated_fraction": 0.3,
|
|
"coverage_fraction": 0.9,
|
|
},
|
|
}
|
|
)
|
|
record = encode_frame(
|
|
frame_id=3,
|
|
t_rel_s=0.0,
|
|
grid_id=0,
|
|
dx=np.zeros(0, dtype=np.float32),
|
|
dy=np.zeros(0, dtype=np.float32),
|
|
magnitude=np.zeros(0, dtype=np.float32),
|
|
active=np.zeros(0, dtype=bool),
|
|
force=sample,
|
|
field_quality=sample.field_quality(),
|
|
)
|
|
decoded = decode_frame(record)
|
|
assert decoded.force is not None and decoded.force.normal == pytest.approx(1.5)
|
|
assert decoded.field_quality == pytest.approx((0.3, 0.9))
|
|
# Quality is evidence about the estimate, not part of the force payload.
|
|
assert decoded.force.diagnostics == {}
|
|
|
|
|
|
def test_the_reader_serves_field_quality_and_none_for_old_frames(tmp_path):
|
|
key = result_key(recording="r", sensor_config="s", calibration="c")
|
|
directory = tmp_path / "result"
|
|
writer = AnalysisWriter(directory, key=key)
|
|
writer.begin()
|
|
grid = writer.grid_keyframe(np.zeros(0, np.int32), np.zeros(0, np.int32))
|
|
for frame_id, quality in ((1, (0.4, 0.8)), (2, None)):
|
|
writer.add_frame(
|
|
encode_frame(
|
|
frame_id=frame_id,
|
|
t_rel_s=frame_id / 30.0,
|
|
grid_id=grid,
|
|
dx=np.zeros(0, dtype=np.float32),
|
|
dy=np.zeros(0, dtype=np.float32),
|
|
magnitude=np.zeros(0, dtype=np.float32),
|
|
active=np.zeros(0, dtype=bool),
|
|
field_quality=quality,
|
|
),
|
|
frame_id=frame_id,
|
|
t_rel_s=frame_id / 30.0,
|
|
)
|
|
writer.commit(provenance=Provenance(), counts={})
|
|
|
|
reader = AnalysisReader(directory)
|
|
try:
|
|
first = reader.frame(1)
|
|
second = reader.frame(2)
|
|
assert first.field_quality == pytest.approx((0.4, 0.8))
|
|
assert second.field_quality is None
|
|
finally:
|
|
reader.close()
|
|
|
|
|
|
# ----------------------------------------------------------------------- key
|
|
def test_field_calibration_adds_its_own_cache_component(tmp_path):
|
|
recording = minimal_recording(tmp_path)
|
|
field_file = write_model_file(tmp_path, "field_gain/1")
|
|
spatial_file = write_model_file(tmp_path, "spatial_gain/1")
|
|
|
|
assert calibration_profile(field_file) == ("field_gain/1", FIELD_PROFILE)
|
|
assert calibration_profile(spatial_file)[1] == "fn_delta_autocal_uv/1"
|
|
assert calibration_profile(None) == ("", "")
|
|
|
|
field_service = AnalysisService(
|
|
recording, chain=ChainSpec("finger", calibration=field_file)
|
|
)
|
|
spatial_service = AnalysisService(
|
|
recording, chain=ChainSpec("finger", calibration=spatial_file)
|
|
)
|
|
field_key = field_service.dependency_report()["key"]
|
|
spatial_key = spatial_service.dependency_report()["key"]
|
|
|
|
assert field_key.components["recording"] == spatial_key.components["recording"]
|
|
assert field_key.components["sensor_config"].endswith(FIELD_KEY_SUFFIX)
|
|
assert FIELD_KEY_SUFFIX not in spatial_key.components["sensor_config"]
|
|
assert field_service.dependency_report()["model_profile"] == FIELD_PROFILE
|
|
assert field_key.digest != spatial_key.digest
|
|
|
|
|
|
# ---------------------------------------------------------------------- pass
|
|
def test_field_pass_declares_one_window_and_does_not_use_contact_or_scalar_zero(
|
|
monkeypatch, tmp_path
|
|
):
|
|
sensor = FakeFieldSensor()
|
|
chain = FakeChain(sensor, frames=35)
|
|
recording = minimal_recording(tmp_path)
|
|
reader = FakeReader(field_rows(35), base=recording)
|
|
processor = AnalysisService(
|
|
recording,
|
|
chain=ChainSpec("finger"),
|
|
store_force=True,
|
|
unloaded_reference_frames=30,
|
|
)
|
|
install_pass(monkeypatch, processor, chain, reader)
|
|
|
|
state = new_pass_state(processor, tmp_path)
|
|
frames = list(processor._pass_frames(state))
|
|
|
|
assert len(frames) == 35
|
|
assert sensor.declared == ["video-unloaded-1"], "the window is declared exactly once"
|
|
assert sensor.set_reference_calls == [], "field models must not get a scalar raw zero"
|
|
# 30 pending frames are zero_unavailable; their raw counts stay the measurement.
|
|
for live in frames[:30]:
|
|
assert dict(live.force.status)["fn"] == "zero_unavailable"
|
|
assert live.force.valid and math.isnan(live.force.normal)
|
|
assert live.force.normal_raw == pytest.approx(100.0)
|
|
assert dict(frames[30].force.status)["fn"] == "ok"
|
|
assert frames[30].force.normal == pytest.approx(1.5)
|
|
# The SDK's weighted baseline, not a median of raw counts.
|
|
assert state.field_baseline == pytest.approx(0.25)
|
|
assert state.field_reference_id == "video-unloaded-1"
|
|
assert state.field_extrapolated_count == 35
|
|
# centroid_found was True throughout: the field model has no contact gate.
|
|
|
|
|
|
def test_field_pass_records_provenance_and_per_frame_quality(monkeypatch, tmp_path):
|
|
sensor = FakeFieldSensor()
|
|
chain = FakeChain(sensor, frames=35)
|
|
recording = minimal_recording(tmp_path)
|
|
reader = FakeReader(field_rows(35), base=recording)
|
|
field_file = write_model_file(tmp_path, "field_gain/1")
|
|
cache = ResultCache(tmp_path / "cache")
|
|
processor = AnalysisService(
|
|
recording,
|
|
chain=ChainSpec("finger", calibration=field_file),
|
|
cache=cache,
|
|
store_force=True,
|
|
unloaded_reference_frames=30,
|
|
)
|
|
install_pass(monkeypatch, processor, chain, reader)
|
|
|
|
outcome = processor.run()
|
|
assert outcome.complete, outcome.reason
|
|
|
|
reader_out = AnalysisReader(outcome.directory)
|
|
try:
|
|
provenance = reader_out.provenance
|
|
field = dict(provenance.field)
|
|
assert field["profile"] == FIELD_PROFILE
|
|
assert field["strategy"] == model_module.FIELD_STRATEGY
|
|
assert field["baseline_N"] == pytest.approx(0.25)
|
|
assert field["reference_id"] == "video-unloaded-1"
|
|
assert field["frames"] == 35
|
|
expected_mean = (29 * 0.4 + 0.3 + 5 * 0.2) / 35
|
|
assert field["extrapolated_mean"] == pytest.approx(expected_mean, abs=1e-6)
|
|
assert field["extrapolated_max"] == pytest.approx(0.4)
|
|
assert field["hhd"]["downsample"] == 2
|
|
assert field["hhd"]["grid_shape"] == [2, 2]
|
|
assert reader_out.counts["field_quality_frames"] == 35
|
|
# The same window's frames carry quality in the record; frame ids are the
|
|
# recording's own, not the SDK's diagnostic id.
|
|
first = reader_out.frame(1)
|
|
last = reader_out.frame(35)
|
|
assert first.field_quality == pytest.approx((0.4, 0.9))
|
|
assert last.field_quality == pytest.approx((0.2, 0.97))
|
|
assert reader_out.key_components["sensor_config"].endswith(FIELD_KEY_SUFFIX)
|
|
finally:
|
|
reader_out.close()
|
|
|
|
|
|
def test_field_model_rejects_a_multi_segment_recording(monkeypatch, tmp_path):
|
|
field_file = write_model_file(tmp_path, "field_gain/1")
|
|
recording = minimal_recording(tmp_path, segments=2)
|
|
rows = field_rows(2) + [
|
|
RecordingFrame(i + 1, (i + 2) / 30.0, "", 2, i + 1, True) for i in range(2)
|
|
]
|
|
sensor = FakeFieldSensor()
|
|
chain = FakeChain(sensor, frames=4)
|
|
reader = FakeReader(rows, segments=2, base=recording)
|
|
cache = ResultCache(tmp_path / "cache")
|
|
processor = AnalysisService(
|
|
recording,
|
|
chain=ChainSpec("finger", calibration=field_file),
|
|
cache=cache,
|
|
store_force=True,
|
|
unloaded_reference_frames=30,
|
|
)
|
|
install_pass(monkeypatch, processor, chain, reader)
|
|
|
|
with pytest.raises(ValueError, match="single-segment"):
|
|
processor.run()
|
|
assert not list((tmp_path / "cache").glob("*/meta.json"))
|
|
|
|
|
|
# --------------------------------------------------------------------- live
|
|
def test_live_entries_refuse_a_field_model_file(tmp_path):
|
|
field_file = write_model_file(tmp_path, "field_gain/1")
|
|
rejection = live_field_model_rejection(field_file)
|
|
assert FIELD_PROFILE in rejection and "replay" in rejection.lower()
|
|
|
|
box = SdkAcquisitionWorker()
|
|
with pytest.raises(RuntimeError, match="hhd_normal_l1_field/1"):
|
|
box.start(0, config_ref=tmp_path / "unused.json", cal_path=field_file)
|
|
|
|
finger = FingerAcquisitionWorker()
|
|
with pytest.raises(RuntimeError, match="hhd_normal_l1_field/1"):
|
|
finger.start(0, config_ref=tmp_path / "unused.json", cal_path=field_file)
|
|
|
|
# A hot swap is refused before the device is touched too (the sensor double only
|
|
# has to exist: the refusal comes before any SDK call on it).
|
|
box._pending_calibration = str(field_file)
|
|
with pytest.raises(RuntimeError, match="hhd_normal_l1_field/1"):
|
|
box._consume_pending_calibration(sensor=SimpleNamespace())
|
|
finger._pending_calibration = str(field_file)
|
|
with pytest.raises(RuntimeError, match="hhd_normal_l1_field/1"):
|
|
finger._consume_pending_calibration(sensor=SimpleNamespace())
|
|
|
|
|
|
def test_a_scalar_calibration_is_still_accepted_by_the_live_entry(tmp_path):
|
|
spatial_file = write_model_file(tmp_path, "spatial_gain/1")
|
|
assert live_field_model_rejection(spatial_file) == ""
|
|
assert live_field_model_rejection(None) == ""
|
|
assert live_field_model_rejection(tmp_path / "missing.json") == ""
|
|
|
|
|
|
def test_the_field_reading_carries_quality_without_touching_validity():
|
|
sensor = FakeFieldSensor()
|
|
sensor.declare_unloaded_reference(reference_id="unloaded-1")
|
|
sensor.step()
|
|
reading = read_forces_from_sensor(sensor)
|
|
assert reading["calibration_status"] == {"fn": "zero_unavailable"}
|
|
assert reading["calibration_diagnostics"]["pending_reference_frames"] == 1
|
|
sample = ForceSample.from_reading(reading)
|
|
assert sample.valid, "a missing baseline is not an invalid raw measurement"
|
|
assert math.isnan(sample.normal) and sample.normal_raw == pytest.approx(100.0)
|
|
# Quality is available while the baseline is pending: it describes the support of
|
|
# this frame's integral, not whether the baseline is ready.
|
|
assert sample.field_quality() == pytest.approx((0.4, 0.9))
|
|
for _ in range(30):
|
|
sensor.step()
|
|
ready = read_forces_from_sensor(sensor)
|
|
assert ready["calibration_status"] == {"fn": "ok"}
|
|
ready_sample = ForceSample.from_reading(ready)
|
|
assert ready_sample.normal == pytest.approx(1.5)
|
|
assert ready_sample.field_quality() == pytest.approx((0.2, 0.97))
|
|
|
|
# A reading with no diagnostics at all is *unknown*, never 0 extrapolation.
|
|
reading.pop("calibration_diagnostics")
|
|
assert ForceSample.from_reading(reading).field_quality() is None
|