允许 HHD 下采样率不同的标定保持兼容

按实际运行上下文核对其余处理参数,保留文件指纹与设备身份校验。115 项标定及运行时检查通过。
This commit is contained in:
ykcheungab
2026-10-19 16:30:42 +08:00
parent 5f79c8c7f6
commit c78f7a304f
6 changed files with 135 additions and 8 deletions
+4 -4
View File
@@ -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 "<in-memory calibration>"
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}")
+26 -3
View File
@@ -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)
+12 -1
View File
@@ -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")