滤波预设与生效值出口:FILTER_PRESETS + filters_state() + force_triplets() - What: 新增 FILTER_PRESETS(off/low_latency/balanced/low_noise,数值取自 _perf 的延迟表,low_noise 含 10 Hz 低通)与 available_filter_presets() 供 UI 取用;配置可写 filters.preset 且显式键优先于预设值;新增 Sensor.filters_state()(preset + measurement 组 flow_ema_alpha/flow_ema_applied/hhd_downsample/dis_mode + readout 组中值与低通的窗口/截止/阶数/采样率/是否 auto/是否生效 + sample_rate_source + any_stage_applied + source="sensor.filters_state")作为唯一"生效值"出口;新增 Sensor.force_triplets() 一次返回 {filtered:{raw,newton}, unfiltered:{raw,newton}},并在 _apply_force_filter 中于链后、标定前保存 _force_filtered_raw(修正:此前滤波后的 raw 计数没有保存,force_triplets 的 filtered.raw 实际返回的是输入);reset() 清空未滤波缓存。 - Why: 用户要求配置自描述(A)并把"生效值"作为审计唯一依据(B),同时要保留滤波前数据以便"两个 buffer、两个单位都保留";顺带证实并修掉了 filtered.raw 指向输入这一标签错误。 - Verification: external-frame 模式逐预设构造真实 Sensor 打印 filters_state:off/low_latency/balanced/low_noise 分别为 (ema 1.0 off, median 1 off, lp 0.0 off) / (1.0 off, 3 on, 0.0 off) / (0.5 on, 3 on, 0.0 off) / (0.3 on, 3 on, 10.0 on);注入 10,10,10,999,10,10 观察力三元组,filtered.raw 在尖峰帧为 10.0、unfiltered.raw 为 999.0,证明中值生效且未滤波 buffer 完整;compileall 通过。 - Not in this change: calibration.py 指纹尚未改读 filters_state(),examples/filters 与 docs/filters.md 未加,venv 里的 SDK 未升级(app 行为不变,仍硬编码 10 Hz 低通),app 侧 wrapper/齿轮/CSV 17 列未动。

