From 11f8dc93e56ed75e512bd5d9411c3898127db847 Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Wed, 21 Oct 2026 15:46:01 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20ForceStreamSession=20=E8=A6=86=E7=9B=96?= =?UTF-8?q?=20field=20=E4=B8=8E=20FN-only=20=E6=A0=87=E5=AE=9A=EF=BC=88T06?= =?UTF-8?q?b=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 measurement_requirements，告知调用方 field 模型需要同帧场数据。补充标量、field、FN-only 三类标定与旧 ForceSession 的数值等价测试，以及标定与帧不匹配的情况：field 缺 context 时构造失败，基线帧缺 field 时零点失败，ready 后缺 field 的帧单独标为 invalid_input。与改动前的旧会话基线逐位一致。回归 381 项、lint 通过。 --- docs/force-stream-session.zh.md | 26 +-- src/orisys/force_stream_session.py | 11 ++ tests/test_force_stream_session.py | 252 ++++++++++++++++++++++++++++- 3 files changed, 279 insertions(+), 10 deletions(-) diff --git a/docs/force-stream-session.zh.md b/docs/force-stream-session.zh.md index 5be2f33..924175c 100644 --- a/docs/force-stream-session.zh.md +++ b/docs/force-stream-session.zh.md @@ -1,14 +1,18 @@ -# 新力侧会话（T06a） +# 新力侧会话（T06a/T06b） -本文件是单路 SensorStream 计划 T06a 的 SDK 合同：新的力侧会话直接消费 T05 的 +本文件是单路 SensorStream 计划 T06a/T06b 的 SDK 合同：新的力侧会话直接消费 T05 的 `ImageMeasurement`（T03 `MeasurementFrame` 头 + 同帧 raw/field），参考版本取帧上的 `reference_generation`。旧 `ForceSession` 仍为 App 保留到 T09/T10 迁移，本会话不提供 `process_measurement`、`request_zero_external_reset`、`tracker_reset_seq` 等旧外部复位接口。 ## 接口 -- `ForceStreamSession(sensor_family=..., calibration=None, settings=None)`：构造；T06a 支持 - 无标定与 schema2 标量曲线，`settings` 为 `OutputSettings`（None 用旧默认）。 +- `ForceStreamSession(sensor_family=..., calibration=None, settings=None)`：构造。T06a 支持 + 无标定与 schema2 标量曲线；T06b 补充 schema3 field 模型（`hhd_normal_l1_field/1`，需要 + `context` 与同帧 field）和 FN-only 全局增益模型（`outputs=["fn"]`）；`settings` 为 + `OutputSettings`（None 用旧默认）。 +- `measurement_requirements`：`(want_field, want_contact)`，告诉 T05 调用方是否需要把 + `field` include 进 `ImageMeasurement`（field 模型为 `(True, False)`，标量/FN-only 为 `(False, False)`）。 - `request_zero(reference_id=..., declared_unloaded=True, expected_reference_generation=版本, max_gap_s=None, max_window_s=None)`： 为指定参考版本确认无载归零；可选窗口上限。 - `process(measurement) -> ForceStreamOutput`：消费一个 T05 `ImageMeasurement`，返回逐帧结果 @@ -34,11 +38,15 @@ `reference_generation`；`quality` 为 `ok` / `uncalibrated` / `zero_pending` / `duplicate` / `late` / `session_changed` / `version_changed` / `window_failed` / `invalid`。 迟到、无效和带质量标记的结果不会被丢弃；拒绝帧仍保留 raw 特征。 -- **数值等价与滤波。** T06a 的零点/标定换算/读数滤波复用已发布算法；同一输入、无复位、 - 同一标定时与旧 `ForceSession` 逐帧相等（测试覆盖标量曲线、无标定、median+lowpass 滤波器）。 +- **数值等价与滤波。** 零点/标定换算/读数滤波复用已发布算法；同一输入、无复位、同一标定时 + 与旧 `ForceSession` 逐帧相等（T06a 覆盖标量曲线、无标定、median+lowpass 滤波器；T06b 覆盖 + `context` 与同帧 field）和 FN-only 全局增益模型（`outputs=["fn"]`）；`settings` 为 固定设计帧率与显式滤波切换属于 T07，不在本会话实现。 +- **三类标定与不匹配。** 标量曲线、field、FN-only 各有有效测试；无标定会话保持 raw 与 + `uncalibrated` 状态；field 模型缺 `context` 时构造报 `CalibrationError`，基线帧缺 field 或 + 模型拒绝帧时零点失败并给出 `invalid_input`/原因；FN-only 的 FX/FY 永远保持 NaN + + `uncalibrated`，不冒充三轴标定。 - **线程与只读。** 一个实例只属于一个力侧线程；输入数组只读，输出只含标量/状态/不可变进度。 - T06b 在本合同上补充 field 模型、FN-only 全局增益模型与三类标定的完整等价测试。 ## 示例 @@ -64,8 +72,8 @@ assert row.zero.phase in ("resetting", "collecting") ## 后续 -T06b：field（`hhd_normal_l1_field/1`）与 FN-only（schema3 `outputs=["fn"]`）标定、标定不匹配、 -field 加权基线与标量力域基线的分离，以及对照旧 ForceSession 的完整五场景数值等价。 +T06b 已交付标量/field/FN-only 三类标定与旧 ForceSession 的完整数值等价；spatial_gain/接触 +映射模型不在 T06b 验收范围，需要时另立任务。 T07 才改变输出滤波的帧率与切换语义；T09/T10 迁移 App 后删除旧 `ForceSession`。 另一项后续任务：删除旧 `ForceSession` 之前，先把零点、标定、读数引擎从旧类中抽出来， 避免新会话依赖待删除的类型（T06b 或 T09 之前）。 diff --git a/src/orisys/force_stream_session.py b/src/orisys/force_stream_session.py index 0e3f226..c9168ee 100644 --- a/src/orisys/force_stream_session.py +++ b/src/orisys/force_stream_session.py @@ -194,6 +194,17 @@ class ForceStreamSession: def last_output(self) -> ForceStreamOutput | None: return self._last_output + @property + def measurement_requirements(self) -> tuple[bool, bool]: + """``(want_field, want_contact)`` the installed calibration model asks for. + + A field model (``hhd_normal_l1_field/1``) asks for the same-frame field, so + a T05 caller must include ``field`` in the ``ImageMeasurement``; a + scalar/FN-only model asks for neither. The engine's answer is reused + instead of a second interpretation of the model. + """ + return self._engine.measurement_requirements + @property def reference_generation(self) -> int | None: """The reference generation this session currently accepts frames for.""" diff --git a/tests/test_force_stream_session.py b/tests/test_force_stream_session.py index 305f11b..0ab37f0 100644 --- a/tests/test_force_stream_session.py +++ b/tests/test_force_stream_session.py @@ -15,6 +15,7 @@ import importlib.util from pathlib import Path import numpy as np +import pytest from orisys.force_stream_session import ForceStreamOutput, ForceStreamSession from orisys.image_pipeline import ImageMeasurement @@ -44,6 +45,8 @@ def _measurement( sensor_id: str = "TEST-1", capture_time_ns: int | None = None, source_frame_id: int | None = None, + field: dict | None = None, + valid_mask: np.ndarray | None = None, ) -> ImageMeasurement: stamp = FrameStamp( sensor_id=sensor_id, @@ -61,7 +64,14 @@ def _measurement( source_dt_s=None if frame_id == 1 else FRAME_NS / 1e9, reference_generation=version, ) - return ImageMeasurement(frame=frame, force_raw=raw, torque=0.0, flow=None) + return ImageMeasurement( + frame=frame, + force_raw=raw, + torque=0.0, + flow=None, + field=field, + valid_mask=valid_mask, + ) def _raw_for(frame_id: int) -> tuple[float, float, float]: @@ -419,3 +429,243 @@ def test_version_change_branch_frame_is_registered_for_duplicate_detection() -> assert repeat.duplicate is True assert repeat.quality == "duplicate" assert new.zero == after_change, "duplicate must not be processed again" + + +# ------------------------------------------- T06b: field and FN-only calibration +def _field_payload_for(frame_id: int): + return helpers._field_payload(0.5 + 0.05 * frame_id, 0.5) + + +def _field_models(tmp_path): + document, context = helpers._field_document(tmp_path) + old_model = helpers.parse_model(document, tmp_path / "field-calibration-old.json") + new_model = helpers.parse_model(document, tmp_path / "field-calibration-new.json") + return old_model, new_model, context + + +def _field_measurement(frame_id: int, *, include_field: bool = True, mask=None): + return _measurement( + frame_id, + (0.0, 0.0, 0.0), + field=_field_payload_for(frame_id) if include_field else None, + valid_mask=np.ones((4, 4), dtype=bool) if mask is None else mask, + ) + + +def _field_old(tmp_path): + old_model, new_model, context = _field_models(tmp_path) + source = helpers.FakeMeasurementSource( + lambda frame_id: (0.0, 0.0, 0.0), + field_for=_field_payload_for, + output_mask=np.ones((4, 4), dtype=bool), + ) + session = helpers.ForceSession( + source=source, + sensor_family="finger", + context=context, + calibration=old_model, + settings=helpers.legacy_settings(), + ) + return session, source, new_model, context + + +def _compare_field_with_old(tmp_path) -> None: + old, source, new_model, context = _field_old(tmp_path) + new = ForceStreamSession( + sensor_family="finger", + context=context, + calibration=new_model, + settings=helpers.legacy_settings(), + ) + assert new.measurement_requirements == (True, False) + old.request_zero(reference_id="field-ref", declared_unloaded=True) + new.request_zero( + reference_id="field-ref", + declared_unloaded=True, + expected_reference_generation=0, + ) + old_raw, old_newton, old_status = [], [], [] + new_raw, new_newton, new_status = [], [], [] + for frame_id in range(1, 41): + old_out = helpers.process(old, source, frame_id) + new_out = new.process(_field_measurement(frame_id)) + old_raw.append(np.asarray(old_out.raw)) + old_newton.append(np.asarray(old_out.newton)) + old_status.append(list(old_out.measurement.newton_status)) + new_raw.append(np.asarray(new_out.raw)) + new_newton.append(np.asarray(new_out.newton)) + new_status.append(list(new_out.newton_status)) + assert np.array_equal(np.stack(old_raw), np.stack(new_raw), equal_nan=True) + assert np.array_equal(np.stack(old_newton), np.stack(new_newton), equal_nan=True) + assert old_status == new_status + assert new.zero.phase == "ready" + # The field model outputs FN only; shear axes stay explicitly unavailable. + assert np.isfinite(new_newton[-1][0]) + assert np.isnan(new_newton[-1][1]) and np.isnan(new_newton[-1][2]) + assert new_status[-1][1:] == ["uncalibrated", "uncalibrated"] + + +def _fn_only_models(): + context = helpers.build_context( + { + "camera": {"resolution_stitch": [4, 4]}, + "tracker": {"dis_mode": 1, "hhd_downsample": 1}, + } + ) + document = helpers._global_gain_document(context, gains={"fn": 0.01}) + old_model = helpers.parse_model(document) + new_model = helpers.parse_model(document) + return old_model, new_model, context + + +def _fn_raw(frame_id: int) -> tuple[float, float, float]: + return (100.0 + frame_id, 7.0, -7.0) + + +def _compare_fn_only_with_old() -> None: + old_model, new_model, context = _fn_only_models() + source = helpers.FakeMeasurementSource(_fn_raw) + old = helpers.ForceSession( + source=source, + sensor_family="box", + context=context, + calibration=old_model, + settings=helpers.legacy_settings(), + ) + new = ForceStreamSession( + sensor_family="box", + context=context, + calibration=new_model, + settings=helpers.legacy_settings(), + ) + assert new.measurement_requirements == (False, False) + old.request_zero(reference_id="fn-ref", declared_unloaded=True) + new.request_zero( + reference_id="fn-ref", + declared_unloaded=True, + expected_reference_generation=0, + ) + old_raw, old_newton, old_status = [], [], [] + new_raw, new_newton, new_status = [], [], [] + for frame_id in range(1, 41): + old_out = helpers.process(old, source, frame_id) + new_out = new.process(_measurement(frame_id, _fn_raw(frame_id))) + old_raw.append(np.asarray(old_out.raw)) + old_newton.append(np.asarray(old_out.newton)) + old_status.append(list(old_out.measurement.newton_status)) + new_raw.append(np.asarray(new_out.raw)) + new_newton.append(np.asarray(new_out.newton)) + new_status.append(list(new_out.newton_status)) + assert np.array_equal(np.stack(old_raw), np.stack(new_raw), equal_nan=True) + assert np.array_equal(np.stack(old_newton), np.stack(new_newton), equal_nan=True) + assert old_status == new_status + assert new.zero.phase == "ready" + assert np.isfinite(new_newton[-1][0]) + assert np.isnan(new_newton[-1][1]) and np.isnan(new_newton[-1][2]) + assert new_status[-1][1:] == ["uncalibrated", "uncalibrated"] + + +def test_field_model_matches_old_session_on_this_platform(tmp_path) -> None: + _compare_field_with_old(tmp_path) + + +def test_fn_only_model_matches_old_session_on_this_platform() -> None: + _compare_fn_only_with_old() + + +def test_three_calibration_types_match_the_old_session_on_this_platform( + tmp_path, +) -> None: + # Same interpreter/platform for both chains: no cross-OS golden comparison. + _compare_with_old(helpers.box_cal()) # schema2 scalar curves + _compare_field_with_old(tmp_path) # schema3 field model (FN-only) + _compare_fn_only_with_old() # schema3 global gain (FN-only) + + +def test_field_model_requires_context_and_field_frame(tmp_path) -> None: + from orisys.calibration import CalibrationError + + old_model, new_model, context = _field_models(tmp_path) + with pytest.raises(CalibrationError, match="pipeline context"): + ForceStreamSession(sensor_family="finger", calibration=old_model) + + session = ForceStreamSession( + sensor_family="finger", + context=context, + calibration=new_model, + settings=helpers.legacy_settings(), + ) + session.request_zero( + reference_id="field-ref", + declared_unloaded=True, + expected_reference_generation=0, + ) + missing = session.process(_field_measurement(1, include_field=False)) + assert missing.zero.phase == "failed" + assert "invalid_input" in missing.zero.failure_reason + assert all(np.isnan(value) for value in missing.newton) + assert missing.newton_status[0] == "invalid_input" + + +def test_field_model_uncalibrated_session_stays_explicit(tmp_path) -> None: + _, _, context = _field_models(tmp_path) + new = ForceStreamSession( + sensor_family="finger", + context=context, + calibration=None, + settings=helpers.legacy_settings(), + ) + new.request_zero( + reference_id="field-raw", + declared_unloaded=True, + expected_reference_generation=0, + ) + rows = [_field_measurement(frame_id) for frame_id in range(1, 32)] + results = [new.process(row) for row in rows] + assert results[-1].zero.phase == "ready" + assert all(np.isnan(value) for value in results[-1].newton) + assert set(results[-1].newton_status) == {"uncalibrated"} + assert results[-1].quality == "uncalibrated" + + +def test_fn_only_model_requires_context_and_rejects_invalid_frames() -> None: + from orisys.calibration import CalibrationError + + old_model, new_model, context = _fn_only_models() + with pytest.raises(CalibrationError, match="pipeline context"): + ForceStreamSession(sensor_family="box", calibration=old_model) + + new = ForceStreamSession( + sensor_family="box", + context=context, + calibration=new_model, + settings=helpers.legacy_settings(), + ) + new.request_zero( + reference_id="fn-ref", declared_unloaded=True, expected_reference_generation=0 + ) + invalid = new.process(_measurement(1, (float("nan"), 7.0, -7.0))) + assert invalid.zero.phase == "failed" + assert "invalid input frame" in invalid.zero.failure_reason + assert all(np.isnan(value) for value in invalid.newton) + + +def test_fn_only_model_uncalibrated_session_stays_explicit() -> None: + _, _, context = _fn_only_models() + new = ForceStreamSession( + sensor_family="box", + context=context, + calibration=None, + settings=helpers.legacy_settings(), + ) + new.request_zero( + reference_id="fn-raw", declared_unloaded=True, expected_reference_generation=0 + ) + results = [ + new.process(_measurement(frame_id, _fn_raw(frame_id))) + for frame_id in range(1, 32) + ] + assert results[-1].zero.phase == "ready" + assert all(np.isnan(value) for value in results[-1].newton) + assert set(results[-1].newton_status) == {"uncalibrated"} + assert results[-1].quality == "uncalibrated"