标定指纹改读生效值并按测量/读数分组 - What: calibration.py 新增 FILTER_MEASUREMENT_KEYS（flow_ema_alpha）与 FILTER_READOUT_KEYS（中值窗口、低通截止/阶数/采样率/rate_auto）、filters_fingerprint()（把 Sensor.filters_state() 归一化成 {measurement, readout, preset, source}：非法值丢弃、来源没写的键不出现而不是补默认、空映射返回 {} 表示"无记录"）、filter_mismatches()（按组判读差异）与 _filters_from_config / _flat_filters_from（扁平旧格式与 preset 展开的兼容解析，显式键优先）；pipeline_from_sensor_json() 新增 filters_state= 关键字——传了生效值就用生效值，否则读文件的 filters 段并标注 source="sensor.json.filters"，返回结构化 filters 键，删除原先会凭空补出 0.5/3/10 的扁平键与 enabled 覆盖逻辑；Calibration.to_dict() 写 meta.filters（同时保留扁平键供旧读者按名取用）；parse() 把 filters 读回 pipeline（此前 to_dict 写进 meta 的滤波键在读回时被整段丢弃，指纹实际是"只写不读"）；check() 改用 filter_mismatches 分开报告——measurement 组差异进 warnings 且措辞含 not comparable，readout 组差异进新增字段 CheckResult.notes（只记录、不拦截）；build(extra_meta=…) 把 filters 与扁平滤波键折进结构化组。tests/test_calibration.py 新增 11 项测试（分组、preset 展开与显式键优先、enabled:false 全关、非法值为空而非默认、写入-读回往返、旧扁平文档读回、生效值压过文件、四个预设逐档核对）。docs/filters.md 新增第 7 节。 - Why: 用户决策 B 要求"Sensor.filters_state() 作为唯一生效值出口，calibration.py 指纹与 session.json 都读它、不读配置文件，指纹分 measurement/readout 两组"；动手时确认旧实现三个问题叠加：来源错（读文件、被 filters.enabled 覆盖、文件没写就补默认 0.5/3/10），写进 meta 的指纹在读回时被丢弃，check() 因此永远比不出滤波差异——审计链是断的，不是不精确。 - Verification: SDK 全量 pytest 335 passed / 15 skipped，唯一失败 test_vnormal_longrun_no_outlier_jumps 需要 1200 帧而 data/sample/s1_480p.mp4 只有 577 帧（环境/数据，与本改动无关）；external-frame 模式逐预设构造真实 Sensor，filters_fingerprint(filters_state()) 得 off(1.0,1,0.0) / low_latency(1.0,3,0.0) / balanced(0.5,3,0.0) / low_noise(0.3,3,10.0)；写入→读回往返后 measurement/readout/preset 完整保留且 to_dict 幂等；旧扁平 meta 文档仍读回同一指纹；.repro/verify_cal_filters.py 50 项全绿。 - Not in this change: 未动 app（app 的 meta_from_sensor_json 只取 4 个键，不受影响）；未升级 venv 内 SDK；examples/filters 的四个机器可读预设片段与 app 侧 wrapper/齿轮/CSV 17 列未动。

docs/filters.md

@@ -72,3 +72,44 @@ t["unfiltered"]["raw"], t["unfiltered"]["newton"]   # 中值/低通之前；同
 3. 记录里主列 = 滤波后（与界面一致），未滤波列另存；列结构与格式版本可区分。
 4. 标定/可比性指纹用 `measurement` 组；`readout` 组变化只记录、不拦截。
 5. 双份数据在同一帧、同一份标定系数下产生，差异可完全归因于滤波链。