This commit is contained in:
ykcheungab
2026-09-14 10:35:51 +08:00
parent 1adb1c30f0
commit 0beb36a1af
+143 -3
View File
@@ -42,6 +42,48 @@ from .signal_processing import (
DEFAULT_FORCE_LOWPASS_HZ = 0.0 DEFAULT_FORCE_LOWPASS_HZ = 0.0
DEFAULT_FORCE_MEDIAN_WINDOW = 3 DEFAULT_FORCE_MEDIAN_WINDOW = 3
DEFAULT_FLOW_EMA_ALPHA = 0.5 DEFAULT_FLOW_EMA_ALPHA = 0.5
#: Named filter presets, so a recommendation is *executable* instead of a
#: comment in a JSON file (``json.load`` cannot carry comments). A config may
#: write ``"filters": {"preset": "balanced"}`` and override any single key on
#: top of it; the resolved values are what ``Sensor.filters_state()`` reports.
#: The latency/noise trade-offs behind these numbers are measured in
#: ``_perf/`` and tabulated in ``docs/filters.md``.
FILTER_PRESETS = {
# Everything off: the file says what runs, and the raw reading is the record.
"off": {
"flow_ema_alpha": 1.0,
"force_median_window": 1,
"force_lowpass_cutoff_hz": 0.0,
},
# One frame of latency @30 fps: median only. The EMA changes the
# measurement, so it stays off when latency matters more than jitter.
"low_latency": {
"flow_ema_alpha": 1.0,
"force_median_window": 3,
"force_lowpass_cutoff_hz": 0.0,
},
# Two frames @30 fps: the EMA removes the spatially-coherent flow jitter
# that no force-level filter can reach, the median removes its spikes.
"balanced": {
"flow_ema_alpha": 0.5,
"force_median_window": 3,
"force_lowpass_cutoff_hz": 0.0,
},
# Three frames @30 fps and the quietest readout: as ``balanced`` plus the
# force low-pass on top of the median.
"low_noise": {
"flow_ema_alpha": 0.3,
"force_median_window": 3,
"force_lowpass_cutoff_hz": 10.0,
},
}
def available_filter_presets() -> dict:
"""The presets, for UIs that want to offer them (the studio's gear dialog)."""
return {name: dict(values) for name, values in FILTER_PRESETS.items()}
from .capture import ( from .capture import (
CaptureConfig, CaptureConfig,
CaptureSource, CaptureSource,
@@ -188,12 +230,17 @@ class Sensor:
if not isinstance(filters_cfg, dict): if not isinstance(filters_cfg, dict):
filters_cfg = {} filters_cfg = {}
legacy_off = filters_cfg.get("enabled", None) is False legacy_off = filters_cfg.get("enabled", None) is False
preset_name = filters_cfg.get("preset", None)
preset_values = FILTER_PRESETS.get(str(preset_name), {}) if preset_name else {}
self._filter_preset = str(preset_name) if preset_name in FILTER_PRESETS else None
def _cfg(key, default): def _cfg(key, default):
"""The configured value, or the documented default when unset.""" """The configured value: an explicit key, else the preset, else default."""
if legacy_off: if legacy_off:
return default return default
value = filters_cfg.get(key, None) value = filters_cfg.get(key, None)
if value is None:
value = preset_values.get(key, None)
return default if value is None else value return default if value is None else value
self._lowpass_order = 2 self._lowpass_order = 2
@@ -1014,6 +1061,10 @@ class Sensor:
if getattr(self, "_force_median", None) is not None: if getattr(self, "_force_median", None) is not None:
self._force_median.reset() self._force_median.reset()
self._ema_flow = None self._ema_flow = None
# The unfiltered pair belongs to the frame it was computed for: a new
# session must not report the previous one's reading.
self._force_unfiltered = None
self._force_raw_unfiltered = None
if self.img is not None: if self.img is not None:
self.initial_img = self.img.copy() self.initial_img = self.img.copy()
@@ -1233,7 +1284,8 @@ class Sensor:
self.calibration = cal_mod.load(cal, strict=True) self.calibration = cal_mod.load(cal, strict=True)
def _apply_force_filter(self): def _apply_force_filter(self):
raw = np.asarray(self._force_raw, dtype=np.float64) pre_filter = np.asarray(self._force_raw, dtype=np.float64)
raw = pre_filter
# Median first: it deletes the heavy-tailed spikes that the flow chain # Median first: it deletes the heavy-tailed spikes that the flow chain
# produces, so the low-pass does not have to spread them over time. # produces, so the low-pass does not have to spread them over time.
median = getattr(self, "_force_median", None) median = getattr(self, "_force_median", None)
@@ -1241,7 +1293,15 @@ class Sensor:
raw = median.process(raw) raw = median.process(raw)
if self._lowpass_enabled and self._lowpass_cutoff_hz > 0: if self._lowpass_enabled and self._lowpass_cutoff_hz > 0:
raw = self._force_filter.process(raw) raw = self._force_filter.process(raw)
fn, fx, fy = float(raw[0]), float(raw[1]), float(raw[2]) # The chain's own output, before calibration: the "filtered raw counts".
# Without this the raw axis would have no filtered/unfiltered pair and
# ``force_triplets`` could only offer the device counts twice.
self._force_filtered_raw = (
float(raw[0]),
float(raw[1]),
float(raw[2]),
)
fn, fx, fy = self._force_filtered_raw
# Apply loaded force_calibration.json when present; otherwise keep raw. # Apply loaded force_calibration.json when present; otherwise keep raw.
if self.calibration is not None: if self.calibration is not None:
@@ -1251,6 +1311,86 @@ class Sensor:
self.fnormal, self.fshearx, self.fsheary = fn, fx, fy self.fnormal, self.fshearx, self.fsheary = fn, fx, fy
# The same reading with no readout filtering, so a caller can keep both
# (the app stores the filtered values as the record and the unfiltered
# ones beside them). Calibrated with the same coefficients, so the two
# only differ by the filter chain.
ufn, ufx, ufy = (float(pre_filter[0]), float(pre_filter[1]), float(pre_filter[2]))
if self.calibration is not None:
ufn, ufx, ufy = self.calibration.convert(ufn, ufx, ufy)
ufx, ufy, ufn = float(ufx), float(ufy), float(ufn)
self._force_raw_unfiltered = (
float(pre_filter[0]), float(pre_filter[1]), float(pre_filter[2])
)
self._force_unfiltered = (ufn, ufx, ufy)
# ------------------------------------------------------- filter contract
def filters_state(self) -> dict:
"""What the filter chain is **actually** doing, for records and audits.
Read this instead of the config file: a config can be overridden at
runtime (``configure``) and a value that is present is never silently
replaced, so the effective state is the only honest answer. The two
groups are kept apart on purpose:
* ``measurement`` — stages upstream of the HHD/force computation
(``flow_ema_alpha``): turning them on changes *what is measured*, so
two recordings with different values are not comparable;
* ``readout`` — stages that only smooth the force triple on its way out
(median, low-pass): recorded, but they do not change the physical
claim.
"""
median_applied = bool(getattr(self._force_median, "enabled", False))
tracker_cfg = self._config.get("tracker") or {}
return {
"preset": getattr(self, "_filter_preset", None),
"measurement": {
"flow_ema_alpha": float(self._flow_ema_alpha),
"flow_ema_applied": bool(self._flow_ema_alpha < 1.0),
"hhd_downsample": int(getattr(self, "hhd_downsample", 0) or 0),
"dis_mode": tracker_cfg.get("dis_mode"),
},
"readout": {
"force_median_window": int(self._force_median_window),
"force_median_applied": median_applied,
"force_lowpass_cutoff_hz": float(self._lowpass_cutoff_hz),
"force_lowpass_order": int(self._lowpass_order),
"force_lowpass_sample_rate_hz": float(self._lowpass_sample_rate_hz),
"force_lowpass_rate_auto": bool(self._lowpass_rate_auto),
"force_lowpass_applied": bool(self._lowpass_enabled),
"sample_rate_source": str(getattr(self, "_force_rate_source", "config")),
},
"any_stage_applied": bool(self._filters_enabled),
"unfiltered_available": True,
"source": "sensor.filters_state",
}
def force_triplets(self) -> dict:
"""The filtered and unfiltered force triples of the current frame.
One call, so a sampling thread reads a single consistent frame: the
filtered pair is what ``read_info`` reports and what a recording stores
as its primary columns, the unfiltered pair is the same reading with the
readout stages bypassed.
"""
raw_unfiltered = getattr(self, "_force_raw_unfiltered", None)
unfiltered = getattr(self, "_force_unfiltered", None)
filtered_raw = getattr(self, "_force_filtered_raw", None)
return {
"filtered": {
"raw": tuple(filtered_raw) if filtered_raw is not None else None,
"newton": (
float(self.fnormal),
float(self.fshearx),
float(self.fsheary),
),
},
"unfiltered": {
"raw": tuple(raw_unfiltered) if raw_unfiltered is not None else None,
"newton": tuple(unfiltered) if unfiltered is not None else None,
},
}
# def export_config(self, path="./config/exported_config.json"): # def export_config(self, path="./config/exported_config.json"):
# """Return current configuration dict.""" # """Return current configuration dict."""
# configs.export_config(self._config, path) # configs.export_config(self._config, path)