固定 HHD 下采样并标记三轴力单位
避免运行中改变 HHD 测量配置,并明确部分标定时各轴单位。已通过 SDK 与共享层测试。
This commit is contained in:
@@ -205,7 +205,6 @@ sensor.configure(lowpass_cutoff_hz=5.0, lowpass_sample_rate_hz=30.0)
|
|||||||
* **lowpass_sample_rate_hz** (`float`,可选): 力信号采样率,默认 `30.0`。
|
* **lowpass_sample_rate_hz** (`float`,可选): 力信号采样率,默认 `30.0`。
|
||||||
* **lowpass_order** (`int`,可选): Butterworth 滤波器阶数,默认 `2`。
|
* **lowpass_order** (`int`,可选): Butterworth 滤波器阶数,默认 `2`。
|
||||||
* **lowpass_enabled** (`bool`,可选): 显式开启或关闭低通滤波。
|
* **lowpass_enabled** (`bool`,可选): 显式开启或关闭低通滤波。
|
||||||
* **hhd_downsample** (`int`,可选): HHD 解算下采样倍数,需为 `>= 1` 的整数,配置默认 `3`。
|
|
||||||
|
|
||||||
#### 读取结果
|
#### 读取结果
|
||||||
|
|
||||||
|
|||||||
+32
-19
@@ -1144,7 +1144,6 @@ class Sensor:
|
|||||||
lowpass_enabled=None,
|
lowpass_enabled=None,
|
||||||
flow_ema_alpha=None,
|
flow_ema_alpha=None,
|
||||||
force_median_window=None,
|
force_median_window=None,
|
||||||
hhd_downsample=None,
|
|
||||||
):
|
):
|
||||||
"""Update runtime-safe settings without reconnecting the camera.
|
"""Update runtime-safe settings without reconnecting the camera.
|
||||||
|
|
||||||
@@ -1167,9 +1166,18 @@ class Sensor:
|
|||||||
force_median_window : int, optional
|
force_median_window : int, optional
|
||||||
Causal sliding-median width applied to the raw force triple before
|
Causal sliding-median width applied to the raw force triple before
|
||||||
the low-pass. ``<= 1`` disables it; 3 is the default.
|
the low-pass. ``<= 1`` disables it; 3 is the default.
|
||||||
hhd_downsample : int, optional
|
|
||||||
Integer ``>= 1``. Rebuilds the HHD solver on the next frame.
|
|
||||||
"""
|
"""
|
||||||
|
if flow_ema_alpha is not None:
|
||||||
|
alpha = float(flow_ema_alpha)
|
||||||
|
if not (0.0 < alpha <= 1.0):
|
||||||
|
raise ValueError("flow_ema_alpha must be in (0, 1]")
|
||||||
|
if self.calibration is not None:
|
||||||
|
mismatches, _ = cal_mod.filter_mismatches(
|
||||||
|
self.calibration.pipeline.get("filters"),
|
||||||
|
{"measurement": {"flow_ema_alpha": alpha}},
|
||||||
|
)
|
||||||
|
if mismatches:
|
||||||
|
raise cal_mod.CalibrationError("; ".join(mismatches))
|
||||||
filter_changed = False
|
filter_changed = False
|
||||||
if lowpass_cutoff_hz is not None:
|
if lowpass_cutoff_hz is not None:
|
||||||
self._lowpass_cutoff_hz = float(lowpass_cutoff_hz)
|
self._lowpass_cutoff_hz = float(lowpass_cutoff_hz)
|
||||||
@@ -1209,9 +1217,6 @@ class Sensor:
|
|||||||
filter_changed = True
|
filter_changed = True
|
||||||
|
|
||||||
if flow_ema_alpha is not None:
|
if flow_ema_alpha is not None:
|
||||||
alpha = float(flow_ema_alpha)
|
|
||||||
if not (0.0 < alpha <= 1.0):
|
|
||||||
raise ValueError("flow_ema_alpha must be in (0, 1]")
|
|
||||||
self._flow_ema_alpha = alpha
|
self._flow_ema_alpha = alpha
|
||||||
self._ema_flow = None
|
self._ema_flow = None
|
||||||
|
|
||||||
@@ -1234,17 +1239,6 @@ class Sensor:
|
|||||||
if self.compute_deformation_valid:
|
if self.compute_deformation_valid:
|
||||||
self._apply_force_filter()
|
self._apply_force_filter()
|
||||||
|
|
||||||
if hhd_downsample is not None:
|
|
||||||
try:
|
|
||||||
k = max(1, int(hhd_downsample))
|
|
||||||
except (TypeError, ValueError) as exc:
|
|
||||||
raise ValueError("hhd_downsample must be an integer >= 1") from exc
|
|
||||||
if k != self._hhd_downsample:
|
|
||||||
self._hhd_downsample = k
|
|
||||||
self.__hhd = None
|
|
||||||
self._hhd_grid_shape = None
|
|
||||||
self._hhd_ref_shape = None
|
|
||||||
self._update_hhd_mask_cache()
|
|
||||||
return self
|
return self
|
||||||
|
|
||||||
def _bundle_name_candidates(self) -> list[str]:
|
def _bundle_name_candidates(self) -> list[str]:
|
||||||
@@ -1381,7 +1375,11 @@ class Sensor:
|
|||||||
"stitch_resolution": camera.get("resolution_stitch"),
|
"stitch_resolution": camera.get("resolution_stitch"),
|
||||||
"dis_mode": tracker.get("dis_mode"),
|
"dis_mode": tracker.get("dis_mode"),
|
||||||
"hhd_downsample": tracker.get("hhd_downsample"),
|
"hhd_downsample": tracker.get("hhd_downsample"),
|
||||||
"filters": cal_mod.filters_fingerprint(self._config.get("filters")),
|
"filters": cal_mod.filters_fingerprint(
|
||||||
|
self.filters_state()
|
||||||
|
if hasattr(self, "_flow_ema_alpha")
|
||||||
|
else self._config.get("filters")
|
||||||
|
),
|
||||||
}
|
}
|
||||||
|
|
||||||
@property
|
@property
|
||||||
@@ -1393,6 +1391,22 @@ class Sensor:
|
|||||||
def force_unit(self) -> str:
|
def force_unit(self) -> str:
|
||||||
return "N" if self.force_calibrated else "raw"
|
return "N" if self.force_calibrated else "raw"
|
||||||
|
|
||||||
|
@property
|
||||||
|
def calibrated_axes(self) -> tuple[str, ...]:
|
||||||
|
"""Axes with a loaded force-calibration curve."""
|
||||||
|
channels = self.calibration.channels if self.calibration is not None else {}
|
||||||
|
return tuple(axis for axis in cal_mod.CHANNEL_IDS if axis in channels)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def force_units(self) -> tuple[str, str, str]:
|
||||||
|
"""Units of FN, FX and FY in that order."""
|
||||||
|
axes = self.calibrated_axes
|
||||||
|
return (
|
||||||
|
"N" if "fn" in axes else "raw",
|
||||||
|
"N" if "fx" in axes else "raw",
|
||||||
|
"N" if "fy" in axes else "raw",
|
||||||
|
)
|
||||||
|
|
||||||
def set_calibration(
|
def set_calibration(
|
||||||
self,
|
self,
|
||||||
cal: "cal_mod.Calibration | str | Path | None",
|
cal: "cal_mod.Calibration | str | Path | None",
|
||||||
@@ -1481,7 +1495,6 @@ class Sensor:
|
|||||||
"measurement": {
|
"measurement": {
|
||||||
"flow_ema_alpha": float(self._flow_ema_alpha),
|
"flow_ema_alpha": float(self._flow_ema_alpha),
|
||||||
"flow_ema_applied": bool(self._flow_ema_alpha < 1.0),
|
"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"),
|
"dis_mode": tracker_cfg.get("dis_mode"),
|
||||||
},
|
},
|
||||||
"readout": {
|
"readout": {
|
||||||
|
|||||||
@@ -319,6 +319,40 @@ def test_device_calibration_rejects_partial_axes_and_unsafe_id(tmp_path: Path):
|
|||||||
load_for_device(root, "box", "../BOX-001")
|
load_for_device(root, "box", "../BOX-001")
|
||||||
|
|
||||||
|
|
||||||
|
def test_force_units_report_partial_calibration_without_changing_triplets(tmp_path: Path):
|
||||||
|
path = tmp_path / "partial.json"
|
||||||
|
doc = json.loads(_box_cal_path().read_text(encoding="utf-8"))
|
||||||
|
doc["channels"] = {"normal": doc["channels"]["normal"]}
|
||||||
|
path.write_text(json.dumps(doc), encoding="utf-8")
|
||||||
|
sensor = Sensor("external", config_name="box", cal_path=path, verbose=False, backend="cpu")
|
||||||
|
try:
|
||||||
|
assert sensor.calibrated_axes == ("fn",)
|
||||||
|
assert sensor.force_units == ("N", "raw", "raw")
|
||||||
|
assert set(sensor.force_triplets()) == {"filtered", "unfiltered"}
|
||||||
|
sensor.set_calibration(None)
|
||||||
|
assert sensor.calibrated_axes == ()
|
||||||
|
assert sensor.force_units == ("raw", "raw", "raw")
|
||||||
|
finally:
|
||||||
|
sensor.disconnect()
|
||||||
|
|
||||||
|
|
||||||
|
def test_measurement_filter_change_refuses_incompatible_calibration(tmp_path: Path):
|
||||||
|
path = tmp_path / "filtered.json"
|
||||||
|
doc = json.loads(_box_cal_path().read_text(encoding="utf-8"))
|
||||||
|
doc["pipeline"] = {"filters": {"measurement": {"flow_ema_alpha": 1.0}}}
|
||||||
|
path.write_text(json.dumps(doc), encoding="utf-8")
|
||||||
|
sensor = Sensor("external", config_name="box", cal_path=path, verbose=False, backend="cpu")
|
||||||
|
try:
|
||||||
|
sensor.configure(force_median_window=3)
|
||||||
|
assert sensor.filters_state()["readout"]["force_median_window"] == 3
|
||||||
|
with pytest.raises(CalibrationError, match="flow_ema_alpha"):
|
||||||
|
sensor.configure(flow_ema_alpha=0.5, force_median_window=5)
|
||||||
|
assert sensor.filters_state()["measurement"]["flow_ema_alpha"] == 1.0
|
||||||
|
assert sensor.filters_state()["readout"]["force_median_window"] == 3
|
||||||
|
finally:
|
||||||
|
sensor.disconnect()
|
||||||
|
|
||||||
|
|
||||||
def test_parse_shipped_box_file_directly():
|
def test_parse_shipped_box_file_directly():
|
||||||
cal = parse(json.loads(_box_cal_path().read_text(encoding="utf-8")), source_path=_box_cal_path())
|
cal = parse(json.loads(_box_cal_path().read_text(encoding="utf-8")), source_path=_box_cal_path())
|
||||||
assert cal.schema_version == 2
|
assert cal.schema_version == 2
|
||||||
|
|||||||
@@ -246,24 +246,16 @@ def test_read_info_force_stays_raw_without_calibration_json(monkeypatch):
|
|||||||
sensor.disconnect()
|
sensor.disconnect()
|
||||||
|
|
||||||
|
|
||||||
def test_configure_hhd_downsample_invalidates_solver_cache():
|
def test_hhd_downsample_is_fixed_by_sensor_config():
|
||||||
sensor = _external_sensor()
|
sensor = _external_sensor()
|
||||||
try:
|
try:
|
||||||
sensor._hhd_downsample = 1
|
configured = sensor._config["tracker"]["hhd_downsample"]
|
||||||
sensor._Sensor__hhd = object()
|
assert sensor._hhd_downsample == configured
|
||||||
sensor._hhd_grid_shape = (40, 40)
|
sensor.configure(force_median_window=3)
|
||||||
sensor.valid_mask = np.ones((90, 90), dtype=bool)
|
assert sensor._hhd_downsample == configured
|
||||||
sensor.valid_mask_hhd = sensor.valid_mask
|
assert "hhd_downsample" not in sensor.filters_state()["measurement"]
|
||||||
|
with pytest.raises(TypeError):
|
||||||
sensor.configure(hhd_downsample=3)
|
sensor.configure(hhd_downsample=1)
|
||||||
assert sensor._hhd_downsample == 3
|
|
||||||
assert sensor._Sensor__hhd is None
|
|
||||||
assert sensor._hhd_grid_shape is None
|
|
||||||
assert sensor.valid_mask_hhd is not None
|
|
||||||
assert sensor.valid_mask_hhd.shape == (30, 30)
|
|
||||||
|
|
||||||
with pytest.raises(ValueError):
|
|
||||||
sensor.configure(hhd_downsample="nope")
|
|
||||||
finally:
|
finally:
|
||||||
sensor.disconnect()
|
sensor.disconnect()
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user