feat: ForceStreamSession 覆盖 field 与 FN-only 标定（T06b）

新增 measurement_requirements，告知调用方 field 模型需要同帧场数据。补充标量、field、FN-only 三类标定与旧 ForceSession 的数值等价测试，以及标定与帧不匹配的情况：field 缺 context 时构造失败，基线帧缺 field 时零点失败，ready 后缺 field 的帧单独标为 invalid_input。与改动前的旧会话基线逐位一致。回归 381 项、lint 通过。
2026-10-21 15:46:01 +08:00
parent 14e73d743c
commit 11f8dc93e5
3 changed files with 279 additions and 10 deletions
@@ -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"