From 0d8b79b759ffae70ab9747aa524a253840e833e0 Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Mon, 14 Sep 2026 10:56:04 +0800 Subject: [PATCH] =?UTF-8?q?app=20=E4=BE=A7=E6=BB=A4=E6=B3=A2=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3=E6=94=B9=E4=B8=BA=E9=A9=B1=E5=8A=A8=20SDK=20=E5=B9=B6?= =?UTF-8?q?=E8=AF=BB=E5=9B=9E=E7=94=9F=E6=95=88=E5=80=BC=20-=20What:=20sha?= =?UTF-8?q?red/force/filters.py=20=E4=BB=8E"=E8=87=AA=E5=B7=B1=E6=BB=A4?= =?UTF-8?q?=E4=B8=80=E9=81=8D"=E6=94=B9=E6=88=90=20app=20=E7=9A=84?= =?UTF-8?q?=E6=BB=A4=E6=B3=A2=E6=8E=A7=E5=88=B6=E9=9D=A2=EF=BC=8C=E5=8F=AA?= =?UTF-8?q?=E5=81=9A=E4=B8=89=E4=BB=B6=E4=BA=8B=EF=BC=9A(1)=20FilterSettin?= =?UTF-8?q?gs=20=E6=94=B9=E7=94=A8=20SDK=20=E7=9A=84=E9=94=AE=E4=B8=8E"?= =?UTF-8?q?=E6=8C=89=E5=80=BC=E5=BC=80=E5=85=B3"=E8=AF=AD=E4=B9=89?= =?UTF-8?q?=EF=BC=88flow=5Fema=5Falpha=20/=20median=5Fwindow=20/=20lowpass?= =?UTF-8?q?=5Fhz=20/=20lowpass=5Forder=20/=20sample=5Frate=5Fhz=20/=20show?= =?UTF-8?q?=5Funfiltered=20/=20preset=EF=BC=9B=CE=B1=E2=89=A51=E3=80=81?= =?UTF-8?q?=E7=AA=97=E5=8F=A3=E2=89=A41=E3=80=81=E6=88=AA=E6=AD=A2?= =?UTF-8?q?=E2=89=A40=20=E5=90=84=E8=87=AA=E8=A1=A8=E7=A4=BA=E5=85=B3?= =?UTF-8?q?=EF=BC=89=EF=BC=8Cnormalised()=20=E6=8A=8A=E4=B8=8D=E5=8F=AF?= =?UTF-8?q?=E7=94=A8=E5=80=BC=E4=B8=80=E5=BE=8B=E8=90=BD=E6=88=90"?= =?UTF-8?q?=E5=85=B3"=E8=80=8C=E4=B8=8D=E6=98=AF=E8=90=BD=E6=88=90?= =?UTF-8?q?=E4=BC=9A=E6=BB=A4=E6=B3=A2=E7=9A=84=E9=BB=98=E8=AE=A4=E5=80=BC?= =?UTF-8?q?=EF=BC=8C=E5=B9=B6=E6=96=B0=E5=A2=9E=20problems(sample=5Frate?= =?UTF-8?q?=5Fhz)=20=E6=8C=87=E5=87=BA"=E6=88=AA=E6=AD=A2=20=E2=89=A5=20Ny?= =?UTF-8?q?quist=E3=80=81=E8=AE=BE=E5=A4=87=E6=A0=B9=E6=9C=AC=E4=B8=8D?= =?UTF-8?q?=E4=BC=9A=E6=BB=A4"=E8=BF=99=E7=B1=BB=E8=AF=B7=E6=B1=82?= =?UTF-8?q?=EF=BC=9B(2)=20apply=5Fto(sensor)=20=E5=B0=B1=E6=98=AF=E4=B8=80?= =?UTF-8?q?=E6=AC=A1=20Sensor.configure(...)=EF=BC=8C=E7=94=A8=20CONFIGURE?= =?UTF-8?q?=5FPARAMETERS=20=E4=B8=80=E5=A4=84=E7=BB=B4=E6=8A=A4"=E7=8A=B6?= =?UTF-8?q?=E6=80=81=E9=94=AE=20=E2=86=92=20configure=20=E5=BD=A2=E5=8F=82?= =?UTF-8?q?"=E7=9A=84=E6=98=A0=E5=B0=84=EF=BC=88=E4=BD=8E=E9=80=9A?= =?UTF-8?q?=E9=82=A3=E4=B8=80=E5=AF=B9=E5=9C=A8=E9=85=8D=E7=BD=AE/?= =?UTF-8?q?=E7=8A=B6=E6=80=81=E9=87=8C=E5=8F=AB=20force=5Flowpass=5F*?= =?UTF-8?q?=E3=80=81=E5=9C=A8=E8=B0=83=E7=94=A8=E9=87=8C=E5=8F=AB=20lowpas?= =?UTF-8?q?s=5F*=EF=BC=8C=E6=AD=A4=E5=89=8D=E6=AD=A3=E6=98=AF=E8=BF=99?= =?UTF-8?q?=E9=87=8C=E5=86=99=E9=94=99=E5=AF=BC=E8=87=B4=E6=94=AF=E6=8C=81?= =?UTF-8?q?=E6=80=A7=E5=88=A4=E6=96=AD=E6=8A=8A=E6=89=80=E6=9C=89=E9=94=AE?= =?UTF-8?q?=E9=83=BD=E5=88=A4=E6=88=90=E4=B8=8D=E6=94=AF=E6=8C=81=EF=BC=89?= =?UTF-8?q?=EF=BC=8C=E5=B9=B6=E8=BF=94=E5=9B=9E=E8=AF=A5=20SDK=20=E6=94=AF?= =?UTF-8?q?=E6=8C=81=E4=B8=8D=E4=BA=86=E7=9A=84=E9=94=AE=E8=80=8C=E4=B8=8D?= =?UTF-8?q?=E6=98=AF=E9=BB=98=E9=BB=98=E4=B8=A2=E6=8E=89=E2=80=94=E2=80=94?= =?UTF-8?q?=E8=80=81=20SDK=EF=BC=880.5.2=20=E4=B9=8B=E5=89=8D=EF=BC=89?= =?UTF-8?q?=E6=B2=A1=E6=9C=89=20flow=5Fema=5Falpha/=E4=B8=AD=E5=80=BC?= =?UTF-8?q?=EF=BC=8C=E5=85=B6=E5=85=89=E6=B5=81=20EMA=20=E4=B9=9F=E5=85=B3?= =?UTF-8?q?=E4=B8=8D=E6=8E=89=EF=BC=8C=E5=BF=85=E9=A1=BB=E8=AE=A9=E4=BA=BA?= =?UTF-8?q?=E7=9C=8B=E8=A7=81=EF=BC=9B(3)=20from=5Ffilters=5Fstate(state)?= =?UTF-8?q?=20=E8=AF=BB=E5=9B=9E=E8=AE=BE=E5=A4=87**=E5=AE=9E=E9=99=85**?= =?UTF-8?q?=E5=9C=A8=E8=B7=91=E7=9A=84=E5=80=BC=EF=BC=88=E7=95=8C=E9=9D=A2?= =?UTF-8?q?=E4=B8=8E=E8=AE=B0=E5=BD=95=E9=83=BD=E4=BB=A5=E6=AD=A4=E4=B8=BA?= =?UTF-8?q?=E5=87=86=EF=BC=89=EF=BC=8Cmetadata(sensor=5Fstate)=20=E4=BA=A7?= =?UTF-8?q?=E5=87=BA=20session.json/CSV=20=E5=A4=B4=E7=9A=84=20filters=20?= =?UTF-8?q?=E5=9D=97=EF=BC=8Cmeasurement=20=E7=BB=84=E4=B8=8E=20readout=20?= =?UTF-8?q?=E7=BB=84=E4=B8=A5=E6=A0=BC=E5=88=86=E5=BC=80=E3=80=81=E5=B8=A6?= =?UTF-8?q?=20describe=20=E4=B8=8E=20source=EF=BC=88=E6=9C=89=E8=AE=BE?= =?UTF-8?q?=E5=A4=87=E6=97=B6=E4=B8=BA=20sensor.filters=5Fstate=EF=BC=8C?= =?UTF-8?q?=E5=8F=AA=E6=9C=89=E8=AF=B7=E6=B1=82=E6=97=B6=E4=B8=BA=20app.Fi?= =?UTF-8?q?lterSettings=EF=BC=8C=E4=B8=A4=E8=80=85=E4=B8=8D=E5=8F=AF?= =?UTF-8?q?=E6=B7=B7=E8=AE=A4=EF=BC=89=E3=80=82=E6=96=B0=E5=A2=9E=20descri?= =?UTF-8?q?be=5Fstate()/settings=5Ffrom=5Fmapping()=EF=BC=9BForceFilterCha?= =?UTF-8?q?in=20=E9=99=8D=E7=BA=A7=E4=B8=BA"=E5=8F=82=E8=80=83=E5=AE=9E?= =?UTF-8?q?=E7=8E=B0"=EF=BC=8C=E5=8F=AA=E4=B8=BA=20tests/test=5Fforce=5Ffi?= =?UTF-8?q?lters.py=20=E7=9A=84=E7=AD=89=E4=BB=B7=E6=80=A7=E6=96=AD?= =?UTF-8?q?=E8=A8=80=E5=AD=98=E5=9C=A8=EF=BC=88app=20=E4=B8=8D=E5=86=8D?= =?UTF-8?q?=E8=B0=83=E7=94=A8=E5=AE=83=EF=BC=89=EF=BC=8C=E5=B9=B6=E5=88=A0?= =?UTF-8?q?=E6=8E=89=E9=9A=8F=E4=B9=8B=E6=97=A0=E7=94=A8=E7=9A=84=20=5Fodd?= =?UTF-8?q?()=E3=80=82force/=5F=5Finit=5F=5F.py=20=E5=AF=BC=E5=87=BA=20des?= =?UTF-8?q?cribe=5Fstate=E3=80=82tests/test=5Fforce=5Ffilters.py=20?= =?UTF-8?q?=E9=87=8D=E5=86=99=E4=B8=BA=2019=20=E9=A1=B9=EF=BC=9A=E6=8C=89?= =?UTF-8?q?=E5=80=BC=E5=BC=80=E5=85=B3=E3=80=81=E4=B8=8D=E5=8F=AF=E7=94=A8?= =?UTF-8?q?=E5=80=BC=3D=E5=85=B3=E3=80=81Nyquist=20=E9=A2=84=E8=AD=A6?= =?UTF-8?q?=E3=80=81=E4=B8=80=E6=AC=A1=20configure=20=E8=B0=83=E7=94=A8?= =?UTF-8?q?=E4=B8=94=E9=94=AE=E5=90=8D=E6=AD=A3=E7=A1=AE=E3=80=81=E8=80=81?= =?UTF-8?q?=20SDK=20=E7=BC=BA=E9=94=AE=E8=A2=AB=E4=B8=8A=E6=8A=A5=E8=80=8C?= =?UTF-8?q?=E4=B8=8D=E6=98=AF=E5=BF=BD=E7=95=A5=E3=80=81=E6=98=BE=E7=A4=BA?= =?UTF-8?q?=E5=BC=80=E5=85=B3=E7=BB=9D=E4=B8=8D=E5=8F=91=E7=BB=99=E8=AE=BE?= =?UTF-8?q?=E5=A4=87=E3=80=81=E8=AF=BB=E5=9B=9E=E7=94=9F=E6=95=88=E5=80=BC?= =?UTF-8?q?=E4=B8=8E=20sample=5Frate=20=E5=BE=80=E8=BF=94=E3=80=81metadata?= =?UTF-8?q?=20=E4=B8=A4=E7=BB=84=E5=88=86=E7=A6=BB=E4=B8=8E=E6=9D=A5?= =?UTF-8?q?=E6=BA=90=E6=A0=87=E6=B3=A8=E3=80=81=E7=A9=BA=E7=9A=84=20prefer?= =?UTF-8?q?ences=20=E6=98=A0=E5=B0=84=E7=AD=89=E4=BA=8E"=E4=BB=80=E4=B9=88?= =?UTF-8?q?=E9=83=BD=E4=B8=8D=E6=BB=A4"=E3=80=81=E5=8F=82=E8=80=83?= =?UTF-8?q?=E9=93=BE=E4=BB=8D=E4=B8=8E=20SDK=20=E6=BB=A4=E6=B3=A2=E5=99=A8?= =?UTF-8?q?=E9=80=90=E4=BD=8D=E4=B8=80=E8=87=B4=E3=80=82=20-=20Why:=20?= =?UTF-8?q?=E7=94=A8=E6=88=B7=E8=A6=81=E6=B1=82"=E7=BB=9F=E4=B8=80?= =?UTF-8?q?=E9=87=87=E9=9B=86=E6=BB=A4=E6=B3=A2=E5=92=8C=E6=98=BE=E7=A4=BA?= =?UTF-8?q?=E6=BB=A4=E6=B3=A2=E3=80=81=E6=98=BE=E7=A4=BA=E7=9A=84=E5=92=8C?= =?UTF-8?q?=E8=AE=B0=E5=BD=95=E7=9A=84=E5=BF=85=E9=A1=BB=E4=B8=80=E8=87=B4?= =?UTF-8?q?=E3=80=81=E4=BF=9D=E7=95=99=E6=BB=A4=E6=B3=A2=E5=89=8D=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E4=B8=A4=E4=B8=AA=20buffer"=EF=BC=8C=E5=B9=B6?= =?UTF-8?q?=E5=86=B3=E5=AE=9A=E6=95=B4=E6=9D=A1=E9=93=BE=E7=95=99=E5=9C=A8?= =?UTF-8?q?=20SDK=E3=80=81app=20=E9=80=9A=E8=BF=87=20Sensor.configure=20?= =?UTF-8?q?=E9=A9=B1=E5=8A=A8=E3=80=81filters.py=20=E4=BF=9D=E7=95=99?= =?UTF-8?q?=E4=B8=BA=20wrapper=20=E4=B8=8E=20CSV=20=E5=85=83=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=87=BA=E5=8F=A3=E3=80=82=E4=B8=8A=E4=B8=80=E7=89=88?= =?UTF-8?q?=E7=9A=84=20FilterSettings=20=E7=94=A8=20median=5Fenabled/lowpa?= =?UTF-8?q?ss=5Fenabled=20=E5=B8=83=E5=B0=94=20+=20app=20=E8=87=AA?= =?UTF-8?q?=E5=B7=B1=E5=AE=9E=E7=8E=B0=E6=BB=A4=E6=B3=A2=EF=BC=8C=E6=97=A2?= =?UTF-8?q?=E4=B8=8E=20SDK=20=E7=9A=84"=E6=8C=89=E5=80=BC=E5=BC=80?= =?UTF-8?q?=E5=85=B3"=E4=B8=A4=E5=A5=97=E8=AF=8D=E6=B1=87=EF=BC=8C?= =?UTF-8?q?=E4=B9=9F=E4=BC=9A=E8=AE=A9=E5=90=8C=E4=B8=80=E4=BB=BD=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=9C=A8=E7=95=8C=E9=9D=A2=E4=B8=8E=20CSV=20=E4=B9=8B?= =?UTF-8?q?=E9=97=B4=E5=87=BA=E7=8E=B0=E7=AC=AC=E4=BA=8C=E6=9D=A1=E6=BB=A4?= =?UTF-8?q?=E6=B3=A2=E8=B7=AF=E5=BE=84=E3=80=82=20-=20Verification:=20pyte?= =?UTF-8?q?st=20shared/tests/test=5Fforce=5Ffilters.py=20=E2=86=92=2019=20?= =?UTF-8?q?passed=EF=BC=88=E5=90=AB=E4=B8=8E=20orisys.signal=5Fprocessing.?= =?UTF-8?q?LowpassFilter=20=E7=9A=84=E9=80=90=E4=BD=8D=E7=AD=89=E4=BB=B7?= =?UTF-8?q?=E6=96=AD=E8=A8=80=EF=BC=89=EF=BC=9Bshared=20=E5=85=A8=E9=87=8F?= =?UTF-8?q?=20153=20passed=EF=BC=9BSDK=20=E5=B7=B2=E5=8D=87=E7=BA=A7?= =?UTF-8?q?=E5=88=B0=200.5.2=EF=BC=88editable=EF=BC=89=EF=BC=8Cstudio=2085?= =?UTF-8?q?=20=E9=A1=B9=E6=B5=8B=E8=AF=95=E4=B8=8E=20ui=5Fcheck=20554=20?= =?UTF-8?q?=E9=A1=B9=E6=A3=80=E6=9F=A5=E5=9C=A8=E5=85=B6=E4=B8=8A=E5=85=A8?= =?UTF-8?q?=E7=BB=BF=E3=80=82=20-=20Not=20in=20this=20change:=20sdk=5Fwork?= =?UTF-8?q?er/finger=5Fworker=20=E4=BB=8D=E5=9C=A8=E7=A1=AC=E7=BC=96?= =?UTF-8?q?=E7=A0=81=20lowpass=5Fcutoff=5Fhz=3D10.0=EF=BC=88=E4=B8=8B?= =?UTF-8?q?=E4=B8=80=E6=AD=A5=E6=94=B9=E4=B8=BA=20apply=5Fto=EF=BC=89?= =?UTF-8?q?=E3=80=81=E6=9B=B2=E7=BA=BF=E5=9B=BE=E4=BB=8D=E5=9C=A8=E6=98=BE?= =?UTF-8?q?=E7=A4=BA=E7=AB=AF=E5=81=9A=20filtfilt=E3=80=81=E9=BD=BF?= =?UTF-8?q?=E8=BD=AE=E5=AF=B9=E8=AF=9D=E6=A1=86=E4=B8=8E=20plot=5F*=20pref?= =?UTF-8?q?erences=E3=80=81CSV=20=E7=9A=84=2017=20=E5=88=97=E4=B8=8E?= =?UTF-8?q?=E6=9C=AA=E6=BB=A4=E6=B3=A2=20buffer=E3=80=81session.json=20?= =?UTF-8?q?=E5=86=99=E5=85=A5=20filters=20=E5=9D=97=E9=83=BD=E8=BF=98?= =?UTF-8?q?=E6=B2=A1=E6=8E=A5=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 | 4 +- shared/src/orisys_app_shared/force/filters.py | 400 ++++++++++++++---- shared/tests/test_force_filters.py | 320 +++++++++++--- 3 files changed, 583 insertions(+), 141 deletions(-) diff --git a/shared/src/orisys_app_shared/force/__init__.py b/shared/src/orisys_app_shared/force/__init__.py index 2d61e45..83a5aa5 100644 --- a/shared/src/orisys_app_shared/force/__init__.py +++ b/shared/src/orisys_app_shared/force/__init__.py @@ -14,8 +14,9 @@ from orisys_app_shared.force.calibration_recorder import ( CalibrationRecorder, ) from orisys_app_shared.force.filters import ( - ForceFilterChain, FilterSettings, + ForceFilterChain, + describe_state, settings_from_mapping, ) from orisys_app_shared.force.history import MetricHistory, PlotState, lowpass_filter @@ -47,6 +48,7 @@ __all__ = [ "SessionSlice", "SessionStore", "build_force_calibration_v2", + "describe_state", "export_force_calibration_v2", "lowpass_filter", "meta_from_sensor_json", diff --git a/shared/src/orisys_app_shared/force/filters.py b/shared/src/orisys_app_shared/force/filters.py index 364c9d2..6d844f9 100644 --- a/shared/src/orisys_app_shared/force/filters.py +++ b/shared/src/orisys_app_shared/force/filters.py @@ -1,119 +1,370 @@ -"""One causal filter chain for the force readings, and the values it replaced. +"""The app's filter control surface: what was asked for, and what is running. -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. +The filter *chain* lives in the SDK — ``compute_flow`` applies the flow EMA +before the HHD, and ``_apply_force_filter`` applies the median and the low-pass +*before* the device calibration. Filtering anywhere else would produce numbers no +recording could reproduce, and would leave the unfiltered newtons unrecoverable +(the calibration has already been applied by then). So this module does three +things, and nothing else: -Both stages are causal and per-sample, because acquisition cannot look ahead: +* turn the operator's choice into one ``Sensor.configure(...)`` call + (:meth:`FilterSettings.apply_to`); +* read the **effective** values back from ``Sensor.filters_state()`` + (:meth:`FilterSettings.from_filters_state`), so the dialog and the record show + what the device is doing rather than what was typed into it; +* write the settings into the recording metadata (:meth:`FilterSettings.metadata`), + keeping the **measurement** group (``flow_ema_alpha``: it changes the flow, so + two recordings with different values are not comparable) apart from the + **readout** group (median, low-pass: they only smooth the force on its way out). -* **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. +``ForceFilterChain`` at the bottom is a reference implementation kept for +``tests/test_force_filters.py``, which pins that this app understands the SDK's +chain exactly. The app itself never filters. """ from __future__ import annotations -import json -from dataclasses import dataclass -from pathlib import Path -from typing import Any, Iterable, Optional +import inspect +import logging +from dataclasses import dataclass, replace +from typing import Any, Iterable, Mapping, Optional import numpy as np -#: Channels of the chain: three raw counts followed by the three newton values. +logger = logging.getLogger(__name__) + +#: Channels of the reference chain: three raw counts then the three newtons. CHANNELS = 6 +#: The SDK settings this app drives, in chain order (measurement first). These +#: are the sensor's own *state* keys: the same words appear in ``sensor.json``, +#: ``filters_state()``, ``session.json`` and the preferences, so a reader never +#: has to translate between two vocabularies. +MEASUREMENT_KEYS: tuple[str, ...] = ("flow_ema_alpha",) +READOUT_KEYS: tuple[str, ...] = ( + "force_median_window", + "force_lowpass_cutoff_hz", + "force_lowpass_order", + "force_lowpass_sample_rate_hz", +) -def _odd(value: Any, *, minimum: int = 3, maximum: int = 31) -> int: +#: State key -> ``Sensor.configure`` parameter. Not the same on both sides: the +#: low-pass pair is ``force_lowpass_*`` in the config/state and ``lowpass_*`` in +#: the call. One table, so "can the SDK take this?" and "what do we pass?" can +#: never disagree. +CONFIGURE_PARAMETERS: dict[str, str] = { + "flow_ema_alpha": "flow_ema_alpha", + "force_median_window": "force_median_window", + "force_lowpass_cutoff_hz": "lowpass_cutoff_hz", + "force_lowpass_order": "lowpass_order", + "force_lowpass_sample_rate_hz": "lowpass_sample_rate_hz", +} + +#: A sensible cutoff to offer when the operator switches the low-pass on. +SUGGESTED_LOWPASS_HZ = 10.0 + + +def _as_float(value: Any, default: float) -> float: try: - number = int(value) + return float(value) except (TypeError, ValueError): - return 0 - if number < minimum: - return 0 - if number % 2 == 0: - number += 1 - return min(number, maximum) + return default + + +def _as_int(value: Any, default: int) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _sample_rate(value: Any) -> float | str: + """``"auto"``, or a positive rate in Hz. Anything else is ``"auto"``.""" + if isinstance(value, str): + text = value.strip().lower() + return "auto" if text in ("", "auto", "measured") else "auto" + rate = _as_float(value, 0.0) + return rate if rate > 0 else "auto" @dataclass(frozen=True) class FilterSettings: - """What the operator chose. ``enabled`` flags decide what actually runs.""" + """The filter settings the operator chose, in the SDK's own vocabulary. - median_enabled: bool = False - median_window: int = 5 - lowpass_enabled: bool = True - lowpass_hz: float = 10.0 + Every stage is switched by its **own value** (``flow_ema_alpha >= 1``, + ``median_window <= 1``, ``lowpass_hz <= 0`` all mean "off"), so there is no + second boolean to disagree with the number. Values that cannot be used are + resolved to "off" by :meth:`normalised`, never to a default that filters. + """ + + #: Flow-EMA exponent. **Measurement**: it changes the flow, the contact + #: statistics and ``_force_raw``, so two recordings with different values are + #: not comparable. + flow_ema_alpha: float = 1.0 + #: Sliding-median window in samples. **Readout**. + median_window: int = 1 + #: Low-pass cutoff in Hz. **Readout**. + lowpass_hz: float = 0.0 lowpass_order: int = 2 - sample_rate_hz: float = 30.0 - #: Show the unfiltered readings next to the filtered ones on the cards. + #: ``"auto"`` follows the measured frame rate (30 fps until measured); a + #: number pins the rate the coefficients were designed for. + sample_rate_hz: float | str = "auto" + #: Display only — draw the unfiltered curves beside the filtered ones. Never + #: sent to the sensor, and never part of what makes two records comparable. show_unfiltered: bool = False + #: The preset this came from, when it came from one (kept so a record can say + #: "balanced" instead of only the numbers it expanded to). + preset: Optional[str] = None + # ------------------------------------------------------------- resolving 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)), + """Clamp to what the SDK can run, with unusable values meaning "off".""" + alpha = _as_float(self.flow_ema_alpha, 1.0) + if not 0.0 < alpha < 1.0: + alpha = 1.0 + window = max(1, _as_int(self.median_window, 1)) + cutoff = max(0.0, _as_float(self.lowpass_hz, 0.0)) + order = min(6, max(1, _as_int(self.lowpass_order, 2))) + return replace( + self, + flow_ema_alpha=alpha, + median_window=window, + lowpass_hz=cutoff, + lowpass_order=order, + sample_rate_hz=_sample_rate(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) + @property + def flow_ema_on(self) -> bool: + return 0.0 < _as_float(self.flow_ema_alpha, 1.0) < 1.0 - def as_dict(self) -> dict: + @property + def median_on(self) -> bool: + return _as_int(self.median_window, 1) > 1 + + @property + def lowpass_on(self) -> bool: + return _as_float(self.lowpass_hz, 0.0) > 0.0 + + def active(self) -> bool: + """Whether any stage filters at all (an off chain is a pass-through).""" + return bool(self.flow_ema_on or self.median_on or self.lowpass_on) + + def as_state(self) -> dict[str, Any]: + """The settings in the shape ``Sensor.filters_state()`` reports.""" + settings = self.normalised() + rate = settings.sample_rate_hz + readout: dict[str, Any] = { + "force_median_window": settings.median_window, + "force_lowpass_cutoff_hz": settings.lowpass_hz, + "force_lowpass_order": settings.lowpass_order, + "force_lowpass_rate_auto": rate == "auto", + } + if rate != "auto": + readout["force_lowpass_sample_rate_hz"] = float(rate) + return { + "measurement": {"flow_ema_alpha": settings.flow_ema_alpha}, + "readout": readout, + } + + def _values_by_state_key(self) -> dict[str, Any]: + """This request, keyed the way the SDK's state names things.""" 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, + "flow_ema_alpha": settings.flow_ema_alpha, + "force_median_window": settings.median_window, + "force_lowpass_cutoff_hz": settings.lowpass_hz, + "force_lowpass_order": settings.lowpass_order, + "force_lowpass_sample_rate_hz": settings.sample_rate_hz, + } + + def as_dict(self) -> dict[str, Any]: + """A flat mapping for the preferences file (SDK keys, plus display).""" + settings = self.normalised() + return { + "flow_ema_alpha": settings.flow_ema_alpha, + "force_median_window": settings.median_window, + "force_lowpass_cutoff_hz": settings.lowpass_hz, + "force_lowpass_order": settings.lowpass_order, + "force_lowpass_sample_rate_hz": settings.sample_rate_hz, "show_unfiltered": settings.show_unfiltered, + "preset": settings.preset, } def describe(self) -> str: - """A short human line for the recording metadata.""" + """A short line for the record, upstream stage first.""" + return describe_state(self.as_state()) + + def problems(self, sample_rate_hz: Optional[float] = None) -> tuple[str, ...]: + """Why the device would not do what these settings ask. + + The SDK drops a low-pass whose cutoff is at or above Nyquist instead of + silently re-tuning it, so the dialog has to say so *before* the session + starts — otherwise the file would claim a filter the device is not + running. + """ 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})" + issues: list[str] = [] + if settings.lowpass_on and sample_rate_hz: + nyquist = 0.5 * _as_float(sample_rate_hz, 0.0) + if nyquist > 0 and settings.lowpass_hz >= nyquist: + issues.append( + f"lowpass {settings.lowpass_hz:g} Hz is at or above Nyquist " + f"({nyquist:g} Hz): the device would not filter" + ) + return tuple(issues) + + # ---------------------------------------------------------- driving the SDK + def unsupported_by(self, sensor: object) -> tuple[str, ...]: + """The SDK keys this sensor's ``configure()`` does not accept. + + Reported rather than dropped: a stage the operator switched on but the + installed SDK cannot run is exactly the disagreement that ends up in a + recorded file. Older SDKs (before 0.5.2) have no ``flow_ema_alpha`` and + no median, and their flow EMA is not switchable at all. + """ + try: + parameters = inspect.signature(sensor.configure).parameters + except (AttributeError, TypeError, ValueError): # pragma: no cover - exotic fake + return tuple(CONFIGURE_PARAMETERS) + return tuple( + key + for key, parameter in CONFIGURE_PARAMETERS.items() + if parameter not in parameters + ) + + def configure_kwargs(self, skip: Iterable[str] = ()) -> dict[str, Any]: + """The ``Sensor.configure(...)`` call these settings amount to.""" + skipped = set(skip) + return { + CONFIGURE_PARAMETERS[key]: value + for key, value in self._values_by_state_key().items() + if key not in skipped + } + + def apply_to(self, sensor: object) -> tuple[str, ...]: + """Push these settings into the device; return what it could not take. + + Called from the acquisition thread right after the sensor opens, so a + session is filtered the way the operator asked for from its first frame. + """ + unsupported = self.unsupported_by(sensor) + if unsupported: + logger.warning( + "The installed orisys SDK cannot apply %s; the effective filter " + "state is in Sensor.filters_state()", + ", ".join(unsupported), ) - return " + ".join(parts) if parts else "none" + sensor.configure(**self.configure_kwargs(unsupported)) + return unsupported + + # ------------------------------------------------------------ reading back + @classmethod + def from_filters_state( + cls, state: Mapping[str, Any], *, show_unfiltered: bool = False + ) -> "FilterSettings": + """The settings a device reports as in effect. + + This is what the dialog displays and what a record stores: a config file + or a request can say one thing while the running device does another. + """ + measurement = dict(state.get("measurement") or {}) + readout = dict(state.get("readout") or {}) + rate: Any = "auto" + if not readout.get("force_lowpass_rate_auto", True): + rate = readout.get("force_lowpass_sample_rate_hz", "auto") + return cls( + flow_ema_alpha=_as_float(measurement.get("flow_ema_alpha"), 1.0), + median_window=_as_int(readout.get("force_median_window"), 1), + lowpass_hz=_as_float(readout.get("force_lowpass_cutoff_hz"), 0.0), + lowpass_order=_as_int(readout.get("force_lowpass_order"), 2), + sample_rate_hz=rate, + show_unfiltered=bool(show_unfiltered), + preset=state.get("preset"), + ).normalised() + + def metadata(self, sensor_state: Optional[Mapping[str, Any]] = None) -> dict[str, Any]: + """The ``filters`` block for ``session.json`` / a recording header. + + With a sensor state this records what the device was doing + (``source: sensor.filters_state``); without one it records what the app + asked for and says so, so the two are never mistaken for each other. + """ + if sensor_state: + measurement = dict(sensor_state.get("measurement") or {}) + readout = dict(sensor_state.get("readout") or {}) + preset = sensor_state.get("preset") + source = str(sensor_state.get("source") or "sensor.filters_state") + describe = describe_state(sensor_state) + else: + state = self.as_state() + measurement = state["measurement"] + readout = state["readout"] + preset = self.normalised().preset + source = "app.FilterSettings" + describe = describe_state(state) + block: dict[str, Any] = { + "measurement": measurement, + "readout": readout, + "describe": describe, + "source": source, + "show_unfiltered": bool(self.show_unfiltered), + } + if preset: + block["preset"] = str(preset) + return block -def settings_from_mapping(data: Optional[dict]) -> FilterSettings: +def describe_state(state: Mapping[str, Any]) -> str: + """``none`` / ``flowEMA(0.5) + median(3) + lowpass(10Hz,order2)``. + + Written upstream-first, which is also the order the stages run and the order + the two groups are listed in: the measurement stage first. + """ + measurement = dict(state.get("measurement") or {}) + readout = dict(state.get("readout") or {}) + parts: list[str] = [] + alpha = _as_float(measurement.get("flow_ema_alpha"), 1.0) + if 0.0 < alpha < 1.0: + parts.append(f"flowEMA({alpha:g})") + window = _as_int(readout.get("force_median_window"), 1) + if window > 1: + parts.append(f"median({window})") + cutoff = _as_float(readout.get("force_lowpass_cutoff_hz"), 0.0) + if cutoff > 0: + order = _as_int(readout.get("force_lowpass_order"), 2) + parts.append(f"lowpass({cutoff:g}Hz,order{order})") + return " + ".join(parts) if parts else "none" + + +def settings_from_mapping(data: Optional[Mapping[str, Any]]) -> FilterSettings: """Build settings from a preferences/metadata mapping, with defaults.""" source = dict(data or {}) - defaults = FilterSettings() + base = 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), + flow_ema_alpha=source.get("flow_ema_alpha", base.flow_ema_alpha), + median_window=source.get("force_median_window", base.median_window), + lowpass_hz=source.get("force_lowpass_cutoff_hz", base.lowpass_hz), + lowpass_order=source.get("force_lowpass_order", base.lowpass_order), + sample_rate_hz=source.get("force_lowpass_sample_rate_hz", base.sample_rate_hz), + show_unfiltered=source.get("show_unfiltered", base.show_unfiltered), + preset=source.get("preset", base.preset), ).normalised() +# --------------------------------------------------------------------------- +# Reference implementation: kept so the tests can prove this app understands the +# SDK's chain exactly. Nothing in the app calls it. +# --------------------------------------------------------------------------- class ForceFilterChain: - """The median + low-pass chain, one instance per acquisition session.""" + """The median + low-pass chain the SDK runs, reimplemented for comparison. + + Both stages are causal and per-sample, because acquisition cannot look + ahead: the median removes single-sample spikes without smearing a step, and + the low-pass is a Butterworth IIR with persistent state (``sosfilt`` with + ``zi``). Order is median then low-pass, so a spike cannot ring the IIR. + """ def __init__(self, settings: Optional[FilterSettings] = None) -> None: self._settings = (settings or FilterSettings()).normalised() @@ -128,11 +379,12 @@ class ForceFilterChain: self._sos = None self._zi = None settings = self._settings - if not settings.lowpass_enabled: + if not settings.lowpass_on: return from scipy.signal import butter - nyquist = 0.5 * settings.sample_rate_hz + rate = _as_float(settings.sample_rate_hz, 30.0) + nyquist = 0.5 * rate cutoff = min(float(settings.lowpass_hz), nyquist * 0.95) if cutoff <= 0: return @@ -182,7 +434,7 @@ class ForceFilterChain: return sample.copy() settings = self._settings out = sample.copy() - if settings.median_enabled: + if settings.median_on: window = int(settings.median_window) for index in range(CHANNELS): series = self._median[index] diff --git a/shared/tests/test_force_filters.py b/shared/tests/test_force_filters.py index 98ff3c7..58ae275 100644 --- a/shared/tests/test_force_filters.py +++ b/shared/tests/test_force_filters.py @@ -1,32 +1,254 @@ -"""The force filter chain: what it removes, and what it must not do. +"""The app's filter control surface: what it asks the SDK for, and what it records. -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. +The chain itself lives in the SDK, so these tests are about the three things +this app is responsible for: handing the operator's choice to +``Sensor.configure`` in the sensor's own vocabulary, reading the *effective* +state back instead of trusting the request, and writing a record that keeps the +measurement stage apart from the readout stages. + +``ForceFilterChain`` is checked here too, because it is what makes the SDK's +chain a known quantity rather than a black box: at the same settings it must +reproduce the SDK's filter exactly. """ from __future__ import annotations import numpy as np +import pytest from orisys_app_shared.force.filters import ( CHANNELS, FilterSettings, ForceFilterChain, + describe_state, settings_from_mapping, ) +class FakeSensor: + """A Sensor with the current SDK's ``configure`` surface.""" + + def __init__(self) -> None: + self.calls: list[dict] = [] + + def configure( + self, + *, + flow_ema_alpha=None, + force_median_window=None, + lowpass_cutoff_hz=None, + lowpass_order=None, + lowpass_sample_rate_hz=None, + ) -> None: + self.calls.append( + { + "flow_ema_alpha": flow_ema_alpha, + "force_median_window": force_median_window, + "lowpass_cutoff_hz": lowpass_cutoff_hz, + "lowpass_order": lowpass_order, + "lowpass_sample_rate_hz": lowpass_sample_rate_hz, + } + ) + + 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) +# ------------------------------------------------------------- value semantics +def test_every_stage_is_switched_by_its_own_value(): + assert FilterSettings().active() is False + assert FilterSettings(flow_ema_alpha=1.0).flow_ema_on is False + assert FilterSettings(flow_ema_alpha=0.5).flow_ema_on is True + assert FilterSettings(median_window=1).median_on is False + assert FilterSettings(median_window=3).median_on is True + assert FilterSettings(lowpass_hz=0.0).lowpass_on is False + assert FilterSettings(lowpass_hz=10.0).lowpass_on is True + # ...and any one of them is enough to make the chain do something. + assert FilterSettings(median_window=3).active() is True + + +def test_an_unusable_value_means_off_not_a_default_that_filters(): + assert FilterSettings(flow_ema_alpha="wide").normalised().flow_ema_alpha == 1.0 + assert FilterSettings(flow_ema_alpha=0.0).normalised().flow_ema_alpha == 1.0 + assert FilterSettings(flow_ema_alpha=2.0).normalised().flow_ema_alpha == 1.0 + assert FilterSettings(median_window="many").normalised().median_window == 1 + assert FilterSettings(lowpass_hz=-5.0).normalised().lowpass_hz == 0.0 + assert FilterSettings(lowpass_hz="lots").normalised().lowpass_hz == 0.0 + assert FilterSettings(sample_rate_hz=-1.0).normalised().sample_rate_hz == "auto" + assert FilterSettings(lowpass_order=99).normalised().lowpass_order == 6 + + +def test_problems_reports_a_cutoff_the_device_would_drop(): + """The SDK drops a cutoff at or above Nyquist; the dialog must say so.""" + assert FilterSettings(lowpass_hz=10.0).problems(30.0) == () + above = FilterSettings(lowpass_hz=20.0).problems(30.0) + assert len(above) == 1 and "Nyquist" in above[0] + # A low-pass that is off cannot be above anything. + assert FilterSettings(lowpass_hz=0.0).problems(30.0) == () + + +# ------------------------------------------------------------- driving the SDK +def test_applying_settings_is_one_configure_call_in_sdk_names(): + sensor = FakeSensor() + unsupported = FilterSettings( + flow_ema_alpha=0.5, median_window=3, lowpass_hz=10.0, sample_rate_hz="auto" + ).apply_to(sensor) + assert unsupported == () + assert sensor.calls == [ + { + "flow_ema_alpha": 0.5, + "force_median_window": 3, + "lowpass_cutoff_hz": 10.0, + "lowpass_order": 2, + "lowpass_sample_rate_hz": "auto", + } + ] + + +def test_an_sdk_that_cannot_take_a_key_is_reported_not_ignored(): + class OldSensor: + def configure(self, *, lowpass_cutoff_hz=None, lowpass_sample_rate_hz=None) -> None: + self.cutoff = lowpass_cutoff_hz + + sensor = OldSensor() + unsupported = FilterSettings(flow_ema_alpha=0.5, median_window=3).apply_to(sensor) + assert unsupported == ( + "flow_ema_alpha", + "force_median_window", + "force_lowpass_order", ) + # What the old SDK *can* take still gets there. + assert sensor.cutoff == 0.0 + + +def test_the_display_toggle_is_never_sent_to_the_sensor(): + kwargs = FilterSettings(show_unfiltered=True).configure_kwargs() + assert "show_unfiltered" not in kwargs + assert kwargs == { + "flow_ema_alpha": 1.0, + "force_median_window": 1, + "lowpass_cutoff_hz": 0.0, + "lowpass_order": 2, + "lowpass_sample_rate_hz": "auto", + } + + +# ------------------------------------------------------------ reading the truth +def test_the_effective_state_is_what_the_dialog_shows(): + """A request and a device can disagree; the device wins.""" + state = { + "preset": "balanced", + "measurement": {"flow_ema_alpha": 0.5, "flow_ema_applied": True}, + "readout": { + "force_median_window": 3, + "force_median_applied": True, + "force_lowpass_cutoff_hz": 0.0, + "force_lowpass_order": 2, + "force_lowpass_rate_auto": True, + "force_lowpass_applied": False, + }, + "source": "sensor.filters_state", + } + settings = FilterSettings.from_filters_state(state, show_unfiltered=True) + assert settings.flow_ema_alpha == 0.5 + assert settings.median_window == 3 + assert settings.lowpass_hz == 0.0 + assert settings.sample_rate_hz == "auto" + assert settings.show_unfiltered is True + assert settings.preset == "balanced" + assert settings.describe() == "flowEMA(0.5) + median(3)" + # Round trip: what came from the device configures the device. + assert settings.configure_kwargs()["flow_ema_alpha"] == 0.5 + + +def test_a_pinned_sample_rate_survives_the_round_trip(): + state = { + "measurement": {"flow_ema_alpha": 1.0}, + "readout": { + "force_median_window": 1, + "force_lowpass_cutoff_hz": 10.0, + "force_lowpass_order": 4, + "force_lowpass_rate_auto": False, + "force_lowpass_sample_rate_hz": 60.0, + }, + } + settings = FilterSettings.from_filters_state(state) + assert settings.sample_rate_hz == 60.0 + assert settings.lowpass_order == 4 + assert settings.as_state()["readout"]["force_lowpass_rate_auto"] is False + assert settings.as_state()["readout"]["force_lowpass_sample_rate_hz"] == 60.0 + + +# ---------------------------------------------------------------- the record +def test_metadata_records_the_device_state_and_says_where_it_came_from(): + state = { + "preset": "low_noise", + "measurement": {"flow_ema_alpha": 0.3}, + "readout": { + "force_median_window": 3, + "force_lowpass_cutoff_hz": 10.0, + "force_lowpass_order": 2, + "force_lowpass_rate_auto": True, + }, + "source": "sensor.filters_state", + } + block = FilterSettings.from_filters_state(state).metadata(state) + assert block["source"] == "sensor.filters_state" + assert block["preset"] == "low_noise" + # The two groups stay apart: this is what makes two records comparable or not. + assert block["measurement"] == {"flow_ema_alpha": 0.3} + assert block["readout"]["force_lowpass_cutoff_hz"] == 10.0 + assert block["describe"] == "flowEMA(0.3) + median(3) + lowpass(10Hz,order2)" + + +def test_metadata_without_a_device_is_labelled_as_a_request(): + block = FilterSettings(median_window=3).metadata() + assert block["source"] == "app.FilterSettings" + assert block["describe"] == "median(3)" + assert block["show_unfiltered"] is False + + +def test_describe_state_ignores_stages_that_are_off(): + assert describe_state({}) == "none" + assert describe_state({"measurement": {"flow_ema_alpha": 1.0}}) == "none" + assert ( + describe_state( + { + "measurement": {"flow_ema_alpha": 0.5}, + "readout": {"force_median_window": 1, "force_lowpass_cutoff_hz": 0.0}, + } + ) + == "flowEMA(0.5)" + ) + + +def test_settings_round_trip_through_a_preferences_mapping(): + settings = settings_from_mapping( + { + "flow_ema_alpha": 0.3, + "force_median_window": 7, + "force_lowpass_cutoff_hz": 12.5, + "show_unfiltered": True, + "preset": "low_noise", + } + ) + assert settings.flow_ema_alpha == 0.3 + assert settings.median_window == 7 + assert settings.lowpass_hz == 12.5 + assert settings.show_unfiltered is True + assert settings.preset == "low_noise" + assert settings.describe() == "flowEMA(0.3) + median(7) + lowpass(12.5Hz,order2)" + # persisting and reloading must not change anything. + assert settings_from_mapping(settings.as_dict()) == settings + # An empty mapping is "nothing filters", not "our favourite filter". + assert settings_from_mapping(None) == FilterSettings() + assert settings_from_mapping(None).active() is False + + +# ------------------------------------------- the reference chain (SDK parity) +def test_a_disabled_chain_is_a_pass_through(): + chain = ForceFilterChain(FilterSettings()) assert not chain.active sample = [1.0, 2.0, 3.0, 4.0, 5.0, 6.0] assert chain.process(sample).tolist() == sample @@ -34,9 +256,7 @@ def test_a_disabled_chain_is_a_pass_through(): 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) - ) + chain = ForceFilterChain(FilterSettings(median_window=3)) outputs = [ float(chain.process(_zeros(9.0 if index == 2 else 0.0))[0]) for index in range(5) @@ -48,18 +268,21 @@ def test_the_median_removes_a_single_sample_spike(): 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_median_window_is_bounded(): + # "Off" is a window of one; anything that cannot be a window at all becomes one. + assert FilterSettings(median_window=1).normalised().median_window == 1 + assert FilterSettings(median_window=0).normalised().median_window == 1 + assert FilterSettings(median_window=-3).normalised().median_window == 1 + assert FilterSettings(median_window=1).median_on is False + # A window of two is a real (if mild) median — the SDK runs it, so we do too; + # only one sample means the stage is off. + assert FilterSettings(median_window=2).median_on is True + assert ForceFilterChain(FilterSettings(median_window=1)).active 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) - ) + chain = ForceFilterChain(FilterSettings(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 @@ -74,19 +297,14 @@ def test_the_lowpass_settles_on_a_step_without_ringing_at_the_start(): 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]) + chain = ForceFilterChain(FilterSettings(median_window=3)) + for _ in range(3): + 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) - ) + chain = ForceFilterChain(FilterSettings(lowpass_hz=5.0, sample_rate_hz=30.0)) for _ in range(30): chain.process(_zeros(5.0)) chain.reset() @@ -95,20 +313,19 @@ def test_reset_clears_the_memory_between_sessions(): ) -def test_the_chain_reproduces_the_sdk_filter_exactly(): - """Replacing the SDK's low-pass with ours must not move a single number. +def test_the_reference_chain_reproduces_the_sdk_filter_exactly(): + """The SDK chain is not a black box: we can reproduce it number for number. ``Sensor._apply_force_filter`` filters ``_force_raw`` with ``orisys.signal_processing.LowpassFilter`` (Butterworth, ``sosfilt`` with - persistent state, steady-state reset) and then applies the device - calibration. Our chain uses the same recipe, so at the same settings the - recorded values stay identical — this test is what makes that a guarantee - rather than a hope. + persistent state, steady-state reset) and *then* applies the device + calibration. Running the same recipe here is what turns "the app does not + filter, the SDK does" from a claim into something a test can check. """ try: from orisys.signal_processing import LowpassFilter except Exception: # pragma: no cover - the SDK is not installed here - return + pytest.skip("orisys SDK is not importable") rng = np.random.default_rng(3) signal = np.cumsum(rng.normal(0.0, 1.0, (400, 3)), axis=0) * 10.0 + 500.0 @@ -118,9 +335,7 @@ def test_the_chain_reproduces_the_sdk_filter_exactly(): sdk.reset(signal[0]) sdk_out = np.array([sdk.process(row) for row in signal]) - chain = ForceFilterChain( - FilterSettings(lowpass_enabled=True, lowpass_hz=10.0, sample_rate_hz=30.0) - ) + chain = ForceFilterChain(FilterSettings(lowpass_hz=10.0, sample_rate_hz=30.0)) # The chain carries six channels (raw + newtons); the first three are the # raw counts the SDK filters, so give all six the same steady state. chain.reset(list(signal[0]) * 2) @@ -129,30 +344,3 @@ def test_the_chain_reproduces_the_sdk_filter_exactly(): assert np.array_equal(sdk_out, chain_out), ( f"the chain differs from the SDK filter by {np.abs(sdk_out - chain_out).max()}" ) - - 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 - - -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)" - assert settings_from_mapping(None).as_dict()["lowpass_hz"] == FilterSettings().lowpass_hz