diff --git a/docs/force-stream-session.zh.md b/docs/force-stream-session.zh.md index 74196bc..de5b05f 100644 --- a/docs/force-stream-session.zh.md +++ b/docs/force-stream-session.zh.md @@ -17,6 +17,9 @@ 为指定参考版本确认无载归零；可选窗口上限。 - `process(measurement) -> ForceStreamOutput`：消费一个 T05 `ImageMeasurement`，返回逐帧结果 （`frame` 携带 T03 帧头与 `reference_generation`，以及 `raw/newton/status/quality/zero`）。 + `measurement_raw`/`measurement_newton` 是同帧、**未经输出滤波**的测量三元组（T09c）：接受帧取 + `ForceOutput.measurement` 的未滤波 raw/N；诊断帧的 `measurement_raw` 等于该帧 `force_raw`、 + `measurement_newton` 为 NaN。直通设置下它们与 `raw`/`newton` 逐位相同。 ## 规则 diff --git a/src/orisys/force_stream_session.py b/src/orisys/force_stream_session.py index da37bd5..5bf216f 100644 --- a/src/orisys/force_stream_session.py +++ b/src/orisys/force_stream_session.py @@ -93,6 +93,9 @@ class ForceStreamOutput: warmup_required_frames: int = 0 warmup_elapsed_frames: int = 0 warming_up: bool = False + #: T09c: the same frame's unfiltered measurement, before the output chain. + measurement_raw: tuple[float, float, float] = (float("nan"),) * 3 + measurement_newton: tuple[float, float, float] = (float("nan"),) * 3 @property def reference_generation(self) -> int: @@ -359,6 +362,8 @@ class ForceStreamSession: accepted: bool = True, duplicate: bool = False, late: bool = False, + measurement_raw: tuple[float, float, float] | None = None, + measurement_newton: tuple[float, float, float] | None = None, ) -> ForceStreamOutput: output = ForceStreamOutput( frame=frame, @@ -381,6 +386,16 @@ class ForceStreamSession: warmup_required_frames=self._filter.warmup_required_frames, warmup_elapsed_frames=self._filter.warmup_elapsed_frames, warming_up=self._filter.warming_up, + measurement_raw=( + (float("nan"),) * 3 + if measurement_raw is None + else tuple(float(value) for value in measurement_raw) + ), + measurement_newton=( + (float("nan"),) * 3 + if measurement_newton is None + else tuple(float(value) for value in measurement_newton) + ), ) self._last_output = output return output @@ -409,6 +424,8 @@ class ForceStreamSession: accepted=False, duplicate=duplicate, late=late, + measurement_raw=raw, + measurement_newton=(float("nan"),) * 3, ) @staticmethod @@ -652,4 +669,8 @@ class ForceStreamSession: quality=quality, diagnostics=tuple(output.measurement.diagnostics), accepted=True, + measurement_raw=tuple(float(value) for value in output.measurement.sdk.raw), + measurement_newton=tuple( + float(value) for value in output.measurement.newton + ), ) diff --git a/tests/test_force_stream_session.py b/tests/test_force_stream_session.py index 0ab37f0..55b6fcb 100644 --- a/tests/test_force_stream_session.py +++ b/tests/test_force_stream_session.py @@ -669,3 +669,84 @@ def test_fn_only_model_uncalibrated_session_stays_explicit() -> None: 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_unfiltered_measurement_fields_survive_the_output_filter() -> None: + inputs = [100.0, 100.0, 100.0, 900.0, 100.0, 120.0, 140.0, 160.0] + filtered = _new_chain(settings=helpers.legacy_settings(median_window=5)) + passthrough = _new_chain(settings=helpers.legacy_settings(median_window=1)) + for frame_id, value in enumerate(inputs, start=1): + one = filtered.process(_measurement(frame_id, (value, 0.0, 0.0))) + two = passthrough.process(_measurement(frame_id, (value, 0.0, 0.0))) + assert one.measurement_raw == (value, 0.0, 0.0) + assert two.raw == (value, 0.0, 0.0) + assert two.measurement_raw == one.measurement_raw + assert one.measurement_newton == two.measurement_newton + assert filtered.process(_measurement(9, (900.0, 0.0, 0.0))).raw[0] != 900.0 + assert passthrough.process(_measurement(9, (900.0, 0.0, 0.0))).raw[0] == 900.0 + + +def test_diagnostic_rows_keep_the_unfiltered_measurement() -> None: + session = _new_chain(settings=helpers.legacy_settings(median_window=5)) + session.process(_measurement(20, _raw_for(20))) + duplicate = session.process(_measurement(20, _raw_for(20))) + assert duplicate.quality == "duplicate" + assert duplicate.measurement_raw == tuple(float(value) for value in _raw_for(20)) + assert duplicate.raw == duplicate.measurement_raw + assert all(np.isnan(value) for value in duplicate.measurement_newton) + assert all(np.isnan(value) for value in duplicate.newton) + + +def test_calibrated_measurement_newton_is_unfiltered() -> None: + filt = _new_chain( + helpers.box_cal(), + helpers.legacy_settings( + median_window=5, lowpass_hz=3.0, lowpass_order=2, sample_rate_hz=30.0 + ), + ) + passthrough = _new_chain( + helpers.box_cal(), helpers.legacy_settings(median_window=1) + ) + for session in (filt, passthrough): + session.request_zero( + reference_id="r", + declared_unloaded=True, + expected_reference_generation=0, + ) + rows = [] + pass_rows = [] + for frame_id in range(1, 80): + raw = list(_raw_for(1)) + if frame_id == 60: + raw[0] += 20000.0 + rows.append(filt.process(_measurement(frame_id, tuple(raw)))) + pass_rows.append(passthrough.process(_measurement(frame_id, tuple(raw)))) + for filtered, plain in zip(rows, pass_rows): + for measured, reference in zip( + filtered.measurement_newton, plain.newton + ): + assert (np.isnan(measured) and np.isnan(reference)) or measured == reference + assert filtered.measurement_raw == plain.raw + spike = rows[59] + assert spike.newton[0] != spike.measurement_newton[0] + + +def test_late_frame_carries_unfiltered_measurement_fields() -> None: + session = _new_chain( + helpers.box_cal(), + helpers.legacy_settings( + median_window=5, lowpass_hz=3.0, lowpass_order=2, sample_rate_hz=30.0 + ), + ) + session.request_zero( + reference_id="r", declared_unloaded=True, expected_reference_generation=0 + ) + for frame_id in range(1, 32): + session.process(_measurement(frame_id, _raw_for(frame_id))) + raw = _raw_for(32) + late = session.process( + _measurement(32, raw, capture_time_ns=31 * FRAME_NS) + ) + assert late.late is True + assert late.measurement_raw == tuple(float(value) for value in raw) + assert all(np.isnan(value) for value in late.measurement_newton)