diff --git a/docs/calibration-model-contract.zh.md b/docs/calibration-model-contract.zh.md index d1f9282..2e0483b 100644 --- a/docs/calibration-model-contract.zh.md +++ b/docs/calibration-model-contract.zh.md @@ -1,7 +1,7 @@ # 统一力标定模型契约 Purpose: reference. Scope: SDK 标定文件选择、SN 绑定和推断接口。 -状态：2026-10-18 已实现 schema 3 的空间/全局增益；ML 后端仅保留契约，尚未实现。 +状态：2026-10-19 已实现 schema 3 的空间、全局与场加权增益；ML 后端仅保留契约，尚未实现。 保持 `cal_path` 与现有单点文件；新增空间模型和机器学习模型使用同一入口。 运行中的 Finger 可调用 `runtime.set_calibration(file, device_id=sn)`；调用者须在处理线程的帧边界执行。 该接口按实际几何、展开域及处理参数重新校验，成功切换后清除会话零点；校验失败保留原模型。 @@ -154,46 +154,9 @@ FingerRuntime 使用 `artifacts` 中 `role="coordinate_map"` 的只读 NPZ： 输入 profile 为 `force_delta/1`，由 SDK 按同一归零语义准备对应 raw 增量。 这不同于旧曲线的绝对 raw 输入；兼容适配器不得替旧文件额外扣零。 -## 机器学习扩展 +## 其他模型类型 -同一 SN、路径和外层字段不变，只替换 `model`。下面只预留 `onnx/1` 示例， -不表示 SDK 已支持 ONNX；后端实现和输入 profile 验证通过前必须拒绝加载。 - -```json -{ - "type": "onnx/1", - "input_profile": "fn_delta_autocal_uv/1", - "output_frame": "sdk_force/1", - "outputs": ["fn"], - "params": { - "input_tensor": "features", - "output_tensor": "force_N", - "feature_order": ["delta_raw", "u", "v"], - "dtype": "float32", - "shape": [1, 3], - "valid_force_range_N": [0.0, 3.0] - }, - "artifacts": [ - { - "role": "weights", - "path": "DD02-GL-001/model.onnx", - "sha256": "0000000000000000000000000000000000000000000000000000000000000000" - } - ] -} -``` - -此标量 profile 的规范输入即上述顺序、类型、形状；输出 float32 `[1, len(outputs)]`， -轴序严格按 `outputs`。归一化及单位转换封装在已验证的模型图内，不由应用另做。 -`fn_delta_autocal_uv/1` 只允许 `outputs=["fn"]`，不能凭 FN 特征宣称三轴已标定。 -需要图像、完整光流或历史序列时新增输入 profile，明确 H/W、通道、掩码、单位和时间窗口； -没有对应特征的处理器拒绝加载，不能用零张量凑齐输入。 - -`artifacts` 只在有外部文件时必填；路径相对所选 JSON 的父目录，禁止绝对路径、URL、 -`..` 或链接逃逸。校验 SHA-256、角色和文件完整性后才初始化受支持的后端。 -不加载 pickle、任意 Python 或模型自带扩展库；缺推断依赖须明确报错，不自动安装。 -文件与所有附件共同构成内容指纹，不能只哈希 JSON。 -附件间依赖也须逐项声明和校验，未声明的外部权重不可加载。 +场加权 FN 模型 `field_gain/1`（输入 profile `hhd_normal_l1_field/1`）的字段、映射、空载基线与诊断契约见[场加权模型契约](field-gain-model.zh.md)。ML 扩展（首个示例 `onnx/1`）仍是预留示例，见[机器学习模型契约](ml-model-contract.zh.md)；后端实现和输入 profile 验证通过前必须拒绝加载。 ## 每帧推断与有效性 @@ -240,12 +203,14 @@ schema 3 按声明轴校验，允许 FN-only；旧 SDK 遇 schema 3 必须拒绝 |---|---| | `calibration.py` | schema 分流、身份和附件校验；旧曲线适配器 | | `sensors/calibration.py` | 统一选择器与指纹；支持 SN 请求 | -| 新 `calibration_models.py` | 受控类型表、空间增益与模型生命周期；ML 后端另行实现 | +| 新 `calibration_models.py` | 受控类型表、空间/全局/场加权模型与生命周期；ML 后端另行实现 | | `sensor.py`、`finger/runtime.py`、处理器适配层 | 提供同帧位置/零点，缓存逐轴结果和状态 | | Sensor 与 `ForceReadout` | 复用读数滤波器；schema 3 在推断后逐轴滤波，旧路径不变 | | 离线导出器 | 实验模型到 schema 3；SDK 不依赖实验目录 | 当前导出入口为 workspace 的 `scripts/finger_multical/export_sdk.py`，输出已有文件时拒绝覆盖。 +场加权实验文件 `data/dd02-gl/force_calibration.v3.field.json` 由同目录 `export_field.py` +从上述交付只读转换（不重训、不改排除点、不覆盖原件），契约见[场加权模型契约](field-gain-model.zh.md)。 `data/dd02-gl/force_calibration.v3.json` 与同名 `.coordinates.npz` 是本次实验交付， `identity.sn="dd02-gl"` 为实验标识，未从硬件读取；原三边缘点与非单调点警告保留于 provenance。 它不是生产包默认文件，需显式 `cal_path` 选择。 @@ -253,4 +218,6 @@ schema 3 按声明轴校验，允许 FN-only；旧 SDK 遇 schema 3 必须拒绝 实现验收必须覆盖：旧结果与单位不变；显式路径/SN/默认/关闭选择；错误 SN 不串用； 3/14/更多点、退化网格与凸包外；节点和插值预测；零点不重复扣除；缺坐标/无接触； FN-only 的 FX/FY 不冒充 N；未知模型、附件损坏和依赖缺失；状态模型每帧只调用一次及 reset。 +场加权另需常增益等价、两区域贡献、边界投影、映射错误与 30 帧加权基线生命周期， +见 `tests/test_field_calibration.py`。 另以独立录像验证物理准确度，契约或单元测试通过不代表标定合格。 diff --git a/docs/field-gain-model.zh.md b/docs/field-gain-model.zh.md new file mode 100644 index 0000000..7e0539a --- /dev/null +++ b/docs/field-gain-model.zh.md @@ -0,0 +1,102 @@ +# 场加权力标定模型契约 + +用途：reference。范围：SDK schema 3 `field_gain/1` 的输入、映射、空载基线、诊断与兼容边界。 +[统一力标定模型契约](calibration-model-contract.zh.md)负责文件选择、身份与逐帧生命周期； +本文只补充场加权模型特有语义。状态：2026-10-19 已实现并有回归测试，离线录像对照见 +[实施计划](../../docs/workflow/field-weighted-calibration-plan.zh.md)。 + +## 定位与边界 + +从现有多点增益生成空间图，对 HHD 无旋向量场做加权求和，作为 FN 的实验估计。 +不依赖深度、接触中心、最大光流点或单接触判定；实时入口尚未接入，必须显式拒绝。 +现有增益对应各加载位置的整场响应，不是已验证的局部系数。交付证明软件契约与可复现估计， +不宣称局部压力、多接触力可加性或精度改善。负的基线差保留；显示阈值只由显示层决定。 + +## 文件字段 + +`model.type` 为 `field_gain/1`，`input_profile` 为 `hhd_normal_l1_field/1`，`outputs` 仅 `["fn"]`。 +`pipeline` 沿用 schema 3 的测量身份，并增加闭合的 `pipeline.field`（由 SDK +`FIELD_IDENTITY` 生成）：`grid`、`shape`、`mask_shape`、`vector`、`reduction`、`dtype`、 +`unit`、`mask`、`lifecycle`、`area_compensation`。它们进入特征签名；改变任一值即需升级 profile。 +`params` 必需键： + +| 键 | 规则 | +|---|---| +| `points_uv`、`gains_N_per_raw`、`triangles` | 沿用空间增益的完整凸包三角网校验；点数由数组决定 | +| `valid_force_range_N` | 必填 `[low, high]`；超范围保留数值并标 `out_of_range` | +| `mapping` | 固定 `{"inside": "barycentric", "outside": "boundary_projection", "tie_break": "boundary_edge_order"}` | +| `baseline` | 固定 `{"frames": 30, "statistic": "median"}`，仅实现这一策略 | + +`artifacts` 必须有且仅有一个 `role="coordinate_map"` 附件；`map_sha256` 必须等于附件字节摘要。 +附件提供输出分辨率 `uv_by_output_px` 与 `valid_mask`，SDK 打开模型时校验形状与有限性。 + +## 增益图与加权求和 + +先把支撑点增益映射到 HHD 向量场的计算网格，再逐帧归约；缓存身份包含模型指纹、坐标附件、 +HHD 网格形状与物理有效掩膜，任一变化都重建。映射规则： + +- 三角网内按重心插值取增益；查询落在共享边时按文件中的三角顺序取第一个命中的三角形。 +- 凸包外投影到最近边界线段，再按投影参数插值两端增益；距离相同时取边界边排序中靠前者。 +- 输出网格内只在物理有效像素上统计；有效掩膜内缺坐标或增益是模型映射错误，报 + `coordinate_unmapped`，不当作零或域外增益。 +- 按块面积降采样到 HHD 网格：增益取覆盖有效像素的均值；`coverage` 是覆盖面积比， + `inside` 是凸包内面积比；混合覆盖格点保留比例，不平均无效坐标。 + +逐帧对同一帧的 HHD 无旋向量场 `d` 计算 +`P = A * sum_{HHD 激活格点}(gain * (|d_x| + |d_y|))`。HHD raw 汇总正是 +`A * sum_{激活格点}(|d_x|+|d_y|)`，所以常增益必须精确等于 `gain * raw`；`coverage` 只用于 +外推比例与诊断，不得再乘进 `P`，否则掩膜边缘的部分覆盖格点会被重复折扣。 + +HHD 激活但 `coverage=0`（没有任何有效像素映射到该格点）是模型映射错误，报 +`coordinate_unmapped`，不得静默记零。`A` 取自 `hhd/api.py` 现有的 +参考网格面积补偿（参考网格像素数/当前网格像素数），调用方不得再乘一次或重复归一化。 +不得改变 FN raw、HHD 分解或剪切分量。空场时 `P=0` 且外推比例为 0； +负的 `FN = P - B0` 保留为有效数值（可能超量程）。 + +## 空载基线与生命周期 + +用户先声明无载窗口，SDK 才对后续帧累计加权 `P`： + +- `Sensor.declare_unloaded_reference(reference_id=...)`（`FingerRuntime` 同名转发）开始窗口； + `reference_id` 必填并记录。窗口要求连续 30 个有效帧；任一无效输入或映射错误会中断连续计数， + 向前 30 帧零点未就绪。 +- `B0` 为这 30 个 `P` 的中位数；就绪前的帧状态为 `zero_unavailable`，不输出牛顿值。 + 就绪后 `FN = P - B0`。 +- 空载确认不以接触定位为门槛，但要求输入有限；换模型、重置、倒播或新会话使基线失效。 +- 场模型拒绝旧标量零点：`Sensor.set_calibration_reference()` 对 `hhd_normal_l1_field/1` + 抛 `CalibrationError`，调用方必须使用加权窗口接口，不能把旧 FN raw 零点当作 `B0`。 + +## 结果诊断 + +`CalibrationResult.diagnostics` 逐帧携带版本化诊断；旧结果或缺字段时按未知处理，不得补零： + +| 键 | 含义 | +|---|---| +| `model_fingerprint` | 模型文档与附件的内容指纹 | +| `mapping` | 生效的边界策略 | +| `baseline_N`、`reference_id` | 已就绪基线及其声明标识；未就绪为 `None` | +| `pending_reference_frames` | 当前连续窗口已累计的有效帧数 | +| `extrapolated_fraction`、`coverage_fraction` | 按 `q` 加权的凸包外比例与覆盖比例 | +| `area_scale`、`frame_id` | 本帧面积补偿与帧标识 | + +`extrapolated_fraction` 是估计质量，不是有效性：它不改变 `status`，也不得让上层丢弃牛顿值。 +零点缺失、输入错误仍无效；超量程沿用现有显式状态。 + +## 兼容与实现位置 + +旧 `spatial_gain/1`、`global_gain/1` 与旧文件语义不变；schema 3 按声明轴校验，旧 SDK 遇 +`field_gain/1` 必须拒绝。`hhd_downsample` 仍按既有兼容例外处理：只允许该键数值不同， +其余处理参数、几何、展开域、SN 与传感器类型仍须匹配，不能借此跳过签名校验。 + +| 位置 | 责任 | +|---|---| +| `src/orisys/field_calibration.py` | profile/策略常量、增益图构建、加权归约 | +| `src/orisys/calibration_models.py` | 类型注册、参数校验、基线状态与诊断 | +| `src/orisys/hhd/api.py` | 同帧 `get_field()`：`d` 拷贝、物理掩膜、面积补偿 | +| `src/orisys/sensor.py`、`src/orisys/finger/runtime.py` | 同帧场推断、加权零点声明、上下文绑定 | +| 工作区 `scripts/finger_multical/export_field.py` | 现有空间模型只读导出为实验文件，不重训、不改排除点 | + +回归证据：`tests/test_field_calibration.py` 覆盖常增益等价、两区域贡献、重叠向量先合成再 L1、 +边界投影、掩膜/NaN/空场、映射错误与 30 帧基线生命周期；`tests/test_model_runtime.py` 覆盖 +HHD 同帧场接线、加权零点声明与无接触推断。命令与 run_id 记录在实施计划的执行结果中。 +实验导出文件不覆盖原件，provenance 声明来源哈希、边界/基线策略和实验假设。 diff --git a/docs/ml-model-contract.zh.md b/docs/ml-model-contract.zh.md new file mode 100644 index 0000000..f2e4f9b --- /dev/null +++ b/docs/ml-model-contract.zh.md @@ -0,0 +1,44 @@ +# 机器学习力标定模型契约 + +用途：reference。范围：SDK schema 3 中预留的 ML 模型后端（首个示例 `onnx/1`）与输入 profile +约束。文件选择、身份、逐帧生命周期与有效性状态由[统一力标定模型契约](calibration-model-contract.zh.md) +负责。状态：仅保留契约，SDK 尚未实现 ONNX 后端；实现与验证通过前必须拒绝加载。 + +同一 SN、路径和外层字段不变，只替换 `model`。下面只预留 `onnx/1` 示例， +不表示 SDK 已支持 ONNX。 + +```json +{ + "type": "onnx/1", + "input_profile": "fn_delta_autocal_uv/1", + "output_frame": "sdk_force/1", + "outputs": ["fn"], + "params": { + "input_tensor": "features", + "output_tensor": "force_N", + "feature_order": ["delta_raw", "u", "v"], + "dtype": "float32", + "shape": [1, 3], + "valid_force_range_N": [0.0, 3.0] + }, + "artifacts": [ + { + "role": "weights", + "path": "DD02-GL-001/model.onnx", + "sha256": "0000000000000000000000000000000000000000000000000000000000000000" + } + ] +} +``` + +此标量 profile 的规范输入即上述顺序、类型、形状；输出 float32 `[1, len(outputs)]`， +轴序严格按 `outputs`。归一化及单位转换封装在已验证的模型图内，不由应用另做。 +`fn_delta_autocal_uv/1` 只允许 `outputs=["fn"]`，不能凭 FN 特征宣称三轴已标定。 +需要图像、完整光流或历史序列时新增输入 profile，明确 H/W、通道、掩码、单位和时间窗口； +没有对应特征的处理器拒绝加载，不能用零张量凑齐输入。 + +`artifacts` 只在有外部文件时必填；路径相对所选 JSON 的父目录，禁止绝对路径、URL、 +`..` 或链接逃逸。校验 SHA-256、角色和文件完整性后才初始化受支持的后端。 +不加载 pickle、任意 Python 或模型自带扩展库；缺推断依赖须明确报错，不自动安装。 +文件与所有附件共同构成内容指纹，不能只哈希 JSON。 +附件间依赖也须逐项声明和校验，未声明的外部权重不可加载。 diff --git a/src/orisys/calibration_context.py b/src/orisys/calibration_context.py index 0534fb4..ae4e089 100644 --- a/src/orisys/calibration_context.py +++ b/src/orisys/calibration_context.py @@ -53,8 +53,13 @@ def domain_fingerprint(domain) -> str: def build_context(config, filters=None, *, geometry_sha256=None, source_scale=None, coordinates=None, output_domain_sha256=None, processing_backend=None, - preprocessing=None) -> dict: - """Build the closed v1 feature identity from effective values, excluding readout filters.""" + preprocessing=None, field=None) -> dict: + """Build the closed v1 feature identity from effective values, excluding readout filters. + + ``field`` carries the closed profile identity for field models (shape, dtype, + unit, mask and lifecycle). It is omitted for scalar profiles so their + signatures stay byte-identical to previously exported files. + """ camera = config.get("camera", {}) tracker = dict(config.get("tracker", {})) raw_downsample = tracker.get("hhd_downsample", 1) @@ -94,5 +99,8 @@ def build_context(config, filters=None, *, geometry_sha256=None, source_scale=No "source_scale": source_scale, "preprocessing": preprocessing or {}, } + if field is not None: + result["field"] = deepcopy(field) + effective["field"] = deepcopy(field) 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 472aff5..d8fb6c6 100644 --- a/src/orisys/calibration_models.py +++ b/src/orisys/calibration_models.py @@ -7,13 +7,32 @@ import json import math from collections.abc import Mapping from copy import deepcopy -from dataclasses import dataclass +from dataclasses import dataclass, field as dataclass_field from pathlib import Path, PurePosixPath, PureWindowsPath from typing import Any import numpy as np from scipy.spatial import ConvexHull, QhullError +from .field_calibration import ( + BASELINE_POLICY, + FIELD_IDENTITY, + FIELD_PROFILE, + MAPPING_POLICY, + ZERO_FRAME_COUNT, + FieldGainMap, + build_field_gain_map, + coerce_field, + mask_sha256, + reduce_field, +) + +_PROFILE_FOR_TYPE = { + "spatial_gain/1": "fn_delta_autocal_uv/1", + "global_gain/1": "force_delta/1", + "field_gain/1": FIELD_PROFILE, +} + def _require(condition: bool, message: str) -> None: if not condition: @@ -98,11 +117,15 @@ def _mesh(params: dict[str, Any]) -> tuple[np.ndarray, np.ndarray, np.ndarray]: return vertices, np.asarray(gains, dtype=float), np.asarray(triangles, dtype=int) -def _pipeline(data: Any, spatial: bool) -> None: +def _pipeline(data: Any, profile: str) -> None: + spatial = profile != "force_delta/1" required = {"stitch_resolution", "dis_mode", "hhd_downsample", "filters", "feature_signature_sha256"} - _keys(data, required | ({"geometry_sha256", "coordinates", "source_scale"} if spatial else set()), - {"geometry_sha256", "coordinates", "output_domain_sha256", "source_scale"}) + if spatial: + required |= {"geometry_sha256", "coordinates", "source_scale"} + if profile == FIELD_PROFILE: + required.add("field") + _keys(data, required, {"geometry_sha256", "coordinates", "output_domain_sha256", "source_scale"}) size = data["stitch_resolution"] _require(isinstance(size, list) and len(size) == 2 and all(type(n) is int and n > 0 for n in size), "invalid stitch_resolution") @@ -127,12 +150,17 @@ def _pipeline(data: Any, spatial: bool) -> None: _require(isinstance(coord["origin_xy"], list) and len(coord["origin_xy"]) == 2 and all(_number(x) for x in coord["origin_xy"]), "invalid coordinate origin") _require(_sha(coord["map_sha256"]), "invalid map SHA-256") + if "field" in required: + _require(data["field"] == FIELD_IDENTITY, "unsupported field identity") @dataclass(frozen=True) class CalibrationResult: values_N: dict[str, float | None] status: dict[str, str] + #: Versioned extra evidence (field model: baseline, extrapolation fraction). + #: Consumers must tolerate missing keys so old cached results stay readable. + diagnostics: dict[str, Any] = dataclass_field(default_factory=dict) class ModelCalibration: @@ -154,16 +182,25 @@ class ModelCalibration: self._params = self._document["model"]["params"] self.coordinate_map = self.coordinate_valid_mask = None self._artifact_hashes: list[str] = [] - if self.model_type == "spatial_gain/1": + self._points = self._gains = self._triangles = None + self._field_maps: dict[tuple[tuple[int, int], str], FieldGainMap] = {} + self._field_reference_id: str | None = None + self._field_zero_values: list[float] = [] + self._field_baseline: float | None = None + if self.model_type in ("spatial_gain/1", "field_gain/1"): self._points, self._gains, self._triangles = _mesh(self._params) self._load_artifacts(document["model"].get("artifacts", [])) + if self.model_type == "field_gain/1": + _require(self.coordinate_map is not None, + "field_gain/1 requires a coordinate_map artifact") def _load_artifacts(self, artifacts: Any) -> None: _require(isinstance(artifacts, list), "artifacts must be an array") _require(len(artifacts) <= 1, "only one coordinate_map artifact is supported") for artifact in artifacts: _keys(artifact, {"role", "path", "sha256"}) - _require(self.model_type == "spatial_gain/1" and artifact["role"] == "coordinate_map", + _require(self.model_type in ("spatial_gain/1", "field_gain/1") + and artifact["role"] == "coordinate_map", "unsupported artifact role") name = artifact["path"] _require(isinstance(name, str) and bool(name) and ":" not in name and @@ -229,7 +266,108 @@ class ModelCalibration: _require(pipeline.get(key) == value, f"calibration pipeline mismatch: {key}") def reset(self, reference_id: Any = None) -> None: - """Gain backends have no history; preserve the lifecycle for future models.""" + """Clear per-stream state; the field model also voids its weighted baseline.""" + del reference_id + self._field_reference_id = None + self._field_zero_values = [] + self._field_baseline = None + + @property + def field_baseline_N(self) -> float | None: + """Current weighted unloaded baseline, or ``None`` while it is unavailable.""" + return self._field_baseline + + def begin_reference(self, reference_id: Any) -> None: + """Declare a new unloaded window; the next 30 valid frames set the baseline.""" + _require(self.model_type == "field_gain/1", + "a weighted reference requires field_gain/1") + _require(isinstance(reference_id, str) and bool(reference_id.strip()), + "reference_id must be a non-empty string") + self._field_reference_id = reference_id + self._field_zero_values = [] + self._field_baseline = None + + def field_gain_map(self, grid_shape: tuple[int, int], + physical_mask: np.ndarray) -> FieldGainMap: + """Gain/coverage map for one HHD grid and one output-resolution mask.""" + _require(self.model_type == "field_gain/1", + "field gain maps require field_gain/1") + mask = np.asarray(physical_mask, dtype=bool) + _require(mask.ndim == 2 and mask.shape == self.coordinate_map.shape[:2], + "physical mask must match the coordinate map output grid") + return self._field_map((int(grid_shape[0]), int(grid_shape[1])), mask) + + def _field_map(self, grid_shape: tuple[int, int], physical_mask: np.ndarray) -> FieldGainMap: + key = (grid_shape, mask_sha256(physical_mask)) + cached = self._field_maps.get(key) + if cached is not None: + return cached + gain_map = build_field_gain_map( + self._points, self._gains, self._triangles, + self.coordinate_map, self.coordinate_valid_mask, + grid_shape, physical_mask, + ) + if len(self._field_maps) >= 4: + self._field_maps.pop(next(iter(self._field_maps))) + self._field_maps[key] = gain_map + return gain_map + + def _break_field_window(self) -> None: + """An invalid frame breaks the consecutive declared unloaded window.""" + if self._field_baseline is None: + self._field_zero_values = [] + + def _predict_field(self, frame_id: int, features: Mapping[str, Any]) -> CalibrationResult: + def invalid(reason: str, **diagnostics: Any) -> CalibrationResult: + diagnostics.setdefault("frame_id", frame_id) + return CalibrationResult({axis: None for axis in self.outputs}, + {axis: reason for axis in self.outputs}, diagnostics) + + try: + field = coerce_field(features.get("field")) + except (TypeError, ValueError) as exc: + self._break_field_window() + return invalid("invalid_input", error=str(exc)) + mask = features.get("output_valid_mask") + try: + mask = np.asarray(mask, dtype=bool) + _require(mask.ndim == 2 and mask.shape == self.coordinate_map.shape[:2], + "output_valid_mask must match the coordinate map output grid") + except (TypeError, ValueError) as exc: + self._break_field_window() + return invalid("invalid_input", error=str(exc)) + try: + gain_map = self._field_map(field.grid_shape, mask) + reduction = reduce_field(field, gain_map) + except ValueError as exc: + self._break_field_window() + return invalid("coordinate_unmapped", error=str(exc)) + diagnostics: dict[str, Any] = { + "model_fingerprint": self.fingerprint, + "mapping": dict(MAPPING_POLICY), + "extrapolated_fraction": reduction.extrapolated_fraction, + "coverage_fraction": reduction.coverage_fraction, + "area_scale": field.area_scale, + } + if self._field_baseline is None: + diagnostics["reference_id"] = self._field_reference_id + if self._field_reference_id is None: + diagnostics["pending_reference_frames"] = 0 + return invalid("zero_unavailable", **diagnostics) + self._field_zero_values.append(reduction.weighted_N) + if len(self._field_zero_values) < ZERO_FRAME_COUNT: + diagnostics["pending_reference_frames"] = len(self._field_zero_values) + return invalid("zero_unavailable", **diagnostics) + self._field_baseline = float(np.median(self._field_zero_values)) + self._field_zero_values = [] + diagnostics["baseline_N"] = self._field_baseline + diagnostics["pending_reference_frames"] = 0 + return invalid("zero_unavailable", **diagnostics) + diagnostics["baseline_N"] = self._field_baseline + value = reduction.weighted_N - self._field_baseline + low, high = self._params["valid_force_range_N"] + status = "ok" if low <= value <= high else "out_of_range" + return CalibrationResult({"fn": float(value)}, {"fn": status}, diagnostics) def convert(self, fn: float, fx: float, fy: float) -> None: """Reject legacy scalar calls which lack frame, position and zero features.""" @@ -245,6 +383,8 @@ class ModelCalibration: if (not isinstance(features, Mapping) or type(frame_id) is not int or (timestamp_s is not None and not _number(timestamp_s))): return invalid("invalid_input") + if self.model_type == "field_gain/1": + return self._predict_field(frame_id, features) if features.get("zero_valid") is not True: return invalid("zero_unavailable") if features.get("contact_found") is False: @@ -296,23 +436,43 @@ def parse_model(data: dict[str, Any], source_path: str | Path | None = None) -> _require(isinstance(data.get("provenance", {}), dict), "provenance must be an object") model = data["model"] _keys(model, {"type", "input_profile", "output_frame", "outputs", "params"}, {"artifacts"}) - _require(model["type"] in ("spatial_gain/1", "global_gain/1"), f"unsupported calibration model: {model['type']}") - spatial = model["type"] == "spatial_gain/1" - _require(model["input_profile"] == ("fn_delta_autocal_uv/1" if spatial else "force_delta/1"), + _require(model["type"] in _PROFILE_FOR_TYPE, + f"unsupported calibration model: {model['type']}") + model_type = model["type"] + spatial = model_type == "spatial_gain/1" + field_weighted = model_type == "field_gain/1" + _require(model["input_profile"] == _PROFILE_FOR_TYPE[model_type], "unsupported input_profile for model") _require(model["output_frame"] == "sdk_force/1", "unsupported output_frame") outputs = model["outputs"] _require(isinstance(outputs, list) and bool(outputs) and all(a in ("fn", "fx", "fy") for a in outputs) and len(set(outputs)) == len(outputs), "invalid model outputs") - _require(not spatial or outputs == ["fn"], "spatial model supports fn only") - _pipeline(data["pipeline"], spatial) + _require(not (spatial or field_weighted) or outputs == ["fn"], "spatial model supports fn only") + _pipeline(data["pipeline"], model["input_profile"]) params = model["params"] - _keys(params, {"gains_N_per_raw", "valid_force_range_N"} | ({"points_uv", "triangles"} if spatial else set())) + required_params = {"gains_N_per_raw", "valid_force_range_N"} + if spatial or field_weighted: + required_params |= {"points_uv", "triangles"} + if field_weighted: + required_params |= {"mapping", "baseline"} + _keys(params, required_params) force_range = params["valid_force_range_N"] _require(isinstance(force_range, list) and len(force_range) == 2 and all(_number(x) for x in force_range) and force_range[0] < force_range[1], "invalid force range") if not spatial: gains = params["gains_N_per_raw"] - _keys(gains, set(outputs)) - _require(all(_number(g) and g > 0 for g in gains.values()), "gains must be finite positive numbers") + if field_weighted: + _require(isinstance(gains, list) and len(gains) == len(params["points_uv"]), + "field gains must have one value per support point") + else: + _keys(gains, set(outputs)) + _require(all(_number(g) and g > 0 for g in gains.values()) if isinstance(gains, dict) + else all(_number(g) and g > 0 for g in gains), + "gains must be finite positive numbers") + if field_weighted: + _require(params["mapping"] == MAPPING_POLICY, "unsupported field mapping policy") + _require(params["baseline"] == BASELINE_POLICY, "unsupported field baseline policy") + artifacts = model.get("artifacts") + _require(isinstance(artifacts, list) and len(artifacts) == 1, + "field_gain/1 requires one coordinate_map artifact") return ModelCalibration(data, source_path) diff --git a/src/orisys/field_calibration.py b/src/orisys/field_calibration.py new file mode 100644 index 0000000..14d54ff --- /dev/null +++ b/src/orisys/field_calibration.py @@ -0,0 +1,347 @@ +"""Field-weighted force estimation from a validated support mesh and HHD grid. + +The SDK builds a gain map on the finger output grid, downsamples it by valid +area onto the HHD computation grid, and reduces the curl-free HHD component +with ``q = |d_x| + |d_y|``. The map is a deterministic function of the model +mesh, the coordinate artifact and the physical valid mask, so callers cache it +by those identities instead of rebuilding it every frame. +""" + +from __future__ import annotations + +import hashlib +from collections import Counter +from dataclasses import dataclass +from typing import Any, Mapping + +import numpy as np + +#: Closed identity of the field input profile. It is part of the exported and +#: runtime feature signature, so changing any value is a schema/profile change. +FIELD_PROFILE = "hhd_normal_l1_field/1" +FIELD_IDENTITY: dict[str, Any] = { + "grid": "hhd_native", + "shape": [None, None, 2], + "mask_shape": [None, None], + "vector": "irrotational", + "reduction": "l1", + "dtype": "float32", + "unit": "raw", + "mask": "physical_valid", + "lifecycle": "same_frame", + "area_compensation": "reference_grid", +} +#: How a gain is read inside and outside the support hull. Deterministic on +#: purpose: equal-distance points choose the lowest sorted boundary edge. +MAPPING_POLICY: dict[str, Any] = { + "inside": "barycentric", + "outside": "boundary_projection", + "tie_break": "boundary_edge_order", +} +#: Session baseline reduction; the caller declares the unloaded window. +BASELINE_POLICY: dict[str, Any] = {"frames": 30, "statistic": "median"} +ZERO_FRAME_COUNT = int(BASELINE_POLICY["frames"]) + +_INSIDE_TOLERANCE = 1e-10 +_GAIN_TOLERANCE = 1e-9 + + +def _require(condition: bool, message: str) -> None: + if not condition: + raise ValueError(message) + + +@dataclass(frozen=True) +class FieldFrame: + """One frame of the HHD curl-free field on the native computation grid.""" + + d: np.ndarray + valid_mask: np.ndarray + area_scale: float + grid_shape: tuple[int, int] + reference_shape: tuple[int, int] | None = None + + +@dataclass(frozen=True) +class FieldGainMap: + """Area-averaged gains on the HHD grid, with coverage as an area fraction. + + ``coverage`` is the fraction of each cell's pixels that carry a mapped + coordinate and a gain; ``inside`` is the fraction whose gain came from + barycentric interpolation. ``coverage - inside`` is the boundary-projected + (extrapolated) fraction used for the per-frame quality diagnostic. + """ + + grid_shape: tuple[int, int] + gain: np.ndarray + coverage: np.ndarray + inside: np.ndarray + physical_mask_sha256: str + + def diagnostic(self) -> dict[str, float]: + return { + "coverage": float(self.coverage.mean()), + "inside": float(self.inside.mean()), + "extrapolated": float(np.clip(self.coverage - self.inside, 0.0, 1.0).mean()), + } + + +@dataclass(frozen=True) +class FieldReduction: + """Weighted field sum and its per-frame extrapolation diagnostic.""" + + weighted_N: float + extrapolated_fraction: float + coverage_fraction: float + + +def coerce_field(value: Any) -> FieldFrame: + """Validate the same-frame field payload published by ``hhd/api.py``.""" + _require(isinstance(value, Mapping), "field input must be a mapping") + d = value.get("d") + mask = value.get("valid_mask") + area_scale = value.get("area_scale") + grid_shape = value.get("grid_shape") + reference_shape = value.get("reference_shape") + _require(isinstance(d, np.ndarray) and d.ndim == 3 and d.shape[2] == 2, + "field d must be an (H, W, 2) array") + _require(d.dtype == np.float32, "field d dtype must be float32") + _require(isinstance(mask, np.ndarray) and mask.dtype == np.bool_ and mask.shape == d.shape[:2], + "field valid_mask must be an (H, W) bool array") + _require(isinstance(area_scale, (int, float)) and not isinstance(area_scale, bool) + and np.isfinite(area_scale) and area_scale > 0, + "field area_scale must be finite and positive") + _require(isinstance(grid_shape, tuple) and len(grid_shape) == 2 + and all(type(size) is int and size > 0 for size in grid_shape), + "field grid_shape must be two positive integers") + _require(tuple(grid_shape) == tuple(d.shape[:2]), "field grid_shape does not match d") + if reference_shape is not None: + _require(isinstance(reference_shape, tuple) and len(reference_shape) == 2 + and all(type(size) is int and size > 0 for size in reference_shape), + "field reference_shape must be two positive integers") + if not np.isfinite(d[mask]).all(): + raise ValueError("field contains non-finite active vectors") + return FieldFrame( + d=d, + valid_mask=mask, + area_scale=float(area_scale), + grid_shape=(int(grid_shape[0]), int(grid_shape[1])), + reference_shape=None if reference_shape is None else (int(reference_shape[0]), + int(reference_shape[1])), + ) + + +def mask_sha256(mask: np.ndarray) -> str: + """Stable identity of a boolean mask for cache keys.""" + packed = np.packbits(np.ascontiguousarray(mask, dtype=bool)) + return hashlib.sha256(packed.tobytes()).hexdigest() + + +def _boundary_edges(triangles: np.ndarray) -> list[tuple[int, int]]: + """Hull edges of a validated complete triangulation, in a fixed order.""" + counts: Counter[tuple[int, int]] = Counter() + for triangle in triangles: + indices = [int(index) for index in triangle] + for start, end in ((0, 1), (1, 2), (2, 0)): + edge = tuple(sorted((indices[start], indices[end]))) + counts[edge] += 1 + boundary = [edge for edge, count in counts.items() if count == 1] + boundary.sort() + _require(bool(boundary), "support mesh has no boundary edges") + return boundary + + +def _inside_gain(points: np.ndarray, gains: np.ndarray, triangles: np.ndarray, + uv: np.ndarray, pending: np.ndarray) -> tuple[np.ndarray, np.ndarray]: + """Barycentric gain for pending pixels, first matching triangle in file order.""" + gain = np.full(len(uv), np.nan) + inside = np.zeros(len(uv), dtype=bool) + for triangle in triangles: + candidate = ~inside & pending + if not candidate.any(): + break + p0, p1, p2 = (points[int(index)] for index in triangle) + query = uv[candidate] + v0, v1 = p1 - p0, p2 - p0 + det = float(v0[0] * v1[1] - v1[0] * v0[1]) + _require(abs(det) > 0, "support mesh has a degenerate triangle") + offset = query - p0 + w1 = (offset[:, 0] * v1[1] - offset[:, 1] * v1[0]) / det + w2 = (v0[0] * offset[:, 1] - v0[1] * offset[:, 0]) / det + weights = np.stack((1.0 - w1 - w2, w1, w2), axis=1) + hit = weights.min(axis=1) >= -_INSIDE_TOLERANCE + if not hit.any(): + continue + hit_indices = np.flatnonzero(candidate)[hit] + gain[hit_indices] = ( + weights[hit, 0] * gains[int(triangle[0])] + + weights[hit, 1] * gains[int(triangle[1])] + + weights[hit, 2] * gains[int(triangle[2])] + ) + inside[hit_indices] = True + return gain, inside + + +def _boundary_gain(points: np.ndarray, gains: np.ndarray, triangles: np.ndarray, + uv: np.ndarray, pending: np.ndarray) -> np.ndarray: + """Project pending points onto the nearest hull edge and interpolate its gains. + + Equal distances keep the first edge in sorted ``(index, index)`` order, so + the result does not depend on triangle storage order. The projection + parameter is clamped to the segment; gains stay inside the support range. + """ + gain = np.full(len(uv), np.nan) + indices = np.flatnonzero(pending) + if indices.size == 0: + return gain + query = uv[indices] + best_distance = np.full(len(indices), np.inf) + for start, end in _boundary_edges(triangles): + first, second = points[start], points[end] + direction = second - first + denominator = float(np.dot(direction, direction)) + _require(denominator > 0, "support mesh has a zero-length boundary edge") + parameter = np.clip(((query - first) @ direction) / denominator, 0.0, 1.0) + projected = first + parameter[:, None] * direction + distance = np.sum((query - projected) ** 2, axis=1) + better = distance < best_distance - 1e-12 + best_distance[better] = distance[better] + gain[indices[better]] = gains[start] + parameter[better] * (gains[end] - gains[start]) + _require(np.isfinite(gain[indices]).all(), "support mesh boundary projection failed") + return gain + + +def _block_starts(divisions: int, size: int) -> tuple[np.ndarray, np.ndarray]: + starts = np.array([index * size // divisions for index in range(divisions)], dtype=np.intp) + ends = np.array([(index + 1) * size // divisions for index in range(divisions)], dtype=np.intp) + return starts, ends + + +def build_field_gain_map(points: np.ndarray, gains: np.ndarray, triangles: np.ndarray, + coordinate_map: np.ndarray, coordinate_valid_mask: np.ndarray, + grid_shape: tuple[int, int], + physical_mask: np.ndarray) -> FieldGainMap: + """Build the gain/coverage map for one output mask and one HHD grid shape. + + Raises :class:`ValueError` when the physical mask reaches pixels that the + model's coordinate artifact does not map; that is a model mapping error and + must not silently become a zero or an extrapolated gain. + """ + _require(isinstance(points, np.ndarray) and points.ndim == 2 and points.shape[1] == 2, + "support points must be an (N, 2) array") + _require(isinstance(gains, np.ndarray) and gains.shape == (len(points),), + "support gains must have one value per point") + _require(isinstance(triangles, np.ndarray) and triangles.ndim == 2 and triangles.shape[1] == 3, + "support triangles must be an (T, 3) array") + _require(isinstance(coordinate_map, np.ndarray) and coordinate_map.ndim == 3 + and coordinate_map.shape[2] == 2, "coordinate map must be an (H, W, 2) array") + _require(isinstance(coordinate_valid_mask, np.ndarray) + and coordinate_valid_mask.dtype == np.bool_ + and coordinate_valid_mask.shape == coordinate_map.shape[:2], + "coordinate valid mask must be an (H, W) bool array") + output_shape = tuple(int(size) for size in coordinate_map.shape[:2]) + _require(isinstance(grid_shape, tuple) and len(grid_shape) == 2 + and all(type(size) is int and size > 0 for size in grid_shape), + "grid shape must be two positive integers") + _require(grid_shape[0] <= output_shape[0] and grid_shape[1] <= output_shape[1], + "HHD grid cannot exceed the output grid") + _require(isinstance(physical_mask, np.ndarray) and physical_mask.dtype == np.bool_ + and physical_mask.shape == output_shape, + "physical mask must match the output grid") + if np.any(physical_mask & ~coordinate_valid_mask): + raise ValueError("model mapping error: physical mask has pixels without mapped coordinates") + + uv = np.asarray(coordinate_map[physical_mask], dtype=np.float64).reshape(-1, 2) + _require(np.isfinite(uv).all(), "model mapping error: mapped coordinates are not finite") + pixels = int(uv.shape[0]) + gain_flat = np.full(pixels, np.nan) + inside_flat = np.zeros(pixels, dtype=bool) + pending = np.ones(pixels, dtype=bool) + if pixels: + inside_gain, inside_flat = _inside_gain(points, gains, triangles, uv, pending) + gain_flat[inside_flat] = inside_gain[inside_flat] + pending = ~inside_flat + if pending.any(): + boundary_gain = _boundary_gain(points, gains, triangles, uv, pending) + gain_flat[pending] = boundary_gain[pending] + low, high = float(gains.min()), float(gains.max()) + _require(np.isfinite(gain_flat).all(), "gain map contains unmapped valid pixels") + _require(np.all(gain_flat >= low - _GAIN_TOLERANCE) + and np.all(gain_flat <= high + _GAIN_TOLERANCE), + "interpolated gain outside the support range") + + valid_flat = np.ones(pixels, dtype=np.float64) + inside_values = inside_flat.astype(np.float64) + gain_values = gain_flat + + def _reduce(values: np.ndarray) -> np.ndarray: + grid = np.zeros(output_shape, dtype=np.float64) + grid[physical_mask] = values + rows, _ = _block_starts(grid_shape[0], output_shape[0]) + row_blocks = np.add.reduceat(grid, rows, axis=0) + cols, _ = _block_starts(grid_shape[1], output_shape[1]) + return np.add.reduceat(row_blocks, cols, axis=1) + + area = _reduce(valid_flat) + gain_sum = _reduce(gain_values) + inside_area = _reduce(inside_values) + row_starts, row_ends = _block_starts(grid_shape[0], output_shape[0]) + col_starts, col_ends = _block_starts(grid_shape[1], output_shape[1]) + block_area = np.outer(row_ends - row_starts, col_ends - col_starts).astype(np.float64) + _require(np.all(block_area > 0), "HHD grid cells cannot be empty") + coverage = area / block_area + inside = inside_area / block_area + gain = np.divide(gain_sum, area, out=np.zeros_like(gain_sum), where=area > 0) + _require(np.isfinite(gain).all() and np.isfinite(coverage).all(), + "field gain map reduction failed") + coverage.setflags(write=False) + inside.setflags(write=False) + gain.setflags(write=False) + return FieldGainMap( + grid_shape=(int(grid_shape[0]), int(grid_shape[1])), + gain=gain.astype(np.float64), + coverage=coverage, + inside=inside, + physical_mask_sha256=mask_sha256(physical_mask), + ) + + +def reduce_field(field: FieldFrame, gain_map: FieldGainMap) -> FieldReduction: + """Apply ``P = A * sum(gain * (|d_x| + |d_y|))`` over active HHD cells. + + The HHD raw aggregate is ``A * sum(active * q)``; multiplying the same sum by + a constant gain must therefore give exactly ``gain * raw``. ``coverage`` is + diagnostic-only here: folding it into ``P`` would discount a partially covered + mask-edge cell that the raw aggregate still counts in full. An active cell with + no mapped output pixel has no valid gain, so it is reported as a model mapping + error instead of silently contributing zero. + + ``area_scale`` comes from ``hhd/api.py``'s existing reference-grid + compensation, so the caller must not apply another area factor. + """ + _require(field.grid_shape == gain_map.grid_shape, + "field grid shape does not match the gain map; rebuild the cache") + d = np.asarray(field.d, dtype=np.float64) + active = np.asarray(field.valid_mask, dtype=bool) + q = np.where(active, np.abs(d[:, :, 0]) + np.abs(d[:, :, 1]), 0.0) + q = np.where(np.isfinite(q), q, 0.0) + if np.any(active & (gain_map.coverage <= 0.0)): + raise ValueError("model mapping error: active HHD cells have no mapped pixels") + total = float(np.sum(gain_map.gain * q) * field.area_scale) + coverage_mass = q * gain_map.coverage + source_sum = float(np.sum(q)) + coverage_sum = float(np.sum(coverage_mass)) + if source_sum > 0.0: + extrapolated = float(np.sum(q * np.clip(gain_map.coverage - gain_map.inside, 0.0, 1.0)) + / source_sum) + extrapolated = float(np.clip(extrapolated, 0.0, 1.0)) + coverage_fraction = float(np.clip(coverage_sum / source_sum, 0.0, 1.0)) + else: + extrapolated = 0.0 + coverage_fraction = 0.0 + _require(np.isfinite(total), "weighted field sum is not finite") + return FieldReduction( + weighted_N=total, + extrapolated_fraction=extrapolated, + coverage_fraction=coverage_fraction, + ) diff --git a/src/orisys/finger/runtime.py b/src/orisys/finger/runtime.py index ce6fd7c..f68d398 100644 --- a/src/orisys/finger/runtime.py +++ b/src/orisys/finger/runtime.py @@ -14,6 +14,7 @@ from ..capture import ( open_capture, ) from ..configs import load_config +from ..field_calibration import FIELD_IDENTITY, FIELD_PROFILE from ..frame_processors import FrameUnwarper from ..sensor import Sensor from ..unroll import ( @@ -326,9 +327,10 @@ class FingerRuntime: model = sensor.calibration uv_map = getattr(model, "coordinate_map", None) - if model.input_profile == "fn_delta_autocal_uv/1": + if model.input_profile in ("fn_delta_autocal_uv/1", FIELD_PROFILE): if uv_map is None or uv_map.shape[:2] != (geometry.out_height, geometry.out_width): raise ValueError("calibration coordinate map does not match finger output shape") + field_identity = FIELD_IDENTITY if model.input_profile == FIELD_PROFILE else None sensor.bind_calibration_context(build_context( sensor._config, sensor.filters_state(), geometry_sha256=sha256_file(mesh_path), source_scale=geometry.source_scale, @@ -336,6 +338,7 @@ class FingerRuntime: output_domain_sha256=domain_fingerprint(domain), processing_backend="uv_unroll", preprocessing={"flip_source_vertical": flip_source_vertical, "mirror": mirror}, + field=field_identity, )) capture_config = CaptureConfig.from_camera_dict( @@ -399,19 +402,25 @@ class FingerRuntime: selected = cal_mod.validate_for_device(selected, "finger", serial) context = None if getattr(selected, "schema_version", None) == 3: - if selected.input_profile == "fn_delta_autocal_uv/1": + if selected.input_profile in ("fn_delta_autocal_uv/1", FIELD_PROFILE): uv_map = selected.coordinate_map if uv_map is None or uv_map.shape[:2] != (self.geometry.out_height, self.geometry.out_width): raise ValueError("calibration coordinate map does not match finger output shape") + field_identity = FIELD_IDENTITY if selected.input_profile == FIELD_PROFILE else None context = build_context( self.geometry.config, sensor.filters_state(), geometry_sha256=sha256_file(self.geometry.export_paths.mesh), source_scale=self.geometry.source_scale, coordinates=selected.pipeline.get("coordinates"), output_domain_sha256=domain_fingerprint(self.domain), processing_backend="uv_unroll", preprocessing={"flip_source_vertical": self.frame_processor.flip_source_vertical, - "mirror": sensor.mirror}) + "mirror": sensor.mirror}, + field=field_identity) sensor.set_calibration(selected, context=context) + def declare_unloaded_reference(self, reference_id): + """Declare the unloaded window for a field model; the SDK medians 30 weighted frames.""" + self.tactile.raw_sensor.declare_unloaded_reference(reference_id=reference_id) + @property def calibration_status(self): """Copy the current per-axis calibration validity state.""" diff --git a/src/orisys/hhd/api.py b/src/orisys/hhd/api.py index a7cf2a6..1fb9cdc 100644 --- a/src/orisys/hhd/api.py +++ b/src/orisys/hhd/api.py @@ -155,6 +155,37 @@ class nHHD: return self.d return cp.asnumpy(self._d_gpu) + def get_field(self, valid_mask=None): + """Curl-free HHD component and its reference-grid area scale for one frame. + + The SDK calibration layer owns the field *profile* identity; this module + only publishes the numerical payload. It belongs to the frame that was + just decomposed: the next ``decompose`` call overwrites the solver's + arrays, so callers copy whatever they keep. + + Returns: + dict with ``d`` (float32 HxWx2 copy), ``valid_mask`` (HxW bool), + ``area_scale`` (reference pixels / current pixels, the same factor + ``get_force`` already applies), ``grid_shape`` and ``reference_shape``. + """ + direction = getattr(self, 'd', None) + if direction is None: + raise ValueError("field requested before decompose") + shape = direction.shape[:2] + mask = self._normalize_mask(valid_mask, shape) if valid_mask is not None else None + field = numpy.array(direction, dtype=numpy.float32, copy=True) + if mask is not None: + field[~mask] = 0.0 + area_scale, _ = self._force_scale(shape) + reference = self._force_ref_shape or shape + return { + 'd': field, + 'valid_mask': numpy.ones(shape, dtype=bool) if mask is None else mask.copy(), + 'area_scale': float(area_scale), + 'grid_shape': (int(shape[0]), int(shape[1])), + 'reference_shape': (int(reference[0]), int(reference[1])), + } + def _solve_poisson_backend(self, field, verbose=False, valid_mask=None): """Dispatch Poisson solve using the selected backend. diff --git a/src/orisys/sensor.py b/src/orisys/sensor.py index 3960a30..7e1cbc8 100644 --- a/src/orisys/sensor.py +++ b/src/orisys/sensor.py @@ -17,12 +17,13 @@ import numpy as np from . import calibration as cal_mod from . import configs from .backend_availability import check_cupy_available, check_opencl_available -from .force_readout import ( - FILTER_PRESETS, - ForceReadout, - available_filter_presets, -) +# Explicit re-exports: ``calibration.py`` resolves named presets through this +# module and the preset tests import the table from here. +from .force_readout import ForceReadout +from .force_readout import FILTER_PRESETS as FILTER_PRESETS +from .force_readout import available_filter_presets as available_filter_presets from .frame_processors import create_frame_processor +from .field_calibration import FIELD_PROFILE from .get_depth import Contact3D from .hhd import nHHD @@ -790,7 +791,13 @@ class Sensor: if self._uses_calibration_model(): # Spatial inference must use this HHD frame's contact, even when the # caller did not request contact diagnostics from StreamPipeline. - if self.calibration.input_profile == "fn_delta_autocal_uv/1": + # The field model needs no locator: it consumes this frame's field. + if self.calibration.input_profile == FIELD_PROFILE: + self.apply_calibration_frame( + frame_id=self._model_frame_id, + field=self.__hhd.get_field(valid_mask=mask_ds), + ) + elif self.calibration.input_profile == "fn_delta_autocal_uv/1": self.compute_contact() else: self.apply_calibration_frame(frame_id=self._model_frame_id) @@ -809,7 +816,9 @@ class Sensor: """ if self._uses_calibration_model() and not self.compute_deformation_valid: raise cal_mod.CalibrationError("compute current deformation before schema 3 contact") - if (self._uses_calibration_model() and self.compute_contact_valid + if (self._uses_calibration_model() + and self.calibration.input_profile == "fn_delta_autocal_uv/1" + and self.compute_contact_valid and self._calibration_result_frame == self._model_frame_id): return self._model_contact_result multiple_foe = False @@ -845,7 +854,8 @@ class Sensor: print("Warning: contact centroid lies outside the valid UV mask.") self.compute_contact_valid = True - if self._uses_calibration_model(): + if (self._uses_calibration_model() + and self.calibration.input_profile == "fn_delta_autocal_uv/1"): self._model_contact_result = result uv = self._calibration_contact_uv() self.apply_calibration_frame( @@ -1290,6 +1300,12 @@ class Sensor: def set_calibration_reference(self, raw, *, reference_id): """Declare a measured unloaded raw reference for this stream, not the model file.""" + if (self._uses_calibration_model() + and self.calibration.input_profile == FIELD_PROFILE): + raise cal_mod.CalibrationError( + "field model needs a declared weighted unloaded window, not a scalar zero; " + "use declare_unloaded_reference()" + ) if not isinstance(raw, dict) or not reference_id: raise ValueError("raw must map axes to finite values and reference_id is required") values = {} @@ -1303,6 +1319,26 @@ class Sensor: if self._uses_calibration_model(): self.calibration.reset(self._calibration_reference_id) + def declare_unloaded_reference(self, *, reference_id): + """Start the declared unloaded window for a field model. + + The model medians the next 30 consecutive valid weighted field sums; + until then every frame reports ``zero_unavailable``. Contact location is + deliberately not consulted, and the scalar raw reference path stays + rejected for this profile. + """ + if not self._uses_calibration_model(): + raise cal_mod.CalibrationError("unloaded reference requires a schema 3 calibration") + if self.calibration.input_profile != FIELD_PROFILE: + raise cal_mod.CalibrationError( + "weighted unloaded reference requires hhd_normal_l1_field/1") + if not isinstance(reference_id, str) or not reference_id.strip(): + raise ValueError("reference_id must be a non-empty string") + self._reset_calibration_session() + self._calibration_reference = {} + self._calibration_reference_id = str(reference_id) + self.calibration.begin_reference(str(reference_id)) + def _calibration_contact_uv(self): if not self.centroid_found: return None @@ -1320,8 +1356,9 @@ class Sensor: def apply_calibration_frame( self, *, frame_id, timestamp_s=None, raw=None, uv=None, contact_found=None, mapping_valid=False, single_contact=True, + field=None, output_mask=None, ): - """Infer once from an unfiltered raw triple and same-frame optional contact UV.""" + """Infer once from an unfiltered raw triple, contact UV or same-frame field.""" if not self._uses_calibration_model(): raise cal_mod.CalibrationError("apply_calibration_frame requires schema 3") if self._calibration_context is None: @@ -1353,6 +1390,12 @@ class Sensor: } if contact_found is not None: features["contact_found"] = bool(contact_found) + if field is not None: + features["field"] = field + mask = self.valid_mask if output_mask is None else output_mask + if mask is None: + mask = np.ones(self.calibration.coordinate_map.shape[:2], dtype=bool) + features["output_valid_mask"] = np.asarray(mask, dtype=bool) result = self.calibration.predict( frame_id=frame_id, timestamp_s=timestamp_s, features=features, reference_id=self._calibration_reference_id, diff --git a/tests/test_field_calibration.py b/tests/test_field_calibration.py new file mode 100644 index 0000000..e08fa9d --- /dev/null +++ b/tests/test_field_calibration.py @@ -0,0 +1,464 @@ +"""Field gain model: gain mapping, weighted reduction and unloaded baseline.""" + +from __future__ import annotations + +import hashlib +from copy import deepcopy + +import numpy as np +import pytest + +from orisys.calibration import CalibrationError, parse, validate_for_device +from orisys.calibration_context import build_context +from orisys.calibration_models import parse_model +from orisys.field_calibration import ( + BASELINE_POLICY, + FIELD_IDENTITY, + MAPPING_POLICY, + build_field_gain_map, + coerce_field, + reduce_field, +) + +FULL_SQUARE = [[0.0, 0.0], [3.0, 0.0], [0.0, 3.0], [3.0, 3.0]] +SMALL_SQUARE = [[0.0, 0.0], [2.0, 0.0], [2.0, 2.0], [0.0, 2.0]] + + +def _artifact(tmp_path, *, shape=(4, 4), uv=None, valid=None): + height, width = shape + if uv is None: + rows, cols = np.mgrid[0:height, 0:width] + uv = np.stack((cols, rows), axis=-1).astype(np.float64) + if valid is None: + valid = np.ones((height, width), dtype=bool) + path = tmp_path / "map.npz" + np.savez(path, uv_by_output_px=uv, valid_mask=valid, origin_xy=np.array([0, 0])) + return path, hashlib.sha256(path.read_bytes()).hexdigest() + + +def _pipeline(*, digest): + pipeline = { + "stitch_resolution": [4, 4], + "dis_mode": 1, + "hhd_downsample": 1, + "filters": {"measurement": {"flow_ema_alpha": 1.0}}, + "source_scale": 0.5, + "geometry_sha256": "b" * 64, + "coordinates": {"frame": "autocal_flat", "unit": "px", + "origin_xy": [0, 0], "map_sha256": digest}, + "feature_signature_sha256": "a" * 64, + "field": deepcopy(FIELD_IDENTITY), + } + return pipeline + + +def _document(tmp_path, *, gains=None, points=None, triangles=None, uv=None, + coord_valid=None, artifact=True): + height, width = (4, 4) if uv is None else uv.shape[:2] + path, digest = _artifact(tmp_path, shape=(height, width), uv=uv, valid=coord_valid) + pipeline = _pipeline(digest=digest) + params = { + "points_uv": points if points is not None else FULL_SQUARE, + "gains_N_per_raw": gains if gains is not None else [1.0, 2.0, 3.0, 4.0], + "triangles": triangles if triangles is not None else [[0, 1, 2], [1, 3, 2]], + "valid_force_range_N": [0.0, 100.0], + "mapping": deepcopy(MAPPING_POLICY), + "baseline": deepcopy(BASELINE_POLICY), + } + document = { + "schema_version": 3, + "unit": "N", + "identity": {"sensor_type": "finger", "sn": "DD02-GL"}, + "pipeline": pipeline, + "model": { + "type": "field_gain/1", + "input_profile": "hhd_normal_l1_field/1", + "output_frame": "sdk_force/1", + "outputs": ["fn"], + "params": params, + }, + } + if artifact: + document["model"]["artifacts"] = [ + {"role": "coordinate_map", "path": path.name, "sha256": digest} + ] + return document + + +def _field(dx=0.5, dy=0.5, *, shape=(2, 2), area_scale=4.0, mask=None, dtype=np.float32): + d = np.zeros((*shape, 2), dtype=dtype) + d[:, :, 0] = dx + d[:, :, 1] = dy + if mask is None: + mask = np.ones(shape, dtype=bool) + mask = np.asarray(mask, dtype=bool) + d[~mask] = 0.0 + return {"d": d, "valid_mask": mask, "area_scale": area_scale, "grid_shape": tuple(shape)} + + +def _features(field, *, mask_shape=(4, 4), output_mask=None): + if output_mask is None: + output_mask = np.ones(mask_shape, dtype=bool) + return {"field": field, "output_valid_mask": output_mask} + + +def _model_with_gain(tmp_path, gain=1.0): + document = _document(tmp_path, gains=[gain] * 4) + return parse_model(document, tmp_path / "calibration.json") + + +def _feed_ready(model, frames=30, field=None): + model.begin_reference("unloaded") + field = field if field is not None else _field() + for _ in range(frames): + model.predict(1, 0.0, _features(field)) + assert model.field_baseline_N is not None + + +def test_parse_accepts_field_model_and_declares_profile(tmp_path): + model = parse_model(_document(tmp_path), tmp_path / "calibration.json") + assert model.model_type == "field_gain/1" + assert model.input_profile == "hhd_normal_l1_field/1" + assert model.channels == {"fn": "N"} + assert model.to_dict()["model"]["params"]["mapping"] == MAPPING_POLICY + assert len(model.fingerprint) == 64 + gain_map = model.field_gain_map((2, 2), np.ones((4, 4), dtype=bool)) + assert gain_map.gain.shape == (2, 2) + assert np.allclose(gain_map.coverage, 1.0) + + +def test_constant_gain_matches_hand_computed_weighted_sum(tmp_path): + model = _model_with_gain(tmp_path, gain=1.0) + _feed_ready(model) + result = model.predict(31, 0.0, _features(_field())) + # q=1 on each of 4 cells, coverage 1, gain 1, A=4: P=16, baseline=16, FN=0. + assert result.values_N["fn"] == pytest.approx(0.0) + assert result.status["fn"] == "ok" + assert result.diagnostics["baseline_N"] == pytest.approx(16.0) + result = model.predict(32, 0.0, _features(_field(1.5, 1.5))) + # q=3 per cell -> P=48, FN=48-16=32. + assert result.values_N["fn"] == pytest.approx(32.0) + + +def test_two_regions_contribute_and_vectors_combine_before_l1(tmp_path): + model = _model_with_gain(tmp_path, gain=1.0) + _feed_ready(model) + d = np.zeros((2, 2, 2), dtype=np.float32) + d[0, 0] = (1.0, 0.0) + d[1, 1] = (0.0, 0.5) + mask = np.ones((2, 2), dtype=bool) + field = {"d": d, "valid_mask": mask, "area_scale": 4.0, "grid_shape": (2, 2)} + result = model.predict(31, 0.0, _features(field)) + # q sums 1.5, two cells -> P=6; baseline 16 -> -10. + assert result.values_N["fn"] == pytest.approx(-10.0) + + combined = np.zeros((2, 2, 2), dtype=np.float32) + combined[0, 0] = (0.75, 0.0) + # A negative and a positive vector at the same cell combine to 0.5, not 1.0. + first = combined.copy() + second = np.zeros_like(combined) + second[0, 0] = (-0.25, 0.0) + first_map = {"d": first, "valid_mask": mask, "area_scale": 4.0, "grid_shape": (2, 2)} + second_map = {"d": second, "valid_mask": mask, "area_scale": 4.0, "grid_shape": (2, 2)} + combined_map = {"d": first + second, "valid_mask": mask, "area_scale": 4.0, "grid_shape": (2, 2)} + gain = model.field_gain_map((2, 2), np.ones((4, 4), dtype=bool)) + reduction = reduce_field(coerce_field(combined_map), gain) + assert reduction.weighted_N == pytest.approx(4.0 * 0.5) + assert reduction.weighted_N != pytest.approx( + reduce_field(coerce_field(first_map), gain).weighted_N + + reduce_field(coerce_field(second_map), gain).weighted_N + ) + + +def test_boundary_projection_and_mask_coverage(tmp_path): + model = parse_model(_document(tmp_path, points=SMALL_SQUARE, + triangles=[[0, 1, 2], [0, 2, 3]]), + tmp_path / "calibration.json") + gains = model.field_gain_map((4, 4), np.ones((4, 4), dtype=bool)) + # (1,1) is inside the small hull; (3,3) projects to the nearest edge. + assert gains.gain[1, 1] == pytest.approx(2.0) + assert gains.coverage[1, 1] == pytest.approx(1.0) + assert gains.inside[1, 1] == pytest.approx(1.0) + assert gains.gain[3, 3] == pytest.approx(3.0) + assert gains.inside[3, 3] == pytest.approx(0.0) + assert gains.coverage[3, 3] == pytest.approx(1.0) + + # A physical mask that keeps only the left half removes the right cells entirely. + constant = parse_model(_document(tmp_path, points=SMALL_SQUARE, gains=[1.0] * 4, + triangles=[[0, 1, 2], [0, 2, 3]]), + tmp_path / "calibration.json") + left_half = np.zeros((4, 4), dtype=bool) + left_half[:, :2] = True + gains = constant.field_gain_map((2, 2), left_half) + assert np.allclose(gains.coverage, [[1.0, 0.0], [1.0, 0.0]]) + field = _field(mask=np.array([[True, False], [True, False]])) + reduction = reduce_field(coerce_field(field), gains) + # Two active cells with q=1, A=4: P=8. Inactive cells never contribute. + assert reduction.weighted_N == pytest.approx(8.0) + + +def test_partial_coverage_constant_gain_still_matches_raw_sum(tmp_path): + """R1: coverage is diagnostic, not a second mask weight. + + HHD raw counts a partially covered mask-edge cell in full; folding coverage + into ``P`` would discount it again and break ``P == gain * raw``. + """ + model = _model_with_gain(tmp_path, gain=0.5) + mask = np.ones((4, 4), dtype=bool) + mask[0, 0] = False + gain_map = model.field_gain_map((2, 2), mask) + assert gain_map.coverage[0, 0] == pytest.approx(0.75) + field = _field(0.5, 0.5, shape=(2, 2), area_scale=4.0) + raw = 4.0 * float(np.sum(np.abs(field["d"]))) + reduction = reduce_field(coerce_field(field), gain_map) + assert reduction.weighted_N == pytest.approx(0.5 * raw) + assert reduction.coverage_fraction == pytest.approx((0.75 + 1.0 + 1.0 + 1.0) / 4.0) + + +def test_active_hhd_cell_without_coverage_is_a_mapping_error(tmp_path): + """R1: an active cell with no mapped pixel must not be silently counted as zero.""" + model = _model_with_gain(tmp_path, gain=1.0) + left_half = np.zeros((4, 4), dtype=bool) + left_half[:, :2] = True + gain_map = model.field_gain_map((2, 2), left_half) + assert np.allclose(gain_map.coverage, [[1.0, 0.0], [1.0, 0.0]]) + with pytest.raises(ValueError, match="mapping error"): + reduce_field(coerce_field(_field(mask=np.ones((2, 2), dtype=bool))), gain_map) + result = model.predict(1, 0.0, _features( + _field(mask=np.ones((2, 2), dtype=bool)), output_mask=left_half)) + assert result.status == {"fn": "coordinate_unmapped"} + assert result.values_N == {"fn": None} + + +def test_mapping_and_input_errors_are_explicit(tmp_path): + uv = np.stack(np.mgrid[0:4, 0:4][::-1], axis=-1).astype(np.float64) + valid = np.ones((4, 4), dtype=bool) + valid[0, 0] = False + model = parse_model(_document(tmp_path, uv=uv, coord_valid=valid), + tmp_path / "calibration.json") + with pytest.raises(ValueError, match="mapping error"): + model.field_gain_map((2, 2), np.ones((4, 4), dtype=bool)) + result = model.predict(1, 0.0, _features(_field())) + assert result.status == {"fn": "coordinate_unmapped"} + assert result.values_N == {"fn": None} + + broken = _field() + broken["d"][0, 0, 0] = np.nan + result = model.predict(2, 0.0, _features(broken)) + assert result.status == {"fn": "invalid_input"} + assert result.values_N == {"fn": None} + + wrong_dtype = _field(dtype=np.float64) + with pytest.raises(ValueError, match="dtype"): + coerce_field(wrong_dtype) + wrong_shape = _field() + wrong_shape["grid_shape"] = (3, 2) + with pytest.raises(ValueError, match="grid_shape"): + coerce_field(wrong_shape) + + +def test_empty_field_reports_zero_and_negative_baseline_difference(tmp_path): + model = _model_with_gain(tmp_path, gain=1.0) + _feed_ready(model) + empty = _field(0.0, 0.0) + result = model.predict(31, 0.0, _features(empty)) + assert result.status["fn"] == "out_of_range" + assert result.values_N["fn"] == pytest.approx(-16.0) + assert result.diagnostics["extrapolated_fraction"] == pytest.approx(0.0) + + +def test_weighted_baseline_needs_declaration_and_thirty_consecutive_frames(tmp_path): + model = _model_with_gain(tmp_path, gain=1.0) + result = model.predict(1, 0.0, _features(_field())) + assert result.status == {"fn": "zero_unavailable"} + assert result.diagnostics["pending_reference_frames"] == 0 + model.begin_reference("unloaded") + for _ in range(29): + result = model.predict(1, 0.0, _features(_field())) + assert result.status == {"fn": "zero_unavailable"} + result = model.predict(1, 0.0, _features(_field())) + assert result.status == {"fn": "zero_unavailable"} + assert result.diagnostics["baseline_N"] == pytest.approx(16.0) + assert result.diagnostics["reference_id"] == "unloaded" + ready = model.predict(2, 0.0, _features(_field())) + assert ready.status["fn"] == "ok" + + model.begin_reference("second") + invalid = _field() + invalid["d"][0, 0, 0] = np.nan + for _ in range(29): + model.predict(3, 0.0, _features(_field())) + assert model.predict(3, 0.0, _features(invalid)).status == {"fn": "invalid_input"} + result = model.predict(4, 0.0, _features(_field())) + assert result.diagnostics["pending_reference_frames"] == 1 + assert result.status == {"fn": "zero_unavailable"} + + +def test_reset_and_model_identity_void_the_weighted_baseline(tmp_path): + document = _document(tmp_path, gains=[1.0] * 4) + model = parse_model(document, tmp_path / "calibration.json") + _feed_ready(model) + assert model.field_baseline_N is not None + model.reset(None) + assert model.field_baseline_N is None + result = model.predict(31, 0.0, _features(_field())) + assert result.status == {"fn": "zero_unavailable"} + fresh = parse_model(document, tmp_path / "calibration.json") + assert fresh.field_baseline_N is None + assert fresh.predict(31, 0.0, _features(_field())).status == {"fn": "zero_unavailable"} + + +def test_field_document_rejections(tmp_path): + mutations = [ + lambda data: data["model"].update(input_profile="fn_delta_autocal_uv/1"), + lambda data: data["model"].update(outputs=["fn", "fx"]), + lambda data: data["model"].pop("artifacts"), + lambda data: data["model"]["params"].update(mapping={"inside": "nearest"}), + lambda data: data["model"]["params"].update(baseline={"frames": 29, "statistic": "median"}), + lambda data: data["pipeline"].pop("field"), + lambda data: data["pipeline"].update(field={"grid": "other"}), + lambda data: data["model"]["params"].update(extra=True), + lambda data: data["model"]["params"].update(gains_N_per_raw=[1.0, 2.0, 3.0, 0.0]), + lambda data: data["pipeline"].pop("geometry_sha256"), + ] + for mutate in mutations: + data = _document(tmp_path) + mutate(data) + with pytest.raises(ValueError): + parse_model(data, tmp_path / "calibration.json") + + +def test_field_model_is_loadable_through_calibration_entry_points(tmp_path): + document = _document(tmp_path) + model = parse(document, source_path=tmp_path / "calibration.json") + assert model.schema_version == 3 + assert parse_model(document, tmp_path / "calibration.json").model_type == "field_gain/1" + with pytest.raises(CalibrationError, match="identity mismatch"): + validate_for_device(model, "finger", "OTHER-SN") + + +def test_context_includes_field_identity_and_rejects_scalar_profile(tmp_path): + path, digest = _artifact(tmp_path) + coordinates = {"frame": "autocal_flat", "unit": "px", + "origin_xy": [0, 0], "map_sha256": digest} + config = { + "type": "compound_eye", + "camera": {"resolution_raw": [4, 4], "resolution_stitch": [4, 4]}, + "tracker": {"dis_mode": 1, "hhd_downsample": 1}, + "filters": {"measurement": {"flow_ema_alpha": 1.0}}, + "contact": {}, + } + context = build_context(config, config["filters"], geometry_sha256="b" * 64, + source_scale=0.5, coordinates=coordinates, + processing_backend="uv_unroll", + preprocessing={"flip_source_vertical": True, "mirror": False}, + field=FIELD_IDENTITY) + document = _document(tmp_path) + document["pipeline"] = dict(context) + model = parse_model(document, tmp_path / "calibration.json") + model.validate_for_context(context) + scalar_context = build_context(config, config["filters"], geometry_sha256="b" * 64, + source_scale=0.5, coordinates=coordinates, + processing_backend="uv_unroll") + with pytest.raises(ValueError, match="field"): + model.validate_for_context(scalar_context) + + +def test_gain_map_is_deterministic_across_builds(tmp_path): + model = _model_with_gain(tmp_path, gain=2.0) + first = model.field_gain_map((2, 2), np.ones((4, 4), dtype=bool)) + model._field_maps.clear() + second = model.field_gain_map((2, 2), np.ones((4, 4), dtype=bool)) + assert np.array_equal(first.gain, second.gain) + assert np.array_equal(first.coverage, second.coverage) + assert first.physical_mask_sha256 == second.physical_mask_sha256 + with pytest.raises(ValueError, match="mask"): + build_field_gain_map(model._points, model._gains, model._triangles, + model.coordinate_map, model.coordinate_valid_mask, + (2, 2), np.ones((3, 3), dtype=bool)) + + +def test_field_model_ignores_contact_state(tmp_path): + model = _model_with_gain(tmp_path, gain=1.0) + model.begin_reference("unloaded") + for _ in range(30): + model.predict(1, 0.0, _features(_field())) + features = _features(_field()) + features.update(contact_found=False, zero_valid=False) + result = model.predict(31, 0.0, features) + assert result.status["fn"] == "ok" + assert result.values_N["fn"] == pytest.approx(0.0) + + +def _hhd(grid_shape, reference_shape, downsample): + from orisys.hhd import nHHD + + return nHHD(cuda=False, grid=grid_shape, spacings=(18 * downsample, 18 * downsample), + poisson_backend="cpu", force_reference_grid=reference_shape) + + +def _identity_uv(shape): + rows, cols = np.mgrid[0:shape[0], 0:shape[1]] + return np.stack((cols, rows), axis=-1).astype(np.float64) + + +def test_hhd_field_payload_uses_actual_reference_area_scale(): + reference = (9, 8) + for downsample in (1, 2, 3, 4): + grid = (reference[0] // downsample, reference[1] // downsample) + solver = _hhd(grid, reference, downsample) + solver.get_force(np.zeros((*grid, 2), dtype=np.float32)) + payload = solver.get_field() + assert payload["area_scale"] == pytest.approx( + reference[0] * reference[1] / (grid[0] * grid[1])) + assert payload["grid_shape"] == grid + assert payload["reference_shape"] == reference + assert payload["d"].dtype == np.float32 + assert payload["d"].shape == (*grid, 2) + payload["d"][0, 0, 0] = 123.0 + assert solver.d[0, 0, 0] != 123.0 + + +def test_constant_gain_matches_raw_hhd_sum(tmp_path): + shape = (6, 5) + rng = np.random.default_rng(7) + solver = _hhd(shape, shape, 1) + _, forces = solver.get_force(rng.normal(size=(*shape, 2)).astype(np.float32)) + raw_sn = float(forces[0]) + payload = solver.get_field() + uv = _identity_uv(shape) + document = _document(tmp_path, gains=[0.5] * 4, uv=uv, + points=[[0.0, 0.0], [shape[1] - 1.0, 0.0], + [0.0, shape[0] - 1.0], [shape[1] - 1.0, shape[0] - 1.0]], + triangles=[[0, 1, 2], [1, 3, 2]]) + model = parse_model(document, tmp_path / "calibration.json") + gain_map = model.field_gain_map(shape, np.ones(shape, dtype=bool)) + assert np.allclose(gain_map.coverage, 1.0) + reduction = reduce_field(coerce_field(payload), gain_map) + # Constant gain: the field-weighted sum equals the raw HHD sum times the gain. + assert reduction.weighted_N == pytest.approx(0.5 * raw_sn, rel=1e-5, abs=1e-6) + + +def test_constant_field_integral_across_downsample_and_nondivisible_grids(tmp_path): + reference = (9, 8) + expected = 0.5 * 2.0 * reference[0] * reference[1] + for downsample in (1, 2, 3, 4): + grid = (reference[0] // downsample, reference[1] // downsample) + solver = _hhd(grid, reference, downsample) + solver.get_force(np.zeros((*grid, 2), dtype=np.float32)) + payload = solver.get_field() + payload["d"] = np.ones((*grid, 2), dtype=np.float32) + payload["valid_mask"] = np.ones(grid, dtype=bool) + uv = _identity_uv(reference) + document = _document( + tmp_path, gains=[0.5] * 4, uv=uv, + points=[[0.0, 0.0], [reference[1] - 1.0, 0.0], + [0.0, reference[0] - 1.0], [reference[1] - 1.0, reference[0] - 1.0]], + triangles=[[0, 1, 2], [1, 3, 2]], + ) + model = parse_model(document, tmp_path / "calibration.json") + gain_map = model.field_gain_map(grid, np.ones(reference, dtype=bool)) + reduction = reduce_field(coerce_field(payload), gain_map) + # q=2 per cell; A*cell_count == reference pixel count for every integer + # downsample, including grids that do not divide the reference exactly. + assert reduction.weighted_N == pytest.approx(expected, rel=1e-5, abs=1e-6) diff --git a/tests/test_model_runtime.py b/tests/test_model_runtime.py index 4b483ba..27bfee0 100644 --- a/tests/test_model_runtime.py +++ b/tests/test_model_runtime.py @@ -345,3 +345,113 @@ def test_bad_zero_reference_is_rejected_without_reset(reference): with pytest.raises(ValueError): sensor.set_calibration_reference(reference, reference_id="bad") assert sensor._calibration_reference == {"fn": 10.} + + +class FieldRecordingModel: + schema_version = 3 + channels = {"fn": "N"} + input_profile = "hhd_normal_l1_field/1" + + def __init__(self): + self.calls = [] + self.resets = [] + self.begin_calls = [] + self.coordinate_map = np.zeros((2, 2, 2), dtype=np.float32) + self.coordinate_valid_mask = np.ones((2, 2), dtype=bool) + + def reset(self, reference_id): + self.resets.append(reference_id) + + def begin_reference(self, reference_id): + self.begin_calls.append(reference_id) + + def validate_for_context(self, context): + if context != {"signature": "actual"}: + raise CalibrationError("pipeline mismatch") + + def predict(self, **kwargs): + self.calls.append(kwargs) + features = kwargs["features"] + if "field" not in features: + return SimpleNamespace(values_N={"fn": None}, status={"fn": "invalid_input"}, + diagnostics={}) + return SimpleNamespace(values_N={"fn": 1.0}, status={"fn": "ok"}, diagnostics={}) + + +def test_hhd_field_model_infers_from_same_frame_field_without_contact(): + sensor = make_sensor() + sensor.calibration = FieldRecordingModel() + sensor._calibration_context = {"signature": "actual"} + sensor.cuda = sensor.verbose = False + sensor.valid_mask = sensor.valid_mask_hhd = None + sensor._hhd_downsample = 1 + sensor._hhd_grid_shape = sensor._hhd_ref_shape = (2, 2) + sensor._model_frame_id = 0 + field = np.zeros((2, 2, 2), dtype=np.float32) + calls = [] + + def get_field(valid_mask=None): + calls.append(valid_mask) + return {"d": field, "valid_mask": np.ones((2, 2), dtype=bool), "area_scale": 1.0, + "grid_shape": (2, 2), "reference_shape": (2, 2)} + + sensor._Sensor__hhd = SimpleNamespace( + get_force=lambda *args, **kwargs: (None, (10., 0., 0., 0.)), + get_normal=lambda: field, get_shear=lambda: field, get_field=get_field, + ) + sensor.compute_hhd(np.zeros((2, 2, 2), dtype=np.float32)) + features = sensor.calibration.calls[-1]["features"] + assert features["field"]["grid_shape"] == (2, 2) + assert features["output_valid_mask"].shape == (2, 2) + assert "contact_found" not in features + assert sensor.calibration_status == {"fn": "ok"} + assert sensor.fnormal == pytest.approx(1.0) + assert calls == [None] + + +def test_weighted_unloaded_reference_is_declared_not_a_scalar_zero(): + sensor = make_sensor() + sensor.calibration = FieldRecordingModel() + sensor._calibration_context = {"signature": "actual"} + sensor.declare_unloaded_reference(reference_id="video-unloaded-100") + assert sensor.calibration.begin_calls == ["video-unloaded-100"] + assert sensor._calibration_reference_id == "video-unloaded-100" + with pytest.raises(CalibrationError, match="scalar zero"): + sensor.set_calibration_reference({"fn": 0.}, reference_id="bad") + with pytest.raises(CalibrationError, match="hhd_normal_l1_field"): + make_sensor().declare_unloaded_reference(reference_id="scalar-model") + + +def test_finger_runtime_binds_field_identity_and_forwards_the_window(monkeypatch, tmp_path): + import orisys.calibration_context as context_mod + from orisys.field_calibration import FIELD_IDENTITY + from orisys.finger.runtime import FingerRuntime + + sensor = make_sensor() + sensor.calibration = FieldRecordingModel() + sensor._calibration_context = {"signature": "actual"} + sensor.device_id = None + sensor.mirror = False + runtime = FingerRuntime.__new__(FingerRuntime) + runtime.tactile = SimpleNamespace(raw_sensor=sensor) + mesh = tmp_path / "mesh.npz" + mesh.write_bytes(b"actual geometry") + runtime.geometry = SimpleNamespace(config={}, export_paths=SimpleNamespace(mesh=mesh), + source_scale=.42, out_height=2, out_width=2) + runtime.domain = object() + runtime.frame_processor = SimpleNamespace(flip_source_vertical=True) + monkeypatch.setattr(context_mod, "domain_fingerprint", lambda domain: "domain") + contexts = [] + + def context(config, filters, **kwargs): + contexts.append(kwargs) + return {"signature": "actual"} + + monkeypatch.setattr(context_mod, "build_context", context) + selected = FieldRecordingModel() + selected.pipeline = {"coordinates": {"frame": "autocal_flat"}} + runtime.set_calibration(selected) + assert contexts[0]["field"] == FIELD_IDENTITY + runtime.declare_unloaded_reference("runtime-window") + assert sensor.calibration.begin_calls == ["runtime-window"] +