新增单链力滤波引擎(中值 + 因果低通),为"显示=记录"做准备 - What: shared/src/orisys_app_shared/force/filters.py 新增 FilterSettings(中值开关/窗口、低通开关/截止/阶数、采样率、show_unfiltered,窗口强制奇数、<3 自动关闭、空设置直通)与 ForceFilterChain:6 通道(3 原始计数 + 3 牛顿值)、先中值后低通、两级均因果且逐样本、每通道独立 IIR 状态,reset(first) 用稳态初值启动以避免会话开头的假瞬态;settings_from_mapping 支持从 prefs/metadata 构造,describe() 给出一行摘要写进录制元数据;force/__init__ 导出这三个名字。新增 shared/tests/test_force_filters.py(7 例:直通、中值去单点尖峰但不削持续变化、窗口奇数化与下限、阶跃无起始瞬态、通道状态独立、reset 清状态、prefs 往返),并加入 drive_shared 的模块表。 - Why: 用户问"滤波在哪里调",查清现状是两处固化 10 Hz 低通(SDK 侧影响记录、显示侧再用 filtfilt 二次滤波导致显示≠记录)且 median 完全不存在;用户要求把采集滤波与显示滤波统一成一条链、显示值必须等于记录值、并保留滤波前数据(另开一个 buffer,且写进 forces.csv)。本轮先落地这条链并覆盖测试。 - Verification: drive_shared.py 141 项 0 失败(含新增 7 项)、ui_check.py 554 项 0 失败、test_ui_smoke.py 71 项 0 失败、locale_studio.py 262 键齐全。 - Not in this change: 采集路径未改动(SDK 低通仍是 10 Hz、forces.csv 仍是 11 列)——把带状态的滤波器半途接进记录会让数据只滤一半,因此下一步一次性完成:关 SDK 侧滤波、worker 应用链并让 packet 同时携带滤波前后两套值、forces.csv 追加 6 列未滤波读数(11→17、RECORD_FORMAT_VERSION+1)、SessionStore 第二 buffer、去掉显示端 filtfilt、曲线表头新增齿轮按钮并扩展曲线设置对话框(中值/低通/显示未滤波,运行中禁用滤波控件)。完整清单见 .repro/filter_plan.md。
This commit is contained in:
@@ -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
|
||||
Reference in New Issue
Block a user