+
+## 7. 指纹：标定文件里怎么记（已实现）
+
+`orisys.calibration` 负责把滤波设置写进标定文件的 `meta`，并在读回时还原：
+
+```python
+from orisys.calibration import filters_fingerprint, filter_mismatches, pipeline_from_sensor_json
+
+pipeline = pipeline_from_sensor_json(config_path, filters_state=sensor.filters_state())
+# {"stitch_resolution": …, "dis_mode": …, "hhd_downsample": …, "config_ref": …,
+#  "filters": {"measurement": {"flow_ema_alpha": 0.5},
+#              "readout": {"force_median_window": 3, "force_lowpass_cutoff_hz": 0.0},
+#              "preset": "balanced", "source": "sensor.filters_state"}}
+```
+
+写入的文件长这样（`meta` 里同时保留扁平键供旧读者按名取用）：
+
+```json
+"meta": {
+  "filters": {
+    "measurement": {"flow_ema_alpha": 0.5},
+    "readout": {"force_median_window": 3, "force_lowpass_cutoff_hz": 0.0},
+    "preset": "balanced",
+    "source": "sensor.filters_state"
+  },
+  "flow_ema_alpha": 0.5,
+  "force_median_window": 3
+}
+```
+
+规则：
+
+- **生效值优先**：传了 `filters_state=` 就用它；没传才读文件的 `filters` 段，并标成
+  `"source": "sensor.json.filters"`（那是"意图"，不一定等于设备正在做的事）。
+- **不猜**：来源没写的键**不出现**在指纹里。`{}` 表示"没有任何记录"，绝不表示"大概是关的"。
+  旧文档写的扁平键（`meta.flow_ema_alpha` …）在读回时会被折进对应的组，因此仍可比。
+- **`enabled: false`** 只作为"全部关"的废弃别名保留（`flow_ema_alpha=1`、中值 `1`、低通 `0`）。
+- **分组判读**：`calibration.check(doc, pipeline=…)` 把两组的差异分开报告 —— measurement 组的
+  差异进 `warnings`（措辞含 `not comparable`，即"两份记录不可比"），readout 组的差异进
+  `notes`（只记录，不拦截）。`CheckResult.notes` 是为此新增的字段。
+

src/orisys/calibration.py

@@ -47,11 +47,186 @@ _KNOWN_TOP_KEYS = frozenset(
     {"schema_version", "unit", "channels", "meta", "identity", "pipeline", "provenance", "measurements", "quality"}
 )
 
+#: Filter stages that change **what is measured**: they sit upstream of the
+#: force computation, so two recordings with different values are not comparable.
+#: They belong to the measurement-pipeline identity, for the same reason
+#: ``dis_mode`` and ``hhd_downsample`` do.
+FILTER_MEASUREMENT_KEYS: tuple[str, ...] = ("flow_ema_alpha",)
+#: Filter stages that only smooth the force triple on its way out. They are
+#: recorded so a file explains its own numbers, but a difference between two
+#: files does not invalidate a comparison.
+FILTER_READOUT_KEYS: tuple[str, ...] = (
+    "force_median_window",
+    "force_lowpass_cutoff_hz",
+    "force_lowpass_order",
+    "force_lowpass_sample_rate_hz",
+    "force_lowpass_rate_auto",
+)
+#: Every recorded filter value, measurement group first.
+FILTER_KEYS: tuple[str, ...] = FILTER_MEASUREMENT_KEYS + FILTER_READOUT_KEYS
+#: What "off" looks like, per stage. ``enabled: false`` means exactly this.
+_OFF_FILTER_VALUES: dict[str, Any] = {
+    "flow_ema_alpha": 1.0,
+    "force_median_window": 1,
+    "force_lowpass_cutoff_hz": 0.0,
+}
+_INT_FILTER_KEYS = frozenset({"force_median_window", "force_lowpass_order"})
+_FLOAT_FILTER_KEYS = frozenset(
+    {"flow_ema_alpha", "force_lowpass_cutoff_hz", "force_lowpass_sample_rate_hz"}
+)
+
 
 class CalibrationError(ValueError):
     """Invalid or unusable force calibration document / path."""
 
 
