orisys-tech/orisys-sdk
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增 HHD 场加权标定模型与加权空载基线

Browse Source 
field_gain/1 用坐标增益图对无旋向量场做 L1 场加权，以连续 30 帧加权中位数替代旧 raw 零点，实时入口暂不支持。目标测试 102 项与 lint_gate 通过；契约见 orisys-sdk/docs/field-gain-model.zh.md。
This commit is contained in:
ykcheungab
2026-10-19 19:49:25 +08:00
parent c78f7a304f
commit 0a042c7fbf
11 changed files with 1355 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/calibration-model-contract.zh.md
+8 -41
View File
@@ -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`。
另以独立录像验证物理准确度，契约或单元测试通过不代表标定合格。
docs/field-gain-model.zh.md
+102
View File
@@ -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 声明来源哈希、边界/基线策略和实验假设。
docs/ml-model-contract.zh.md
+44
View File
@@ -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。
附件间依赖也须逐项声明和校验，未声明的外部权重不可加载。
src/orisys/calibration_context.py
+10 -2
View File
@@ -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)
src/orisys/calibration_models.py
+175 -15
View File
@@ -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)
src/orisys/field_calibration.py
+347
View File
@@ -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,
)
src/orisys/finger/runtime.py
+12 -3
View File
@@ -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."""
src/orisys/hhd/api.py
+31
View File
@@ -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.
src/orisys/sensor.py
+52 -9
View File
@@ -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,
tests/test_field_calibration.py
+464
View File
@@ -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)
tests/test_model_runtime.py
+110
View File
@@ -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"]