diff --git a/apps/studio/docs/replay-analysis.md b/apps/studio/docs/replay-analysis.md index 24e1b7b..3c6a314 100644 --- a/apps/studio/docs/replay-analysis.md +++ b/apps/studio/docs/replay-analysis.md @@ -38,10 +38,9 @@ zlib level 6。 `(max − min) / 510`,实测 6.6e-5(该帧峰值 0.034)。这不到深度视图所用 256 级色标的一步,所以 重建出来的帧**不可能被染成与原帧不同的颜色**。 -稀疏存储实测过并否决:每像素 4 B 索引在最坏帧上要 910 B,而整幅量化平面只要 486 B —— 一片几乎 -全零的字节平面在 zlib 下几乎免费,像素索引不是。限制在岛内也实测过并否决:418 B 对 486 B, -在 4,048 B 的记录里省 5.2 B/帧,代价是重建要依赖有效性掩码 keyframe(box 链路根本没有), -而且一旦有深度值落在岛外就会**静默丢掉一个测量**。 +稀疏存储实测后否决:最坏帧每像素 4 B 索引要 910 B,整幅量化平面只要 486 B —— 几乎全零的 +字节平面在 zlib 下几乎免费。限制在岛内也否决:418 B 对 486 B,每帧省 5.2 B,却要依赖有效性 +掩码 keyframe(box 链路没有),且岛外深度值会静默丢测量。 `provenance` 随结果存:算法版本、传感器配置及其内容哈希、标定 id、单位、参考帧 id、首个输出 帧 id、解码偏移、逐通道的"重算/记录"标记(`flow`、`depth` 重算;`forces` 见第 1.1 节)、 @@ -90,7 +89,8 @@ box 链路没有 `valid_mask`,`max_arrows=400` 的上限因此按幅值排序 `magnitude` 重新推导 `active`,不需要重建结果;只有采样**网格**(`flow_grid_spacing`、 `flow_edge_margin`、`max_arrows`)变化才会移动候选位置,那份策略记在 provenance 里供比对而 不塞进键里。SDK 版本同样不在键里(补丁版本不该让键变化),它记在 provenance 的 `versions`, -由 `AnalysisReader.stale_reasons()` 报出。 +由 `AnalysisReader.stale_reasons()` 报出。场加权实验模型的键分量与逐帧质量见 +[replay-field-experiment.zh.md](replay-field-experiment.zh.md)。 ## 4. 参考帧规则 diff --git a/apps/studio/docs/replay-field-experiment.zh.md b/apps/studio/docs/replay-field-experiment.zh.md new file mode 100644 index 0000000..f5cbdb1 --- /dev/null +++ b/apps/studio/docs/replay-field-experiment.zh.md @@ -0,0 +1,62 @@ +# Studio 回放:场加权实验模型 + +用途:reference。范围:Studio 回放分析对 SDK `field_gain/1`(输入 profile +`hhd_normal_l1_field/1`)的接线、缓存身份、逐帧质量诊断与实时拒绝边界。 +格式、完整性、失效键与播放语义分别由[replay-analysis.md](replay-analysis.md)与 +[replay-analysis-part-b.md](replay-analysis-part-b.md)负责;模型语义见 +[场模型契约](../../../orisys-sdk/docs/field-gain-model.zh.md),任务范围见 +[实施计划](../../../docs/workflow/field-weighted-calibration-plan.zh.md)。 + +## 缓存身份 + +场加权模型是新的分析行为,但不是新的二进制格式。`result_key` 的 `sensor_config` 分量在 +标定文件实际为 `field_gain/1` 时追加 `;field-weighted=1`;spatial/global 模型与旧结果的键 +不变,因此旧回放结果仍能命中。标定内容身份仍来自 schema 3 指纹(含 `pipeline.field`、 +几何与坐标附件),算法版本不因本功能整体升级。 + +provenance 另存 `field` 块:`profile`、`strategy`、实验来源、`baseline_N`、 +`reference_id`、参与统计帧数、`extrapolated_mean/max` 与**运行时实际**的 +`hhd.downsample`、`hhd.grid_shape`。缓存键只用标定指纹与配置内容哈希,不用声明值假装实际值; +实际网格只进 provenance。 + +## 回放 pass + +- 打开链后读 `sensor.calibration.input_profile` 判定模型;field 模型只在显式 `cal_path` + 下使用,profile 默认文件不假定为实验模型。 +- 新模型要求**单 segment 录像**:加权基线无法在段边界重建,多 segment 直接 `ValueError` + 并留下原因,不静默丢基线。 +- 空载窗口仍需操作员在分析设置里声明(30 帧);每个 pass 在参考帧 `reset` 之后、该帧 + `process_chain` 之前调用一次 `declare_unloaded_reference(reference_id="video-unloaded-")`。 + SDK 累计连续 30 个有效加权 `P` 并取中位数;前 30 帧状态 `zero_unavailable`, + 没有牛顿值但保留 raw 读数。不得调用 `set_calibration_reference` 传旧 raw 标量零点, + 也不得把 `centroid_found` 当门槛。 +- 第 30 个有效帧后确认 `baseline_N` 已就绪;视频结束仍未就绪沿用现有 `ValueError` 语义。 +- 逐帧质量取自 `sensor.calibration_result.diagnostics`:`extrapolated_fraction`、 + `coverage_fraction`、`pending_reference_frames`、`baseline_N`、`reference_id`、 + `model_fingerprint`。质量是估计证据,不参与 `calibration_status`,不得因此丢弃牛顿值。 + +## 逐帧存储与读取 + +场加权质量追加在帧记录最后:`FRAME_FLAG_FIELD` 后接 `extrapolated_fraction` 与 +`coverage_fraction` 两个 float32,沿用 `FRAME_FLAG_FORCE` 的追加先例,格式版本不变。 +`AnalysisReader.frame().field_quality` 返回二元组;旧结果没有该 flag 时返回 `None`, +上层显示“未知”,不得补零或拒绝结果。`counts.field_quality_frames` 记录带质量帧数。 + +`ForceSample.from_reading` 继续只凭 raw 有限性判断 `valid`;质量缺失或外推比例高低都不改变 +有效性。`out_of_range` 与负的基线差保留数值。 + +## 显示与实时拒绝 + +- `replay_controller` 在曲线来源与提示中使用 provenance 的 `field` 块:实验模型标签、 + 加权基线、平均外推比例;缺字段显示未知。中英 locales 成对新增键。 +- 实时入口未接入该模型:`SdkAcquisitionWorker.start()`/`FingerAcquisitionWorker.start()` + 与热切换 `_consume_pending_calibration()` 在打开设备前检查显式 `cal_path`,遇 + schema 3 `field_gain/1` 抛可诊断的 `RuntimeError`;由现有运行错误处理显示。 + `calibration_service`/实时 UI 若继续重构,保留同一检查函数供调用。 + +## 验收与限制 + +回归见 `shared/tests/test_field_calibration_app.py`、`apps/studio/tests/test_field_experiment_display.py` +及扩展现有四件套:窗口只声明一次、前 30 帧无牛顿值、第 30 帧后就绪、无接触仍有效、 +多 segment 拒绝、质量 round-trip 与旧结果未知、field 键与 spatial 键不同、实时入口拒绝。 +本次不宣称物理精度或在先模型的精度改善;基线窗口是否平稳、外推比例大小都必须随结果如实报告。 diff --git a/apps/studio/src/orisys_studio/replay_controller.py b/apps/studio/src/orisys_studio/replay_controller.py index 07fa1bd..2a411ce 100644 --- a/apps/studio/src/orisys_studio/replay_controller.py +++ b/apps/studio/src/orisys_studio/replay_controller.py @@ -42,6 +42,7 @@ from orisys_app_shared.acquisition.image_backend import ( ) from orisys_app_shared.i18n import t from orisys_app_shared.force.source import AnalysisCurveSource +from orisys_app_shared.analysis.model import FIELD_PROFILE from orisys_app_shared.paths import recordings_dir from PySide6.QtCore import QObject, QTimer, Signal from PySide6.QtWidgets import QFileDialog, QWidget @@ -66,6 +67,59 @@ PLAYBACK_TICK_MS = 33 ReplayHost = ReplayShell +def field_experiment_note(session: Any) -> tuple[str, str]: + """What a field-weighted result is, as ``(status line, tooltip)``. + + A field result is an **experiment**: its force is an area-weighted integral of the + HHD curl-free field with a baseline measured from a declared unloaded window, and + part of its support may be extrapolated. The replay has to say that where the + operator looks — the status line and the chart's tooltip — using the result's own + provenance. A result that carries the field profile but no diagnostics (an older + cache, or a pass that could not collect them) says *unknown*: a missing + extrapolated fraction is not zero extrapolation. + + Returns ``("", "")`` for anything that is not a field result, so a scalar model's + replay is unchanged. + """ + reader = getattr(session, "reader", session) + if reader is None: + return "", "" + provenance = getattr(reader, "provenance", None) + field = getattr(provenance, "field", None) + if not isinstance(field, dict) or not field: + return "", "" + if str(field.get("profile") or "") != FIELD_PROFILE: + return "", "" + frames = int(field.get("frames") or 0) + baseline = field.get("baseline_N") + extrapolated = field.get("extrapolated_mean") + if frames <= 0 or baseline is None or extrapolated is None: + return t("chart.field_unknown"), t("chart.field_unknown_tip") + hhd = dict(field.get("hhd") or {}) + grid_shape = hhd.get("grid_shape") + grid = ( + f"{int(grid_shape[0])}x{int(grid_shape[1])}" + if isinstance(grid_shape, (list, tuple)) and len(grid_shape) == 2 + else "?" + ) + downsample = hhd.get("downsample") + note = t( + "chart.field_experiment", + baseline=f"{float(baseline):.3f}", + frames=frames, + extrapolated=f"{100.0 * float(extrapolated):.1f}", + ) + tip = t( + "chart.field_experiment_tip", + fingerprint=str(field.get("model_fingerprint") or "?")[:16], + strategy=str(field.get("strategy") or "?"), + reference=str(field.get("reference_id") or "?"), + grid=grid, + downsample="?" if downsample is None else str(downsample), + ) + return note, tip + + @dataclass class ReplayActions: """The five menu/shortcut entries whose meaning depends on the replay state.""" @@ -1018,6 +1072,18 @@ class ReplayController(QObject): # A result just became readable, so it is what the operator is shown — unless a # derived view was already on screen (a depth view chosen while the pass ran). self.serve_stored_view() + # A field-weighted result is an experiment; name its source and quality where + # the operator reads the measurement. A scalar result answers ("", "") and + # nothing is added. + note, tip = field_experiment_note(session) + if note: + self._host.notify(note) + try: + self._charts.setToolTip(tip or note) + except Exception: # noqa: BLE001 - a tooltip must not break adoption + logger.debug( + "could not attach the field experiment tooltip", exc_info=True + ) def serve_stored_view(self) -> bool: """Put the cached view on screen when a result becomes readable. diff --git a/apps/studio/tests/test_field_experiment_display.py b/apps/studio/tests/test_field_experiment_display.py new file mode 100644 index 0000000..af89c41 --- /dev/null +++ b/apps/studio/tests/test_field_experiment_display.py @@ -0,0 +1,72 @@ +"""S4: the replay names the field-weighted experiment a result came from. + +The field model is an experiment, and its numbers are estimates: the result records +its model fingerprint, the declared unloaded reference and the mean extrapolated +fraction. The replay has to say so — and say "unknown" for a result that predates +the diagnostics, never 0. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +from orisys_app_shared.analysis.model import FIELD_PROFILE +from orisys_app_shared.i18n import set_locale, t + +from orisys_studio.replay_controller import field_experiment_note + + +def _session(field: dict) -> SimpleNamespace: + return SimpleNamespace( + reader=SimpleNamespace(provenance=SimpleNamespace(field=field)) + ) + + +FIELD = { + "profile": FIELD_PROFILE, + "strategy": "field-weighted-baseline-1", + "model_fingerprint": "abcdef0123456789" * 4, + "reference_id": "video-unloaded-36", + "baseline_N": 0.25, + "frames": 35, + "extrapolated_mean": 0.3, + "extrapolated_max": 0.4, + "coverage_mean": 0.95, + "hhd": {"downsample": 2, "grid_shape": [150, 140]}, +} + + +def teardown_function(): + set_locale("zh_CN") + + +def test_a_field_result_names_its_baseline_and_extrapolation(): + set_locale("zh_CN") + note, tip = field_experiment_note(_session(dict(FIELD))) + assert note == t( + "chart.field_experiment", + baseline="0.250", + frames=35, + extrapolated="30.0", + ) + assert "field-weighted-baseline-1" in tip + assert "abcdef0123456789" in tip + assert "150x140" in tip or "150 × 140" in tip + + set_locale("en_US") + english, _ = field_experiment_note(_session(dict(FIELD))) + assert english and english != note + assert "30.0" in english + + +def test_a_field_result_without_diagnostics_says_unknown(): + set_locale("zh_CN") + note, tip = field_experiment_note(_session({"profile": FIELD_PROFILE})) + assert note == t("chart.field_unknown") + assert tip == t("chart.field_unknown_tip") + + +def test_a_scalar_or_missing_result_has_no_field_note(): + assert field_experiment_note(_session({})) == ("", "") + assert field_experiment_note(None) == ("", "") + assert field_experiment_note(SimpleNamespace(reader=None)) == ("", "") diff --git a/apps/studio/tests/test_recording_recalibration.py b/apps/studio/tests/test_recording_recalibration.py index e59e055..4c1c8fd 100644 --- a/apps/studio/tests/test_recording_recalibration.py +++ b/apps/studio/tests/test_recording_recalibration.py @@ -171,3 +171,50 @@ def test_explicit_raw_choice_disables_default_calibration_before_processing(monk chain = ChainSpec(sensor_type=family, calibration_disabled=True) service_module._OpenChain(chain, video, tmp_path).open() assert applied == [None] + + +def test_a_field_model_setup_gets_its_own_key_and_the_unloaded_window( + monkeypatch, replay_recording, tmp_path +): + """S4: the field model is a different experiment, so it is a different cache entry. + + The strategy component is added by the analysis service from the chosen file's + ``model.input_profile``; the 30-frame declaration travels with the setup. A + spatial result's key keeps its old components, so its existing cache still hits. + """ + from orisys_app_shared.analysis import bare_video + from orisys_app_shared.analysis import service as service_module + + field_file = tmp_path / "field.json" + field_file.write_text( + json.dumps( + { + "schema_version": 3, + "model": { + "type": "field_gain/1", + "input_profile": "hhd_normal_l1_field/1", + }, + } + ), + encoding="utf-8", + ) + monkeypatch.setattr( + service_module, + "calibration_profile", + lambda path: ("field_gain/1", "hhd_normal_l1_field/1"), + ) + monkeypatch.setattr(bare_video, "calibration_identity", lambda path: "field@abc") + + service = AnalysisService( + Path(replay_recording), + chain=ChainSpec(sensor_type="finger", calibration=field_file), + store_force=True, + unloaded_reference_frames=30, + ) + facts = service.dependency_report() + assert facts["model_profile"] == "hhd_normal_l1_field/1" + assert facts["key"].components["sensor_config"].endswith(";field-weighted=1") + assert facts["key"].components["calibration"] == "field@abc" + assert service.unloaded_reference_frames == 30 + # The recording itself was not touched by key computation. + assert (Path(replay_recording) / "session.json").is_file() diff --git a/shared/src/orisys_app_shared/acquisition/finger_worker.py b/shared/src/orisys_app_shared/acquisition/finger_worker.py index 97260d1..d7d9a76 100644 --- a/shared/src/orisys_app_shared/acquisition/finger_worker.py +++ b/shared/src/orisys_app_shared/acquisition/finger_worker.py @@ -43,6 +43,7 @@ from orisys_app_shared.acquisition.sdk_worker import ( _capture_sequence, calibration_digest, capture_status_of, + live_field_model_rejection, loaded_calibration_version, loaded_path, process_chain, @@ -408,6 +409,9 @@ class FingerAcquisitionWorker(RecordSinkMixin): ) -> FixedCameraMode: # Non-blocking: a previous session that is stuck opening must never hold # the caller (the GUI thread) hostage. + rejection = live_field_model_rejection(cal_path) + if rejection: + raise RuntimeError(rejection) self.stop() previous = self._thread ok, detail = orisys_importable() @@ -643,6 +647,9 @@ class FingerAcquisitionWorker(RecordSinkMixin): session = self._session_id if path is None or sensor is None: return + rejection = live_field_model_rejection(path) + if rejection: + raise RuntimeError(rejection) if not hasattr(sensor, "set_calibration"): raise RuntimeError("sensor does not support set_calibration") if device_id: diff --git a/shared/src/orisys_app_shared/acquisition/sdk_worker.py b/shared/src/orisys_app_shared/acquisition/sdk_worker.py index 09d5acb..64f4edf 100644 --- a/shared/src/orisys_app_shared/acquisition/sdk_worker.py +++ b/shared/src/orisys_app_shared/acquisition/sdk_worker.py @@ -95,7 +95,7 @@ def _high_resolution_timer(): if winmm is not None: try: winmm.timeEndPeriod(1) - except Exception: # noqa: BLE001 + except Exception: # noqa: BLE001, S110 - a failed timer restore is not fatal pass # Re-exported so ``from ...sdk_worker import SdkPacket`` keeps working; the @@ -109,6 +109,7 @@ __all__ = [ "load_fixed_mode", "loaded_calibration_version", "loaded_path", + "live_field_model_rejection", "orisys_importable", "process_chain", "read_depth", @@ -394,6 +395,8 @@ def read_forces_from_sensor(sensor, *, config_ref: str | Path | None = None): if arr.size == 0: depth = None calibrated = _sensor_force_calibrated(sensor, config_ref=config_ref) + result = getattr(sensor, "calibration_result", None) + diagnostics = getattr(result, "diagnostics", None) return { "normal_force": float(fn), "shear_x": float(fx), @@ -410,6 +413,13 @@ def read_forces_from_sensor(sensor, *, config_ref: str | Path | None = None): "invalid_reason": error, "raw_available": read_raw, "calibration_status": dict(getattr(sensor, "calibration_status", {}) or {}), + # Evidence from the calibration model itself (the field model's + # extrapolated/coverage fractions, baseline and fingerprint). It is carried, + # never interpreted here: whether a value is displayed is the model status's + # decision, and quality is not validity. + "calibration_diagnostics": dict(diagnostics or {}) + if isinstance(diagnostics, dict) + else {}, } @@ -692,6 +702,41 @@ def loaded_calibration_version(sensor) -> str: return calibration_digest(calibration) +#: The field-weighted model's input profile. Kept as a literal here because the +#: acquisition package must not import SDK modules unless a run actually opens one; +#: ``analysis.model`` owns the app's copy of the same name for the replay path. +_FIELD_INPUT_PROFILE = "hhd_normal_l1_field/1" + + +def live_field_model_rejection(path: str | Path | None) -> str: + """Why a chosen calibration file must not be used for a **live** run, or ``""``. + + The field-weighted model (``field_gain/1`` / ``hhd_normal_l1_field/1``) is a + replay-only experiment: it needs a declared 30-frame unloaded window and the + same-frame HHD field, neither of which the live acquisition loop supplies. The + live entry therefore refuses it by name, before the device is opened, instead of + loading it and failing per frame. An unreadable file yields ``""`` — the open + path already reports that with its own reason, and this helper must not replace + one failure with another. + """ + if path is None or not str(path).strip(): + return "" + try: + import orisys.calibration as sdk_calibration + + model = sdk_calibration.load(str(path), strict=True) + except Exception: # noqa: BLE001 - the open path reports an unloadable file itself + return "" + if str(getattr(model, "input_profile", "") or "") != _FIELD_INPUT_PROFILE: + return "" + return ( + "the field-weighted calibration model (hhd_normal_l1_field/1) is replay-only: " + "it needs a declared 30-frame unloaded window and the same-frame HHD field, " + "which the live chain does not provide. Choose a scalar calibration for live " + f"operation ({Path(str(path)).name})." + ) + + class SdkAcquisitionWorker(RecordSinkMixin): """Background loop: Sensor.get_img + deformation + force readout.""" @@ -838,6 +883,9 @@ class SdkAcquisitionWorker(RecordSinkMixin): # Non-blocking: a previous session that is stuck opening the device must # never hold the caller (the GUI thread) hostage. del device_id # box/mini open path has no Finger-style serial check + rejection = live_field_model_rejection(cal_path) + if rejection: + raise RuntimeError(rejection) self.stop() previous = self._thread ok, detail = orisys_importable() @@ -1023,6 +1071,9 @@ class SdkAcquisitionWorker(RecordSinkMixin): session = self._session_id if path is None or sensor is None: return + rejection = live_field_model_rejection(path) + if rejection: + raise RuntimeError(rejection) if not hasattr(sensor, "set_calibration"): raise RuntimeError("sensor does not support set_calibration") sensor.set_calibration(path) @@ -1050,7 +1101,7 @@ class SdkAcquisitionWorker(RecordSinkMixin): return try: sensor.disconnect() - except Exception: + except Exception: # noqa: BLE001, S110 - teardown must not mask the pass result pass def _close_all_sensors(self) -> None: diff --git a/shared/src/orisys_app_shared/analysis/model.py b/shared/src/orisys_app_shared/analysis/model.py index a4e88b4..cccf2e3 100644 --- a/shared/src/orisys_app_shared/analysis/model.py +++ b/shared/src/orisys_app_shared/analysis/model.py @@ -57,6 +57,17 @@ differently now. This is the one channel whose ``Provenance.sources`` entry is ``recomputed`` on the *synthetic* path and ``recorded`` beside a real ``forces.csv``: the two must never be confused, which is what the field is for. +**The field model's quality is stored beside the force, not inside it.** The +field-weighted model is an experiment: its estimate is an area-weighted integral of the +HHD curl-free field over a mapped gain, and part of that support may be extrapolated +from the convex hull boundary. The SDK reports that as an extrapolated fraction; the +pass stores it per frame behind :data:`FRAME_FLAG_FIELD` with an 8-byte +:data:`FIELD_PAYLOAD` (extrapolated, coverage). It is *evidence about the estimate*, +never a validity statement: a frame with a non-zero extrapolated fraction still has a +newton value. A frame without the flag has **unknown** quality (a scalar model, an +older result) and must be displayed that way, not as 0. The total per-frame cost is +8 B against the 4,048 B a frame already costs. + Sparse storage was measured and rejected: 4-byte indices per non-zero pixel cost 910 B for the worst non-empty frame against 486 B for the whole quantised map, because a mostly-zero byte plane is nearly free for zlib while pixel indices are not. Island @@ -92,9 +103,14 @@ __all__ = [ "CODEC_ZLIB", "DEPTH_BITS", "DEPTH_HEADER", + "FIELD_KEY_SUFFIX", + "FIELD_PAYLOAD", + "FIELD_PROFILE", + "FIELD_STRATEGY", "FORCE_PAYLOAD", "FRAME_FLAG_DEPTH", "FRAME_FLAG_DROPPED", + "FRAME_FLAG_FIELD", "FRAME_FLAG_FORCE", "FRAME_FLAG_VALIDITY", "FRAME_HEADER", @@ -145,6 +161,20 @@ ALGORITHM_VERSION = "flow-hhd-contact-2" #: unit :class:`~orisys_app_shared.scene3d.visual_policy.FlowSamples` documents. UNIT = "uv_px" +#: The field-weighted model's input profile (SDK calibration contract). The app only +#: names it: the model is loaded and run by the SDK, and this string is what tells the +#: pass that the unloaded reference is a 30-frame weighted window rather than a scalar +#: raw zero. It also keys the result, because that window is a property of this video. +FIELD_PROFILE = "hhd_normal_l1_field/1" +#: Cache-key component appended to ``sensor_config`` when the chosen calibration is the +#: field-weighted model. A spatial result and a field result must never share a key: +#: the force on the same footage is a different experiment with a different baseline. +#: It is added only for an explicitly chosen file whose ``model.input_profile`` is +#: :data:`FIELD_PROFILE`; the scalar models' keys are unchanged. +FIELD_KEY_SUFFIX = ";field-weighted=1" +#: Version of the app's field strategy, written to the provenance for display/audit. +FIELD_STRATEGY = "field-weighted-baseline-1" + #: ``frame_id``, ``t_rel_s``, grid keyframe id, flags, candidate count, depth #: non-zero count, depth minimum, depth maximum. The three trailing pad bytes keep #: the value arrays 8-byte aligned inside a record. @@ -160,6 +190,12 @@ FRAME_FLAG_DEPTH = 0x04 #: pass was not asked to collect it (a recording, whose forces live in its own #: ``forces.csv``) or the chain exposed none. FRAME_FLAG_FORCE = 0x08 +#: The frame carries the field model's quality payload (:data:`FIELD_PAYLOAD`), +#: appended after the force payload. This is *evidence about the estimate* — how much +#: of its support was extrapolated — not a validity statement: the newton value is +#: still governed by the calibration status and must not be dropped because the +#: extrapolated fraction is non-zero. +FRAME_FLAG_FIELD = 0x10 #: The force payload: FN, FX, FY in newtons, then the raw triplet, then one byte of #: validity and one of "a calibration curve was applied". Appended **last** in the @@ -168,6 +204,13 @@ FRAME_FLAG_FORCE = 0x08 #: always ignored trailing bytes. FORCE_PAYLOAD = struct.Struct(" bool: @@ -537,6 +585,23 @@ class ForceSample: def raw(self) -> tuple[float, float, float]: return (self.normal_raw, self.shear_x_raw, self.shear_y_raw) + def field_quality(self) -> tuple[float, float] | None: + """This frame's ``(extrapolated_fraction, coverage_fraction)``, or ``None``. + + Only the field model reports these. ``None`` means *unknown* (an older result, + a scalar model, or a frame whose model gave no field evidence) and must never + be displayed as 0. + """ + extrapolated = self.diagnostics.get("extrapolated_fraction") + coverage = self.diagnostics.get("coverage_fraction") + try: + values = (float(extrapolated), float(coverage)) + except (TypeError, ValueError): + return None + if not all(value == value and abs(value) != float("inf") for value in values): + return None + return tuple(min(1.0, max(0.0, value)) for value in values) # type: ignore[return-value] + def values(self, unit: str = "raw") -> tuple[float, float, float]: """The triplet in ``unit``: newtons when a calibration was applied, else raw. @@ -565,6 +630,7 @@ class ForceSample: return _NAN return float(reading.get(key, _NAN)) + diagnostics = reading.get("calibration_diagnostics") return cls( normal=newton("fn", "normal_force"), shear_x=newton("fx", "shear_x"), @@ -575,6 +641,7 @@ class ForceSample: valid=bool(reading.get("valid", True)), calibrated=bool(reading.get("force_calibrated", False)), status=tuple(sorted(statuses.items())), + diagnostics=dict(diagnostics or {}), ) def encode(self) -> bytes: @@ -628,8 +695,9 @@ def encode_frame( depth_nonzero: int = 0, dropped: bool = False, force: ForceSample | None = None, + field_quality: Sequence[float] | None = None, ) -> bytes: - """One frame record: a fixed header, then the value arrays, depth and force. + """One frame record: a fixed header, then the value arrays, depth, force and quality. Arrays are little-endian regardless of host order, so a result written on one machine reads back identically on another. ``dx``/``dy``/``magnitude`` are stored @@ -645,6 +713,12 @@ def encode_frame( ``force`` is the chain's own sample for this frame, or ``None`` when the pass did not collect one (``FRAME_FLAG_FORCE``). It is written last, after every field an older reader knows, so the record stays readable by a build without this flag. + + ``field_quality`` is the field model's ``(extrapolated_fraction, coverage_fraction)`` + for this frame, or ``None`` when the model reported none (a scalar model, an older + result, or a frame without field evidence). It is appended *after* the force payload + for the same reason the force payload follows depth: an older reader never parses + past a field it knows, and a chunk's offset table delimits the record. """ dx = np.ascontiguousarray(dx, dtype=" 0xFFFF: raise ValueError(f"{count} candidates do not fit the frame header") + quality_payload = b"" + if field_quality is not None: + values = tuple(float(value) for value in field_quality) + if len(values) != 2 or not all(value == value for value in values): + raise ValueError( + "field quality must be two finite fractions (extrapolated, coverage)" + ) + quality_payload = FIELD_PAYLOAD.pack( + min(1.0, max(0.0, values[0])), min(1.0, max(0.0, values[1])) + ) flags = 0 if validity_digest is not None: flags |= FRAME_FLAG_VALIDITY @@ -667,6 +751,8 @@ def encode_frame( flags |= FRAME_FLAG_DROPPED if force is not None: flags |= FRAME_FLAG_FORCE + if quality_payload: + flags |= FRAME_FLAG_FIELD depth_payload = b"" depth_low = depth_high = 0.0 if depth is not None: @@ -696,6 +782,7 @@ def encode_frame( parts.append(depth_payload) if force is not None: parts.append(force.encode()) + parts.append(quality_payload) return b"".join(parts) @@ -720,6 +807,9 @@ class DecodedFrame: #: ``None`` when the frame carried no force (an older result, a recording whose forces #: live in its own ``forces.csv``, or a chain that produced none). force: ForceSample | None = None + #: ``None`` when the frame carried no field quality (an older result, a scalar model, + #: or no field evidence). Not the same fact as "the extrapolated fraction was 0". + field_quality: tuple[float, float] | None = None def depth_map(self) -> np.ndarray | None: """This frame's depth map, or ``None`` when it had no depth channel.""" @@ -771,6 +861,14 @@ def decode_frame( if len(view) < cursor + FORCE_PAYLOAD.size: raise ValueError("a frame says it carries force but is shorter than one sample") force = ForceSample.decode(view, cursor) + cursor += FORCE_PAYLOAD.size + field_quality = None + if flags & FRAME_FLAG_FIELD: + if len(view) < cursor + FIELD_PAYLOAD.size: + raise ValueError("a frame says it carries field quality but is shorter than it") + field_quality = tuple( + float(value) for value in FIELD_PAYLOAD.unpack_from(view, cursor) + ) return DecodedFrame( frame_id=int(frame_id), t_rel_s=float(t_rel), @@ -785,6 +883,7 @@ def decode_frame( validity_digest=validity, depth=depth, force=force, + field_quality=field_quality, ) @@ -795,18 +894,22 @@ def frame_payload_bytes( with_depth: bool = False, depth_stored: int = 0, with_force: bool = False, + with_field_quality: bool = False, ) -> int: """Stored bytes of one frame record at ``candidates`` sample locations. ``with_depth`` adds the 6-byte depth header every frame that carries a channel pays; ``depth_stored`` adds a payload of that many bytes on top, for a caller sizing the - memory bound rather than the common case. ``with_force`` adds the fixed force sample. + memory bound rather than the common case. ``with_force`` adds the fixed force sample; + ``with_field_quality`` adds the fixed field quality payload. """ total = FRAME_HEADER.size + int(candidates) * 12 + (int(candidates) + 7) // 8 if with_validity: total += 8 if with_force: total += FORCE_PAYLOAD.size + if with_field_quality: + total += FIELD_PAYLOAD.size if with_depth: total += DEPTH_HEADER.size + max(0, int(depth_stored)) return total @@ -872,6 +975,11 @@ class FramePayload: #: which is why the provenance — not this field — is what a chart must label a trace #: from. force: ForceSample | None = None + #: The field model's ``(extrapolated_fraction, coverage_fraction)`` for this frame, + #: or ``None`` when the result carries none (a scalar model, an older result). This + #: is quality evidence, not validity: a finite newton value with a non-zero + #: extrapolated fraction is still a value. + field_quality: tuple[float, float] | None = None @property def has_depth(self) -> bool: @@ -928,6 +1036,7 @@ class FramePayload: depth_nonzero=self.depth_nonzero, dropped=self.dropped, force=self.force, + field_quality=self.field_quality, ) @@ -1035,9 +1144,20 @@ class Provenance: #: that time is not throughput. open_seconds: float = 0.0 warnings: list[str] = field(default_factory=list) + #: True when the pass reached the end of every declared segment and committed. complete: bool = False stopped_reason: str = "" counts: dict[str, Any] = field(default_factory=dict) + #: The field-weighted experiment behind this result, when it has one. Keys: + #: ``profile``/``strategy`` (what ran), ``model_fingerprint`` (content identity of + #: the model), ``reference_id``/``baseline_N`` (the declared unloaded window), and + #: ``frames``/``extrapolated_mean``/``extrapolated_max``/``coverage_mean`` (per-frame + #: quality statistics) plus ``hhd`` (effective grid shape and downsample the + #: weighted sum ran on). Empty for a scalar model or an older result; a consumer + #: must display that as *unknown*, never as zero extrapolation. + #: Declared last on purpose: an attribute named ``field`` would shadow the + #: ``dataclasses.field`` helper for every attribute after it in this class body. + field: dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: return dict(self.__dict__) diff --git a/shared/src/orisys_app_shared/analysis/reader.py b/shared/src/orisys_app_shared/analysis/reader.py index a11ea3e..4d0ee35 100644 --- a/shared/src/orisys_app_shared/analysis/reader.py +++ b/shared/src/orisys_app_shared/analysis/reader.py @@ -533,12 +533,22 @@ class AnalysisReader: depth_max=depth_max, depth_stored_bytes=depth_stored, force=decoded.force, + field_quality=decoded.field_quality, ) def force(self, frame_id: int) -> ForceSample | None: """One frame's stored force sample, or ``None`` when the frame carries none.""" return self.frame(int(frame_id)).force + def field_quality(self, frame_id: int) -> tuple[float, float] | None: + """One frame's ``(extrapolated_fraction, coverage_fraction)``, or ``None``. + + ``None`` is the truthful answer for a result that predates the field-model + diagnostics (or for a scalar model): the extrapolated fraction was not + recorded, which is not the same as zero. Display it as unknown. + """ + return self.frame(int(frame_id)).field_quality + @property def has_force(self) -> bool: """Whether *any* frame in this result carries a force sample. diff --git a/shared/src/orisys_app_shared/analysis/service.py b/shared/src/orisys_app_shared/analysis/service.py index d4b309d..f1864ad 100644 --- a/shared/src/orisys_app_shared/analysis/service.py +++ b/shared/src/orisys_app_shared/analysis/service.py @@ -61,6 +61,9 @@ from orisys_app_shared.analysis.cache import ( ) from orisys_app_shared.analysis.model import ( ALGORITHM_VERSION, + FIELD_KEY_SUFFIX, + FIELD_PROFILE, + FIELD_STRATEGY, KEYFRAME_MASK, ForceSample, Provenance, @@ -86,6 +89,7 @@ __all__ = [ "ChainSpec", "LiveFrame", "Progress", + "calibration_profile", "chain_spec_for", "recording_config_sha256", "resolve_finger_export_dir", @@ -136,6 +140,116 @@ def recording_config_sha256(config_ref: str | Path | None) -> str: return "" +def calibration_profile(path: str | Path | None) -> tuple[str, str]: + """``(model.type, model.input_profile)`` of a calibration file, or ``("", "")``. + + A plain JSON lookup on purpose: the result key has to be computed *before* anything + opens the SDK (a cache hit is a file read, and a pass that turns out to be cached + must not pay for a chain open). Schema 1/2 documents have no ``model`` block and + answer ``("", "")``, which is the same answer an unreadable file gives — the open + path reports that failure with its own reason. + """ + if path is None: + return ("", "") + try: + data = json.loads(Path(str(path)).read_text(encoding="utf-8-sig")) + except (OSError, ValueError): + return ("", "") + if not isinstance(data, dict): + return ("", "") + model = data.get("model") + if not isinstance(model, dict): + return ("", "") + return ( + str(model.get("type") or ""), + str(model.get("input_profile") or ""), + ) + + +def _chain_raw_sensor(chain: Any) -> Any: + """The SDK ``Sensor`` behind a chain (a finger chain wraps it in a tactile one).""" + return getattr(chain.sensor, "raw_sensor", chain.sensor) + + +def _chain_field_model(chain: Any) -> Any: + """The loaded calibration model when it is the field-weighted one, else ``None``.""" + model = getattr(_chain_raw_sensor(chain), "calibration", None) + if str(getattr(model, "input_profile", "") or "") != FIELD_PROFILE: + return None + return model + + +def _field_diagnostics(sensor: Any) -> dict[str, Any]: + """The calibration model's diagnostics for the frame it just answered. + + The SDK exposes them as ``Sensor.calibration_result.diagnostics``; a sensor that + has no model (or a model that reports none) answers ``{}``. Missing diagnostics + mean *unknown quality*, never zero extrapolation. + """ + result = getattr(sensor, "calibration_result", None) + diagnostics = getattr(result, "diagnostics", None) + return dict(diagnostics) if isinstance(diagnostics, dict) else {} + + +def _hhd_identity(sensor: Any) -> dict[str, Any]: + """The HHD grid the weighted sum actually ran on, for the result's provenance. + + Read from the open sensor (the effective values), not from the configuration + document: the SDK may accept a calibration whose declared ``hhd_downsample`` + differs from the runtime one, and the area factor ``A`` depends on the grid that + actually ran. + """ + identity: dict[str, Any] = {} + downsample = getattr(sensor, "_hhd_downsample", None) + if downsample is not None: + try: + identity["downsample"] = int(downsample) + except (TypeError, ValueError): + pass + for key, attribute in (("grid_shape", "_hhd_grid_shape"), ("reference_shape", "_hhd_ref_shape")): + value = getattr(sensor, attribute, None) + if value is None: + continue + try: + identity[key] = [int(size) for size in value] + except (TypeError, ValueError): + continue + return identity + + +def _accept_field_diagnostics( + state: _PassState, sensor: Any, *, zero_frames: int +) -> None: + """Keep this frame's weighted-window status and quality statistics in the state.""" + diagnostics = _field_diagnostics(sensor) + if zero_frames and state.field_reference_id and state.field_baseline is None: + baseline = diagnostics.get("baseline_N") + if baseline is not None and np.isfinite(float(baseline)): + state.field_baseline = float(baseline) + pending = diagnostics.get("pending_reference_frames") + if pending is not None: + state.provenance.counts["field_pending_reference_frames"] = int(pending) + extrapolated = diagnostics.get("extrapolated_fraction") + if extrapolated is not None: + try: + value = min(1.0, max(0.0, float(extrapolated))) + except (TypeError, ValueError): + value = None + if value is not None: + state.field_extrapolated_sum += value + state.field_extrapolated_max = max(state.field_extrapolated_max, value) + state.field_extrapolated_count += 1 + coverage = diagnostics.get("coverage_fraction") + if coverage is not None: + try: + value = min(1.0, max(0.0, float(coverage))) + except (TypeError, ValueError): + value = None + if value is not None: + state.field_coverage_sum += value + state.field_coverage_count += 1 + + def resolve_finger_export_dir(config_ref: str | Path) -> Path | None: """``assets/30fps-0703`` beside a finger ``sensor.json``, when it is complete. @@ -509,6 +623,18 @@ class _PassState: frames_expected: int = 0 frames_written: int = 0 started: float = 0.0 + #: The field-weighted experiment this pass ran, when the loaded model is + #: ``field_gain/1``. Empty/false for every scalar model. + field_profile: str = "" + field_fingerprint: str = "" + field_reference_id: str = "" + field_baseline: float | None = None + field_hhd: dict[str, Any] = field(default_factory=dict) + field_extrapolated_sum: float = 0.0 + field_extrapolated_max: float = 0.0 + field_extrapolated_count: int = 0 + field_coverage_sum: float = 0.0 + field_coverage_count: int = 0 class AnalysisService: @@ -631,6 +757,14 @@ class AnalysisService: from orisys_app_shared.analysis.bare_video import calibration_identity calibration = calibration_identity(self.chain.calibration) + # The chosen file's model, read as JSON (never by opening the SDK): the field + # model's unloaded baseline is measured from *this* video's frames, so a field + # result and a scalar result on the same footage must not share a cache entry. + model_type, model_profile = ( + ("", "") + if self.chain.calibration_disabled + else calibration_profile(self.chain.calibration) + ) zero_frames = self._zero_reference_frames() key = result_key( recording=identity, @@ -638,6 +772,7 @@ class AnalysisService: self.chain.identity() + (";unloaded-reference=30" if zero_frames else "") + (";force=recomputed" if self.store_force else "") + + (FIELD_KEY_SUFFIX if model_profile == FIELD_PROFILE else "") ), calibration=calibration, reference_frame=self.reference_frame, @@ -650,6 +785,8 @@ class AnalysisService: "segment_sizes": sizes, "recording_identity": identity, "calibration_id": calibration, + "model_type": model_type, + "model_profile": model_profile, "key": key, } @@ -680,6 +817,63 @@ class AnalysisService: f"frames ({span}); a reset that cannot be applied is not a result" ) + def _require_field_single_segment(self, facts: dict[str, Any]) -> None: + """Refuse a field-model pass over more than one segment. + + The field model's baseline is one declared, consecutive 30-frame window of the + *same* stream. A segment boundary reopens the chain (and the tracker), so a + window that crossed it would be a different measurement wearing the same + reference id. Refusing is the only truthful answer; silently re-declaring would + discard the first segment's baseline and silently rebuilding it would invent + one. + """ + if facts.get("model_profile") != FIELD_PROFILE: + return + segments = list(facts.get("segments") or ()) + if len(segments) <= 1: + return + raise ValueError( + "field_gain/1 requires a single-segment recording: its weighted unloaded " + "baseline is one declared window of the same stream and cannot be rebuilt " + f"across a segment boundary ({len(segments)} segments in {self.recording})" + ) + + @staticmethod + def _field_provenance(state: _PassState) -> dict[str, Any]: + """The field-weighted experiment behind this result, as stored evidence. + + Kept even for an abandoned pass: a cancelled analysis still has to say which + model it was running and whether its unloaded window completed. + """ + extrapolated_mean = ( + state.field_extrapolated_sum / state.field_extrapolated_count + if state.field_extrapolated_count + else None + ) + coverage_mean = ( + state.field_coverage_sum / state.field_coverage_count + if state.field_coverage_count + else None + ) + return { + "profile": state.field_profile, + "strategy": FIELD_STRATEGY, + "model_fingerprint": state.field_fingerprint, + "reference_id": state.field_reference_id, + "baseline_N": state.field_baseline, + "frames": state.field_extrapolated_count, + "extrapolated_mean": ( + None if extrapolated_mean is None else round(extrapolated_mean, 6) + ), + "extrapolated_max": ( + round(state.field_extrapolated_max, 6) + if state.field_extrapolated_count + else None + ), + "coverage_mean": None if coverage_mean is None else round(coverage_mean, 6), + "hhd": dict(state.field_hhd), + } + # ------------------------------------------------------------------------ pass def recompute(self) -> Iterator[LiveFrame]: """Re-run the chain over every segment and yield what it computes now. @@ -692,6 +886,7 @@ class AnalysisService: key: ResultKey = facts["key"] grouped = self.rows_by_segment(facts["info"].frames) self._require_reference(grouped) + self._require_field_single_segment(facts) working = self._make_working_directory(key) state = _PassState( provenance=self._provenance(key, facts), @@ -724,6 +919,7 @@ class AnalysisService: # Checked before anything is written, so a caller error does not leave a # directory behind at all. self._require_reference(grouped) + self._require_field_single_segment(facts) working = self._make_working_directory(key) writer = AnalysisWriter( self.cache.directory(key), key=key, chunk_frames=self.chunk_frames @@ -762,6 +958,10 @@ class AnalysisService: state.reasons.append(f"{type(exc).__name__}: {exc}") finally: self._discard_working_directory(working) + if state.field_profile: + # Written for complete *and* abandoned results: a cancelled field pass still + # has to name its model and whether the unloaded window had completed. + provenance.field = self._field_provenance(state) if ( not state.reasons and self.reference_frame is not None @@ -784,6 +984,9 @@ class AnalysisService: writer.bytes_written / state.frames_written, 2 ) counts["open_seconds"] = provenance.open_seconds + if state.field_profile: + counts["field_frames"] = state.field_extrapolated_count + counts["field_baseline_ready"] = state.field_baseline is not None provenance.decode_order_offsets = dict(state.offsets) distinct = sorted(set(state.offsets.values())) provenance.decode_order_offset = distinct[0] if len(distinct) == 1 else None @@ -843,6 +1046,25 @@ class AnalysisService: provenance.warnings.append(f"segment_frame_gap:{number}") with _OpenChain(self.chain, path, state.working) as chain: zero_samples = [] + raw_sensor = _chain_raw_sensor(chain) + field_model = _chain_field_model(chain) + field_weighted = field_model is not None + if field_weighted: + # The baseline is one window of the same stream; a second segment + # would reopen the tracker and could only re-declare it. Refuse + # rather than silently keep the first window or rebuild a new one. + if len(state.segments) > 1: + raise ValueError( + "field_gain/1 requires a single-segment recording: its " + "weighted unloaded baseline is one declared window of the " + "same stream and cannot be rebuilt across a segment boundary" + ) + if not state.field_profile: + state.field_profile = FIELD_PROFILE + state.field_fingerprint = str( + getattr(field_model, "fingerprint", "") or "" + ) + state.field_hhd = _hhd_identity(raw_sensor) offset, measured = self._segment_offset(chain, number, state.offsets) state.offsets[str(number)] = offset state.sources.add("measured" if measured else "declared") @@ -888,25 +1110,79 @@ class AnalysisService: provenance.counts["reset_count"] = ( provenance.counts.get("reset_count", 0) + 1 ) + if ( + field_weighted + and zero_frames + and not state.field_reference_id + and ( + self.reference_frame is None + or int(row.frame_id) >= int(self.reference_frame) + ) + ): + # The tracker reset cleared the model's session too, so the + # declaration must follow it and *precede* this frame's chain + # call: otherwise the first frame of the window is consumed + # before the model knows a window started. + declare = getattr(raw_sensor, "declare_unloaded_reference", None) + if not callable(declare): + raise ValueError( + "this SDK cannot declare the field model's weighted " + "unloaded window" + ) + state.field_reference_id = ( + f"video-unloaded-{int(row.frame_id)}" + ) + declare(reference_id=state.field_reference_id) + provenance.counts["field_reference_frames"] = zero_frames depth, error = process_chain(chain.sensor) if error: provenance.warnings.append(f"chain:{error}") state.reasons.append(f"chain_failed:{row.frame_id}") return - if zero_frames and (self.reference_frame is None or row.frame_id >= self.reference_frame) and len(zero_samples) < zero_frames: - sensor = getattr(chain.sensor, "raw_sensor", chain.sensor) - model = getattr(sensor, "calibration", None) - if getattr(model, "schema_version", None) != 3: - raise ValueError("unloaded reference requires a schema 3 calibration") - keys = (sensor.info.FNORMAL_RAW, sensor.info.FSHEARX_RAW, sensor.info.FSHEARY_RAW) - values = sensor.read_info(*keys) - if getattr(sensor, "centroid_found", False) or not np.all(np.isfinite(values)): - raise ValueError("unloaded reference contains contact or invalid readings") + if field_weighted: + _accept_field_diagnostics( + state, raw_sensor, zero_frames=zero_frames + ) + if ( + not field_weighted + and zero_frames + and ( + self.reference_frame is None + or row.frame_id >= self.reference_frame + ) + and len(zero_samples) < zero_frames + ): + if getattr(raw_sensor, "calibration", None) is None or getattr( + raw_sensor.calibration, "schema_version", None + ) != 3: + raise ValueError( + "unloaded reference requires a schema 3 calibration" + ) + keys = ( + raw_sensor.info.FNORMAL_RAW, + raw_sensor.info.FSHEARX_RAW, + raw_sensor.info.FSHEARY_RAW, + ) + values = raw_sensor.read_info(*keys) + if getattr(raw_sensor, "centroid_found", False) or not np.all( + np.isfinite(values) + ): + raise ValueError( + "unloaded reference contains contact or invalid readings" + ) zero_samples.append(values) if len(zero_samples) == zero_frames: - raw = {axis: float(np.median([value[index] for value in zero_samples])) - for index, axis in enumerate(("fn", "fx", "fy")) if axis in model.channels} - sensor.set_calibration_reference(raw, reference_id=f"video-unloaded-{row.frame_id}") + raw = { + axis: float( + np.median([value[index] for value in zero_samples]) + ) + for index, axis in enumerate(("fn", "fx", "fy")) + if axis in raw_sensor.calibration.channels + } + raw_sensor.set_calibration_reference( + raw, + reference_id=f"video-unloaded-{row.frame_id}", + ) provenance.counts["unloaded_reference_frames"] = zero_frames provenance.counts["unloaded_reference_fn_raw"] = raw.get("fn") force = self._read_force(chain) @@ -945,8 +1221,16 @@ class AnalysisService: depth=depth_array, force=force, ) - if zero_frames and len(zero_samples) != zero_frames: - raise ValueError("video ended before 30 unloaded reference frames were available") + if field_weighted: + if zero_frames and state.field_baseline is None: + raise ValueError( + "video ended before the 30-frame weighted unloaded baseline " + "was available" + ) + elif zero_frames and len(zero_samples) != zero_frames: + raise ValueError( + "video ended before 30 unloaded reference frames were available" + ) # ------------------------------------------------------------------- internals def _zero_reference_frames(self) -> int: @@ -1055,6 +1339,7 @@ class AnalysisService: self._accumulate( provenance.counts, samples, live.depth_nonzero, live.depth_abs_sum ) + field_quality = None if live.force is None else live.force.field_quality() record = encode_frame( frame_id=row.frame_id, t_rel_s=row.t_rel_s, @@ -1068,6 +1353,7 @@ class AnalysisService: depth_nonzero=live.depth_nonzero, dropped=row.dropped, force=live.force, + field_quality=field_quality, ) counts = provenance.counts counts["frame_record_bytes"] = counts.get("frame_record_bytes", 0) + len(record) @@ -1153,6 +1439,13 @@ class AnalysisService: counts["force_zero_unavailable_frames"] = counts.get( "force_zero_unavailable_frames", 0 ) + (1 if zero_missing else 0) + quality = force.field_quality() + if quality is not None: + counts["field_quality_frames"] = counts.get("field_quality_frames", 0) + 1 + counts["field_extrapolated_last"] = quality[0] + counts["field_extrapolated_max"] = max( + float(counts.get("field_extrapolated_max", 0.0)), quality[0] + ) def _provenance(self, key: ResultKey, facts: dict[str, Any]) -> Provenance: session = facts["session"] diff --git a/shared/tests/_field_fakes.py b/shared/tests/_field_fakes.py new file mode 100644 index 0000000..ba625b6 --- /dev/null +++ b/shared/tests/_field_fakes.py @@ -0,0 +1,292 @@ +"""Fake SDK surface for the S4 field-model tests (shared by two test modules). + +The real SDK field model is exercised by ``orisys-sdk/tests/test_field_calibration.py``; +the app tests describe the *calls and data* the app owns. Keeping the fakes in one +place means the pass test and the integration test cannot drift apart. +""" + +from __future__ import annotations + +import hashlib +import json +import math +from pathlib import Path +from types import SimpleNamespace + +import numpy as np + +from orisys_app_shared.acquisition import sdk_worker +from orisys_app_shared.analysis import service as service_module +from orisys_app_shared.analysis.model import Provenance +from orisys_app_shared.record.playback import RecordingFrame + +FIELD_PROFILE = "hhd_normal_l1_field/1" + + +# ------------------------------------------------------------------ calibration +def write_model_file(tmp_path: Path, model_type: str) -> Path: + """A schema 3 calibration document the SDK accepts, with its coordinate map. + + ``field_gain/1`` and ``spatial_gain/1`` differ only in the model block, so a test + can compare two documents that really load through the SDK's parser. + """ + from orisys.field_calibration import BASELINE_POLICY, FIELD_IDENTITY, MAPPING_POLICY + + rows, cols = np.mgrid[0:4, 0:4] + uv = np.stack((cols, rows), axis=-1).astype(np.float64) + attachment = tmp_path / "map.npz" + np.savez( + attachment, + uv_by_output_px=uv, + valid_mask=np.ones((4, 4), dtype=bool), + origin_xy=np.array([0, 0]), + ) + digest = hashlib.sha256(attachment.read_bytes()).hexdigest() + pipeline = { + "stitch_resolution": [4, 4], + "dis_mode": 1, + "hhd_downsample": 1, + "filters": {"measurement": {"flow_ema_alpha": 1.0}}, + "feature_signature_sha256": "a" * 64, + "geometry_sha256": "b" * 64, + "source_scale": 0.42, + "coordinates": { + "frame": "autocal_flat", + "unit": "px", + "origin_xy": [0, 0], + "map_sha256": digest, + }, + } + params = { + "points_uv": [[0, 0], [3, 0], [0, 3], [3, 3]], + "gains_N_per_raw": [1.0, 1.0, 1.0, 1.0], + "triangles": [[0, 1, 2], [1, 3, 2]], + "valid_force_range_N": [0.0, 3.0], + } + if model_type == "field_gain/1": + pipeline["field"] = dict(FIELD_IDENTITY) + profile = FIELD_PROFILE + params["mapping"] = dict(MAPPING_POLICY) + params["baseline"] = dict(BASELINE_POLICY) + else: + profile = "fn_delta_autocal_uv/1" + document = { + "schema_version": 3, + "unit": "N", + "identity": {"sensor_type": "finger", "sn": "dd02-gl"}, + "pipeline": pipeline, + "model": { + "type": model_type, + "input_profile": profile, + "output_frame": "sdk_force/1", + "outputs": ["fn"], + "params": params, + "artifacts": [ + {"role": "coordinate_map", "path": attachment.name, "sha256": digest} + ], + }, + } + path = tmp_path / ("field.json" if model_type == "field_gain/1" else "spatial.json") + path.write_text(json.dumps(document), encoding="utf-8") + return path + + +# --------------------------------------------------------------------- recording +def minimal_recording(root: Path, *, segments: int = 1) -> Path: + """The two text files and the segment bytes a reader needs, and nothing else.""" + recording = root / "recording" + recording.mkdir(parents=True, exist_ok=True) + (recording / "frames.csv").write_text( + "frame_id,t_rel_s,acquired_at,segment,segment_frame,in_video,dropped\n" + "1,0.0,,1,1,1,0\n2,0.033,,1,2,1,0\n", + encoding="utf-8", + ) + for number in range(1, segments + 1): + (recording / f"raw_{number:04d}.mp4").write_bytes(b"") + (recording / "session.json").write_text( + json.dumps( + { + "session_id": 1, + "started_at": "2026-01-01T00:00:00+0000", + "ended_at": "", + "device": {"sensor_type": "finger", "config_ref": ""}, + "stats": {"frames": 2, "video_frames": 2, "segments": segments}, + } + ), + encoding="utf-8", + ) + return recording + + +def field_rows(count: int, *, segment: int = 1) -> list[RecordingFrame]: + return [ + RecordingFrame(i + 1, i / 30.0, "", segment, i + 1, True) + for i in range(count) + ] + + +# ------------------------------------------------------------------------- SDK +class FakeFieldModel: + schema_version = 3 + input_profile = FIELD_PROFILE + model_type = "field_gain/1" + channels = {"fn": "N"} + fingerprint = "abcdef0123456789" * 4 + + def __init__(self) -> None: + self.field_baseline_N: float | None = None + + +class FakeFieldSensor: + """The slice of ``Sensor`` the pass touches, simulating the SDK's window.""" + + def __init__(self, *, zero_frames: int = 30) -> None: + self.info = SimpleNamespace( + FNORMAL=0, + FSHEARX=1, + FSHEARY=2, + FNORMAL_RAW=3, + FSHEARX_RAW=4, + FSHEARY_RAW=5, + ) + self.calibration = FakeFieldModel() + self.force_calibrated = True + self.centroid_found = True # the field model must not gate on it + self.flow = np.zeros((4, 4, 2), dtype=np.float32) + self.valid_mask = None + self.calibration_status = {"fn": "zero_unavailable"} + self.calibration_result = None + self.declared: list[str] = [] + self.set_reference_calls: list[tuple] = [] + self.pending = 0 + self.frame_index = 0 + self.fn_value = float("nan") + self.zero_frames = zero_frames + self._hhd_downsample = 2 + self._hhd_grid_shape = (2, 2) + self._hhd_ref_shape = (4, 4) + + def declare_unloaded_reference(self, *, reference_id: str) -> None: + self.declared.append(str(reference_id)) + self.pending = 0 + self.calibration.field_baseline_N = None + + def set_calibration_reference(self, raw, *, reference_id) -> None: # pragma: no cover + self.set_reference_calls.append((raw, reference_id)) + + def read_info(self, *keys): + values = { + 0: self.fn_value, + 1: float("nan"), + 2: float("nan"), + 3: 100.0, + 4: 2.0, + 5: 3.0, + } + return tuple(values[key] for key in keys) + + def step(self): + """One frame of the SDK's field model: 30-frame window, then ok frames.""" + self.frame_index += 1 + diagnostics: dict = {} + if self.declared: + if self.pending < self.zero_frames: + self.pending += 1 + diagnostics = { + "pending_reference_frames": self.pending, + "extrapolated_fraction": 0.4, + "coverage_fraction": 0.9, + "model_fingerprint": self.calibration.fingerprint, + "frame_id": self.frame_index, + } + if self.pending == self.zero_frames: + self.calibration.field_baseline_N = 0.25 + diagnostics["baseline_N"] = 0.25 + diagnostics["pending_reference_frames"] = 0 + diagnostics["extrapolated_fraction"] = 0.3 + self.calibration_status = {"fn": "zero_unavailable"} + self.fn_value = float("nan") + else: + diagnostics = { + "baseline_N": 0.25, + "extrapolated_fraction": 0.2, + "coverage_fraction": 0.97, + "model_fingerprint": self.calibration.fingerprint, + } + self.calibration_status = {"fn": "ok"} + self.fn_value = 1.5 + else: + diagnostics = {"model_fingerprint": self.calibration.fingerprint} + self.calibration_result = SimpleNamespace( + diagnostics=diagnostics, + status=dict(self.calibration_status), + values_N={"fn": None if math.isnan(self.fn_value) else self.fn_value}, + ) + return None + + +class FakeChain: + def __init__(self, sensor, *, frames: int = 35) -> None: + self.sensor = sensor + self.frames = int(frames) + self.index = 0 + self.open_seconds = 0.0 + self.debug_files: list[str] = [] + self.decode_position = 0 + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def read(self): + if self.index >= self.frames: + return False, None + self.index += 1 + return True, np.zeros((4, 4), dtype=np.uint8) + + def feed(self, frame): + return frame + + +class FakeReader: + def __init__(self, rows, *, segments: int = 1, base: Path) -> None: + self._rows = list(rows) + self._segments = [ + base / f"raw_{number:04d}.mp4" for number in range(1, segments + 1) + ] + + def session(self): + return { + "session_id": 1, + "started_at": "2026-01-01T00:00:00+0000", + "ended_at": "", + "device": {"sensor_type": "finger", "config_ref": ""}, + "stats": {"frames": len(self._rows), "segments": len(self._segments)}, + } + + def info(self): + return SimpleNamespace(frames=self._rows, video_frames=len(self._rows)) + + def segments(self): + return list(self._segments) + + def segment_path(self, number: int): + index = int(number) - 1 + return self._segments[index] if 0 <= index < len(self._segments) else None + + +def install_pass(monkeypatch, processor, chain: FakeChain, reader: FakeReader) -> None: + """Make the service walk the fake chain instead of opening the SDK.""" + monkeypatch.setattr(processor, "reader", lambda: reader) + monkeypatch.setattr(processor, "_segment_offset", lambda *args: (0, True)) + monkeypatch.setattr(service_module, "_OpenChain", lambda *args: chain) + monkeypatch.setattr(sdk_worker, "process_chain", lambda sensor: (sensor.step(), "")) + + +def new_pass_state(processor, tmp_path: Path, *, count: int = 35): + rows = field_rows(count) + return service_module._PassState( + Provenance(), [1], {1: list(rows)}, tmp_path + ) diff --git a/shared/tests/test_analysis_force_validity.py b/shared/tests/test_analysis_force_validity.py index 2a6dab9..a5e8cd3 100644 --- a/shared/tests/test_analysis_force_validity.py +++ b/shared/tests/test_analysis_force_validity.py @@ -10,6 +10,8 @@ from __future__ import annotations import math +import pytest + from orisys_app_shared.analysis.model import ALGORITHM_VERSION, ForceSample from orisys_app_shared.analysis.service import AnalysisService @@ -97,3 +99,45 @@ def test_results_stored_under_the_old_validity_rule_are_not_served(): # Version 1 marked every frame invalid when the model lacked a zero; the cache key # carries this string, so those results must miss the cache and be recomputed. assert ALGORITHM_VERSION != "flow-hhd-contact-1" + + +def test_extrapolated_fraction_is_quality_not_validity(): + """A field estimate's extrapolation is evidence about its support, not validity. + + The SDK reports the field model's extrapolated/coverage fraction in + ``calibration_diagnostics`` and keeps the per-axis status ``ok``; the app stores + the fractions as quality and must keep the newton value. Reading them as + invalidity would silently drop every experimental value. + """ + reading = _reading({"fn": "ok"}, normal=3.5) + reading["calibration_diagnostics"] = { + "extrapolated_fraction": 0.42, + "coverage_fraction": 0.9, + "baseline_N": 0.25, + "model_fingerprint": "f" * 64, + } + sample = ForceSample.from_reading(reading) + assert sample.normal == pytest.approx(3.5) + assert sample.valid + assert sample.diagnostics["extrapolated_fraction"] == pytest.approx(0.42) + assert sample.field_quality() == pytest.approx((0.42, 0.9)) + + # The force payload is unchanged by the new evidence: quality is stored in its + # own frame payload, so an older result and this one read the same force fields. + decoded = ForceSample.decode(sample.encode()) + assert decoded.normal == pytest.approx(3.5) + assert decoded.diagnostics == {} + + +def test_a_baseline_wait_still_keeps_the_raw_measurement(): + """The first 30 declared frames are ``zero_unavailable``: raw, not newton.""" + reading = _reading({"fn": "zero_unavailable"}, normal=float("nan")) + reading["calibration_diagnostics"] = { + "pending_reference_frames": 7, + "extrapolated_fraction": 0.35, + } + sample = ForceSample.from_reading(reading) + assert sample.valid + assert sample.raw() == (1234.0, -5.0, 7.0) + assert math.isnan(sample.normal) + assert sample.diagnostics["pending_reference_frames"] == 7 diff --git a/shared/tests/test_field_calibration_app.py b/shared/tests/test_field_calibration_app.py new file mode 100644 index 0000000..c892afd --- /dev/null +++ b/shared/tests/test_field_calibration_app.py @@ -0,0 +1,339 @@ +"""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 diff --git a/shared/tests/test_multical_integration.py b/shared/tests/test_multical_integration.py index 301c49d..894add8 100644 --- a/shared/tests/test_multical_integration.py +++ b/shared/tests/test_multical_integration.py @@ -182,3 +182,39 @@ def test_video_unloaded_reference_is_explicit_and_requires_the_whole_interval(mo assert len(list(processor._pass_frames(state))) == frame_count assert applied == [{"fn": 100.}] assert state.provenance.counts["unloaded_reference_frames"] == 30 + + +def test_a_field_model_reading_keeps_raw_and_reports_weighted_quality(tmp_path): + """The field model is a different zeroing protocol, not a different validity rule. + + Its reading carries the SDK's weighted diagnostics; the app must keep the raw + counts while the baseline is pending and report the extrapolated fraction as + quality (0.0 is a real value, "no diagnostics" is unknown). + """ + from orisys_app_shared.acquisition.sdk_worker import read_forces_from_sensor + from orisys_app_shared.analysis.model import ForceSample + + from _field_fakes import FakeFieldSensor # tests' own SDK double + + 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 and sample.normal_raw == pytest.approx(100.0) + assert math.isnan(sample.normal) + # Quality is a statement about the frame's support, available while the baseline + # is still pending; only absent diagnostics mean "unknown". + assert sample.field_quality() == pytest.approx((0.4, 0.9)) + reading.pop("calibration_diagnostics") + assert ForceSample.from_reading(reading).field_quality() is None + + 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))