+def _coerce_filter_value(key: str, value: Any) -> Any:
+    """A filter value in its recorded type, or ``None`` when unusable."""
+    if value is None:
+        return None
+    if key == "force_lowpass_rate_auto":
+        return bool(value)
+    try:
+        if key in _INT_FILTER_KEYS:
+            return int(value)
+        if key in _FLOAT_FILTER_KEYS:
+            return float(value)
+    except (TypeError, ValueError):
+        return None
+    return value
+
+
+def _filters_from_config(block: Mapping[str, Any] | Any) -> dict[str, dict[str, Any]]:
+    """Best-effort filter groups from a flat ``filters`` block.
+
+    Only what the mapping states — directly or through a named preset, with an
+    explicit key winning over the preset — is reported. Nothing is filled in from
+    a default: a default is how a document ends up claiming a filter the device
+    is not running. ``enabled: false`` keeps its deprecated meaning of "every
+    stage off".
+    """
+    flat = dict(block) if isinstance(block, Mapping) else {}
+    preset_name = flat.get("preset") if isinstance(flat.get("preset"), str) else None
+    merged: dict[str, Any] = {}
+    if preset_name:
+        try:
+            from .sensor import FILTER_PRESETS  # deferred: sensor imports this module
+        except Exception:  # pragma: no cover - defensive
+            FILTER_PRESETS = {}
+        merged.update(FILTER_PRESETS.get(preset_name, {}) or {})
+    for key in FILTER_KEYS:
+        if flat.get(key) is not None:
+            merged[key] = flat[key]
+    if isinstance(flat.get("force_lowpass_sample_rate_hz"), str):
+        # "auto" / "measured" is a rate *source*, not a number; keep it as one.
+        merged["force_lowpass_rate_auto"] = True
+        merged.pop("force_lowpass_sample_rate_hz", None)
+    if flat.get("enabled", None) is False:
+        order = merged.get("force_lowpass_order")
+        rate_auto = merged.get("force_lowpass_rate_auto")
+        merged = dict(_OFF_FILTER_VALUES)
+        # Order and rate tracking stay meaningful even with the stage off.
+        if order is not None:
+            merged["force_lowpass_order"] = order
+        if rate_auto is not None:
+            merged["force_lowpass_rate_auto"] = rate_auto
+    return {
+        group: {key: _coerce_filter_value(key, merged.get(key)) for key in keys}
+        for group, keys in (
+            ("measurement", FILTER_MEASUREMENT_KEYS),
+            ("readout", FILTER_READOUT_KEYS),
+        )
+    }
+
+
+def _flat_filters_from(*sources: Mapping[str, Any] | Any) -> dict[str, Any]:
+    """The flat form older documents used: filter keys at the top of meta."""
+    out: dict[str, Any] = {}
+    for source in sources:
+        if not isinstance(source, Mapping):
+            continue
+        for key in FILTER_KEYS:
+            if key not in out and source.get(key) is not None:
+                out[key] = source[key]
+    return out
+
+
+def filters_fingerprint(state: Mapping[str, Any] | Any) -> dict[str, Any]:
+    """Split a filter state into its ``measurement`` and ``readout`` groups.
+
+    The honest input is ``Sensor.filters_state()`` — the values that actually
+    took effect, which is not necessarily what the config file says. A flat
+    mapping is accepted as a fallback (a sensor.json ``filters`` block, or the
+    flat form older documents wrote) and read as the intent it is.
+
+    Values a source does not state are absent from the result rather than
+    guessed, so ``{}`` means "nothing recorded" and never "probably off".
+    """
+    if not isinstance(state, Mapping) or not state:
+        return {}
+    measurement = state.get("measurement")
+    readout = state.get("readout")
+    if isinstance(measurement, Mapping) or isinstance(readout, Mapping):
+        groups = {
+            "measurement": {
+                key: _coerce_filter_value(key, measurement.get(key))
+                for key in FILTER_MEASUREMENT_KEYS
+            }
+            if isinstance(measurement, Mapping)
+            else {},
+            "readout": {
+                key: _coerce_filter_value(key, readout.get(key))
+                for key in FILTER_READOUT_KEYS
+            }
+            if isinstance(readout, Mapping)
+            else {},
+        }
+        source = str(state.get("source") or "sensor.filters_state")
+    else:
+        groups = _filters_from_config(state)
+        source = "sensor.json.filters"
+    out: dict[str, Any] = {}
+    for group in ("measurement", "readout"):
+        values = {
+            key: value for key, value in (groups.get(group) or {}).items() if value is not None
+        }
+        if values:
+            out[group] = values
+    if not out:
+        return {}
+    if state.get("preset"):
+        out["preset"] = str(state["preset"])
+    out["source"] = source
+    return out
+
+
+def filter_mismatches(
+    recorded: Mapping[str, Any] | Any, expected: Mapping[str, Any] | Any
+) -> tuple[list[str], list[str]]:
+    """Compare two filter fingerprints → ``(measurement, readout)`` differences.
+
+    A measurement difference means the two records are not comparable, so it is
+    reported as a warning. A readout difference only changes how the same
+    measurement was smoothed, so it is a note: recorded, never an obstacle.
+    """
+    left = filters_fingerprint(recorded)
+    right = filters_fingerprint(expected)
+    warns: list[str] = []
+    notes: list[str] = []
+    for group, keys, bucket, label in (
+        ("measurement", FILTER_MEASUREMENT_KEYS, warns, "measurement (not comparable)"),
+        ("readout", FILTER_READOUT_KEYS, notes, "readout"),
+    ):
+        for key in keys:
+            if key in (left.get(group) or {}) and key in (right.get(group) or {}):
+                if left[group][key] != right[group][key]:
+                    bucket.append(
+                        f"filter {label} mismatch: {key} "
+                        f"{left[group][key]!r} != {right[group][key]!r}"
+                    )
+    return warns, notes
+
+
 def canonicalize_channel(name: str) -> str:
     """Map FN / normal / shear_x / … to canonical ``fn`` / ``fx`` / ``fy``."""
     key = str(name or "").strip()
