采集端统一由 SDK 滤波、显示端不再二次滤波，session.json 记录生效值 - What: sdk_worker/finger_worker 删除硬编码的 sensor.configure(lowpass_cutoff_hz=LOWPASS_CUTOFF_HZ, lowpass_sample_rate_hz=...) —— 它会让界面上的曲线与 CSV 走两条不同的滤波路径；改为在会话打开设备后调用 FilterSettings.apply_to(sensor)，并把两件事分别记下来：self._filter_state = effective_filters(sensor)（设备报的**实际**生效值，供界面与记录使用）与 self._filters_unsupported（该 SDK 应用不了的键，老 SDK 必须被看见而不是被吞掉）；新增 worker 的 set_filter_settings()/filter_settings()/filter_state()/filters_unsupported() 访问器，start() 时清空上一会话的 filter_state（与 _calibration_version 同样的理由：新会话不能继承旧会话的说法）。finger_worker 保留原有的"runtime 可能拒绝 configure"的容错，但改成记日志并把全部键标为未应用。app_window 新增 _filter_settings()（从 preferences 的 plot_* 键读，空/坏值即"什么都不滤"）与 _filter_metadata()（向**正在跑的** worker 要设备实际状态，写成 session.json 的 filters 块：measurement/readout/describe/source，SDK 缺键时附 unsupported_by_sdk）；_start_session 在 start() 之前把设置交给 worker。ui/force_plots.py 把 MetricHistory 的 lowpass_cutoff_hz 从 LOWPASS_CUTOFF_HZ 改成 0.0：SDK 已经滤过一遍，且这个 buffer 就是进记录的那个，显示端再滤会让曲线画出 CSV 里根本没有的数据（原来用的还是零相位 filtfilt，非因果）。acquisition/constants.py 删掉 LOWPASS_CUTOFF_HZ（"app 在 10 Hz 滤波"这件事不存在了），FORCE_SAMPLE_RATE_HZ 保留。tests/test_force_filters.py 增 3 项：worker 未开设备时 filter_state() 为 {}（"没问过"不等于"没在滤"）、effective_filters() 对没有 filters_state() 的 SDK 返回 {}、以及对着真实 SDK 的端到端断言（external-frame 构造 box Sensor → apply_to → filters_state 回报 flowEMA(0.5)+median(3)+lowpass(10Hz,order2) → force_triplets 两个 buffer 都在 → 全关也可表达）。apps/studio/tools/ui_check.py 的 _StuckWorker 补上新合约方法（set_filter_settings/filter_settings/filter_state/filters_unsupported），否则窗口一启动就 AttributeError。 - Why: 用户的核心要求是"显示的和记录的必须一致，包括所有滤波"，并决定整条链留在 SDK、app 只通过 Sensor.configure 驱动。此前 app 硬编码 10 Hz 低通（SDK 侧）**加上**显示端 filtfilt（app 侧），同一次采集的两份数据经过两条不同的滤波路径；session.json 里的 filter 块也只是照抄常量，说的不是设备在做的事。 - Verification: shared 全量 156 passed（含新增的真实 SDK 端到端断言）；studio 85 项测试通过（修好 _StuckWorker 合约后）；ui_check 554 项检查、4 个场景 0 失败；编译通过。 - Not in this change: 曲线表头的齿轮按钮与滤波设置对话框（因此现在还只能改 preferences.json 的 plot_* 键来换滤波）、CSV 的 17 列与未滤波第二 buffer、回放界面显示滤波前曲线、发布说明里"默认不再滤波"的措辞。

shared/src/orisys_app_shared/acquisition/__init__.py

@@ -1,13 +1,9 @@
 """Acquisition helpers."""
 
