diff --git a/docs/filters.md b/docs/filters.md index 880b9d5..48ae3b3 100644 --- a/docs/filters.md +++ b/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` 是为此新增的字段。 + diff --git a/src/orisys/calibration.py b/src/orisys/calibration.py index 3c2bbc0..7075dd3 100644 --- a/src/orisys/calibration.py +++ b/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", - "flow_ema_alpha", "force_median_window", "force_lowpass_cutoff_hz", - "config_ref"): + for key in ("stitch_resolution", "dis_mode", "hhd_downsample", "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", - "flow_ema_alpha", "force_median_window", "force_lowpass_cutoff_hz"): + for key in ("stitch_resolution", "dis_mode", "hhd_downsample"): 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]: - """Pull stitch / tracker / filter fields from a sensor.json for export pipeline/meta.""" +def pipeline_from_sensor_json( + 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, - "force_median_window": None, - "force_lowpass_cutoff_hz": None, + "filters": {}, "config_ref": None, } - if path is None: - return pipeline - p = Path(path) - pipeline["config_ref"] = str(p) - if not p.is_file(): - return pipeline - try: - data = json.loads(p.read_text(encoding="utf-8")) - except Exception: - return pipeline - if not isinstance(data, dict): + if path is None and filters_state is None: return pipeline + p = Path(path) if path is not None else None + if p is not None: + pipeline["config_ref"] = str(p) + data: dict[str, Any] = {} + if p is not None and p.is_file(): + try: + loaded = json.loads(p.read_text(encoding="utf-8")) + except Exception: + loaded = None + if isinstance(loaded, dict): + 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 - # belong to the pipeline identity just like dis_mode / hhd_downsample. - if filters: - enabled = bool(filters.get("enabled", True)) - for key in ("flow_ema_alpha", "force_lowpass_cutoff_hz"): - 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 + # Signal conditioning is part of the measurement-pipeline identity: these + # stages change the reported force values, so they are recorded next to + # dis_mode / hhd_downsample. The effective state wins over the file. + pipeline["filters"] = filters_fingerprint( + filters_state if filters_state else data.get("filters") + ) 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", - "flow_ema_alpha", "force_median_window", "force_lowpass_cutoff_hz"): + if k in ("stitch_resolution", "dis_mode", "hhd_downsample", "config_ref"): 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 diff --git a/tests/test_calibration.py b/tests/test_calibration.py index 0c8c93d..71302a8 100644 --- a/tests/test_calibration.py +++ b/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)