@@ -175,6 +350,9 @@ class CheckResult:
     ok: bool
     errors: list[str] = field(default_factory=list)
     warnings: list[str] = field(default_factory=list)
+    #: Differences worth recording that are not a reason to refuse anything,
+    #: such as a readout-only filter change between two documents.
+    notes: list[str] = field(default_factory=list)
 
 
 @dataclass(frozen=True)
@@ -242,11 +420,19 @@ class Calibration:
         meta: dict[str, Any] = {}
         meta.update(self.provenance)
         meta.update({k: v for k, v in self.identity.items() if k not in meta})
-        for key in ("stitch_resolution", "dis_mode", "hhd_downsample",
+        for key in ("stitch_resolution", "dis_mode", "hhd_downsample", "config_ref"):
-                   "flow_ema_alpha", "force_median_window", "force_lowpass_cutoff_hz",
-                   "config_ref"):
             if key in self.pipeline and self.pipeline[key] is not None:
                 meta[key] = self.pipeline[key]
+        fingerprint = filters_fingerprint(self.pipeline.get("filters"))
+        if fingerprint:
+            # Structured, so a reader can tell a change that makes two records
+            # incomparable (measurement) from one that only changes how the same
+            # measurement was smoothed (readout).
+            meta["filters"] = fingerprint
+            # The flat keys are kept as well for readers that look them up by
+            # name; the groups above are the authoritative form.
+            for group in ("measurement", "readout"):
+                meta.update(fingerprint.get(group) or {})
         if self.measurements:
             meta["samples"] = self.measurements
         if self.origin and "origin" not in meta:
@@ -310,6 +496,7 @@ def check(doc: dict | Calibration, *, pipeline: dict | None = None) -> CheckResu
     """Validate a document or Calibration. Errors refuse apply/write."""
     errors: list[str] = []
     warns: list[str] = []
+    notes: list[str] = []
 
     if isinstance(doc, Calibration):
         data = doc.to_dict()
@@ -389,12 +576,18 @@ def check(doc: dict | Calibration, *, pipeline: dict | None = None) -> CheckResu
         warns.append("empty sn on factory calibration")
 
     if pipeline is not None and isinstance(doc, Calibration):
-        for key in ("stitch_resolution", "dis_mode", "hhd_downsample",
+        for key in ("stitch_resolution", "dis_mode", "hhd_downsample"):
-                "flow_ema_alpha", "force_median_window", "force_lowpass_cutoff_hz"):
             if key in doc.pipeline and key in pipeline and doc.pipeline[key] != pipeline[key]:
                 warns.append(f"pipeline mismatch: {key}")
