From 2fd2891a47fb997944c3e8a5afa27d4d31f75ea8 Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Mon, 14 Sep 2026 09:57:03 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=8D=95=E9=93=BE=E5=8A=9B?= =?UTF-8?q?=E6=BB=A4=E6=B3=A2=E5=BC=95=E6=93=8E=EF=BC=88=E4=B8=AD=E5=80=BC?= =?UTF-8?q?=20+=20=E5=9B=A0=E6=9E=9C=E4=BD=8E=E9=80=9A=EF=BC=89=EF=BC=8C?= =?UTF-8?q?=E4=B8=BA"=E6=98=BE=E7=A4=BA=3D=E8=AE=B0=E5=BD=95"=E5=81=9A?= =?UTF-8?q?=E5=87=86=E5=A4=87=20-=20What:=20shared/src/orisys=5Fapp=5Fshar?= =?UTF-8?q?ed/force/filters.py=20=E6=96=B0=E5=A2=9E=20FilterSettings?= =?UTF-8?q?=EF=BC=88=E4=B8=AD=E5=80=BC=E5=BC=80=E5=85=B3/=E7=AA=97?= =?UTF-8?q?=E5=8F=A3=E3=80=81=E4=BD=8E=E9=80=9A=E5=BC=80=E5=85=B3/?= =?UTF-8?q?=E6=88=AA=E6=AD=A2/=E9=98=B6=E6=95=B0=E3=80=81=E9=87=87?= =?UTF-8?q?=E6=A0=B7=E7=8E=87=E3=80=81show=5Funfiltered=EF=BC=8C=E7=AA=97?= =?UTF-8?q?=E5=8F=A3=E5=BC=BA=E5=88=B6=E5=A5=87=E6=95=B0=E3=80=81<3=20?= =?UTF-8?q?=E8=87=AA=E5=8A=A8=E5=85=B3=E9=97=AD=E3=80=81=E7=A9=BA=E8=AE=BE?= =?UTF-8?q?=E7=BD=AE=E7=9B=B4=E9=80=9A=EF=BC=89=E4=B8=8E=20ForceFilterChai?= =?UTF-8?q?n=EF=BC=9A6=20=E9=80=9A=E9=81=93=EF=BC=883=20=E5=8E=9F=E5=A7=8B?= =?UTF-8?q?=E8=AE=A1=E6=95=B0=20+=203=20=E7=89=9B=E9=A1=BF=E5=80=BC?= =?UTF-8?q?=EF=BC=89=E3=80=81=E5=85=88=E4=B8=AD=E5=80=BC=E5=90=8E=E4=BD=8E?= =?UTF-8?q?=E9=80=9A=E3=80=81=E4=B8=A4=E7=BA=A7=E5=9D=87=E5=9B=A0=E6=9E=9C?= =?UTF-8?q?=E4=B8=94=E9=80=90=E6=A0=B7=E6=9C=AC=E3=80=81=E6=AF=8F=E9=80=9A?= =?UTF-8?q?=E9=81=93=E7=8B=AC=E7=AB=8B=20IIR=20=E7=8A=B6=E6=80=81=EF=BC=8C?= =?UTF-8?q?reset(first)=20=E7=94=A8=E7=A8=B3=E6=80=81=E5=88=9D=E5=80=BC?= =?UTF-8?q?=E5=90=AF=E5=8A=A8=E4=BB=A5=E9=81=BF=E5=85=8D=E4=BC=9A=E8=AF=9D?= =?UTF-8?q?=E5=BC=80=E5=A4=B4=E7=9A=84=E5=81=87=E7=9E=AC=E6=80=81=EF=BC=9B?= =?UTF-8?q?settings=5Ffrom=5Fmapping=20=E6=94=AF=E6=8C=81=E4=BB=8E=20prefs?= =?UTF-8?q?/metadata=20=E6=9E=84=E9=80=A0=EF=BC=8Cdescribe()=20=E7=BB=99?= =?UTF-8?q?=E5=87=BA=E4=B8=80=E8=A1=8C=E6=91=98=E8=A6=81=E5=86=99=E8=BF=9B?= =?UTF-8?q?=E5=BD=95=E5=88=B6=E5=85=83=E6=95=B0=E6=8D=AE=EF=BC=9Bforce/=5F?= =?UTF-8?q?=5Finit=5F=5F=20=E5=AF=BC=E5=87=BA=E8=BF=99=E4=B8=89=E4=B8=AA?= =?UTF-8?q?=E5=90=8D=E5=AD=97=E3=80=82=E6=96=B0=E5=A2=9E=20shared/tests/te?= =?UTF-8?q?st=5Fforce=5Ffilters.py=EF=BC=887=20=E4=BE=8B=EF=BC=9A=E7=9B=B4?= =?UTF-8?q?=E9=80=9A=E3=80=81=E4=B8=AD=E5=80=BC=E5=8E=BB=E5=8D=95=E7=82=B9?= =?UTF-8?q?=E5=B0=96=E5=B3=B0=E4=BD=86=E4=B8=8D=E5=89=8A=E6=8C=81=E7=BB=AD?= =?UTF-8?q?=E5=8F=98=E5=8C=96=E3=80=81=E7=AA=97=E5=8F=A3=E5=A5=87=E6=95=B0?= =?UTF-8?q?=E5=8C=96=E4=B8=8E=E4=B8=8B=E9=99=90=E3=80=81=E9=98=B6=E8=B7=83?= =?UTF-8?q?=E6=97=A0=E8=B5=B7=E5=A7=8B=E7=9E=AC=E6=80=81=E3=80=81=E9=80=9A?= =?UTF-8?q?=E9=81=93=E7=8A=B6=E6=80=81=E7=8B=AC=E7=AB=8B=E3=80=81reset=20?= =?UTF-8?q?=E6=B8=85=E7=8A=B6=E6=80=81=E3=80=81prefs=20=E5=BE=80=E8=BF=94?= =?UTF-8?q?=EF=BC=89=EF=BC=8C=E5=B9=B6=E5=8A=A0=E5=85=A5=20drive=5Fshared?= =?UTF-8?q?=20=E7=9A=84=E6=A8=A1=E5=9D=97=E8=A1=A8=E3=80=82=20-=20Why:=20?= =?UTF-8?q?=E7=94=A8=E6=88=B7=E9=97=AE"=E6=BB=A4=E6=B3=A2=E5=9C=A8?= =?UTF-8?q?=E5=93=AA=E9=87=8C=E8=B0=83"=EF=BC=8C=E6=9F=A5=E6=B8=85?= =?UTF-8?q?=E7=8E=B0=E7=8A=B6=E6=98=AF=E4=B8=A4=E5=A4=84=E5=9B=BA=E5=8C=96?= =?UTF-8?q?=2010=20Hz=20=E4=BD=8E=E9=80=9A=EF=BC=88SDK=20=E4=BE=A7?= =?UTF-8?q?=E5=BD=B1=E5=93=8D=E8=AE=B0=E5=BD=95=E3=80=81=E6=98=BE=E7=A4=BA?= =?UTF-8?q?=E4=BE=A7=E5=86=8D=E7=94=A8=20filtfilt=20=E4=BA=8C=E6=AC=A1?= =?UTF-8?q?=E6=BB=A4=E6=B3=A2=E5=AF=BC=E8=87=B4=E6=98=BE=E7=A4=BA=E2=89=A0?= =?UTF-8?q?=E8=AE=B0=E5=BD=95=EF=BC=89=E4=B8=94=20median=20=E5=AE=8C?= =?UTF-8?q?=E5=85=A8=E4=B8=8D=E5=AD=98=E5=9C=A8=EF=BC=9B=E7=94=A8=E6=88=B7?= =?UTF-8?q?=E8=A6=81=E6=B1=82=E6=8A=8A=E9=87=87=E9=9B=86=E6=BB=A4=E6=B3=A2?= =?UTF-8?q?=E4=B8=8E=E6=98=BE=E7=A4=BA=E6=BB=A4=E6=B3=A2=E7=BB=9F=E4=B8=80?= =?UTF-8?q?=E6=88=90=E4=B8=80=E6=9D=A1=E9=93=BE=E3=80=81=E6=98=BE=E7=A4=BA?= =?UTF-8?q?=E5=80=BC=E5=BF=85=E9=A1=BB=E7=AD=89=E4=BA=8E=E8=AE=B0=E5=BD=95?= =?UTF-8?q?=E5=80=BC=E3=80=81=E5=B9=B6=E4=BF=9D=E7=95=99=E6=BB=A4=E6=B3=A2?= =?UTF-8?q?=E5=89=8D=E6=95=B0=E6=8D=AE=EF=BC=88=E5=8F=A6=E5=BC=80=E4=B8=80?= =?UTF-8?q?=E4=B8=AA=20buffer=EF=BC=8C=E4=B8=94=E5=86=99=E8=BF=9B=20forces?= =?UTF-8?q?.csv=EF=BC=89=E3=80=82=E6=9C=AC=E8=BD=AE=E5=85=88=E8=90=BD?= =?UTF-8?q?=E5=9C=B0=E8=BF=99=E6=9D=A1=E9=93=BE=E5=B9=B6=E8=A6=86=E7=9B=96?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=E3=80=82=20-=20Verification:=20drive=5Fshare?= =?UTF-8?q?d.py=20141=20=E9=A1=B9=200=20=E5=A4=B1=E8=B4=A5=EF=BC=88?= =?UTF-8?q?=E5=90=AB=E6=96=B0=E5=A2=9E=207=20=E9=A1=B9=EF=BC=89=E3=80=81ui?= =?UTF-8?q?=5Fcheck.py=20554=20=E9=A1=B9=200=20=E5=A4=B1=E8=B4=A5=E3=80=81?= =?UTF-8?q?test=5Fui=5Fsmoke.py=2071=20=E9=A1=B9=200=20=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E3=80=81locale=5Fstudio.py=20262=20=E9=94=AE=E9=BD=90=E5=85=A8?= =?UTF-8?q?=E3=80=82=20-=20Not=20in=20this=20change:=20=E9=87=87=E9=9B=86?= =?UTF-8?q?=E8=B7=AF=E5=BE=84=E6=9C=AA=E6=94=B9=E5=8A=A8=EF=BC=88SDK=20?= =?UTF-8?q?=E4=BD=8E=E9=80=9A=E4=BB=8D=E6=98=AF=2010=20Hz=E3=80=81forces.c?= =?UTF-8?q?sv=20=E4=BB=8D=E6=98=AF=2011=20=E5=88=97=EF=BC=89=E2=80=94?= =?UTF-8?q?=E2=80=94=E6=8A=8A=E5=B8=A6=E7=8A=B6=E6=80=81=E7=9A=84=E6=BB=A4?= =?UTF-8?q?=E6=B3=A2=E5=99=A8=E5=8D=8A=E9=80=94=E6=8E=A5=E8=BF=9B=E8=AE=B0?= =?UTF-8?q?=E5=BD=95=E4=BC=9A=E8=AE=A9=E6=95=B0=E6=8D=AE=E5=8F=AA=E6=BB=A4?= =?UTF-8?q?=E4=B8=80=E5=8D=8A=EF=BC=8C=E5=9B=A0=E6=AD=A4=E4=B8=8B=E4=B8=80?= =?UTF-8?q?=E6=AD=A5=E4=B8=80=E6=AC=A1=E6=80=A7=E5=AE=8C=E6=88=90=EF=BC=9A?= =?UTF-8?q?=E5=85=B3=20SDK=20=E4=BE=A7=E6=BB=A4=E6=B3=A2=E3=80=81worker=20?= =?UTF-8?q?=E5=BA=94=E7=94=A8=E9=93=BE=E5=B9=B6=E8=AE=A9=20packet=20?= =?UTF-8?q?=E5=90=8C=E6=97=B6=E6=90=BA=E5=B8=A6=E6=BB=A4=E6=B3=A2=E5=89=8D?= =?UTF-8?q?=E5=90=8E=E4=B8=A4=E5=A5=97=E5=80=BC=E3=80=81forces.csv=20?= =?UTF-8?q?=E8=BF=BD=E5=8A=A0=206=20=E5=88=97=E6=9C=AA=E6=BB=A4=E6=B3=A2?= =?UTF-8?q?=E8=AF=BB=E6=95=B0=EF=BC=8811=E2=86=9217=E3=80=81RECORD=5FFORMA?= =?UTF-8?q?T=5FVERSION+1=EF=BC=89=E3=80=81SessionStore=20=E7=AC=AC?= =?UTF-8?q?=E4=BA=8C=20buffer=E3=80=81=E5=8E=BB=E6=8E=89=E6=98=BE=E7=A4=BA?= =?UTF-8?q?=E7=AB=AF=20filtfilt=E3=80=81=E6=9B=B2=E7=BA=BF=E8=A1=A8?= =?UTF-8?q?=E5=A4=B4=E6=96=B0=E5=A2=9E=E9=BD=BF=E8=BD=AE=E6=8C=89=E9=92=AE?= =?UTF-8?q?=E5=B9=B6=E6=89=A9=E5=B1=95=E6=9B=B2=E7=BA=BF=E8=AE=BE=E7=BD=AE?= =?UTF-8?q?=E5=AF=B9=E8=AF=9D=E6=A1=86=EF=BC=88=E4=B8=AD=E5=80=BC/?= =?UTF-8?q?=E4=BD=8E=E9=80=9A/=E6=98=BE=E7=A4=BA=E6=9C=AA=E6=BB=A4?= =?UTF-8?q?=E6=B3=A2=EF=BC=8C=E8=BF=90=E8=A1=8C=E4=B8=AD=E7=A6=81=E7=94=A8?= =?UTF-8?q?=E6=BB=A4=E6=B3=A2=E6=8E=A7=E4=BB=B6=EF=BC=89=E3=80=82=E5=AE=8C?= =?UTF-8?q?=E6=95=B4=E6=B8=85=E5=8D=95=E8=A7=81=20.repro/filter=5Fplan.md?= =?UTF-8?q?=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/orisys_app_shared/force/__init__.py | 8 + shared/src/orisys_app_shared/force/filters.py | 205 ++++++++++++++++++ shared/tests/test_force_filters.py | 113 ++++++++++ 3 files changed, 326 insertions(+) create mode 100644 shared/src/orisys_app_shared/force/filters.py create mode 100644 shared/tests/test_force_filters.py diff --git a/shared/src/orisys_app_shared/force/__init__.py b/shared/src/orisys_app_shared/force/__init__.py index 485afea..2d61e45 100644 --- a/shared/src/orisys_app_shared/force/__init__.py +++ b/shared/src/orisys_app_shared/force/__init__.py @@ -13,6 +13,11 @@ from orisys_app_shared.force.calibration_recorder import ( CalibrationPoint, CalibrationRecorder, ) +from orisys_app_shared.force.filters import ( + ForceFilterChain, + FilterSettings, + settings_from_mapping, +) from orisys_app_shared.force.history import MetricHistory, PlotState, lowpass_filter from orisys_app_shared.force.session import ( NA, @@ -33,6 +38,8 @@ __all__ = [ "CalibrationPoint", "CalibrationRecorder", "DisplayForces", + "FilterSettings", + "ForceFilterChain", "MetricHistory", "PlotState", "SelectionStats", @@ -44,5 +51,6 @@ __all__ = [ "lowpass_filter", "meta_from_sensor_json", "select_display_forces", + "settings_from_mapping", "summarise", ] diff --git a/shared/src/orisys_app_shared/force/filters.py b/shared/src/orisys_app_shared/force/filters.py new file mode 100644 index 0000000..364c9d2 --- /dev/null +++ b/shared/src/orisys_app_shared/force/filters.py @@ -0,0 +1,205 @@ +"""One causal filter chain for the force readings, and the values it replaced. + +The chain runs **once, in the acquisition loop**, so the numbers on the screen +and the numbers in ``forces.csv`` are the same filtered values — there is no +second, display-only smoothing that would make the chart disagree with the +record. The unfiltered readings are kept beside the filtered ones, so a filter +can be judged instead of trusted. + +Both stages are causal and per-sample, because acquisition cannot look ahead: + +* **median** — a window of ``median_window`` samples (odd, ≥3), applied per + channel; it removes single-sample spikes without smearing a step; +* **low-pass** — a Butterworth IIR with persistent state (``sosfilt`` with + ``zi``), the same shape as the SDK's own ``LowpassFilter``. + +Order is median then low-pass: the median protects the IIR from a spike that +would otherwise ring for several samples. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Iterable, Optional + +import numpy as np + +#: Channels of the chain: three raw counts followed by the three newton values. +CHANNELS = 6 + + +def _odd(value: Any, *, minimum: int = 3, maximum: int = 31) -> int: + try: + number = int(value) + except (TypeError, ValueError): + return 0 + if number < minimum: + return 0 + if number % 2 == 0: + number += 1 + return min(number, maximum) + + +@dataclass(frozen=True) +class FilterSettings: + """What the operator chose. ``enabled`` flags decide what actually runs.""" + + median_enabled: bool = False + median_window: int = 5 + lowpass_enabled: bool = True + lowpass_hz: float = 10.0 + lowpass_order: int = 2 + sample_rate_hz: float = 30.0 + #: Show the unfiltered readings next to the filtered ones on the cards. + show_unfiltered: bool = False + + def normalised(self) -> "FilterSettings": + window = _odd(self.median_window) + return FilterSettings( + median_enabled=bool(self.median_enabled) and window >= 3, + median_window=window or 5, + lowpass_enabled=bool(self.lowpass_enabled) and float(self.lowpass_hz) > 0, + lowpass_hz=max(0.0, float(self.lowpass_hz)), + lowpass_order=max(1, min(6, int(self.lowpass_order))), + sample_rate_hz=max(1.0, float(self.sample_rate_hz)), + show_unfiltered=bool(self.show_unfiltered), + ) + + def active(self) -> bool: + """Whether any stage actually filters (a disabled chain is a pass-through).""" + settings = self.normalised() + return bool(settings.median_enabled or settings.lowpass_enabled) + + def as_dict(self) -> dict: + settings = self.normalised() + return { + "median_enabled": settings.median_enabled, + "median_window": settings.median_window, + "lowpass_enabled": settings.lowpass_enabled, + "lowpass_hz": settings.lowpass_hz, + "lowpass_order": settings.lowpass_order, + "sample_rate_hz": settings.sample_rate_hz, + "show_unfiltered": settings.show_unfiltered, + } + + def describe(self) -> str: + """A short human line for the recording metadata.""" + settings = self.normalised() + parts = [] + if settings.median_enabled: + parts.append(f"median({settings.median_window})") + if settings.lowpass_enabled: + parts.append( + f"lowpass({settings.lowpass_hz:g}Hz,order{settings.lowpass_order})" + ) + return " + ".join(parts) if parts else "none" + + +def settings_from_mapping(data: Optional[dict]) -> FilterSettings: + """Build settings from a preferences/metadata mapping, with defaults.""" + source = dict(data or {}) + defaults = FilterSettings() + return FilterSettings( + median_enabled=source.get("median_enabled", defaults.median_enabled), + median_window=source.get("median_window", defaults.median_window), + lowpass_enabled=source.get("lowpass_enabled", defaults.lowpass_enabled), + lowpass_hz=source.get("lowpass_hz", defaults.lowpass_hz), + lowpass_order=source.get("lowpass_order", defaults.lowpass_order), + sample_rate_hz=source.get("sample_rate_hz", defaults.sample_rate_hz), + show_unfiltered=source.get("show_unfiltered", defaults.show_unfiltered), + ).normalised() + + +class ForceFilterChain: + """The median + low-pass chain, one instance per acquisition session.""" + + def __init__(self, settings: Optional[FilterSettings] = None) -> None: + self._settings = (settings or FilterSettings()).normalised() + self._median: list[list[float]] = [[] for _ in range(CHANNELS)] + self._sos = None + #: One IIR state per channel: the six readings are independent signals. + self._zi: Optional[list[np.ndarray]] = None + self._build() + + # ------------------------------------------------------------------ setup + def _build(self) -> None: + self._sos = None + self._zi = None + settings = self._settings + if not settings.lowpass_enabled: + return + from scipy.signal import butter + + nyquist = 0.5 * settings.sample_rate_hz + cutoff = min(float(settings.lowpass_hz), nyquist * 0.95) + if cutoff <= 0: + return + self._sos = butter( + settings.lowpass_order, + cutoff / nyquist, + btype="low", + output="sos", + ) + self._zi = None + + def configure(self, settings: FilterSettings) -> None: + """Re-tune the chain; the caller resets the stream when it does.""" + self._settings = settings.normalised() + self._median = [[] for _ in range(CHANNELS)] + self._build() + + def settings(self) -> FilterSettings: + return self._settings + + @property + def active(self) -> bool: + return self._settings.active() + + def reset(self, first: Optional[Iterable[float]] = None) -> None: + """Drop the filter memory (a new session must not inherit the old one).""" + self._median = [[] for _ in range(CHANNELS)] + self._zi = None + if first is not None and self._sos is not None: + # Start each IIR at its first sample's steady state instead of at + # zero: a step from 0 to a real reading otherwise rings for a few + # samples, which would show up as a transient at session start. + from scipy.signal import sosfilt_zi + + values = np.asarray(list(first), dtype=np.float64).reshape(-1) + if values.size == CHANNELS: + steady = sosfilt_zi(self._sos) + self._zi = [steady * float(value) for value in values] + + # ----------------------------------------------------------------- filter + def process(self, values) -> np.ndarray: + """Filter one 6-channel sample: three raw counts then three newtons.""" + sample = np.asarray(list(values), dtype=np.float64).reshape(-1) + if sample.size != CHANNELS: + raise ValueError(f"expected {CHANNELS} channels, got {sample.size}") + if not self.active: + return sample.copy() + settings = self._settings + out = sample.copy() + if settings.median_enabled: + window = int(settings.median_window) + for index in range(CHANNELS): + series = self._median[index] + series.append(float(sample[index])) + if len(series) > window: + del series[0] + out[index] = float(np.median(series)) + if self._sos is not None: + from scipy.signal import sosfilt + + if self._zi is None: + self.reset(out) + states = self._zi if self._zi is not None else [None] * CHANNELS + for index in range(CHANNELS): + filtered, states[index] = sosfilt( + self._sos, np.array([out[index]]), zi=states[index] + ) + out[index] = float(filtered[0]) + self._zi = states + return out diff --git a/shared/tests/test_force_filters.py b/shared/tests/test_force_filters.py new file mode 100644 index 0000000..2538225 --- /dev/null +++ b/shared/tests/test_force_filters.py @@ -0,0 +1,113 @@ +"""The force filter chain: what it removes, and what it must not do. + +The chain runs once, in the acquisition loop, so the numbers on the screen and +the numbers in ``forces.csv`` are the same filtered values; the unfiltered +readings are kept beside them. These tests pin the parts that are easy to get +wrong: odd windows, a disabled chain being a pass-through, per-channel state, +and a step not ringing at session start. +""" + +from __future__ import annotations + +import numpy as np + +from orisys_app_shared.force.filters import ( + CHANNELS, + FilterSettings, + ForceFilterChain, + settings_from_mapping, +) + + +def _zeros(value: float = 0.0) -> list[float]: + return [value] * CHANNELS + + +def test_a_disabled_chain_is_a_pass_through(): + chain = ForceFilterChain( + FilterSettings(median_enabled=False, lowpass_enabled=False) + ) + assert not chain.active + sample = [1.0, 2.0, 3.0, 4.0, 5.0, 6.0] + assert chain.process(sample).tolist() == sample + + +def test_the_median_removes_a_single_sample_spike(): + """One bad readout must not reach the chart, the store or the CSV.""" + chain = ForceFilterChain( + FilterSettings(median_enabled=True, median_window=3, lowpass_enabled=False) + ) + outputs = [ + float(chain.process(_zeros(9.0 if index == 2 else 0.0))[0]) + for index in range(5) + ] + assert outputs[2] == 0.0, "the spike survived the median" + # ...and a sustained change is *not* removed: this is not a clamp. + for _ in range(4): + sustained = float(chain.process(_zeros(9.0))[0]) + assert sustained == 9.0, sustained + + +def test_the_median_window_is_forced_odd_and_bounded(): + assert FilterSettings(median_window=4).normalised().median_window == 5 + assert FilterSettings(median_window=1).normalised().median_window == 5 + # A window below three samples cannot remove anything: the stage turns off. + assert FilterSettings(median_enabled=True, median_window=2).normalised().median_enabled is False + + +def test_the_lowpass_settles_on_a_step_without_ringing_at_the_start(): + """A session must not open with a transient invented by the filter.""" + chain = ForceFilterChain( + FilterSettings(lowpass_enabled=True, lowpass_hz=5.0, sample_rate_hz=30.0) + ) + first = float(chain.process(_zeros(3.0))[0]) + assert abs(first - 3.0) < 1e-9, f"the filter invented a transient: {first}" + # A step is damped and then reached. A second-order digital Butterworth + # overshoots a step by a few percent — that is its response, not a defect — + # so the bound is "a few percent", and the settled value is the step. + values = [ + float(chain.process(_zeros(0.0 if index < 5 else 3.0))[0]) + for index in range(80) + ] + assert max(values) <= 3.0 * 1.1, f"the low-pass overshot by more than 10%: {max(values)}" + assert values[-1] > 2.9, values[-1] + + +def test_each_channel_keeps_its_own_filter_state(): + chain = ForceFilterChain( + FilterSettings(median_enabled=True, median_window=3, lowpass_enabled=False) + ) + chain.process([1.0, 100.0, 2.0, 200.0, 3.0, 300.0]) + chain.process([1.0, 100.0, 2.0, 200.0, 3.0, 300.0]) + out = chain.process([1.0, 100.0, 2.0, 200.0, 3.0, 300.0]) + assert out.tolist() == [1.0, 100.0, 2.0, 200.0, 3.0, 300.0] + + +def test_reset_clears_the_memory_between_sessions(): + chain = ForceFilterChain( + FilterSettings(lowpass_enabled=True, lowpass_hz=5.0, sample_rate_hz=30.0) + ) + for _ in range(30): + chain.process(_zeros(5.0)) + chain.reset() + assert abs(float(chain.process(_zeros(0.0))[0])) < 1e-9, ( + "the new session inherited the previous filter state" + ) + + +def test_settings_round_trip_through_a_preferences_mapping(): + settings = settings_from_mapping( + { + "median_enabled": True, + "median_window": 7, + "lowpass_enabled": True, + "lowpass_hz": 12.5, + "show_unfiltered": True, + } + ) + assert settings.median_enabled and settings.median_window == 7 + assert settings.lowpass_hz == 12.5 and settings.show_unfiltered + assert settings.describe() == "median(7) + lowpass(12.5Hz,order2)" + # An empty mapping is the documented default, not a disabled chain. + default = settings_from_mapping(None) + assert default.as_dict()["lowpass_hz"] == FilterSettings().lowpass_hz