diff --git a/src/orisys/sensor.py b/src/orisys/sensor.py index 1c901be..28751cd 100644 --- a/src/orisys/sensor.py +++ b/src/orisys/sensor.py @@ -42,6 +42,48 @@ from .signal_processing import ( DEFAULT_FORCE_LOWPASS_HZ = 0.0 DEFAULT_FORCE_MEDIAN_WINDOW = 3 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 ( CaptureConfig, CaptureSource, @@ -188,12 +230,17 @@ class Sensor: if not isinstance(filters_cfg, dict): filters_cfg = {} 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): - """The configured value, or the documented default when unset.""" + """The configured value: an explicit key, else the preset, else default.""" if legacy_off: return default value = filters_cfg.get(key, None) + if value is None: + value = preset_values.get(key, None) return default if value is None else value self._lowpass_order = 2 @@ -1014,6 +1061,10 @@ class Sensor: if getattr(self, "_force_median", None) is not None: self._force_median.reset() 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: self.initial_img = self.img.copy() @@ -1233,7 +1284,8 @@ class Sensor: self.calibration = cal_mod.load(cal, strict=True) 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 # produces, so the low-pass does not have to spread them over time. median = getattr(self, "_force_median", None) @@ -1241,7 +1293,15 @@ class Sensor: raw = median.process(raw) if self._lowpass_enabled and self._lowpass_cutoff_hz > 0: 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. if self.calibration is not None: @@ -1251,6 +1311,86 @@ class Sensor: 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"): # """Return current configuration dict.""" # configs.export_config(self._config, path)