+        # The filter groups are compared apart on purpose: ``flow_ema_alpha``
+        # changes what was measured, the readout stages only smooth it.
+        filter_warns, filter_notes = filter_mismatches(
+            doc.pipeline.get("filters"), pipeline.get("filters")
+        )
+        warns.extend(filter_warns)
+        notes.extend(filter_notes)
 
-    return CheckResult(ok=not errors, errors=errors, warnings=warns)
+    return CheckResult(ok=not errors, errors=errors, warnings=warns, notes=notes)
 
 
 def parse(data: dict, *, source_path: Path | None = None) -> Calibration:
@@ -495,6 +688,14 @@ def parse(data: dict, *, source_path: Path | None = None) -> Calibration:
         "hhd_downsample": pipeline_in.get("hhd_downsample", meta.get("hhd_downsample")),
         "config_ref": pipeline_in.get("config_ref", meta.get("config_ref")),
         "config_hash": pipeline_in.get("config_hash", meta.get("config_hash")),
+        # Structured when the document has it, otherwise rebuilt from the flat
+        # keys older documents wrote — reading them back is what makes the
+        # fingerprint checkable at all.
+        "filters": filters_fingerprint(
+            pipeline_in.get("filters")
+            or meta.get("filters")
+            or _flat_filters_from(pipeline_in, meta)
+        ),
     }
     provenance = {
         "origin": provenance_in.get("origin", meta.get("origin", "")),
@@ -603,32 +804,47 @@ def load(
         return None
 
 
-def pipeline_from_sensor_json(path: str | Path | None) -> dict[str, Any]:
+def pipeline_from_sensor_json(
-    """Pull stitch / tracker / filter fields from a sensor.json for export pipeline/meta."""
+    path: str | Path | None,
+    *,
+    filters_state: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Pull stitch / tracker / filter fields from a sensor.json for export pipeline/meta.
+
+    ``filters_state`` is the device's ``Sensor.filters_state()``. Pass it
+    whenever a device is available: a config file states an intent that a runtime
+    ``configure()`` may have moved on from, so only the effective values are
+    worth recording. Without it the file's ``filters`` block is reported for what
+    it is — the intent — and values it does not state are left out rather than
+    filled in with a default.
+
+    The result carries the filter settings in one structured ``"filters"`` key
+    (``measurement`` / ``readout`` / ``preset`` / ``source``), not as loose
+    top-level keys: the groups decide whether two records are comparable, so they
+    have to survive a round trip intact.
+    """
     pipeline: dict[str, Any] = {
         "stitch_resolution": None,
         "dis_mode": None,
         "hhd_downsample": None,
-        "flow_ema_alpha": None,
+        "filters": {},
-        "force_median_window": None,
-        "force_lowpass_cutoff_hz": None,
         "config_ref": None,
     }
-    if path is None:
+    if path is None and filters_state is None:
         return pipeline
-    p = Path(path)
+    p = Path(path) if path is not None else None
+    if p is not None:
         pipeline["config_ref"] = str(p)
-    if not p.is_file():
+    data: dict[str, Any] = {}
-        return pipeline
+    if p is not None and p.is_file():
         try:
-        data = json.loads(p.read_text(encoding="utf-8"))
+            loaded = json.loads(p.read_text(encoding="utf-8"))
         except Exception:
-        return pipeline
+            loaded = None
-    if not isinstance(data, dict):
+        if isinstance(loaded, dict):
-        return pipeline
+            data = loaded
     camera = data.get("camera") if isinstance(data.get("camera"), dict) else {}
     tracker = data.get("tracker") if isinstance(data.get("tracker"), dict) else {}
-    filters = data.get("filters") if isinstance(data.get("filters"), dict) else {}
     stitch = camera.get("resolution_stitch")
     if isinstance(stitch, (list, tuple)) and len(stitch) >= 2:
         pipeline["stitch_resolution"] = [int(stitch[0]), int(stitch[1])]
@@ -642,29 +858,12 @@ def pipeline_from_sensor_json(path: str | Path | None) -> dict[str, Any]:
             pipeline["hhd_downsample"] = int(tracker["hhd_downsample"])
         except (TypeError, ValueError):
             pipeline["hhd_downsample"] = tracker.get("hhd_downsample")
-    # Signal-conditioning settings change the reported force values, so they
+    # Signal conditioning is part of the measurement-pipeline identity: these
-    # belong to the pipeline identity just like dis_mode / hhd_downsample.
+    # stages change the reported force values, so they are recorded next to
-    if filters:
+    # dis_mode / hhd_downsample. The effective state wins over the file.
-        enabled = bool(filters.get("enabled", True))
+    pipeline["filters"] = filters_fingerprint(
-        for key in ("flow_ema_alpha", "force_lowpass_cutoff_hz"):
+        filters_state if filters_state else data.get("filters")
-            if key in filters:
-                try:
-                    pipeline[key] = float(filters[key]) if enabled else 0.0
-                except (TypeError, ValueError):
-                    pipeline[key] = filters.get(key)
-            elif key == "flow_ema_alpha":
-                pipeline[key] = 0.5 if enabled else 0.0
-            else:
-                pipeline[key] = 10.0 if enabled else 0.0
-        if "force_median_window" in filters:
-            try:
-                pipeline["force_median_window"] = (
-                    int(filters["force_median_window"]) if enabled else 1
     )
-            except (TypeError, ValueError):
-                pipeline["force_median_window"] = filters.get("force_median_window")
-        else:
-            pipeline["force_median_window"] = 3 if enabled else 1
     return pipeline
 
 
@@ -747,10 +946,18 @@ def build(
         for k, v in extra_meta.items():
             if k == "samples":
                 continue
-            if k in ("stitch_resolution", "dis_mode", "hhd_downsample", "config_ref",
+            if k in ("stitch_resolution", "dis_mode", "hhd_downsample", "config_ref"):
-                "flow_ema_alpha", "force_median_window", "force_lowpass_cutoff_hz"):
                 if k not in pipeline or pipeline[k] is None:
                     pipeline[k] = v
+            elif k == "filters":
+                if not pipeline.get("filters"):
+                    pipeline["filters"] = filters_fingerprint(v)
+            elif k in FILTER_KEYS:
+                # Flat form: fold the loose key into the structured groups.
+                merged = deepcopy(pipeline.get("filters") or {})
+                group = "measurement" if k in FILTER_MEASUREMENT_KEYS else "readout"
+                merged.setdefault(group, {})[k] = v
+                pipeline["filters"] = filters_fingerprint(merged)
             elif k in ("sensor_type", "sn", "hardware_rev", "gel_id"):
                 if k not in identity or not identity[k]:
                     identity[k] = v

tests/test_calibration.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import shutil
 from pathlib import Path
 
 import pytest
@@ -13,8 +14,11 @@ from orisys.calibration import (
     CalibrationError,
     build,
     check,
+    filter_mismatches,
+    filters_fingerprint,
     load,
     parse,
+    pipeline_from_sensor_json,
     write,
 )
 
@@ -241,3 +245,186 @@ def test_parse_shipped_box_file_directly():
     cal = parse(json.loads(_box_cal_path().read_text(encoding="utf-8")), source_path=_box_cal_path())
     assert cal.schema_version == 1
     assert cal.convert(10000.0, 36000.0, -36000.0) == pytest.approx((0.377, 1.19, -1.19))
+
+
+# ------------------------------------------------------- filter fingerprint
+ALL_OFF_FILTERS = {
+    "flow_ema_alpha": 1.0,
+    "force_median_window": 1,
+    "force_lowpass_cutoff_hz": 0.0,
+    "force_lowpass_order": 2,
+    "force_lowpass_sample_rate_hz": "auto",
+}
+
+
+def _sensor_with_filters(tmp_path: Path, filters: dict) -> Sensor:
+    """An external-frame Sensor whose bundle carries ``filters``."""
+    bundle = _box_cal_path().parent
+    copy = tmp_path / "box"
+    shutil.copytree(bundle, copy)
+    path = copy / "sensor.json"
+    cfg = json.loads(path.read_text(encoding="utf-8"))
+    cfg["filters"] = filters
+    path.write_text(json.dumps(cfg, indent=2), encoding="utf-8")
+    return Sensor(
+        "external",
+        config_name=str(path),
+        cal_path=str(copy / "force_calibration.json"),
+        verbose=False,
+        backend="cpu",
+    )
+
+
+def test_the_fingerprint_keeps_measurement_and_readout_apart():
+    """A change to one group must not be confused with a change to the other."""
+    fp = filters_fingerprint(ALL_OFF_FILTERS)
+    assert fp["measurement"] == {"flow_ema_alpha": 1.0}
+    assert fp["readout"] == {
+        "force_median_window": 1,
+        "force_lowpass_cutoff_hz": 0.0,
+        "force_lowpass_order": 2,
+        "force_lowpass_rate_auto": True,
+    }
+
+
+def test_a_preset_expands_without_inventing_anything():
+    fp = filters_fingerprint({"preset": "low_noise"})
+    assert fp["preset"] == "low_noise"
+    assert fp["measurement"] == {"flow_ema_alpha": 0.3}
+    assert fp["readout"] == {"force_median_window": 3, "force_lowpass_cutoff_hz": 10.0}
+    # An explicit key wins over the preset.
+    explicit = filters_fingerprint({"preset": "low_noise", "force_lowpass_cutoff_hz": 0.0})
+    assert explicit["readout"]["force_lowpass_cutoff_hz"] == 0.0
+    # Keys no source states are absent, never filled in from a default.
+    assert filters_fingerprint({}) == {}
+    assert filters_fingerprint(None) == {}
+    assert filters_fingerprint({"flow_ema_alpha": "not-a-number"}) == {}
+
+
+def test_the_deprecated_master_switch_still_means_all_off():
+    fp = filters_fingerprint(dict(ALL_OFF_FILTERS, enabled=False, flow_ema_alpha=0.3))
+    assert fp["measurement"] == {"flow_ema_alpha": 1.0}
+    assert fp["readout"]["force_median_window"] == 1
+    assert fp["readout"]["force_lowpass_cutoff_hz"] == 0.0
+
+
+def test_a_measurement_change_warns_and_a_readout_change_only_notes():
+    recorded = filters_fingerprint({"preset": "balanced"})
+    expected = filters_fingerprint({"preset": "low_noise"})
+    warns, notes = filter_mismatches(recorded, expected)
+    assert any("not comparable" in w and "flow_ema_alpha" in w for w in warns)
+    assert not any("flow_ema_alpha" in n for n in notes)
+    assert any("force_lowpass_cutoff_hz" in n for n in notes)
+    assert filter_mismatches(recorded, recorded) == ([], [])
+
+
+def test_check_reports_the_two_groups_apart():
+    samples = {
+        key: [
+            CalSample(weight_n=0.0, raw_mean=0.0, n=10),
+            CalSample(weight_n=1.0, raw_mean=1000.0, n=10),
+        ]
+        for key in ("fn", "fx", "fy")
+    }
+    doc = build(
+        samples,
+        identity={"sensor_type": "box", "sn": "FP-1"},
+        pipeline={"filters": filters_fingerprint({"preset": "balanced"})},
+    )
+    result = check(doc, pipeline={"filters": filters_fingerprint({"preset": "low_noise"})})
+    assert result.ok, "a filter difference must not refuse a calibration"
+    assert any("not comparable" in w for w in result.warnings)
+    assert any("readout" in n for n in result.notes)
+    assert not any("readout" in w for w in result.warnings)
+
+
+def test_the_fingerprint_survives_a_write_and_read(tmp_path: Path):
+    samples = {
+        key: [
+            CalSample(weight_n=0.0, raw_mean=0.0, n=10),
+            CalSample(weight_n=1.0, raw_mean=1000.0, n=10),
+        ]
+        for key in ("fn", "fx", "fy")
+    }
+    doc = build(
+        samples,
+        identity={"sensor_type": "box", "sn": "FP-2"},
+        pipeline={"filters": filters_fingerprint({"preset": "balanced"})},
+    )
+    path = tmp_path / "fp.json"
+    write(doc, path, overwrite=True)
+    raw = json.loads(path.read_text(encoding="utf-8"))
+    assert raw["meta"]["filters"]["measurement"] == {"flow_ema_alpha": 0.5}
+    # The flat keys stay for readers that look them up by name.
+    assert raw["meta"]["flow_ema_alpha"] == 0.5
+
+    loaded = load(path)
+    assert loaded.pipeline["filters"]["measurement"] == {"flow_ema_alpha": 0.5}
+    assert loaded.pipeline["filters"]["readout"]["force_median_window"] == 3
+    assert loaded.pipeline["filters"]["preset"] == "balanced"
+    assert loaded.to_dict()["meta"]["filters"] == doc.to_dict()["meta"]["filters"]
+
+
+def test_a_legacy_flat_fingerprint_is_still_read_back(tmp_path: Path):
+    """Documents written before the groups existed keep being comparable."""
+    samples = {
+        key: [
+            CalSample(weight_n=0.0, raw_mean=0.0, n=10),
+            CalSample(weight_n=1.0, raw_mean=1000.0, n=10),
+        ]
+        for key in ("fn", "fx", "fy")
+    }
+    doc = build(samples, identity={"sensor_type": "box", "sn": "FP-3"})
+    legacy = json.loads(json.dumps(doc.to_dict()))
+    legacy["meta"].pop("filters", None)
+    legacy["meta"].update(
+        {"flow_ema_alpha": 0.3, "force_median_window": 3, "force_lowpass_cutoff_hz": 10.0}
+    )
+    path = tmp_path / "legacy.json"
+    path.write_text(json.dumps(legacy), encoding="utf-8")
+    loaded = load(path)
+    assert loaded.pipeline["filters"]["measurement"] == {"flow_ema_alpha": 0.3}
+    assert loaded.pipeline["filters"]["readout"]["force_median_window"] == 3
+    assert loaded.pipeline["filters"]["readout"]["force_lowpass_cutoff_hz"] == 10.0
+
+
+def test_the_effective_state_beats_the_config_file(tmp_path: Path):
+    """A file states an intent; the device states what it is doing."""
+    bundle_json = _box_cal_path().parent / "sensor.json"
+    from_file = pipeline_from_sensor_json(bundle_json)
+    assert from_file["filters"]["measurement"] == {"flow_ema_alpha": 1.0}
+    assert "flow_ema_alpha" not in from_file, "loose top-level filter keys are gone"
+
+    sensor = _sensor_with_filters(tmp_path, {"preset": "balanced"})
+    try:
+        from_state = pipeline_from_sensor_json(
+            bundle_json, filters_state=sensor.filters_state()
+        )
+    finally:
+        sensor.disconnect()
+    assert from_state["filters"]["measurement"] == {"flow_ema_alpha": 0.5}
+    assert from_state["filters"]["readout"]["force_median_window"] == 3
+    assert from_state["filters"]["source"] == "sensor.filters_state"
+    assert from_state["config_ref"] == str(bundle_json)
+
+
+@pytest.mark.parametrize(
+    "preset, alpha, median, cutoff",
+    [
+        ("off", 1.0, 1, 0.0),
+        ("low_latency", 1.0, 3, 0.0),
+        ("balanced", 0.5, 3, 0.0),
+        ("low_noise", 0.3, 3, 10.0),
+    ],
+)
+def test_every_preset_reports_what_it_does(
+    tmp_path: Path, preset: str, alpha: float, median: int, cutoff: float
+):
+    sensor = _sensor_with_filters(tmp_path, {"preset": preset})
+    try:
+        fp = filters_fingerprint(sensor.filters_state())
+    finally:
+        sensor.disconnect()
+    assert fp["measurement"]["flow_ema_alpha"] == pytest.approx(alpha)
+    assert fp["readout"]["force_median_window"] == median
+    assert fp["readout"]["force_lowpass_cutoff_hz"] == pytest.approx(cutoff)