-from orisys_app_shared.acquisition.constants import (
-    FORCE_SAMPLE_RATE_HZ,
-    LOWPASS_CUTOFF_HZ,
-)
+from orisys_app_shared.acquisition.constants import FORCE_SAMPLE_RATE_HZ
 
 __all__ = [
     "FORCE_SAMPLE_RATE_HZ",
-    "LOWPASS_CUTOFF_HZ",
     "FingerAcquisitionWorker",
     "SdkAcquisitionWorker",
     "SdkPacket",

shared/src/orisys_app_shared/acquisition/constants.py

@@ -1,4 +1,6 @@
 """Shared acquisition constants (no heavy imports)."""
 
-LOWPASS_CUTOFF_HZ = 10.0
+#: The rate the SDK designs its force low-pass coefficients for until the device
+#: measures its own. Also the assumption the plots and a recording's time base
+#: are built on, so it stays a constant rather than a per-run guess.
 FORCE_SAMPLE_RATE_HZ = 30.0

shared/src/orisys_app_shared/acquisition/finger_worker.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import logging
 import threading
 import time
 from pathlib import Path
@@ -9,10 +10,6 @@ from typing import Callable, Optional
 
 import numpy as np
 
-from orisys_app_shared.acquisition.constants import (
-    FORCE_SAMPLE_RATE_HZ,
-    LOWPASS_CUTOFF_HZ,
-)
 from orisys_app_shared.acquisition.sinks import RecordSinkMixin
 # The packet and the geometry come from ``models``, so the finger worker does not
 # depend on the box worker; only genuinely shared helpers are imported from it.
@@ -35,6 +32,9 @@ from orisys_app_shared.acquisition.sdk_worker import (
     process_chain,
     read_forces_from_sensor,
 )
+from orisys_app_shared.force.filters import ALL_KEYS, FilterSettings, effective_filters
+
+logger = logging.getLogger(__name__)
 
 _EXPORT_REQUIRED = ("mesh.npz", "camera_intrinsics.npz", "world_to_cam.npy")
 
@@ -99,6 +99,32 @@ class FingerAcquisitionWorker(RecordSinkMixin):
         self._pending_reset: bool = False
         self._mono_t0: float = 0.0
         self._calibration_version: str = ""
+        #: What the filters should do; set before ``start()`` (see the box worker
+        #: for the reasoning: one session, one way of filtering).
+        self._filter_settings: FilterSettings = FilterSettings()
+        #: ``filters_state()`` of the running session: the effective values.
+        self._filter_state: dict = {}
+        #: Keys the installed SDK could not apply.
+        self._filters_unsupported: tuple[str, ...] = ()
+
+    def set_filter_settings(self, settings: Optional[FilterSettings]) -> None:
+        """Choose the filters for the **next** session."""
+        with self._lock:
+            self._filter_settings = (settings or FilterSettings()).normalised()
+
+    def filter_settings(self) -> FilterSettings:
+        with self._lock:
+            return self._filter_settings
+
+    def filter_state(self) -> dict:
+        """What the running device reports as in effect (``{}`` if it cannot say)."""
+        with self._lock:
+            return dict(self._filter_state)
+
+    def filters_unsupported(self) -> tuple[str, ...]:
+        """Settings the installed SDK could not apply, reported not swallowed."""
+        with self._lock:
+            return tuple(self._filters_unsupported)
 
     @property
     def is_running(self) -> bool:
@@ -145,6 +171,10 @@ class FingerAcquisitionWorker(RecordSinkMixin):
             # zeroing request must not fire into the new session's frames.
             self._calibration_version = ""
             self._pending_reset = False
+            # Nor any filter state: until this session's sensor has been
+            # configured and asked, there is nothing to report about it.
+            self._filter_state = {}
+            self._filters_unsupported = ()
         self._session_id += 1
         session = self._session_id
         self._stop.clear()
@@ -325,13 +355,18 @@ class FingerAcquisitionWorker(RecordSinkMixin):
                 return
             raw_sensor = getattr(runtime.tactile, "raw_sensor", None)
             if raw_sensor is not None and hasattr(raw_sensor, "configure"):
+                # The filter chain lives in the SDK; the app says what the
+                # operator asked for and reads back what the device is doing.
+                with self._lock:
+                    settings = self._filter_settings
                 try:
-                    raw_sensor.configure(
-                        lowpass_cutoff_hz=LOWPASS_CUTOFF_HZ,
-                        lowpass_sample_rate_hz=FORCE_SAMPLE_RATE_HZ,
-                    )
-                except Exception:
-                    pass
+                    unsupported = settings.apply_to(raw_sensor)
+                except Exception:  # noqa: BLE001 - a finger runtime may refuse
+                    logger.exception("Could not apply filter settings to the finger sensor")
+                    unsupported = ALL_KEYS
+                with self._lock:
+                    self._filter_state = effective_filters(raw_sensor)
+                    self._filters_unsupported = tuple(unsupported)
             self._status(f"Streaming finger {mode.label()}")
             # The tactile sensor loads its own bundle calibration while opening;
             # name those coefficients from the first frame, not the request path.

shared/src/orisys_app_shared/acquisition/sdk_worker.py

@@ -18,10 +18,8 @@ import numpy as np
 
 logger = logging.getLogger(__name__)
 
-from orisys_app_shared.acquisition.constants import (
-    FORCE_SAMPLE_RATE_HZ,
-    LOWPASS_CUTOFF_HZ,
-)
+from orisys_app_shared.acquisition.constants import FORCE_SAMPLE_RATE_HZ
+from orisys_app_shared.force.filters import FilterSettings, effective_filters
 from orisys_app_shared.acquisition.models import (
     FixedCameraMode,
     SdkPacket,
@@ -409,6 +407,44 @@ class SdkAcquisitionWorker(RecordSinkMixin):
         #: Content hash of the calibration currently in effect.
         self._calibration_version: str = ""
         self._frame_id: int = 0
+        #: What the filters should do; set before ``start()``. The acquisition
+        #: thread applies it once the sensor is open, because re-configuring a
+        #: device that is being read is not safe mid-session.
+        self._filter_settings: FilterSettings = FilterSettings()
+        #: ``Sensor.filters_state()`` of the running session: the *effective*
+        #: values, which are what the dialog shows and the record keeps.
+        self._filter_state: dict = {}
+        #: Keys the installed SDK cannot apply (an SDK older than 0.5.2).
+        self._filters_unsupported: tuple[str, ...] = ()
+
+    def set_filter_settings(self, settings: Optional[FilterSettings]) -> None:
+        """Choose the filters for the **next** session.
+
+        Not applied to a running one on purpose: the filter controls are locked
+        while recording, so every frame of a session is filtered the same way.
+        """
+        with self._lock:
+            self._filter_settings = (
+                settings or FilterSettings()
+            ).normalised()
+
+    def filter_settings(self) -> FilterSettings:
+        with self._lock:
+            return self._filter_settings
+
+    def filter_state(self) -> dict:
+        """What the running device reports as in effect.
+
+        ``{}`` means the SDK could not be asked (older than 0.5.2) — which is not
+        the same as "nothing is filtering" and must not be recorded as if it were.
+        """
+        with self._lock:
+            return dict(self._filter_state)
+
+    def filters_unsupported(self) -> tuple[str, ...]:
+        """Settings the installed SDK could not apply, reported not swallowed."""
+        with self._lock:
+            return tuple(self._filters_unsupported)
 
     @property
     def is_running(self) -> bool:
@@ -454,6 +490,10 @@ class SdkAcquisitionWorker(RecordSinkMixin):
             # an old reset request must not fire into the new session's frames.
             self._calibration_version = ""
             self._pending_reset = False
+            # Same for the filter state: until this session's sensor has been
+            # configured and asked, there is nothing to report about it.
+            self._filter_state = {}
+            self._filters_unsupported = ()
         self._session_id += 1
         session = self._session_id
         self._stop.clear()
@@ -636,10 +676,14 @@ class SdkAcquisitionWorker(RecordSinkMixin):
             if self._stop.is_set() or session != self._session_id:
                 self._close_sensor(session)
                 return
-            sensor.configure(
-                lowpass_cutoff_hz=LOWPASS_CUTOFF_HZ,
-                lowpass_sample_rate_hz=FORCE_SAMPLE_RATE_HZ,
-            )
+            # The filter chain lives in the SDK; the app only says what the
+            # operator asked for and then reads back what the device is doing.
+            with self._lock:
+                settings = self._filter_settings
+            unsupported = settings.apply_to(sensor)
+            with self._lock:
+                self._filter_state = effective_filters(sensor)
+                self._filters_unsupported = tuple(unsupported)
             # A sensor loads its own bundle calibration while opening, so the
             # coefficients in effect from the very first frame are already known
             # here. Without this the record would name the path passed on the

shared/src/orisys_app_shared/force/__init__.py

@@ -17,6 +17,7 @@ from orisys_app_shared.force.filters import (
     FilterSettings,
     ForceFilterChain,
     describe_state,
+    effective_filters,
     settings_from_mapping,
 )
 from orisys_app_shared.force.history import MetricHistory, PlotState, lowpass_filter
@@ -49,6 +50,7 @@ __all__ = [
     "SessionStore",
     "build_force_calibration_v2",
     "describe_state",
+    "effective_filters",
     "export_force_calibration_v2",
     "lowpass_filter",
     "meta_from_sensor_json",

shared/src/orisys_app_shared/force/filters.py

@@ -60,6 +60,9 @@ CONFIGURE_PARAMETERS: dict[str, str] = {
     "force_lowpass_sample_rate_hz": "lowpass_sample_rate_hz",
 }
 
+#: Every key above, in chain order — "nothing could be applied" is this tuple.
+ALL_KEYS: tuple[str, ...] = tuple(CONFIGURE_PARAMETERS)
+
 #: A sensible cutoff to offer when the operator switches the low-pass on.
 SUGGESTED_LOWPASS_HZ = 10.0
 
@@ -227,7 +230,7 @@ class FilterSettings:
         try:
             parameters = inspect.signature(sensor.configure).parameters
         except (AttributeError, TypeError, ValueError):  # pragma: no cover - exotic fake
-            return tuple(CONFIGURE_PARAMETERS)
+            return ALL_KEYS
         return tuple(
             key
             for key, parameter in CONFIGURE_PARAMETERS.items()
@@ -353,6 +356,24 @@ def settings_from_mapping(data: Optional[Mapping[str, Any]]) -> FilterSettings:
     ).normalised()
 
 
+def effective_filters(sensor: object) -> dict[str, Any]:
+    """What a sensor reports as in effect, for a record or for the dialog.
+
+    ``{}`` means the question could not be asked — an SDK older than 0.5.2 has no
+    ``filters_state()`` — which is different from "nothing is filtering" and has
+    to stay different in the record.
+    """
+    reader = getattr(sensor, "filters_state", None)
+    if not callable(reader):
+        return {}
+    try:
+        state = reader()
+    except Exception:  # noqa: BLE001 - metadata must never break a session
+        logger.exception("Could not read the sensor's filter state")
+        return {}
+    return dict(state) if isinstance(state, Mapping) else {}
+
+
 # ---------------------------------------------------------------------------
 # Reference implementation: kept so the tests can prove this app understands the
 # SDK's chain exactly. Nothing in the app calls it.

shared/tests/test_force_filters.py

@@ -246,6 +246,68 @@ def test_settings_round_trip_through_a_preferences_mapping():
     assert settings_from_mapping(None).active() is False
 
 
+# ------------------------------------------------- the worker's filter contract
+def test_a_worker_reports_no_filter_state_until_a_session_has_one():
+    """``{}`` means "not asked yet", which is not the same as "nothing filters"."""
+    from orisys_app_shared.acquisition.sdk_worker import SdkAcquisitionWorker
+
+    worker = SdkAcquisitionWorker()
+    assert worker.filter_state() == {}
+    assert worker.filters_unsupported() == ()
+    worker.set_filter_settings(FilterSettings(median_window=3))
+    assert worker.filter_settings().median_window == 3
+    # Clearing the preference is "nothing filters", not "keep the last choice".
+    worker.set_filter_settings(None)
+    assert worker.filter_settings() == FilterSettings()
+    assert worker.filter_settings().active() is False
+
+
+def test_effective_filters_says_nothing_when_the_sdk_cannot_answer():
+    from orisys_app_shared.force.filters import effective_filters
+
+    assert effective_filters(object()) == {}
+
+    class Broken:
+        def filters_state(self):
+            raise RuntimeError("no")
+
+    assert effective_filters(Broken()) == {}
+
+
+def test_the_settings_reach_a_real_sensor_and_come_back_as_the_truth():
+    """The whole point of this module, against the SDK itself.
+
+    No hardware: the SDK's external-frame mode constructs a Sensor from its own
+    bundle, which is exactly what the acquisition worker does before applying
+    the session's filters.
+    """
+    orisys = pytest.importorskip("orisys")
+    from orisys_app_shared.force.filters import effective_filters
+
+    sensor = orisys.Sensor("external", config_name="box", verbose=False)
+    try:
+        assert sensor.filters_state()["any_stage_applied"] is False
+        unsupported = FilterSettings(
+            flow_ema_alpha=0.5, median_window=3, lowpass_hz=10.0
+        ).apply_to(sensor)
+        assert unsupported == (), "the installed SDK is missing filter settings"
+        state = effective_filters(sensor)
+        assert state["measurement"]["flow_ema_alpha"] == 0.5
+        assert state["readout"]["force_median_window"] == 3
+        assert state["readout"]["force_lowpass_applied"] is True
+        read_back = FilterSettings.from_filters_state(state)
+        assert read_back.describe() == "flowEMA(0.5) + median(3) + lowpass(10Hz,order2)"
+        # ...and the unfiltered pair is available for the second buffer.
+        triplets = sensor.force_triplets()
+        assert set(triplets) == {"filtered", "unfiltered"}
+        assert set(triplets["unfiltered"]) == {"raw", "newton"}
+        # Turning everything off is expressible, and the device agrees.
+        FilterSettings().apply_to(sensor)
+        assert effective_filters(sensor)["any_stage_applied"] is False
+    finally:
+        sensor.disconnect()
+
+
 # ------------------------------------------- the reference chain (SDK parity)
 def test_a_disabled_chain_is_a_pass_through():
     chain = ForceFilterChain(FilterSettings())