From c78f7a304f7552b43731187c007735d5b6a808f2 Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Mon, 19 Oct 2026 16:30:42 +0800 Subject: [PATCH] =?UTF-8?q?=E5=85=81=E8=AE=B8=20HHD=20=E4=B8=8B=E9=87=87?= =?UTF-8?q?=E6=A0=B7=E7=8E=87=E4=B8=8D=E5=90=8C=E7=9A=84=E6=A0=87=E5=AE=9A?= =?UTF-8?q?=E4=BF=9D=E6=8C=81=E5=85=BC=E5=AE=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 按实际运行上下文核对其余处理参数，保留文件指纹与设备身份校验。115 项标定及运行时检查通过。 --- docs/calibration-model-contract.zh.md | 6 +++ src/orisys/calibration.py | 8 +-- src/orisys/calibration_context.py | 29 +++++++++-- src/orisys/calibration_models.py | 13 ++++- tests/test_calibration.py | 12 +++++ tests/test_calibration_context.py | 75 +++++++++++++++++++++++++++ 6 files changed, 135 insertions(+), 8 deletions(-) diff --git a/docs/calibration-model-contract.zh.md b/docs/calibration-model-contract.zh.md index 61f3098..d1f9282 100644 --- a/docs/calibration-model-contract.zh.md +++ b/docs/calibration-model-contract.zh.md @@ -121,6 +121,12 @@ runtime.set_calibration_reference({"fn": measured_zero_raw}, reference_id="unloa 特征定义或单位改变须升级 `feature_version`，配置覆盖后重新计算，不能用配置文件名代替。 实现由 `calibration_context.build_context` 统一生成，特征版本为 `orisys_hhd_raw/1`。 +`hhd_downsample` 是 schema 1/2/3 的兼容例外，允许数值偏差；实际参数、系数和文件指纹不变。 +schema 3 的 `build_context` 运行时上下文仅替换下采样率重算比对签名； +其余处理参数、几何、展开域、SN 和传感器类型仍须匹配。 +序列化与普通映射仍校验实际指纹。 +验证：`test_calibration_context.py`、`test_calibration.py`。 + ## 空间增益模型 `spatial_gain/1` 固定输出 `fn`，使用 `fn_delta_autocal_uv/1`： diff --git a/src/orisys/calibration.py b/src/orisys/calibration.py index 73f002f..ad70906 100644 --- a/src/orisys/calibration.py +++ b/src/orisys/calibration.py @@ -52,7 +52,7 @@ _KNOWN_TOP_KEYS = frozenset( #: Filter stages that change **what is measured**: they sit upstream of the #: force computation, so two recordings with different values are not comparable. #: They belong to the measurement-pipeline identity, for the same reason -#: ``dis_mode`` and ``hhd_downsample`` do. +#: ``dis_mode`` does. HHD resolution is retained in provenance but may differ. FILTER_MEASUREMENT_KEYS: tuple[str, ...] = ("flow_ema_alpha",) #: Filter stages that only smooth the force triple on its way out. They are #: recorded so a file explains its own numbers, but a difference between two @@ -876,7 +876,7 @@ def validate_for_device( *, pipeline: Mapping[str, Any] | None = None, ) -> Calibration: - """Reject a calibration for another device or measurement pipeline.""" + """Reject a different device or pipeline, allowing HHD resolution differences.""" path = cal.source_path or "" if cal.identity.get("sensor_type") != sensor_type or cal.identity.get("sn") != device_id: raise CalibrationError(f"calibration identity mismatch: {path}") @@ -886,7 +886,7 @@ def validate_for_device( # Full feature/geometry validation happens when the processor binds its # actual context. Device construction only knows this legacy subset. if pipeline is not None: - for key in ("stitch_resolution", "dis_mode", "hhd_downsample"): + for key in ("stitch_resolution", "dis_mode"): if pipeline.get(key) is not None and cal.pipeline.get(key) != pipeline[key]: raise CalibrationError(f"calibration pipeline mismatch ({key}): {path}") measurement, _ = filter_mismatches(cal.pipeline.get("filters"), pipeline.get("filters")) @@ -898,7 +898,7 @@ def validate_for_device( if set(CHANNEL_IDS) - cal.channels.keys(): raise CalibrationError(f"calibration requires FN, FX and FY: {path}") if pipeline is not None: - for key in ("stitch_resolution", "dis_mode", "hhd_downsample"): + for key in ("stitch_resolution", "dis_mode"): recorded, current = cal.pipeline.get(key), pipeline.get(key) if recorded is not None and current is not None and recorded != current: raise CalibrationError(f"calibration pipeline mismatch ({key}): {path}") diff --git a/src/orisys/calibration_context.py b/src/orisys/calibration_context.py index dbd1791..0534fb4 100644 --- a/src/orisys/calibration_context.py +++ b/src/orisys/calibration_context.py @@ -4,6 +4,7 @@ from __future__ import annotations import hashlib import json +from copy import deepcopy from pathlib import Path import numpy as np @@ -11,6 +12,29 @@ import numpy as np from .calibration import filters_fingerprint +def _feature_signature(identity) -> str: + encoded = json.dumps(identity, sort_keys=True, separators=(",", ":"), allow_nan=False) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + +class CalibrationContext(dict): + """Actual pipeline fields plus the runtime inputs needed to compare HHD settings. + + JSON serialization contains only the actual fields. Retaining the signature + inputs lets old files differ in downsample without ignoring other hashed settings. + """ + + def __init__(self, fields, feature_identity): + super().__init__(fields) + self._feature_identity = deepcopy(feature_identity) + + def signature_for_downsample(self, downsample: int) -> str: + """Recompute the runtime identity with only the requested HHD resolution.""" + identity = deepcopy(self._feature_identity) + identity["tracker"]["hhd_downsample"] = downsample + return _feature_signature(identity) + + def sha256_file(path) -> str: return hashlib.sha256(Path(path).read_bytes()).hexdigest() @@ -70,6 +94,5 @@ def build_context(config, filters=None, *, geometry_sha256=None, source_scale=No "source_scale": source_scale, "preprocessing": preprocessing or {}, } - encoded = json.dumps(effective, sort_keys=True, separators=(",", ":"), allow_nan=False) - result["feature_signature_sha256"] = hashlib.sha256(encoded.encode("utf-8")).hexdigest() - return result + result["feature_signature_sha256"] = _feature_signature(effective) + return CalibrationContext(result, effective) diff --git a/src/orisys/calibration_models.py b/src/orisys/calibration_models.py index 76ec5b6..472aff5 100644 --- a/src/orisys/calibration_models.py +++ b/src/orisys/calibration_models.py @@ -209,8 +209,19 @@ class ModelCalibration: return deepcopy(self._document) def validate_for_context(self, pipeline: Mapping[str, Any]) -> None: + from .calibration_context import CalibrationContext + for key, value in self.pipeline.items(): - if key == "filters": + if key == "hhd_downsample": + # HHD magnitudes are normalized to the full flow grid; resolution + # can introduce numerical deviations without changing the force unit. + continue + if key == "feature_signature_sha256" and isinstance(pipeline, CalibrationContext): + _require( + pipeline.signature_for_downsample(self.pipeline["hhd_downsample"]) == value, + "calibration pipeline mismatch: feature_signature_sha256", + ) + elif key == "filters": actual = pipeline.get(key, {}) _require(isinstance(actual, Mapping) and actual.get("measurement") == value["measurement"], "calibration pipeline mismatch: filters.measurement") diff --git a/tests/test_calibration.py b/tests/test_calibration.py index 0b3007c..1515e49 100644 --- a/tests/test_calibration.py +++ b/tests/test_calibration.py @@ -306,6 +306,18 @@ def test_device_calibration_rejects_wrong_identity_and_pipeline(tmp_path: Path): load_for_device(root, "box", "BOX-001", pipeline={"dis_mode": 1}) +def test_device_calibration_accepts_only_downsample_difference(tmp_path: Path): + path = _device_file(tmp_path, "BOX-001") + doc = json.loads(path.read_text(encoding="utf-8")) + doc["pipeline"] = {"dis_mode": 1, "hhd_downsample": 3} + path.write_text(json.dumps(doc), encoding="utf-8") + assert load_for_device( + tmp_path, "box", "BOX-001", pipeline={"dis_mode": 1, "hhd_downsample": 2}, + ) is not None + with pytest.raises(CalibrationError, match="dis_mode"): + load_for_device(tmp_path, "box", "BOX-001", pipeline={"dis_mode": 9, "hhd_downsample": 2}) + + def test_device_calibration_rejects_partial_axes_and_unsafe_id(tmp_path: Path): root = tmp_path / "calibrations" root.mkdir() diff --git a/tests/test_calibration_context.py b/tests/test_calibration_context.py index a6fa5c1..60eddc2 100644 --- a/tests/test_calibration_context.py +++ b/tests/test_calibration_context.py @@ -1,6 +1,7 @@ """Feature identities use effective measurement settings, never display metadata.""" from copy import deepcopy +import json import numpy as np import pytest @@ -57,3 +58,77 @@ def test_domain_fingerprint_binds_map_content_not_only_dimensions(): changed[key].flat[0] = not changed[key].flat[0] if key == "valid_mask" else 1 assert domain_fingerprint(changed) != original assert domain_fingerprint({key: value.copy() for key, value in domain.items()}) == original + + +def _model(data): + from orisys.calibration import parse + + return parse({ + "schema_version": 3, "unit": "N", + "identity": {"sensor_type": "box", "sn": "A1"}, + "pipeline": build_context(data), + "model": { + "type": "global_gain/1", "input_profile": "force_delta/1", + "output_frame": "sdk_force/1", "outputs": ["fn"], + "params": {"gains_N_per_raw": {"fn": .001}, "valid_force_range_N": [0, 3]}, + }, + }) + + +@pytest.mark.parametrize("downsample", [1, 2, 4]) +def test_downsample_only_difference_accepts_existing_model_without_rewriting_it(downsample): + from orisys.calibration import validate_for_device + + data = config() + data["tracker"]["hhd_downsample"] = 3 + model = _model(data) + original = model.to_dict() + fingerprint = model.fingerprint + data["tracker"]["hhd_downsample"] = downsample + actual = build_context(data) + assert json.loads(json.dumps(actual)) == actual + assert actual["feature_signature_sha256"] != model.pipeline["feature_signature_sha256"] + validate_for_device(model, "box", "A1", pipeline=actual) + model.validate_for_context(actual) + assert model.to_dict() == original + assert model.fingerprint == fingerprint + assert actual["hhd_downsample"] == downsample + + +def test_runtime_context_binds_a_different_downsample_model_and_predicts(): + from orisys.force_readout import ForceReadout + from orisys.sensor import Sensor + + data = config() + data["tracker"]["hhd_downsample"] = 3 + sensor = Sensor.__new__(Sensor) + sensor._config = data + sensor.calibration = _model(data) + sensor._force_raw = (100.0, 0.0, 0.0) + sensor._force_readout = ForceReadout({}) + sensor._model_readouts = {} + sensor._reset_calibration_session() + data["tracker"]["hhd_downsample"] = 2 + sensor.bind_calibration_context(build_context(data)) + sensor.set_calibration_reference({"fn": 100.0}, reference_id="unloaded") + result = sensor.apply_calibration_frame( + frame_id=1, timestamp_s=0.1, raw=(120.0, 0.0, 0.0), contact_found=True, + ) + assert result.values_N["fn"] == pytest.approx(.02) + + +@pytest.mark.parametrize("section,key,value", [ + ("camera", "resolution_raw", [640, 480]), + ("tracker", "dis_mode", 1), + ("tracker", "flow_iterations", 99), + ("contact", "threshold", .9), + ("filters", "flow_ema_alpha", .5), +]) +def test_downsample_tolerance_does_not_hide_other_pipeline_changes(section, key, value): + data = config() + data["tracker"]["hhd_downsample"] = 3 + model = _model(data) + data["tracker"]["hhd_downsample"] = 2 + data.setdefault(section, {})[key] = value + with pytest.raises(ValueError, match="pipeline mismatch"): + model.validate_for_context(build_context(data))