Compare commits
3 Commits
9c951e9d3c
...
e9710e140b
| Author | SHA1 | Date | |
|---|---|---|---|
 ykcheungab
|
e9710e140b | ||
|
ykcheungab
|
ff7614a6a0 | ||
|
ykcheungab
|
f7f4682ff9 |
@@ -0,0 +1,248 @@
|
||||
# 统一力标定模型契约
|
||||
|
||||
Purpose: reference. Scope: SDK 标定文件选择、SN 绑定和推断接口。
|
||||
状态：2026-10-18 已实现 schema 3 的空间/全局增益；ML 后端仅保留契约，尚未实现。
|
||||
保持 `cal_path` 与现有单点文件；新增空间模型和机器学习模型使用同一入口。
|
||||
|
||||
## 现状与边界
|
||||
|
||||
现有实现见 [calibration.py](../src/orisys/calibration.py)、
|
||||
[Sensor](../src/orisys/sensor.py)、[profile resolver](../src/orisys/sensors/calibration.py)。
|
||||
当前读 schema 1/2，按 SN 加载要求三个通道，转换只接收三个力标量。
|
||||
`ForceReadout.apply` 会分别转换滤波前后数据，不能直接挂接带状态的空间或 ML 模型。
|
||||
|
||||
| 所有者 | 责任 |
|
||||
|---|---|
|
||||
| 离线实验 / AutoCal | 采集、训练、验证、导出模型；将实验坐标转换为契约坐标 |
|
||||
| SDK | 选文件、校验身份与输入、准备特征、推断、输出逐轴有效性 |
|
||||
| 应用 | 提供 SN 与文件选择，展示状态；不自己套用第二次增益 |
|
||||
| 设计资产 | 几何与坐标定义；不提供每台设备的实测标定值 |
|
||||
|
||||
SDK 不训练模型、不导入应用代码、不读取训练 CSV，不运行文件内指定的 Python。
|
||||
实验 `finger_multical.experiment_model/1` 必须显式导出，不能直接当 SDK 文件加载。
|
||||
|
||||
## 文件选择：保留路径，文件决定模型
|
||||
|
||||
不增加 `model_type` 构造参数或全局活动模型。`Sensor`、`FingerRuntime.open` 和
|
||||
profile 创建处理器统一复用内部选择器；现有入口参数只映射为选择请求。
|
||||
|
||||
| 请求 | 文件与行为 |
|
||||
|---|---|
|
||||
| 非空 `cal_path` | 精确加载该文件；缺失、错误或不兼容时报错，不换文件 |
|
||||
| `calibration_dir` + `device_id` | 保持 `<dir>/<sensor_type>/<device_id>.json`；缺文件保持 raw 并报告 missing，不回退产品模板 |
|
||||
| 两者都未提供 | 保持产品包默认解析；有 descriptor 时用 `default_calibration`，旧包保留既有候选顺序 |
|
||||
| profile 的 `CalibrationSelection.NONE` | 明确不标定；不赋予 `cal_path="none"` 新语义 |
|
||||
|
||||
`cal_path` 与 `calibration_dir` 仍互斥。构造入口 `device_id` 对应文件 `identity.sn`，
|
||||
不新造另一个 SN 字段。未指定设备的录像可显式选文件，但实际 SN 一致性标记为 unknown。
|
||||
有 `device_id` 时严格匹配 `sensor_type`、`sn`；不根据相机编号猜 SN。
|
||||
保留 `device_id` 单独传入时的参数错误；目录本身不存在也报错，不能当成 SN 文件缺失。
|
||||
|
||||
```python
|
||||
# 沿用现有调用参数；cal_path 可以选择 schema 1/2/3 文件。
|
||||
runtime = FingerRuntime.open(
|
||||
config="finger", video="test.mp4",
|
||||
device_id="DD02-GL-001",
|
||||
cal_path="calibrations/finger/DD02-GL-001.json",
|
||||
)
|
||||
# 归零值来自本次无载窗口，不是模型内固定常量。
|
||||
runtime.set_calibration_reference({"fn": measured_zero_raw}, reference_id="unloaded-1")
|
||||
# 自动按 SN 选择仍使用 calibration_dir="calibrations"，不同时传 cal_path。
|
||||
```
|
||||
|
||||
同一 SN 比较不同模型时，显式指定不同 JSON；默认 SN 文件只代表一个选定版本。
|
||||
不扫描目录选“最新模型”，不在每帧重新加载。
|
||||
|
||||
## 最小文件：schema 3
|
||||
|
||||
复用 `force_calibration.json` 文件角色，增加 schema 3；文件名不参与模型类型判断。
|
||||
以下是三点结构示例，数值和 SHA 占位值不是设备标定结果。
|
||||
|
||||
```json
|
||||
{
|
||||
"schema_version": 3,
|
||||
"unit": "N",
|
||||
"identity": {"sensor_type": "finger", "sn": "DD02-GL-001"},
|
||||
"pipeline": {
|
||||
"stitch_resolution": [700, 750],
|
||||
"dis_mode": 4,
|
||||
"hhd_downsample": 3,
|
||||
"source_scale": 0.42,
|
||||
"filters": {"measurement": {"flow_ema_alpha": 1.0}},
|
||||
"feature_signature_sha256": "0000000000000000000000000000000000000000000000000000000000000000",
|
||||
"geometry_sha256": "0000000000000000000000000000000000000000000000000000000000000000",
|
||||
"coordinates": {
|
||||
"frame": "autocal_flat",
|
||||
"unit": "px",
|
||||
"origin_xy": [511, 1021],
|
||||
"map_sha256": "0000000000000000000000000000000000000000000000000000000000000000"
|
||||
}
|
||||
},
|
||||
"model": {
|
||||
"type": "spatial_gain/1",
|
||||
"input_profile": "fn_delta_autocal_uv/1",
|
||||
"output_frame": "sdk_force/1",
|
||||
"outputs": ["fn"],
|
||||
"params": {
|
||||
"points_uv": [[-145, 604], [145, 604], [5, 430]],
|
||||
"gains_N_per_raw": [0.00015, 0.00020, 0.00025],
|
||||
"triangles": [[0, 1, 2]],
|
||||
"valid_force_range_N": [0.0, 3.0]
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
| 字段 | 规则 |
|
||||
|---|---|
|
||||
| `schema_version` | 整数 3；未知版本拒绝 |
|
||||
| `unit` | 固定 `N` |
|
||||
| `identity` | `sensor_type`、非空 `sn` 必填；文件位置不能替代身份 |
|
||||
| `pipeline` | 输入 profile 要求的实际处理参数、几何和坐标身份；缺少必需项或不匹配则拒绝 |
|
||||
| `model.type` | SDK 已注册的版本化标识；决定参数解析和推断，不是模块路径 |
|
||||
| `input_profile` | SDK 已实现的特征、单位、形状、预处理和时序契约 |
|
||||
| `output_frame` | `sdk_force/1` 延续对应处理 profile 的 FN/FX/FY 符号和基方向；不是世界坐标合力模长 |
|
||||
| `outputs` | 非空、无重复的 `fn/fx/fy` 子集；支持 FN-only，不冒充三轴标定 |
|
||||
| `params` | 模型类型私有参数；通用加载器不解释点数或网络层 |
|
||||
|
||||
`provenance` 可选，保存数据摘要、训练版本和验证指标；不作为运行时输入。
|
||||
其余未知语义字段拒绝，不能靠忽略字段误读新模型。新增模型类型不必升级外层 schema；
|
||||
修改输入或算法语义必须升级相应 `/1` 标识。
|
||||
|
||||
`geometry_sha256` 是所用计算 mesh 文件字节的 SHA-256；`map_sha256` 是 SDK 坐标映射定义的
|
||||
附件字节摘要，包含坐标转换结果，不能只比较图幅。pipeline 的必需键由输入 profile 固定，
|
||||
覆盖该 backend 所有影响测量的设置；只检查示例中的部分配置不算兼容校验。
|
||||
|
||||
新增 profile 必须定义闭合的签名字段表。首个 profile 的 `feature_signature_sha256` 摘要
|
||||
对象包含 `feature_version`、`processing_backend`、有效的图像预处理/光流/HHD 配置和计算资产哈希；
|
||||
不含文件绝对路径、SN、显示资源或力读数滤波。以 UTF-8、键排序、无多余空格的 JSON 计算 SHA-256。
|
||||
特征定义或单位改变须升级 `feature_version`，配置覆盖后重新计算，不能用配置文件名代替。
|
||||
实现由 `calibration_context.build_context` 统一生成，特征版本为 `orisys_hhd_raw/1`。
|
||||
|
||||
## 空间增益模型
|
||||
|
||||
`spatial_gain/1` 固定输出 `fn`，使用 `fn_delta_autocal_uv/1`：
|
||||
|
||||
- 输入为同一帧的未经过力读数滤波的 FN raw、有效会话零点及接触 U/V。
|
||||
上游光流滤波属于 pipeline 身份；不得混用不同特征链。
|
||||
`delta_raw`、U、V 为有限浮点标量，前者单位 raw，后两者为上述有符号像素。
|
||||
- SDK 会话准备 `delta_raw = fn_raw - zero_raw`，模型不再次扣零。
|
||||
零点必须来自明确的归零操作或已标注窗口；模型文件不保存某段录像的启动零点。
|
||||
- U 向右、V 向上，`x=origin_x+U`、`y=origin_y-V`。
|
||||
SDK 提供并校验坐标转换；不能让模型偷偷按比例换算，不能运行时依赖 AutoCal。
|
||||
- `points_uv` 为 N×2，`gains_N_per_raw` 长度 N；N≥3 且非共线、坐标唯一、增益有限且为正。
|
||||
`triangles` 为任意数量的三索引，必须构成支持点凸包内完整、无重叠的非退化三角网。
|
||||
保存连接关系以避免重新剖分引入变化；点数由数组决定，不固定为 9/15。
|
||||
- 对三角形内查询，`g=Σw_i*g_i`，`fn_N=g*delta_raw`；不插值倒数，不外推。
|
||||
未覆盖、位置失败、归零无效均返回无效，不能偷偷改用 M0。
|
||||
- `valid_force_range_N` 必填。估计超范围时保留数值并标记 `out_of_range`，不截断。
|
||||
|
||||
FingerRuntime 使用 `artifacts` 中 `role="coordinate_map"` 的只读 NPZ：
|
||||
`uv_by_output_px` 为 H×W×2 有符号 UV，`valid_mask` 为 H×W bool；可附 `origin_xy` 与
|
||||
`flat_mapping_sha256`。`pipeline.output_domain_sha256` 绑定实际展开的 remap、mask 和 world_xyz；
|
||||
形状、几何或特征身份不同则在打开视频前拒绝。离线直接提供 UV 的模型可以没有映射附件。
|
||||
|
||||
单点/两点数据不自动切换算法；可导出独立的 `global_gain/1`，参数为各输出轴的增益和有效范围，
|
||||
输入 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。
|
||||
附件间依赖也须逐项声明和校验，未声明的外部权重不可加载。
|
||||
|
||||
## 每帧推断与有效性
|
||||
|
||||
内部统一生命周期如下，`calibration.load` 分流到旧曲线或新模型：
|
||||
|
||||
```text
|
||||
load(selected_file, device_identity, pipeline) -> CalibrationModel
|
||||
model.predict(frame_id, timestamp_s, profile_features, reference_id) -> CalibrationResult
|
||||
model.reset(reference_id)
|
||||
|
||||
CalibrationResult:
|
||||
values_N: {axis: finite_float | null}
|
||||
status: {axis: ok | no_contact | invalid_input | zero_unavailable |
|
||||
coordinate_unmapped | outside_coverage | out_of_range}
|
||||
```
|
||||
|
||||
FN 空间模型输入仅允许单一接触；当前定位器只返回一个中心，无法证明不存在多接触，
|
||||
不能宣称已标定总力。全局 `force_delta/1` 不要求定位，未提供接触状态不等同于无接触。
|
||||
无接触、失败帧的校准输出为空，raw 另行保留；缺失值不是 0 N。
|
||||
`out_of_range` 有数值但不在验证范围，其余失败状态无数值；未声明轴不可用、单位仍为 raw。
|
||||
不把某轴 raw 填进 `values_N`，也不只靠全局 `force_unit` 判断三轴单位。
|
||||
schema 3 的旧数值读数接口以 NaN 表示已声明轴的无效输出，并同步暴露上述状态；
|
||||
`force_calibrated` 表示已加载模型能力，不表示当前帧有效。
|
||||
|
||||
一个模型实例属于一个流；按帧顺序每帧推断一次，读数和录像复用结果。
|
||||
清零、重连、倒播或更换模型时重置状态；不同 SN 不共享可变状态。
|
||||
schema 3 先做上游测量与归零、模型推断，再做有效 N 读数的显示滤波；失败帧不进滤波历史。
|
||||
schema 1/2 仍保留旧滤波/转换顺序。状态模型不能塞入会调用两次的旧 `convert` 回调。
|
||||
|
||||
加载诊断至少包含路径、来源、SN 校验状态、模型/输入 profile、轴集合、内容指纹和失败原因。
|
||||
加载不匹配失败与逐帧暂时无效分开；替换时先完成加载校验，失败不得破坏原活动模型。
|
||||
Sensor 提供 `bind_calibration_context`、`set_calibration_reference`、`apply_calibration_frame`；
|
||||
FingerRuntime 自动绑定真实上下文，并转发归零。`FrameSnapshot.force_units` 和
|
||||
`calibration_status` 随帧保存，不能用稍后读取的 Sensor 状态解释旧帧。
|
||||
归零与换模型须在计算线程或暂停处理时调用，不承诺并发修改安全。
|
||||
|
||||
## 兼容、实现位置与验收
|
||||
|
||||
新读取器并存支持 schema 1/2/3；旧文件不重写，SN 校验的旧三通道要求保持。
|
||||
schema 3 按声明轴校验，允许 FN-only；旧 SDK 遇 schema 3 必须拒绝，不能降级猜读。
|
||||
回退通过恢复旧文件的 `cal_path`；SN 默认文件切换前保留旧副本，不覆盖唯一证据。
|
||||
|
||||
| 修改位置 | 最小工作 |
|
||||
|---|---|
|
||||
| `calibration.py` | schema 分流、身份和附件校验；旧曲线适配器 |
|
||||
| `sensors/calibration.py` | 统一选择器与指纹；支持 SN 请求 |
|
||||
| 新 `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.json` 与同名 `.coordinates.npz` 是本次实验交付，
|
||||
`identity.sn="dd02-gl"` 为实验标识，未从硬件读取；原三边缘点与非单调点警告保留于 provenance。
|
||||
它不是生产包默认文件，需显式 `cal_path` 选择。
|
||||
|
||||
实现验收必须覆盖：旧结果与单位不变；显式路径/SN/默认/关闭选择；错误 SN 不串用；
|
||||
3/14/更多点、退化网格与凸包外；节点和插值预测；零点不重复扣除；缺坐标/无接触；
|
||||
FN-only 的 FX/FY 不冒充 N；未知模型、附件损坏和依赖缺失；状态模型每帧只调用一次及 reset。
|
||||
另以独立录像验证物理准确度，契约或单元测试通过不代表标定合格。
|
||||
+105
-14
@@ -14,39 +14,118 @@ Run:
|
||||
python examples/finger_pipeline.py --camera_id 0 --config finger
|
||||
python examples/finger_pipeline.py --video path/to/clip.mp4
|
||||
python examples/finger_pipeline.py --no-show # headless: print only
|
||||
python examples/finger_pipeline.py --config path/to/sensor.json --video clip.mp4 --cal-path model.json --zero
|
||||
|
||||
Keys: 'q' quit, 'r' reset the tracker.
|
||||
Geometry comes from the selected bundle's manifest.json assets_dir; its UV cache follows the
|
||||
geometry directory. Schema 3 requires matching processing config/geometry and a session zero.
|
||||
Use --zero only when the first 30 frames after warmup are known to be unloaded. Without a zero,
|
||||
schema 3 reports zero_unavailable. Use --help-advanced for SN selection and diagnostic overrides.
|
||||
Keys: 'q' quit, 'r' reset the tracker and clear the schema 3 session reference.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import math
|
||||
import os
|
||||
import statistics
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
os.environ.setdefault("OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS", "0")
|
||||
|
||||
import cv2
|
||||
|
||||
from _display import draw_flow_arrows
|
||||
|
||||
from orisys.configs import resolve_sensor_bundle
|
||||
from orisys.finger import FingerRuntime
|
||||
|
||||
|
||||
def configured_export_dir(config):
|
||||
"""Reuse the selected bundle's geometry, including an explicit sensor.json path."""
|
||||
path = Path(config)
|
||||
bundle = resolve_sensor_bundle(str(path.resolve().parent) if path.is_file() else config)
|
||||
if bundle is None:
|
||||
return None
|
||||
if path.is_file() and bundle.sensor_json != path.resolve():
|
||||
return None
|
||||
if bundle.manifest_json is not None and bundle.assets_dir is None:
|
||||
raise ValueError(f"No usable assets_dir in {bundle.manifest_json}; use --export-dir")
|
||||
return str(bundle.assets_dir) if bundle.assets_dir is not None else None
|
||||
|
||||
|
||||
def measure_unloaded_reference(runtime):
|
||||
"""Measure before pipeline threads start; --zero asserts these frames are unloaded."""
|
||||
tactile = runtime.tactile
|
||||
model = tactile.raw_sensor.calibration
|
||||
if getattr(model, "schema_version", None) != 3:
|
||||
raise ValueError("--zero requires a loaded schema 3 calibration")
|
||||
info = tactile.info
|
||||
axes = [(axis, key) for axis, key in
|
||||
(("fn", info.FNORMAL_RAW), ("fx", info.FSHEARX_RAW), ("fy", info.FSHEARY_RAW))
|
||||
if axis in model.channels]
|
||||
samples = {axis: [] for axis, _ in axes}
|
||||
print("Measuring session zero from 30 unloaded frames...")
|
||||
for _ in range(30):
|
||||
ok, _ = runtime.grab_and_process(color=False)
|
||||
if not ok:
|
||||
raise ValueError("Input ended before 30 unloaded frames; session zero was not set")
|
||||
tactile.compute_deformation()
|
||||
values = tactile.read_info(*(key for _, key in axes))
|
||||
if not all(math.isfinite(value) for value in values):
|
||||
raise ValueError("Non-finite unloaded reading; session zero was not set")
|
||||
for (axis, _), value in zip(axes, values):
|
||||
samples[axis].append(value)
|
||||
runtime.set_calibration_reference(
|
||||
{axis: statistics.median(values) for axis, values in samples.items()},
|
||||
reference_id="startup-unloaded-reference")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="UV-unrolled finger tactile pipeline")
|
||||
advanced = "--help-advanced" in sys.argv
|
||||
parser = argparse.ArgumentParser(description="UV-unrolled finger tactile pipeline",
|
||||
epilog="Use --help-advanced for SN selection and diagnostic overrides.")
|
||||
parser.add_argument("--help-advanced", action="help", help=argparse.SUPPRESS)
|
||||
def advanced_help(message):
|
||||
return message if advanced else argparse.SUPPRESS
|
||||
parser.add_argument("--camera_id", type=int, default=0, help="camera index (default 0)")
|
||||
parser.add_argument("-v", "--video", help="video file path; overrides --camera_id")
|
||||
parser.add_argument("-c", "--config", default="finger",
|
||||
help="product config name or path (default: finger)")
|
||||
calibration = parser.add_mutually_exclusive_group()
|
||||
calibration.add_argument("--cal-path", "--cal_path", help="explicit calibration JSON (schema 1/2/3)")
|
||||
calibration.add_argument("--calibration-dir", "--calibration_dir",
|
||||
help=advanced_help("select <directory>/<sensor_type>/<device_id>.json"))
|
||||
parser.add_argument("--device-id", "--device_id", help=advanced_help("actual device SN to validate"))
|
||||
zero = parser.add_mutually_exclusive_group()
|
||||
zero.add_argument("--zero", action="store_true",
|
||||
help="measure session zero; first 30 frames after warmup MUST be unloaded")
|
||||
zero.add_argument("--zero-fn-raw", type=float,
|
||||
help=advanced_help("measured unloaded FN raw for this session (schema 3 only)"))
|
||||
parser.add_argument("--export-dir", help=advanced_help("override bundle geometry export directory"))
|
||||
parser.add_argument("--map-cache", help=advanced_help("override UV map base path; SDK appends scale suffix"))
|
||||
parser.add_argument("--no-show", dest="show", action="store_false",
|
||||
help="do not open the display window")
|
||||
parser.add_argument("-verbose", action="store_true", help="SDK verbose logs")
|
||||
args = parser.parse_args()
|
||||
if args.device_id and not (args.cal_path or args.calibration_dir):
|
||||
parser.error("--device-id requires --cal-path or --calibration-dir")
|
||||
if args.calibration_dir and not args.device_id:
|
||||
parser.error("--calibration-dir requires --device-id")
|
||||
if args.zero_fn_raw is not None and not math.isfinite(args.zero_fn_raw):
|
||||
parser.error("--zero-fn-raw must be finite")
|
||||
|
||||
source = args.video if args.video else args.camera_id
|
||||
|
||||
# ------------------------------------------------------------------ create the object
|
||||
options = {key: getattr(args, key) for key in
|
||||
("cal_path", "calibration_dir", "device_id", "export_dir", "map_cache")
|
||||
if getattr(args, key) is not None}
|
||||
try:
|
||||
runtime = FingerRuntime.open(config=args.config, video=source, verbose=args.verbose)
|
||||
except RuntimeError as exc:
|
||||
if "export_dir" not in options:
|
||||
export_dir = configured_export_dir(args.config)
|
||||
if export_dir is not None:
|
||||
options["export_dir"] = export_dir
|
||||
runtime = FingerRuntime.open(config=args.config, video=source, verbose=args.verbose, **options)
|
||||
except (RuntimeError, ValueError, OSError) as exc:
|
||||
print(str(exc), file=sys.stderr)
|
||||
return 1
|
||||
|
||||
@@ -56,14 +135,22 @@ def main() -> int:
|
||||
f"source={g.source_width}x{g.source_height} "
|
||||
f"output={g.out_width}x{g.out_height}")
|
||||
|
||||
# Fill the flow/HHD caches once so the first published frame is not a warm-up outlier.
|
||||
runtime.warmup(color=False)
|
||||
|
||||
if args.show:
|
||||
print("Keys: 'q' quit, 'r' reset tracker.")
|
||||
print("Keys: 'q' quit, 'r' reset tracker and clear session zero.")
|
||||
|
||||
include = ("image", "flow") if args.show else ("image",)
|
||||
try:
|
||||
# Warmup may reset the session; apply a supplied reference afterwards and before threads start.
|
||||
runtime.warmup(color=False)
|
||||
if getattr(runtime.tactile.raw_sensor.calibration, "schema_version", None) == 3 and not args.device_id:
|
||||
print("Calibration file selected; hardware SN has not been independently verified.")
|
||||
if args.zero:
|
||||
measure_unloaded_reference(runtime)
|
||||
if args.zero_fn_raw is not None:
|
||||
model = runtime.tactile.raw_sensor.calibration
|
||||
if getattr(model, "schema_version", None) != 3:
|
||||
raise ValueError("--zero-fn-raw requires a loaded schema 3 calibration")
|
||||
runtime.set_calibration_reference({"fn": args.zero_fn_raw}, reference_id="cli-unloaded-reference")
|
||||
with runtime.pipeline(contact=True, include=include) as pipe:
|
||||
while True:
|
||||
# ------------------------------------------------------ read one frame
|
||||
@@ -75,12 +162,13 @@ def main() -> int:
|
||||
# ------------------------------------------------------ use the results
|
||||
coverage = runtime.tactile.mask_stats.get("coverage")
|
||||
coverage_text = f"{coverage:.2f}" if isinstance(coverage, float) else "n/a"
|
||||
fn_unit, fx_unit, fy_unit = runtime.tactile.force_units
|
||||
fn_unit, fx_unit, fy_unit = snap.force_units or runtime.tactile.force_units
|
||||
status = ",".join(f"{axis}:{value}" for axis, value in (snap.calibration_status or {}).items())
|
||||
print(
|
||||
f"FPS={snap.fps:5.1f} "
|
||||
f"normal={snap.normal_force:8.4f} {fn_unit} "
|
||||
f"shear=({snap.shear_x:7.4f} {fx_unit}, {snap.shear_y:7.4f} {fy_unit}) "
|
||||
f"mask={coverage_text} age={snap.age_ms:4.1f} ms",
|
||||
f"mask={coverage_text} age={snap.age_ms:4.1f} ms calibration={status or 'legacy/raw'}",
|
||||
end="\r",
|
||||
)
|
||||
|
||||
@@ -89,13 +177,16 @@ def main() -> int:
|
||||
snap.image, snap.flow, threshold=0.4, grid_spacing=10,
|
||||
valid_mask=runtime.tactile.valid_mask))
|
||||
|
||||
key = cv2.waitKey(1) & 0xFF
|
||||
key = cv2.waitKey(1) & 0xFF if args.show else -1
|
||||
if key == ord("q"):
|
||||
print("\nExiting.")
|
||||
break
|
||||
if key == ord("r"):
|
||||
pipe.reset()
|
||||
print("\nTracker reset.")
|
||||
print("\nTracker reset. Schema 3 session zero cleared; a new unloaded reference is required.")
|
||||
except (RuntimeError, ValueError, OSError) as exc:
|
||||
print(str(exc), file=sys.stderr)
|
||||
return 1
|
||||
except KeyboardInterrupt:
|
||||
print("\nInterrupted.")
|
||||
finally:
|
||||
@@ -106,4 +197,4 @@ def main() -> int:
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
sys.exit(main())
|
||||
|
||||
@@ -17,7 +17,7 @@ from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
CURRENT_SCHEMA = 2
|
||||
SUPPORTED_SCHEMAS = (1, 2)
|
||||
SUPPORTED_SCHEMAS = (1, 2, 3)
|
||||
CHANNEL_IDS: tuple[str, ...] = ("fn", "fx", "fy")
|
||||
|
||||
CHANNEL_ALIASES: dict[str, str] = {
|
||||
@@ -500,6 +500,17 @@ def check(doc: dict | Calibration, *, pipeline: dict | None = None) -> CheckResu
|
||||
warns: list[str] = []
|
||||
notes: list[str] = []
|
||||
|
||||
if getattr(doc, "schema_version", None) == 3 or (isinstance(doc, dict) and doc.get("schema_version") == 3):
|
||||
from .calibration_models import parse_model
|
||||
|
||||
try:
|
||||
selected = parse_model(doc) if isinstance(doc, dict) else doc
|
||||
if pipeline is not None:
|
||||
selected.validate_for_context(pipeline)
|
||||
except (CalibrationError, ValueError, TypeError) as exc:
|
||||
return CheckResult(ok=False, errors=[str(exc)])
|
||||
return CheckResult(ok=True, errors=[], warnings=[])
|
||||
|
||||
if isinstance(doc, Calibration):
|
||||
data = doc.to_dict()
|
||||
channels_obj = doc.channels
|
||||
@@ -605,6 +616,14 @@ def parse(data: dict, *, source_path: Path | None = None) -> Calibration:
|
||||
if schema not in SUPPORTED_SCHEMAS:
|
||||
raise CalibrationError(f"unsupported schema_version: {schema}")
|
||||
|
||||
if schema == 3:
|
||||
from .calibration_models import parse_model
|
||||
|
||||
try:
|
||||
return parse_model(data, source_path=source_path)
|
||||
except (ValueError, TypeError, KeyError) as exc:
|
||||
raise CalibrationError(f"invalid schema 3 calibration: {exc}") from exc
|
||||
|
||||
unit = str(data.get("unit", "N"))
|
||||
channels_raw = data.get("channels")
|
||||
if not isinstance(channels_raw, dict):
|
||||
@@ -766,7 +785,7 @@ def load(
|
||||
explicit = source is not None and str(source).strip() != ""
|
||||
|
||||
if explicit:
|
||||
path = Path(str(source)).expanduser()
|
||||
path = select_calibration_path(cal_path=source)
|
||||
if not path.is_file():
|
||||
msg = f"calibration file not found: {path}"
|
||||
if strict or explicit:
|
||||
@@ -801,7 +820,7 @@ def load(
|
||||
try:
|
||||
return parse(data, source_path=path)
|
||||
except CalibrationError:
|
||||
if strict or explicit:
|
||||
if strict or explicit or (isinstance(data, dict) and (data.get("schema_version") == 3 or "model" in data)):
|
||||
raise
|
||||
return None
|
||||
|
||||
@@ -814,6 +833,26 @@ def load_for_device(
|
||||
pipeline: Mapping[str, Any] | None = None,
|
||||
) -> Calibration | None:
|
||||
"""Load ``<directory>/<sensor_type>/<device_id>.json``; missing file means raw."""
|
||||
path = select_calibration_path(calibration_dir=calibration_dir, sensor_type=sensor_type, device_id=device_id)
|
||||
if path is None:
|
||||
return None
|
||||
cal = load(path, strict=True)
|
||||
assert cal is not None
|
||||
return validate_for_device(cal, sensor_type, device_id, pipeline=pipeline)
|
||||
|
||||
|
||||
def select_calibration_path(*, cal_path=None, calibration_dir=None, sensor_type=None,
|
||||
device_id=None, default_path=None) -> Path | None:
|
||||
"""Select a file without guessing a model or falling back across device identities."""
|
||||
explicit = cal_path is not None and bool(str(cal_path).strip())
|
||||
if explicit and calibration_dir is not None:
|
||||
raise CalibrationError("cal_path and calibration_dir are mutually exclusive")
|
||||
if device_id is not None and not explicit and calibration_dir is None:
|
||||
raise CalibrationError("device_id requires calibration_dir or cal_path")
|
||||
if explicit:
|
||||
return Path(cal_path).expanduser()
|
||||
if calibration_dir is None:
|
||||
return Path(default_path) if default_path is not None else None
|
||||
root = Path(calibration_dir).expanduser()
|
||||
if not root.is_dir():
|
||||
raise CalibrationError(f"calibration directory not found: {root}")
|
||||
@@ -827,9 +866,7 @@ def load_for_device(
|
||||
return None
|
||||
if not path.is_file():
|
||||
raise CalibrationError(f"calibration path is not a file: {path}")
|
||||
cal = load(path, strict=True)
|
||||
assert cal is not None
|
||||
return validate_for_device(cal, sensor_type, device_id, pipeline=pipeline)
|
||||
return path
|
||||
|
||||
|
||||
def validate_for_device(
|
||||
@@ -843,6 +880,19 @@ def validate_for_device(
|
||||
path = cal.source_path or "<in-memory calibration>"
|
||||
if cal.identity.get("sensor_type") != sensor_type or cal.identity.get("sn") != device_id:
|
||||
raise CalibrationError(f"calibration identity mismatch: {path}")
|
||||
if cal.schema_version == 3:
|
||||
if cal.unit != "N" or not cal.channels:
|
||||
raise CalibrationError(f"calibration requires declared force axes in N: {path}")
|
||||
# Full feature/geometry validation happens when the processor binds its
|
||||
# actual context. Device construction only knows this legacy subset.
|
||||
if pipeline is not None:
|
||||
for key in ("stitch_resolution", "dis_mode", "hhd_downsample"):
|
||||
if pipeline.get(key) is not None and cal.pipeline.get(key) != pipeline[key]:
|
||||
raise CalibrationError(f"calibration pipeline mismatch ({key}): {path}")
|
||||
measurement, _ = filter_mismatches(cal.pipeline.get("filters"), pipeline.get("filters"))
|
||||
if measurement:
|
||||
raise CalibrationError(f"calibration pipeline mismatch ({'; '.join(measurement)}): {path}")
|
||||
return cal
|
||||
if cal.unit != "N" or any(cal.channels[axis].unit != "N" for axis in cal.channels):
|
||||
raise CalibrationError(f"calibration unit must be N: {path}")
|
||||
if set(CHANNEL_IDS) - cal.channels.keys():
|
||||
@@ -1050,10 +1100,13 @@ def write(
|
||||
path = Path(path)
|
||||
if path.exists() and not overwrite:
|
||||
raise FileExistsError(f"calibration file exists: {path}")
|
||||
if cal.schema_version == 3:
|
||||
# A moved manifest must not acquire unresolved or different attachments.
|
||||
parse(cal.to_dict(), source_path=path)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
doc = cal.to_dict()
|
||||
# Always emit schema 2 for new writes.
|
||||
doc["schema_version"] = CURRENT_SCHEMA
|
||||
# Preserve an explicitly selected model format; legacy builders still emit v2.
|
||||
doc["schema_version"] = 3 if cal.schema_version == 3 else CURRENT_SCHEMA
|
||||
path.write_text(
|
||||
json.dumps(doc, indent=2, ensure_ascii=False) + "\n",
|
||||
encoding="utf-8",
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
"""Measurement identities shared by SDK runtime and offline calibration exporters."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
from .calibration import filters_fingerprint
|
||||
|
||||
|
||||
def sha256_file(path) -> str:
|
||||
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
|
||||
|
||||
|
||||
def domain_fingerprint(domain) -> str:
|
||||
"""Bind a coordinate lookup to the actual unroll, not just its dimensions."""
|
||||
data = domain.to_dict() if hasattr(domain, "to_dict") else domain
|
||||
digest = hashlib.sha256()
|
||||
for key, dtype in (("map_x", "<f4"), ("map_y", "<f4"), ("valid_mask", "u1"), ("world_xyz", "<f4")):
|
||||
array = np.ascontiguousarray(data[key], dtype=dtype)
|
||||
digest.update(key.encode("ascii"))
|
||||
digest.update(str(array.shape).encode("ascii"))
|
||||
digest.update(array.tobytes())
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
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."""
|
||||
camera = config.get("camera", {})
|
||||
tracker = dict(config.get("tracker", {}))
|
||||
raw_downsample = tracker.get("hhd_downsample", 1)
|
||||
try:
|
||||
downsample = max(1, int(raw_downsample)) if raw_downsample is not None else 1
|
||||
except (TypeError, ValueError):
|
||||
downsample = 1
|
||||
tracker["hhd_downsample"] = downsample
|
||||
if source_scale is not None:
|
||||
source_scale = float(source_scale)
|
||||
measurement = filters_fingerprint(config.get("filters", {}) if filters is None else filters).get("measurement", {})
|
||||
# A missing EMA setting has the same effective meaning as the SDK default.
|
||||
measurement.setdefault("flow_ema_alpha", 1.0)
|
||||
result = {
|
||||
"stitch_resolution": camera.get("resolution_stitch"),
|
||||
"dis_mode": tracker.get("dis_mode"),
|
||||
"hhd_downsample": downsample,
|
||||
"filters": {"measurement": measurement},
|
||||
}
|
||||
if geometry_sha256 is not None:
|
||||
result["geometry_sha256"] = geometry_sha256
|
||||
if source_scale is not None:
|
||||
result["source_scale"] = float(source_scale)
|
||||
if coordinates is not None:
|
||||
result["coordinates"] = coordinates
|
||||
if output_domain_sha256 is not None:
|
||||
result["output_domain_sha256"] = output_domain_sha256
|
||||
effective = {
|
||||
"feature_version": "orisys_hhd_raw/1",
|
||||
"processing_backend": processing_backend or config.get("type", "compound_eye"),
|
||||
"camera": {key: camera.get(key) for key in ("resolution_raw", "resolution_stitch")},
|
||||
"tracker": tracker,
|
||||
"contact": config.get("contact", {}),
|
||||
"measurement": measurement,
|
||||
"geometry_sha256": geometry_sha256,
|
||||
"output_domain_sha256": output_domain_sha256,
|
||||
"source_scale": source_scale,
|
||||
"preprocessing": preprocessing or {},
|
||||
}
|
||||
encoded = json.dumps(effective, sort_keys=True, separators=(",", ":"), allow_nan=False)
|
||||
result["feature_signature_sha256"] = hashlib.sha256(encoded.encode("utf-8")).hexdigest()
|
||||
return result
|
||||
@@ -0,0 +1,307 @@
|
||||
"""Validated, versioned force models; acquisition and training stay outside here."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import math
|
||||
from collections.abc import Mapping
|
||||
from copy import deepcopy
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path, PurePosixPath, PureWindowsPath
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
from scipy.spatial import ConvexHull, QhullError
|
||||
|
||||
|
||||
def _require(condition: bool, message: str) -> None:
|
||||
if not condition:
|
||||
raise ValueError(message)
|
||||
|
||||
|
||||
def _keys(value: Any, required: set[str], optional: set[str] | None = None) -> None:
|
||||
_require(isinstance(value, dict), "expected a JSON object")
|
||||
_require(required <= value.keys(), f"missing fields: {required - value.keys()}")
|
||||
_require(value.keys() <= required | (optional or set()), "unknown semantic fields")
|
||||
|
||||
|
||||
def _number(value: Any) -> bool:
|
||||
return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value)
|
||||
|
||||
|
||||
def _sha(value: Any) -> bool:
|
||||
return isinstance(value, str) and len(value) == 64 and all(c in "0123456789abcdef" for c in value)
|
||||
|
||||
|
||||
def _area(polygon: np.ndarray) -> float:
|
||||
if len(polygon) < 3:
|
||||
return 0.0
|
||||
return abs(float(np.dot(polygon[:, 0], np.roll(polygon[:, 1], 1)) -
|
||||
np.dot(polygon[:, 1], np.roll(polygon[:, 0], 1)))) / 2
|
||||
|
||||
|
||||
def _cross(a: np.ndarray, b: np.ndarray) -> float:
|
||||
return float(a[0] * b[1] - a[1] * b[0])
|
||||
|
||||
|
||||
def _intersection_area(first: np.ndarray, second: np.ndarray) -> float:
|
||||
"""Clip two counterclockwise convex polygons, including shared edges."""
|
||||
polygon = list(first)
|
||||
for a, b in zip(second, np.roll(second, -1, axis=0)):
|
||||
previous = polygon
|
||||
polygon = []
|
||||
if not previous:
|
||||
break
|
||||
for start, end in zip(previous, previous[1:] + previous[:1]):
|
||||
ds, de = _cross(b - a, start - a), _cross(b - a, end - a)
|
||||
if (ds >= 0) != (de >= 0):
|
||||
polygon.append(start + (end - start) * ds / (ds - de))
|
||||
if de >= 0:
|
||||
polygon.append(end)
|
||||
return _area(np.asarray(polygon))
|
||||
|
||||
|
||||
def _mesh(params: dict[str, Any]) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
|
||||
points, gains, triangles = (params[k] for k in ("points_uv", "gains_N_per_raw", "triangles"))
|
||||
_require(isinstance(points, list) and len(points) >= 3, "spatial mesh needs at least 3 points")
|
||||
_require(all(isinstance(p, list) and len(p) == 2 and all(_number(x) for x in p) for p in points),
|
||||
"points_uv must be finite Nx2 numbers")
|
||||
_require(len({tuple(p) for p in points}) == len(points), "duplicate spatial points")
|
||||
_require(isinstance(gains, list) and len(gains) == len(points) and
|
||||
all(_number(g) and g > 0 for g in gains), "gains must be finite positive numbers")
|
||||
_require(isinstance(triangles, list) and bool(triangles), "triangles are required")
|
||||
_require(all(isinstance(t, list) and len(t) == 3 and
|
||||
all(type(i) is int and 0 <= i < len(points) for i in t) and len(set(t)) == 3
|
||||
for t in triangles),
|
||||
"invalid triangle indices")
|
||||
vertices = np.asarray(points, dtype=float)
|
||||
try:
|
||||
hull_area = ConvexHull(vertices).volume
|
||||
except QhullError as exc:
|
||||
raise ValueError("spatial mesh points are collinear") from exc
|
||||
tolerance = max(1.0, hull_area) * 1e-9
|
||||
polygons = []
|
||||
for indices in triangles:
|
||||
polygon = vertices[indices]
|
||||
signed = _cross(polygon[1] - polygon[0], polygon[2] - polygon[0])
|
||||
_require(abs(signed) / 2 > tolerance, "degenerate triangle")
|
||||
if signed < 0:
|
||||
polygon = polygon[::-1]
|
||||
_require(all(_intersection_area(polygon, other) <= tolerance for other in polygons),
|
||||
"overlapping spatial triangles")
|
||||
polygons.append(polygon)
|
||||
_require({i for triangle in triangles for i in triangle} == set(range(len(points))),
|
||||
"spatial mesh must use all support points")
|
||||
_require(abs(sum(_area(p) for p in polygons) - hull_area) <= tolerance,
|
||||
"triangulation must cover the complete convex hull")
|
||||
return vertices, np.asarray(gains, dtype=float), np.asarray(triangles, dtype=int)
|
||||
|
||||
|
||||
def _pipeline(data: Any, spatial: bool) -> None:
|
||||
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"})
|
||||
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")
|
||||
for key in ("dis_mode", "hhd_downsample"):
|
||||
_require(type(data[key]) is int and data[key] >= (1 if key == "hhd_downsample" else 0),
|
||||
f"invalid {key}")
|
||||
if "source_scale" in data:
|
||||
_require(_number(data["source_scale"]) and data["source_scale"] > 0, "invalid source_scale")
|
||||
_require(_sha(data["feature_signature_sha256"]), "invalid feature signature SHA-256")
|
||||
_keys(data["filters"], {"measurement"}, {"readout"})
|
||||
_keys(data["filters"]["measurement"], {"flow_ema_alpha"})
|
||||
alpha = data["filters"]["measurement"]["flow_ema_alpha"]
|
||||
_require(_number(alpha) and 0 < alpha <= 1, "invalid flow_ema_alpha")
|
||||
if "geometry_sha256" in data:
|
||||
_require(_sha(data["geometry_sha256"]), "invalid geometry SHA-256")
|
||||
if "output_domain_sha256" in data:
|
||||
_require(_sha(data["output_domain_sha256"]), "invalid output domain SHA-256")
|
||||
if "coordinates" in data:
|
||||
coord = data["coordinates"]
|
||||
_keys(coord, {"frame", "unit", "origin_xy", "map_sha256"})
|
||||
_require(coord["frame"] == "autocal_flat" and coord["unit"] == "px", "unsupported coordinates")
|
||||
_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")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CalibrationResult:
|
||||
values_N: dict[str, float | None]
|
||||
status: dict[str, str]
|
||||
|
||||
|
||||
class ModelCalibration:
|
||||
"""One validated model per stream. Current gain backends are stateless."""
|
||||
|
||||
schema_version = 3
|
||||
unit = "N"
|
||||
|
||||
def __init__(self, document: dict[str, Any], source_path: str | Path | None = None):
|
||||
self._document = deepcopy(document)
|
||||
self.source_path = Path(source_path).resolve() if source_path is not None else None
|
||||
self.identity = deepcopy(document["identity"])
|
||||
self.pipeline = deepcopy(document["pipeline"])
|
||||
self.provenance = deepcopy(document.get("provenance", {}))
|
||||
self.model_type = document["model"]["type"]
|
||||
self.input_profile = document["model"]["input_profile"]
|
||||
self.outputs = tuple(document["model"]["outputs"])
|
||||
self.channels = {axis: "N" for axis in self.outputs}
|
||||
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 = _mesh(self._params)
|
||||
self._load_artifacts(document["model"].get("artifacts", []))
|
||||
|
||||
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",
|
||||
"unsupported artifact role")
|
||||
name = artifact["path"]
|
||||
_require(isinstance(name, str) and bool(name) and ":" not in name and
|
||||
not PureWindowsPath(name).is_absolute() and not PurePosixPath(name).is_absolute() and
|
||||
".." not in PurePosixPath(name.replace("\\", "/")).parts,
|
||||
"artifact path must be relative without parent traversal")
|
||||
_require(self.source_path is not None, "artifact requires calibration source_path")
|
||||
path = (self.source_path.parent / name).resolve()
|
||||
_require(path.is_relative_to(self.source_path.parent) and path.is_file(),
|
||||
"artifact missing or escapes calibration directory")
|
||||
digest = hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
_require(_sha(artifact["sha256"]) and digest == artifact["sha256"], "artifact SHA-256 mismatch")
|
||||
_require(self.pipeline["coordinates"]["map_sha256"] == digest,
|
||||
"coordinate map identity does not match artifact SHA-256")
|
||||
self._artifact_hashes.append(digest)
|
||||
try:
|
||||
with np.load(path, allow_pickle=False) as archive:
|
||||
keys = set(archive.files)
|
||||
_require({"uv_by_output_px", "valid_mask"} <= keys <=
|
||||
{"uv_by_output_px", "valid_mask", "origin_xy", "flat_mapping_sha256"},
|
||||
"invalid coordinate_map keys")
|
||||
uv, mask = archive["uv_by_output_px"], archive["valid_mask"]
|
||||
_require(uv.ndim == 3 and uv.shape[2] == 2 and mask.shape == uv.shape[:2],
|
||||
"invalid coordinate_map shape")
|
||||
_require(mask.dtype == np.bool_ and np.issubdtype(uv.dtype, np.floating),
|
||||
"invalid coordinate_map dtype")
|
||||
_require(np.isfinite(uv[mask]).all(), "non-finite mapped coordinates")
|
||||
if "origin_xy" in keys:
|
||||
_require(np.array_equal(archive["origin_xy"], self.pipeline["coordinates"]["origin_xy"]),
|
||||
"coordinate_map origin mismatch")
|
||||
self.coordinate_map, self.coordinate_valid_mask = uv.copy(), mask.copy()
|
||||
self.coordinate_map.flags.writeable = False
|
||||
self.coordinate_valid_mask.flags.writeable = False
|
||||
except (OSError, TypeError) as exc:
|
||||
raise ValueError("invalid coordinate_map artifact") from exc
|
||||
|
||||
@property
|
||||
def fingerprint(self) -> str:
|
||||
content = json.dumps(self._document, sort_keys=True, separators=(",", ":"), allow_nan=False)
|
||||
return hashlib.sha256((content + "".join(self._artifact_hashes)).encode("utf-8")).hexdigest()
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return deepcopy(self._document)
|
||||
|
||||
def validate_for_context(self, pipeline: Mapping[str, Any]) -> None:
|
||||
for key, value in self.pipeline.items():
|
||||
if key == "filters":
|
||||
actual = pipeline.get(key, {})
|
||||
_require(isinstance(actual, Mapping) and actual.get("measurement") == value["measurement"],
|
||||
"calibration pipeline mismatch: filters.measurement")
|
||||
else:
|
||||
_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."""
|
||||
|
||||
def convert(self, fn: float, fx: float, fy: float) -> None:
|
||||
"""Reject legacy scalar calls which lack frame, position and zero features."""
|
||||
from .calibration import CalibrationError
|
||||
|
||||
raise CalibrationError("schema 3 requires frame features; use predict/apply_calibration_frame")
|
||||
|
||||
def predict(self, frame_id: int, timestamp_s: float | None, features: Mapping[str, Any],
|
||||
reference_id: Any = None) -> CalibrationResult:
|
||||
def invalid(reason: str) -> CalibrationResult:
|
||||
return CalibrationResult({a: None for a in self.outputs}, {a: reason for a in self.outputs})
|
||||
|
||||
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 features.get("zero_valid") is not True:
|
||||
return invalid("zero_unavailable")
|
||||
if features.get("contact_found") is False:
|
||||
return invalid("no_contact")
|
||||
values = {}
|
||||
if self.model_type == "spatial_gain/1":
|
||||
if features.get("contact_found") is not True:
|
||||
return invalid("invalid_input")
|
||||
if features.get("single_contact", True) is not True or features.get("contact_count", 1) != 1:
|
||||
return invalid("invalid_input")
|
||||
if features.get("mapping_valid") is not True:
|
||||
return invalid("coordinate_unmapped")
|
||||
if not all(_number(features.get(k)) for k in ("delta_raw", "u", "v")):
|
||||
return invalid("invalid_input")
|
||||
query = np.asarray([features["u"], features["v"]])
|
||||
for indices in self._triangles:
|
||||
triangle = self._points[indices]
|
||||
weights = np.linalg.solve((triangle[1:] - triangle[0]).T, query - triangle[0])
|
||||
weights = np.asarray([1 - sum(weights), *weights])
|
||||
if np.min(weights) >= -1e-10:
|
||||
values["fn"] = float(np.dot(weights, self._gains[indices]) * features["delta_raw"])
|
||||
break
|
||||
if not values:
|
||||
return invalid("outside_coverage")
|
||||
else:
|
||||
deltas = features.get("delta_raw_by_axis", features.get("force_delta_raw", {}))
|
||||
if not isinstance(deltas, Mapping):
|
||||
return invalid("invalid_input")
|
||||
for axis in self.outputs:
|
||||
delta = deltas.get(axis, features.get("delta_raw") if axis == "fn" else None)
|
||||
values[axis] = delta * self._params["gains_N_per_raw"][axis] if _number(delta) else None
|
||||
low, high = self._params["valid_force_range_N"]
|
||||
statuses = {}
|
||||
for axis, value in values.items():
|
||||
if value is None or not _number(value):
|
||||
values[axis], statuses[axis] = None, "invalid_input"
|
||||
else:
|
||||
statuses[axis] = "ok" if low <= value <= high else "out_of_range"
|
||||
return CalibrationResult(values, statuses)
|
||||
|
||||
|
||||
def parse_model(data: dict[str, Any], source_path: str | Path | None = None) -> ModelCalibration:
|
||||
"""Parse schema 3 without executing document-specified code or installing backends."""
|
||||
_keys(data, {"schema_version", "unit", "identity", "pipeline", "model"}, {"provenance"})
|
||||
_require(type(data["schema_version"]) is int and data["schema_version"] == 3, "expected schema_version 3")
|
||||
_require(data["unit"] == "N", "schema 3 unit must be N")
|
||||
_keys(data["identity"], {"sensor_type", "sn"})
|
||||
_require(all(isinstance(v, str) and bool(v.strip()) for v in data["identity"].values()), "invalid identity")
|
||||
_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"),
|
||||
"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)
|
||||
params = model["params"]
|
||||
_keys(params, {"gains_N_per_raw", "valid_force_range_N"} | ({"points_uv", "triangles"} if spatial else set()))
|
||||
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")
|
||||
return ModelCalibration(data, source_path)
|
||||
@@ -105,6 +105,17 @@ class TactileProcessor:
|
||||
def force_units(self) -> tuple[str, str, str]:
|
||||
return self._sensor.force_units
|
||||
|
||||
@property
|
||||
def calibration_status(self):
|
||||
return self._sensor.calibration_status
|
||||
|
||||
@property
|
||||
def calibration_result(self):
|
||||
return self._sensor.calibration_result
|
||||
|
||||
def set_calibration_reference(self, raw, *, reference_id):
|
||||
return self._sensor.set_calibration_reference(raw, reference_id=reference_id)
|
||||
|
||||
def push_frame(self, frame, *, already_gray=False, valid_mask=None):
|
||||
"""Feed one processed tactile frame into the analysis pipeline."""
|
||||
mask = self._default_valid_mask if valid_mask is None else valid_mask
|
||||
@@ -305,6 +316,26 @@ class FingerRuntime:
|
||||
calibration_dir=calibration_dir, device_id=device_id,
|
||||
verbose=verbose, mirror=mirror,
|
||||
)
|
||||
if getattr(getattr(sensor, "calibration", None), "schema_version", None) == 3:
|
||||
from ..calibration_context import (
|
||||
build_context,
|
||||
domain_fingerprint,
|
||||
sha256_file,
|
||||
)
|
||||
|
||||
model = sensor.calibration
|
||||
uv_map = getattr(model, "coordinate_map", None)
|
||||
if model.input_profile == "fn_delta_autocal_uv/1":
|
||||
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")
|
||||
sensor.bind_calibration_context(build_context(
|
||||
sensor._config, sensor.filters_state(),
|
||||
geometry_sha256=sha256_file(mesh_path), source_scale=geometry.source_scale,
|
||||
coordinates=model.pipeline.get("coordinates"),
|
||||
output_domain_sha256=domain_fingerprint(domain),
|
||||
processing_backend="uv_unroll",
|
||||
preprocessing={"flip_source_vertical": flip_source_vertical, "mirror": mirror},
|
||||
))
|
||||
|
||||
capture_config = CaptureConfig.from_camera_dict(
|
||||
video,
|
||||
@@ -351,6 +382,15 @@ class FingerRuntime:
|
||||
def valid_mask(self):
|
||||
return self.frame_processor.valid_mask
|
||||
|
||||
def set_calibration_reference(self, raw, *, reference_id):
|
||||
"""Set an explicitly measured unloaded reference for this finger stream."""
|
||||
self.tactile.raw_sensor.set_calibration_reference(raw, reference_id=reference_id)
|
||||
|
||||
@property
|
||||
def calibration_status(self):
|
||||
"""Copy the current per-axis calibration validity state."""
|
||||
return dict(self.tactile.raw_sensor.calibration_status)
|
||||
|
||||
def read_source(self):
|
||||
if self.capture_source is not None and self.capture_source.running:
|
||||
return self.capture_source.read(copy=True, wait_s=1.0)
|
||||
|
||||
@@ -154,6 +154,8 @@ class FrameSnapshot:
|
||||
#: compare it with ``image.shape`` to see how much the frame was reduced.
|
||||
image_size_raw: tuple[int, int] | None = None
|
||||
source_metadata: Any = None
|
||||
force_units: tuple[str, str, str] | None = None
|
||||
calibration_status: dict[str, str] | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -654,6 +656,8 @@ class StreamPipeline:
|
||||
shear_x_raw=float(sensor._force_raw[1]),
|
||||
shear_y_raw=float(sensor._force_raw[2]),
|
||||
torque=float(getattr(sensor, "ftor", 0.0)),
|
||||
force_units=tuple(getattr(sensor, "force_units", (getattr(sensor, "force_unit", "raw"),) * 3)),
|
||||
calibration_status=dict(getattr(sensor, "calibration_status", {})),
|
||||
# Both already private per-frame copies owned by this snapshot.
|
||||
image=frame if "image" in include else None,
|
||||
flow=flow if "flow" in include else None,
|
||||
|
||||
+225
-3
@@ -182,6 +182,23 @@ class Sensor:
|
||||
config_fps=config_fps,
|
||||
verbose=self.verbose,
|
||||
)
|
||||
self._model_frame_id = 0
|
||||
self._calibration_result_frame = None
|
||||
self._calibration_reference = {}
|
||||
self._calibration_reference_id = None
|
||||
self._model_readouts = {}
|
||||
self.calibration_result = None
|
||||
self.calibration_status = {}
|
||||
self._calibration_context = None
|
||||
if self._uses_calibration_model():
|
||||
self._reset_calibration_session()
|
||||
if (self._uses_calibration_model() and self.calibration.input_profile == "force_delta/1"
|
||||
and not self._external_only):
|
||||
from .calibration_context import build_context
|
||||
|
||||
self.bind_calibration_context(build_context(
|
||||
self._config, self.filters_state(), preprocessing={"mirror": self.mirror},
|
||||
))
|
||||
|
||||
if self.verbose:
|
||||
if self.calibration is not None and self.calibration.source_path is not None:
|
||||
@@ -726,7 +743,15 @@ class Sensor:
|
||||
self._force_raw = (float(fn), float(fx), float(fy))
|
||||
self.ftor = float(ftor)
|
||||
self._force_readout.update_sample_rate()
|
||||
self._apply_force_filter()
|
||||
if self._uses_calibration_model():
|
||||
self._model_frame_id += 1
|
||||
self.compute_contact_valid = False
|
||||
self.centroid_found = False
|
||||
self._force_readout.apply(self._force_raw)
|
||||
self._force_filtered_raw = self._force_readout.filtered_raw
|
||||
self._force_raw_unfiltered = self._force_raw
|
||||
else:
|
||||
self._apply_force_filter()
|
||||
|
||||
vnormal_ds = self.__hhd.get_normal()
|
||||
vshear_ds = self.__hhd.get_shear()
|
||||
@@ -762,6 +787,14 @@ class Sensor:
|
||||
|
||||
self.compute_deformation_valid = True
|
||||
|
||||
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":
|
||||
self.compute_contact()
|
||||
else:
|
||||
self.apply_calibration_frame(frame_id=self._model_frame_id)
|
||||
|
||||
return True
|
||||
|
||||
def compute_contact(self, vnormal=None):
|
||||
@@ -774,6 +807,11 @@ class Sensor:
|
||||
Returns:
|
||||
dict: Contact result with centroid, contour, and depth_map
|
||||
"""
|
||||
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
|
||||
and self._calibration_result_frame == self._model_frame_id):
|
||||
return self._model_contact_result
|
||||
multiple_foe = False
|
||||
|
||||
# Use provided vnormal or fall back to internal state
|
||||
@@ -807,6 +845,13 @@ class Sensor:
|
||||
print("Warning: contact centroid lies outside the valid UV mask.")
|
||||
|
||||
self.compute_contact_valid = True
|
||||
if self._uses_calibration_model():
|
||||
self._model_contact_result = result
|
||||
uv = self._calibration_contact_uv()
|
||||
self.apply_calibration_frame(
|
||||
frame_id=self._model_frame_id, contact_found=bool(self.centroid_found),
|
||||
uv=uv, mapping_valid=uv is not None,
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
@@ -871,6 +916,8 @@ class Sensor:
|
||||
# session must not report the previous one's reading.
|
||||
self._force_unfiltered = None
|
||||
self._force_raw_unfiltered = None
|
||||
self._reset_calibration_session()
|
||||
self.compute_deformation_valid = False
|
||||
if self.img is not None:
|
||||
self.initial_img = self.img.copy()
|
||||
|
||||
@@ -950,6 +997,10 @@ class Sensor:
|
||||
alpha = float(flow_ema_alpha)
|
||||
if not (0.0 < alpha <= 1.0):
|
||||
raise ValueError("flow_ema_alpha must be in (0, 1]")
|
||||
if self._uses_calibration_model() and alpha != self._force_readout.flow_ema_alpha:
|
||||
raise cal_mod.CalibrationError(
|
||||
"flow_ema_alpha changes schema 3 measurement identity; load a matching pipeline"
|
||||
)
|
||||
if self.calibration is not None:
|
||||
mismatches, _ = cal_mod.filter_mismatches(
|
||||
self.calibration.pipeline.get("filters"),
|
||||
@@ -965,8 +1016,17 @@ class Sensor:
|
||||
flow_ema_alpha=flow_ema_alpha,
|
||||
force_median_window=force_median_window,
|
||||
)
|
||||
for readout in getattr(self, "_model_readouts", {}).values():
|
||||
readout.configure(
|
||||
lowpass_cutoff_hz=lowpass_cutoff_hz,
|
||||
lowpass_sample_rate_hz=lowpass_sample_rate_hz,
|
||||
lowpass_order=lowpass_order,
|
||||
lowpass_enabled=lowpass_enabled,
|
||||
force_median_window=force_median_window,
|
||||
)
|
||||
if filter_changed and self.compute_deformation_valid:
|
||||
self._apply_force_filter()
|
||||
if not self._uses_calibration_model():
|
||||
self._apply_force_filter()
|
||||
|
||||
return self
|
||||
|
||||
@@ -1139,6 +1199,7 @@ class Sensor:
|
||||
def set_calibration(
|
||||
self,
|
||||
cal: "cal_mod.Calibration | str | Path | None",
|
||||
*, context=None,
|
||||
) -> None:
|
||||
"""Attach or replace the active force calibration (live swap after save)."""
|
||||
if cal is None:
|
||||
@@ -1146,8 +1207,12 @@ class Sensor:
|
||||
self.device_calibration_missing = False
|
||||
self._force_cal_sensor_name = None
|
||||
self._force_cal_attempts = None
|
||||
self._reset_calibration_session()
|
||||
self._calibration_context = None
|
||||
self._force_readout.reset()
|
||||
self._apply_force_filter()
|
||||
return
|
||||
if isinstance(cal, cal_mod.Calibration):
|
||||
if isinstance(cal, cal_mod.Calibration) or getattr(cal, "schema_version", None) == 3:
|
||||
selected = cal
|
||||
else:
|
||||
selected = cal_mod.load(cal, strict=True)
|
||||
@@ -1156,9 +1221,30 @@ class Sensor:
|
||||
selected = cal_mod.validate_for_device(
|
||||
selected, family, self.device_id, pipeline=pipeline
|
||||
)
|
||||
if getattr(selected, "schema_version", None) == 3:
|
||||
context = context if context is not None else getattr(self, "_calibration_context", None)
|
||||
if context is None and selected.input_profile == "force_delta/1":
|
||||
from .calibration_context import build_context
|
||||
|
||||
context = build_context(
|
||||
self._config, self.filters_state(), preprocessing={"mirror": self.mirror},
|
||||
)
|
||||
if context is None:
|
||||
raise cal_mod.CalibrationError("actual pipeline context required before replacing model")
|
||||
try:
|
||||
selected.validate_for_context(context)
|
||||
except ValueError as exc:
|
||||
raise cal_mod.CalibrationError(str(exc)) from exc
|
||||
else:
|
||||
context = None
|
||||
self.calibration = selected
|
||||
self._calibration_context = context
|
||||
self.device_calibration_missing = False
|
||||
self._force_cal_attempts = None
|
||||
self._reset_calibration_session()
|
||||
if not self._uses_calibration_model():
|
||||
self._force_readout.reset()
|
||||
self._apply_force_filter()
|
||||
|
||||
def _apply_force_filter(self):
|
||||
convert = self.calibration.convert if self.calibration is not None else None
|
||||
@@ -1169,6 +1255,142 @@ class Sensor:
|
||||
self._force_raw_unfiltered = readout.raw_unfiltered
|
||||
self._force_unfiltered = readout.unfiltered_newton
|
||||
|
||||
def _uses_calibration_model(self):
|
||||
return getattr(self.calibration, "schema_version", None) == 3
|
||||
|
||||
def bind_calibration_context(self, context):
|
||||
"""Verify the actual feature pipeline before allowing schema 3 inference."""
|
||||
if not self._uses_calibration_model():
|
||||
raise cal_mod.CalibrationError("calibration context requires schema 3")
|
||||
try:
|
||||
self.calibration.validate_for_context(context)
|
||||
except ValueError as exc:
|
||||
raise cal_mod.CalibrationError(str(exc)) from exc
|
||||
self._calibration_context = context
|
||||
|
||||
def _reset_calibration_session(self):
|
||||
self._calibration_reference = {}
|
||||
self._calibration_reference_id = None
|
||||
self._calibration_result_frame = None
|
||||
self.calibration_result = None
|
||||
self.calibration_status = {}
|
||||
self.compute_contact_valid = False
|
||||
self.centroid_found = False
|
||||
self.centroid = (0, 0)
|
||||
self._model_contact_result = None
|
||||
for readout in getattr(self, "_model_readouts", {}).values():
|
||||
readout.reset()
|
||||
if self._uses_calibration_model():
|
||||
self.calibration.reset(None)
|
||||
self.calibration_status = {axis: "zero_unavailable" for axis in self.calibrated_axes}
|
||||
values = [float("nan") if axis in self.calibrated_axes else raw
|
||||
for axis, raw in zip(cal_mod.CHANNEL_IDS, self._force_raw)]
|
||||
self.fnormal, self.fshearx, self.fsheary = values
|
||||
self._force_unfiltered = tuple(values)
|
||||
|
||||
def set_calibration_reference(self, raw, *, reference_id):
|
||||
"""Declare a measured unloaded raw reference for this stream, not the model file."""
|
||||
if not isinstance(raw, dict) or not reference_id:
|
||||
raise ValueError("raw must map axes to finite values and reference_id is required")
|
||||
values = {}
|
||||
for axis, value in raw.items():
|
||||
if axis not in cal_mod.CHANNEL_IDS or not np.isfinite(float(value)):
|
||||
raise ValueError("reference axes and values must be valid and finite")
|
||||
values[axis] = float(value)
|
||||
self._reset_calibration_session()
|
||||
self._calibration_reference = values
|
||||
self._calibration_reference_id = str(reference_id)
|
||||
if self._uses_calibration_model():
|
||||
self.calibration.reset(self._calibration_reference_id)
|
||||
|
||||
def _calibration_contact_uv(self):
|
||||
if not self.centroid_found:
|
||||
return None
|
||||
uv_map = getattr(self.calibration, "coordinate_map", None)
|
||||
valid = getattr(self.calibration, "coordinate_valid_mask", None)
|
||||
if uv_map is None or valid is None:
|
||||
return None
|
||||
if not np.all(np.isfinite(self.centroid)):
|
||||
return None
|
||||
x, y = np.rint(self.centroid).astype(int)
|
||||
if not (0 <= y < valid.shape[0] and 0 <= x < valid.shape[1] and valid[y, x]):
|
||||
return None
|
||||
return tuple(float(value) for value in uv_map[y, x])
|
||||
|
||||
def apply_calibration_frame(
|
||||
self, *, frame_id, timestamp_s=None, raw=None, uv=None,
|
||||
contact_found=None, mapping_valid=False, single_contact=True,
|
||||
):
|
||||
"""Infer once from an unfiltered raw triple and same-frame optional contact UV."""
|
||||
if not self._uses_calibration_model():
|
||||
raise cal_mod.CalibrationError("apply_calibration_frame requires schema 3")
|
||||
if self._calibration_context is None:
|
||||
raise cal_mod.CalibrationError("bind the actual calibration context before inference")
|
||||
if type(frame_id) is not int:
|
||||
raise ValueError("frame_id must be an integer")
|
||||
if frame_id == self._calibration_result_frame:
|
||||
return self.calibration_result
|
||||
if self._calibration_result_frame is not None and frame_id < self._calibration_result_frame:
|
||||
self._reset_calibration_session()
|
||||
raw_values = tuple(self._force_raw if raw is None else raw)
|
||||
if len(raw_values) != 3:
|
||||
raise ValueError("raw must contain FN, FX and FY")
|
||||
if raw is not None:
|
||||
self._force_raw = tuple(float(value) for value in raw_values)
|
||||
if np.all(np.isfinite(self._force_raw)):
|
||||
self._force_readout.apply(self._force_raw)
|
||||
self._force_filtered_raw = self._force_readout.filtered_raw
|
||||
raw_by_axis = dict(zip(cal_mod.CHANNEL_IDS, raw_values))
|
||||
reference = self._calibration_reference
|
||||
deltas = {axis: float(value) - reference.get(axis, float("nan"))
|
||||
for axis, value in raw_by_axis.items()}
|
||||
axes = self.calibrated_axes
|
||||
features = {
|
||||
"delta_raw": deltas["fn"], "force_delta_raw": deltas,
|
||||
"zero_valid": all(axis in reference for axis in axes),
|
||||
"mapping_valid": bool(mapping_valid), "single_contact": single_contact,
|
||||
"u": None if uv is None else uv[0], "v": None if uv is None else uv[1],
|
||||
}
|
||||
if contact_found is not None:
|
||||
features["contact_found"] = bool(contact_found)
|
||||
result = self.calibration.predict(
|
||||
frame_id=frame_id, timestamp_s=timestamp_s,
|
||||
features=features, reference_id=self._calibration_reference_id,
|
||||
)
|
||||
filtered = list(self._force_readout.filtered_raw or raw_values)
|
||||
unfiltered = list(raw_values)
|
||||
for index, axis in enumerate(cal_mod.CHANNEL_IDS):
|
||||
if axis not in axes:
|
||||
continue
|
||||
value = result.values_N.get(axis)
|
||||
unfiltered[index] = float("nan") if value is None else float(value)
|
||||
filtered[index] = unfiltered[index]
|
||||
if value is not None and np.isfinite(value):
|
||||
readout = self._model_readouts.get(axis)
|
||||
if readout is None:
|
||||
readout = ForceReadout(self._config.get("filters"),
|
||||
config_fps=self._force_readout.config_fps)
|
||||
base = self._force_readout
|
||||
readout.configure(
|
||||
lowpass_sample_rate_hz=base.lowpass_sample_rate_hz,
|
||||
lowpass_cutoff_hz=base.lowpass_cutoff_hz,
|
||||
lowpass_order=base.lowpass_order,
|
||||
lowpass_enabled=base.lowpass_enabled,
|
||||
force_median_window=base.force_median_window,
|
||||
)
|
||||
self._model_readouts[axis] = readout
|
||||
base = self._force_readout
|
||||
if readout.lowpass_sample_rate_hz != base.lowpass_sample_rate_hz:
|
||||
readout.configure(lowpass_sample_rate_hz=base.lowpass_sample_rate_hz)
|
||||
filtered[index] = readout.apply((value, value, value))[0]
|
||||
self.fnormal, self.fshearx, self.fsheary = filtered
|
||||
self._force_raw_unfiltered = tuple(float(value) for value in raw_values)
|
||||
self._force_unfiltered = tuple(unfiltered)
|
||||
self.calibration_result = result
|
||||
self.calibration_status = dict(result.status)
|
||||
self._calibration_result_frame = frame_id
|
||||
return result
|
||||
|
||||
# ------------------------------------------------------- filter contract
|
||||
def filters_state(self) -> dict:
|
||||
"""What the filter chain is **actually** doing, for records and audits.
|
||||
|
||||
@@ -253,6 +253,13 @@ def create_processor(
|
||||
elif verbose:
|
||||
print(f"[processor] using the supplied sensor for {profile.profile_id}")
|
||||
|
||||
# A resolved selection (including explicit NONE) is authoritative. Otherwise
|
||||
# Sensor's bundle auto-load can silently defeat the caller's selected file.
|
||||
if hasattr(sensor, "set_calibration"):
|
||||
sensor.set_calibration(profile.calibration.document)
|
||||
elif profile.calibration.document is not None:
|
||||
raise ProcessorCreationError("supplied sensor cannot apply the selected calibration")
|
||||
|
||||
if backend_id == BACKEND_COMPOUND_EYE:
|
||||
config_stitcher = profile.config.get("stitcher")
|
||||
stitcher = Stitcher(
|
||||
|
||||
@@ -119,12 +119,13 @@ def resolve_calibration(
|
||||
)
|
||||
|
||||
if mode is CalibrationSelection.EXPLICIT:
|
||||
path = explicit
|
||||
path = cal_mod.select_calibration_path(cal_path=explicit)
|
||||
else:
|
||||
path = declared_default
|
||||
if path is None:
|
||||
fallback = bundle_root / "force_calibration.json"
|
||||
path = fallback if fallback.is_file() else None
|
||||
path = cal_mod.select_calibration_path(default_path=path)
|
||||
|
||||
if path is None:
|
||||
# Not an error: a profile may legitimately ship no calibration. Calibrated force is
|
||||
@@ -197,6 +198,12 @@ def resolve_calibration(
|
||||
|
||||
identity = _identity_of(document)
|
||||
declared_type = str(identity.get("sensor_type") or "")
|
||||
if document.schema_version == 3 and declared_type != product_family:
|
||||
return ResolvedCalibration(
|
||||
selection=mode, path=path,
|
||||
issues=(Issue(code="calibration.model-family-mismatch",
|
||||
message="selected model belongs to another product family", path=path),),
|
||||
)
|
||||
if declared_type:
|
||||
applicability = "declared"
|
||||
if declared_type != product_family:
|
||||
@@ -273,6 +280,8 @@ def calibration_fingerprint(document: "cal_mod.Calibration", path: Path) -> str:
|
||||
a different directory must produce the same fingerprint.
|
||||
"""
|
||||
try:
|
||||
if getattr(document, "schema_version", None) == 3:
|
||||
return document.fingerprint
|
||||
payload = document.to_dict()
|
||||
except Exception: # noqa: BLE001 - fall back to the raw bytes rather than fail
|
||||
payload = path.read_bytes().decode("utf-8", errors="replace")
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
"""Feature identities use effective measurement settings, never display metadata."""
|
||||
|
||||
from copy import deepcopy
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from orisys.calibration_context import build_context, domain_fingerprint
|
||||
|
||||
|
||||
def config():
|
||||
return {"camera": {"resolution_raw": [1280, 720], "resolution_stitch": [700, 750]},
|
||||
"tracker": {"dis_mode": 4}}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("raw,expected", [(None, 1), ("bad", 1), (0, 1), (-3, 1), ("3", 3), (3.7, 3)])
|
||||
def test_hhd_uses_actual_sensor_normalization(raw, expected):
|
||||
data = config()
|
||||
data["tracker"]["hhd_downsample"] = raw
|
||||
assert build_context(data)["hhd_downsample"] == expected
|
||||
|
||||
|
||||
def test_default_values_equal_explicit_effective_settings():
|
||||
implicit = build_context(config())
|
||||
data = config()
|
||||
data["tracker"]["hhd_downsample"] = 1
|
||||
explicit = build_context(data, {"flow_ema_alpha": 1.0})
|
||||
assert implicit["hhd_downsample"] == 1
|
||||
assert implicit["filters"]["measurement"] == {"flow_ema_alpha": 1.0}
|
||||
assert implicit["feature_signature_sha256"] == explicit["feature_signature_sha256"]
|
||||
|
||||
|
||||
def test_measurement_filters_change_identity_but_readout_and_paths_do_not():
|
||||
data = config()
|
||||
original = build_context(data)
|
||||
assert build_context(data, {"flow_ema_alpha": 0.5})["feature_signature_sha256"] != original["feature_signature_sha256"]
|
||||
data.update(name="other", video="D:/other.mp4", cal_path="E:/calibration.json")
|
||||
assert build_context(data, {"force_median_window": 7, "force_lowpass_cutoff_hz": 2}) == original
|
||||
|
||||
|
||||
def test_explicit_empty_filter_mapping_overrides_config_defaults():
|
||||
data = config()
|
||||
data["filters"] = {"flow_ema_alpha": 0.5}
|
||||
assert build_context(data, {})["filters"]["measurement"] == {"flow_ema_alpha": 1.0}
|
||||
|
||||
|
||||
def test_source_scale_numeric_types_have_identical_identity():
|
||||
assert build_context(config(), source_scale=1) == build_context(config(), source_scale=1.0)
|
||||
|
||||
|
||||
def test_domain_fingerprint_binds_map_content_not_only_dimensions():
|
||||
domain = {"map_x": np.zeros((2, 3)), "map_y": np.zeros((2, 3)),
|
||||
"valid_mask": np.ones((2, 3), dtype=bool), "world_xyz": np.zeros((2, 3, 3))}
|
||||
original = domain_fingerprint(domain)
|
||||
for key in domain:
|
||||
changed = deepcopy(domain)
|
||||
changed[key].flat[0] = not changed[key].flat[0] if key == "valid_mask" else 1
|
||||
assert domain_fingerprint(changed) != original
|
||||
assert domain_fingerprint({key: value.copy() for key, value in domain.items()}) == original
|
||||
@@ -0,0 +1,179 @@
|
||||
"""Schema 3 model validation and deterministic inference."""
|
||||
|
||||
import hashlib
|
||||
from copy import deepcopy
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from orisys.calibration import CalibrationError
|
||||
from orisys.calibration_models import parse_model
|
||||
|
||||
|
||||
def document():
|
||||
return {
|
||||
"schema_version": 3, "unit": "N",
|
||||
"identity": {"sensor_type": "finger", "sn": "DD02-GL"},
|
||||
"pipeline": {
|
||||
"stitch_resolution": [700, 750], "dis_mode": 4,
|
||||
"hhd_downsample": 3, "source_scale": 0.5,
|
||||
"filters": {"measurement": {"flow_ema_alpha": 1.0}},
|
||||
"feature_signature_sha256": "a" * 64,
|
||||
"geometry_sha256": "b" * 64,
|
||||
"coordinates": {"frame": "autocal_flat", "unit": "px",
|
||||
"origin_xy": [511, 1021], "map_sha256": "c" * 64},
|
||||
},
|
||||
"model": {
|
||||
"type": "spatial_gain/1", "input_profile": "fn_delta_autocal_uv/1",
|
||||
"output_frame": "sdk_force/1", "outputs": ["fn"],
|
||||
"params": {"points_uv": [[0, 0], [10, 0], [0, 10]],
|
||||
"gains_N_per_raw": [0.001, 0.002, 0.003],
|
||||
"triangles": [[0, 1, 2]], "valid_force_range_N": [0, 3]},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def features(**overrides):
|
||||
return dict(delta_raw=1000, u=2.5, v=2.5, contact_found=True,
|
||||
zero_valid=True, mapping_valid=True, **overrides)
|
||||
|
||||
|
||||
def test_interpolates_gain_not_sensitivity_and_roundtrips():
|
||||
data = document()
|
||||
model = parse_model(data)
|
||||
result = model.predict(1, 0.1, features(), "zero-1")
|
||||
assert result.values_N == {"fn": pytest.approx(1.75)}
|
||||
assert result.status == {"fn": "ok"}
|
||||
assert model.to_dict() == data
|
||||
assert "fx" not in model.channels
|
||||
data["model"]["params"]["gains_N_per_raw"][0] = 999
|
||||
assert model.predict(2, 0.2, features(), "zero-1").values_N["fn"] == 1.75
|
||||
|
||||
|
||||
def test_legacy_scalar_convert_explicitly_rejected():
|
||||
with pytest.raises(CalibrationError, match="schema 3 requires frame features"):
|
||||
parse_model(document()).convert(1, 2, 3)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("change,status", [
|
||||
({"u": 20}, "outside_coverage"), ({"contact_found": False}, "no_contact"),
|
||||
({"zero_valid": False}, "zero_unavailable"),
|
||||
({"mapping_valid": False}, "coordinate_unmapped"),
|
||||
({"delta_raw": float("nan")}, "invalid_input"),
|
||||
({"delta_raw": "1000"}, "invalid_input"),
|
||||
({"single_contact": False}, "invalid_input"),
|
||||
])
|
||||
def test_frame_invalidity(change, status):
|
||||
values = features()
|
||||
values.update(change)
|
||||
result = parse_model(document()).predict(1, 0.1, values, "zero")
|
||||
assert result.status == {"fn": status}
|
||||
assert result.values_N == {"fn": None}
|
||||
|
||||
|
||||
def test_out_of_range_retains_value():
|
||||
values = features()
|
||||
values["delta_raw"] = 3000
|
||||
result = parse_model(document()).predict(1, 0, values, "zero")
|
||||
assert result.values_N["fn"] == 5.25
|
||||
assert result.status["fn"] == "out_of_range"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mutate", [
|
||||
lambda d: d.update(schema_version=True),
|
||||
lambda d: d.update(extra="ignored"),
|
||||
lambda d: d["identity"].update(sn=""),
|
||||
lambda d: d["model"].update(outputs=["fn", "fx"]),
|
||||
lambda d: d["model"]["params"].update(gains_N_per_raw=[0.001, 0, 0.003]),
|
||||
lambda d: d["model"]["params"].update(triangles=[[0, 1, 1]]),
|
||||
lambda d: d["model"]["params"].update(triangles=[[0, 1, 2], [2, 1, 0]]),
|
||||
lambda d: d["pipeline"].pop("geometry_sha256"),
|
||||
lambda d: d["pipeline"]["coordinates"].update(map_sha256="bad"),
|
||||
])
|
||||
def test_rejects_bad_documents(mutate):
|
||||
data = document()
|
||||
mutate(data)
|
||||
with pytest.raises(ValueError):
|
||||
parse_model(data)
|
||||
|
||||
|
||||
def test_rejects_gap_and_overlapping_equal_area_meshes():
|
||||
data = document()
|
||||
params = data["model"]["params"]
|
||||
params.update(points_uv=[[0, 0], [10, 0], [10, 10], [0, 10]],
|
||||
gains_N_per_raw=[0.001] * 4, triangles=[[0, 1, 2]])
|
||||
with pytest.raises(ValueError, match="triang|mesh|hull"):
|
||||
parse_model(data)
|
||||
params["triangles"] = [[0, 1, 2], [0, 1, 3]]
|
||||
with pytest.raises(ValueError, match="overlap"):
|
||||
parse_model(data)
|
||||
params["triangles"] = [[0, 1, 2], [0, 2, 3]]
|
||||
assert parse_model(data).predict(1, 0, features(), "zero").status["fn"] == "ok"
|
||||
|
||||
|
||||
def test_global_gain_and_context():
|
||||
data = document()
|
||||
data["pipeline"].pop("coordinates")
|
||||
data["pipeline"].pop("geometry_sha256")
|
||||
data["model"].update(type="global_gain/1", input_profile="force_delta/1",
|
||||
outputs=["fn", "fx"], params={
|
||||
"gains_N_per_raw": {"fn": 0.002, "fx": 0.004},
|
||||
"valid_force_range_N": [-3, 3]})
|
||||
model = parse_model(data)
|
||||
result = model.predict(1, 0, {"delta_raw_by_axis": {"fn": 100, "fx": -100},
|
||||
"zero_valid": True, "contact_found": True}, "z")
|
||||
assert result.values_N == {"fn": 0.2, "fx": -0.4}
|
||||
model.validate_for_context(data["pipeline"])
|
||||
wrong = deepcopy(data["pipeline"])
|
||||
wrong["source_scale"] = 0.42
|
||||
with pytest.raises(ValueError, match="source_scale"):
|
||||
model.validate_for_context(wrong)
|
||||
model.reset("new-zero")
|
||||
data["pipeline"].pop("source_scale")
|
||||
assert parse_model(data).input_profile == "force_delta/1"
|
||||
assert model.predict(2, None, {"force_delta_raw": {"fn": 100, "fx": -100},
|
||||
"zero_valid": True}, "z").values_N == {"fn": 0.2, "fx": -0.4}
|
||||
assert model.predict(3, None, {"force_delta_raw": {"fn": 100, "fx": -100},
|
||||
"zero_valid": True, "contact_found": False}, "z").status == {
|
||||
"fn": "no_contact", "fx": "no_contact"}
|
||||
|
||||
|
||||
def test_unknown_backend_fails_explicitly():
|
||||
data = document()
|
||||
data["model"]["type"] = "onnx/1"
|
||||
with pytest.raises(ValueError, match="unsupported.*onnx/1"):
|
||||
parse_model(data)
|
||||
|
||||
|
||||
def test_coordinate_artifact_hash_and_paths(tmp_path):
|
||||
path = tmp_path / "map.npz"
|
||||
np.savez(path, uv_by_output_px=np.zeros((2, 3, 2)), valid_mask=np.ones((2, 3), dtype=bool),
|
||||
origin_xy=np.array([511, 1021]))
|
||||
digest = hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
data = document()
|
||||
data["pipeline"]["coordinates"]["map_sha256"] = digest
|
||||
data["model"]["artifacts"] = [{"role": "coordinate_map", "path": "map.npz", "sha256": digest}]
|
||||
model = parse_model(data, tmp_path / "calibration.json")
|
||||
assert model.coordinate_map.shape == (2, 3, 2)
|
||||
assert len(model.fingerprint) == 64
|
||||
assert not model.coordinate_map.flags.writeable
|
||||
for unsafe in ("../map.npz", "https://example.org/x.npz", "C:\\map.npz", "/map.npz"):
|
||||
data["model"]["artifacts"][0]["path"] = unsafe
|
||||
with pytest.raises(ValueError, match="relative"):
|
||||
parse_model(data, tmp_path / "calibration.json")
|
||||
data["model"]["artifacts"][0]["path"] = "map.npz"
|
||||
path.write_bytes(b"corrupt")
|
||||
with pytest.raises(ValueError, match="SHA-256"):
|
||||
parse_model(data, tmp_path / "calibration.json")
|
||||
|
||||
|
||||
def test_many_points_complete_mesh():
|
||||
from scipy.spatial import Delaunay
|
||||
data = document()
|
||||
points = np.array([[x, y] for x in range(5) for y in range(5)])
|
||||
data["model"]["params"].update(points_uv=points.tolist(), gains_N_per_raw=[0.001] * 25,
|
||||
triangles=Delaunay(points).simplices.tolist())
|
||||
model = parse_model(data)
|
||||
values = features()
|
||||
values.update(u=2, v=2)
|
||||
assert model.predict(1, 0, values, "z").values_N["fn"] == 1.0
|
||||
@@ -0,0 +1,149 @@
|
||||
"""Exercise the example's SDK boundary without opening a camera or display."""
|
||||
|
||||
import importlib.util
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def example(monkeypatch):
|
||||
path = Path(__file__).resolve().parents[1] / "examples" / "finger_pipeline.py"
|
||||
monkeypatch.syspath_prepend(str(path.parent))
|
||||
spec = importlib.util.spec_from_file_location("finger_pipeline_example", path)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
@pytest.mark.parametrize("zero_mode", ["raw", "measure", "none"])
|
||||
def test_schema3_reference_is_applied_after_warmup_before_pipeline(example, monkeypatch, capsys, zero_mode):
|
||||
events = []
|
||||
options = {}
|
||||
snapshot = SimpleNamespace(fps=20., normal_force=1., shear_x=2., shear_y=3., age_ms=1.,
|
||||
force_units=("N", "raw", "raw"), calibration_status={"fn": "ok"},
|
||||
image=None, flow=None)
|
||||
|
||||
class Pipe:
|
||||
def __enter__(self):
|
||||
events.append("pipeline")
|
||||
self.frames = iter([snapshot, None])
|
||||
return self
|
||||
|
||||
def __exit__(self, *args):
|
||||
pass
|
||||
|
||||
def wait(self):
|
||||
return next(self.frames)
|
||||
|
||||
runtime = SimpleNamespace(
|
||||
tactile=SimpleNamespace(raw_sensor=SimpleNamespace(calibration=SimpleNamespace(schema_version=3)),
|
||||
mask_stats={}, force_units=("wrong",) * 3),
|
||||
warmup=lambda **kwargs: events.append("warmup"),
|
||||
set_calibration_reference=lambda raw, **kwargs: events.append(("reference", raw, kwargs)),
|
||||
pipeline=lambda **kwargs: Pipe(), close=lambda: events.append("close"))
|
||||
|
||||
def open_runtime(**kwargs):
|
||||
options.update(kwargs)
|
||||
return runtime
|
||||
|
||||
monkeypatch.setattr(example.FingerRuntime, "open", open_runtime)
|
||||
monkeypatch.setattr(example, "measure_unloaded_reference",
|
||||
lambda runtime: events.append("measure"))
|
||||
monkeypatch.setattr(example.cv2, "waitKey", lambda *args: pytest.fail("headless must not call GUI"))
|
||||
zero_args = {"raw": ["--zero-fn-raw", "1765.1"], "measure": ["--zero"], "none": []}[zero_mode]
|
||||
monkeypatch.setattr(sys, "argv", ["finger_pipeline.py", "--video", "test.mp4", "--no-show",
|
||||
"--cal-path", "model.json", "--device-id", "dd02-gl",
|
||||
"--export-dir", "assets", "--map-cache", "map", *zero_args])
|
||||
assert example.main() == 0
|
||||
assert options["cal_path"] == "model.json" and options["device_id"] == "dd02-gl"
|
||||
assert options["export_dir"] == "assets" and options["map_cache"] == "map"
|
||||
reference_events = {"raw": [("reference", {"fn": 1765.1}, {"reference_id": "cli-unloaded-reference"})],
|
||||
"measure": ["measure"], "none": []}[zero_mode]
|
||||
assert events == ["warmup", *reference_events, "pipeline", "close"]
|
||||
output = capsys.readouterr().out
|
||||
assert "calibration=fn:ok" in output and "wrong" not in output
|
||||
|
||||
|
||||
def test_selected_calibration_error_is_reported_without_fallback(example, monkeypatch, capsys):
|
||||
def fail(**kwargs):
|
||||
raise ValueError("calibration identity mismatch")
|
||||
|
||||
monkeypatch.setattr(example.FingerRuntime, "open", fail)
|
||||
monkeypatch.setattr(sys, "argv", ["finger_pipeline.py", "--no-show", "--cal_path", "bad.json"])
|
||||
assert example.main() == 1
|
||||
assert "calibration identity mismatch" in capsys.readouterr().err
|
||||
|
||||
|
||||
@pytest.mark.parametrize("arguments", [["--zero-fn-raw", "nan"], ["--calibration-dir", "calibrations"],
|
||||
["--device-id", "SN1"], ["--cal-path", "a", "--calibration-dir", "b"]])
|
||||
def test_invalid_selection_fails_before_capture(example, monkeypatch, arguments):
|
||||
monkeypatch.setattr(example.FingerRuntime, "open", lambda **kwargs: pytest.fail("must not open capture"))
|
||||
monkeypatch.setattr(sys, "argv", ["finger_pipeline.py", *arguments])
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
example.main()
|
||||
assert exc.value.code == 2
|
||||
|
||||
|
||||
def test_bundle_geometry_follows_explicit_config(example, tmp_path, monkeypatch):
|
||||
bundle = tmp_path / "custom"
|
||||
bundle.mkdir()
|
||||
(bundle / "sensor.json").write_text('{}')
|
||||
(bundle / "manifest.json").write_text('{"assets_dir": "geometry"}')
|
||||
(bundle / "geometry").mkdir()
|
||||
monkeypatch.chdir(tmp_path)
|
||||
assert Path(example.configured_export_dir("custom/sensor.json")) == bundle / "geometry"
|
||||
|
||||
|
||||
def test_missing_declared_geometry_does_not_use_default_bundle(example, tmp_path):
|
||||
(tmp_path / "sensor.json").write_text('{}')
|
||||
(tmp_path / "manifest.json").write_text('{"assets_dir": "missing"}')
|
||||
with pytest.raises(ValueError, match="assets_dir"):
|
||||
example.configured_export_dir(str(tmp_path / "sensor.json"))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("advanced", [False, True])
|
||||
def test_help_separates_debug_overrides(example, monkeypatch, capsys, advanced):
|
||||
monkeypatch.setattr(sys, "argv", ["finger_pipeline.py", "--help-advanced" if advanced else "--help"])
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
example.main()
|
||||
assert exc.value.code == 0
|
||||
output = capsys.readouterr().out
|
||||
assert "--cal-path" in output and "--zero " in output
|
||||
assert ("--zero-fn-raw" in output) == advanced
|
||||
assert ("--export-dir" in output) == advanced
|
||||
assert ("--device-id" in output) == advanced
|
||||
|
||||
|
||||
@pytest.mark.parametrize("failure", [None, "short", "nan", "legacy"])
|
||||
def test_unloaded_reference_is_measured_atomically(example, failure):
|
||||
events = []
|
||||
frames = iter(range(30))
|
||||
count = 0
|
||||
|
||||
def grab(**kwargs):
|
||||
nonlocal count
|
||||
count += 1
|
||||
return not (failure == "short" and count == 30), None
|
||||
|
||||
def read(*keys):
|
||||
value = next(frames)
|
||||
return (float("nan") if failure == "nan" and value == 29 else value,)
|
||||
|
||||
runtime = SimpleNamespace(
|
||||
tactile=SimpleNamespace(
|
||||
raw_sensor=SimpleNamespace(calibration=SimpleNamespace(
|
||||
schema_version=2 if failure == "legacy" else 3, channels={"fn": {}})),
|
||||
info=SimpleNamespace(FNORMAL_RAW=0, FSHEARX_RAW=1, FSHEARY_RAW=2),
|
||||
compute_deformation=lambda: None, read_info=read),
|
||||
grab_and_process=grab,
|
||||
set_calibration_reference=lambda raw, **kwargs: events.append((raw, kwargs)))
|
||||
if failure:
|
||||
with pytest.raises(ValueError):
|
||||
example.measure_unloaded_reference(runtime)
|
||||
assert events == []
|
||||
else:
|
||||
example.measure_unloaded_reference(runtime)
|
||||
assert events == [({"fn": 14.5}, {"reference_id": "startup-unloaded-reference"})]
|
||||
@@ -0,0 +1,120 @@
|
||||
"""File selection, SN isolation and legacy compatibility at the public loader boundary."""
|
||||
|
||||
import json
|
||||
import time
|
||||
from types import SimpleNamespace
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from orisys import calibration
|
||||
from orisys.calibration_context import build_context
|
||||
from orisys.sensors.calibration import resolve_calibration
|
||||
|
||||
|
||||
def model_doc():
|
||||
config = {"type": "single_camera", "camera": {"resolution_stitch": [64, 64]},
|
||||
"tracker": {"dis_mode": 1, "hhd_downsample": 3}}
|
||||
return {"schema_version": 3, "unit": "N", "identity": {"sensor_type": "finger", "sn": "A1"},
|
||||
"pipeline": build_context(config), "model": {
|
||||
"type": "global_gain/1", "input_profile": "force_delta/1", "output_frame": "sdk_force/1",
|
||||
"outputs": ["fn"], "params": {"gains_N_per_raw": {"fn": .001}, "valid_force_range_N": [0, 3]}}}
|
||||
|
||||
|
||||
def test_schema3_public_load_write_and_fn_only_sn(tmp_path):
|
||||
path = tmp_path / "finger" / "A1.json"
|
||||
path.parent.mkdir()
|
||||
path.write_text(json.dumps(model_doc()))
|
||||
model = calibration.load_for_device(tmp_path, "finger", "A1")
|
||||
assert tuple(model.channels) == ("fn",)
|
||||
assert calibration.check(model).ok
|
||||
target = tmp_path / "copy.json"
|
||||
calibration.write(model, target)
|
||||
assert json.loads(target.read_text())["schema_version"] == 3
|
||||
assert calibration.load(target).fingerprint == model.fingerprint
|
||||
with pytest.raises(calibration.CalibrationError, match="identity mismatch"):
|
||||
calibration.validate_for_device(model, "finger", "another-SN")
|
||||
|
||||
|
||||
def test_missing_sn_does_not_pick_any_neighbor_or_default(tmp_path):
|
||||
(tmp_path / "finger").mkdir()
|
||||
(tmp_path / "finger" / "A1.json").write_text(json.dumps(model_doc()))
|
||||
(tmp_path / "force_calibration.json").write_text(json.dumps(model_doc()))
|
||||
assert calibration.load_for_device(tmp_path, "finger", "B2") is None
|
||||
with pytest.raises(calibration.CalibrationError, match="directory"):
|
||||
calibration.load_for_device(tmp_path / "missing", "finger", "B2")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("change", ["unknown_type", "unknown_schema", "malformed"])
|
||||
def test_selected_bad_model_never_falls_back(tmp_path, change):
|
||||
data = model_doc()
|
||||
if change == "unknown_type":
|
||||
data["model"]["type"] = "onnx/1"
|
||||
elif change == "unknown_schema":
|
||||
data["schema_version"] = 99
|
||||
else:
|
||||
data["model"]["params"]["gains_N_per_raw"] = {}
|
||||
path = tmp_path / "selected.json"
|
||||
path.write_text(json.dumps(data))
|
||||
with pytest.raises(calibration.CalibrationError):
|
||||
calibration.load(path, sensor_type="box")
|
||||
|
||||
|
||||
def test_invalid_default_schema3_does_not_silently_become_raw(monkeypatch, tmp_path):
|
||||
from orisys import configs
|
||||
|
||||
data = model_doc()
|
||||
data["model"]["type"] = "onnx/1"
|
||||
path = tmp_path / "force_calibration.json"
|
||||
path.write_text(json.dumps(data))
|
||||
monkeypatch.setattr(configs, "resolve_sensor_bundle", lambda name: SimpleNamespace(force_calibration_json=path))
|
||||
with pytest.raises(calibration.CalibrationError, match="unsupported calibration model"):
|
||||
calibration.load(sensor_type="finger")
|
||||
|
||||
|
||||
def test_path_selection_keeps_explicit_and_sn_modes_separate(tmp_path):
|
||||
with pytest.raises(calibration.CalibrationError, match="mutually exclusive"):
|
||||
calibration.select_calibration_path(cal_path="file.json", calibration_dir=tmp_path)
|
||||
with pytest.raises(calibration.CalibrationError, match="requires"):
|
||||
calibration.select_calibration_path(device_id="A1")
|
||||
assert calibration.select_calibration_path(cal_path=tmp_path / "A1.json") == tmp_path / "A1.json"
|
||||
|
||||
|
||||
def test_profile_declares_only_available_axes_and_checks_family(tmp_path):
|
||||
path = tmp_path / "selected.json"
|
||||
path.write_text(json.dumps(model_doc()))
|
||||
options = dict(bundle_root=tmp_path, declared_default=None, profile_id="test", selection=path)
|
||||
resolved = resolve_calibration(product_family="finger", **options)
|
||||
assert resolved.axes == ("fn",) and resolved.document is not None
|
||||
assert resolved.fingerprint == resolved.document.fingerprint
|
||||
wrong = resolve_calibration(product_family="box", **options)
|
||||
assert wrong.document is None and wrong.issues[0].code == "calibration.model-family-mismatch"
|
||||
|
||||
|
||||
def test_processor_applies_resolved_selection_including_none(monkeypatch):
|
||||
from orisys.sensors import backends
|
||||
|
||||
monkeypatch.setattr(backends, "validate_profile", lambda profile: SimpleNamespace(usable=True))
|
||||
selected = calibration.parse(model_doc())
|
||||
calls = []
|
||||
sensor = SimpleNamespace(set_calibration=calls.append)
|
||||
for model in (selected, None):
|
||||
profile = SimpleNamespace(processing_backend="single_camera", config={"camera": {}},
|
||||
calibration=SimpleNamespace(document=model), profile_id="test")
|
||||
session = backends.create_processor(profile, sensor=sensor)
|
||||
assert session.sensor is sensor
|
||||
assert calls == [selected, None]
|
||||
|
||||
|
||||
def test_snapshot_keeps_per_frame_units_and_validity_copy():
|
||||
from orisys.pipeline import StreamPipeline
|
||||
|
||||
sensor = SimpleNamespace(info=object(), force_unit="N", force_units=("N", "raw", "raw"),
|
||||
calibration_status={"fn": "outside_coverage"}, fnormal=float("nan"),
|
||||
fshearx=2., fsheary=3., _force_raw=(1., 2., 3.))
|
||||
pipeline = StreamPipeline(sensor, lambda: None, include=())
|
||||
snapshot = pipeline._build(np.zeros((2, 2)), None, 0., 0., 0., time.perf_counter())
|
||||
assert snapshot.force_units == ("N", "raw", "raw")
|
||||
assert snapshot.calibration_status == {"fn": "outside_coverage"}
|
||||
sensor.calibration_status["fn"] = "ok"
|
||||
assert snapshot.calibration_status["fn"] == "outside_coverage"
|
||||
@@ -0,0 +1,308 @@
|
||||
"""Schema 3 lifecycle and readout tests without camera or hardware."""
|
||||
|
||||
from types import SimpleNamespace
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from orisys.calibration import CalibrationError
|
||||
from orisys.force_readout import ForceReadout
|
||||
from orisys.sensor import Sensor
|
||||
|
||||
|
||||
class RecordingModel:
|
||||
schema_version = 3
|
||||
channels = {"fn": None}
|
||||
input_profile = "fn_delta_autocal_uv/1"
|
||||
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
self.resets = []
|
||||
self.coordinate_map = np.array([[[0., 0.], [1., 0.]], [[0., 1.], [1., 1.]]])
|
||||
self.coordinate_valid_mask = np.array([[True, False], [True, True]])
|
||||
|
||||
def reset(self, reference_id):
|
||||
self.resets.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"]
|
||||
status = "ok"
|
||||
if not features["zero_valid"]:
|
||||
status = "zero_unavailable"
|
||||
elif not features["contact_found"]:
|
||||
status = "no_contact"
|
||||
elif not features["mapping_valid"]:
|
||||
status = "coordinate_unmapped"
|
||||
value = features["delta_raw"] * .1 if status == "ok" else None
|
||||
return SimpleNamespace(values_N={"fn": value}, status={"fn": status})
|
||||
|
||||
|
||||
def make_sensor(*, median=1):
|
||||
sensor = Sensor.__new__(Sensor)
|
||||
sensor.calibration = RecordingModel()
|
||||
sensor._config = {"filters": {"force_median_window": median}}
|
||||
sensor._force_readout = ForceReadout(sensor._config["filters"])
|
||||
sensor._force_raw = (10., 20., 30.)
|
||||
sensor._model_readouts = {}
|
||||
sensor._calibration_context = None
|
||||
sensor._reset_calibration_session()
|
||||
sensor.bind_calibration_context({"signature": "actual"})
|
||||
return sensor
|
||||
|
||||
|
||||
def predict(sensor, frame, fn, *, contact=True, mapped=True):
|
||||
return sensor.apply_calibration_frame(frame_id=frame, timestamp_s=frame / 30.,
|
||||
raw=(fn, 20., 30.), uv=(0., 0.),
|
||||
contact_found=contact, mapping_valid=mapped)
|
||||
|
||||
|
||||
def test_once_per_frame_and_explicit_zero_subtracted_once():
|
||||
sensor = make_sensor()
|
||||
sensor.set_calibration_reference({"fn": 10.}, reference_id="zero-a")
|
||||
first = predict(sensor, 1, 25.)
|
||||
assert first.values_N["fn"] == pytest.approx(1.5)
|
||||
assert predict(sensor, 1, 99.) is first
|
||||
assert len(sensor.calibration.calls) == 1
|
||||
assert sensor.force_units == ("N", "raw", "raw")
|
||||
assert (sensor.fnormal, sensor.fshearx, sensor.fsheary) == (1.5, 20., 30.)
|
||||
assert sensor._force_raw_unfiltered == (25., 20., 30.)
|
||||
|
||||
|
||||
def test_unknown_zero_and_invalid_contact_never_become_zero_newtons():
|
||||
sensor = make_sensor()
|
||||
assert predict(sensor, 1, 20.).status["fn"] == "zero_unavailable"
|
||||
assert np.isnan(sensor.fnormal)
|
||||
sensor.set_calibration_reference({"fn": 10.}, reference_id="zero-b")
|
||||
assert predict(sensor, 2, 20., contact=False).status["fn"] == "no_contact"
|
||||
assert np.isnan(sensor.fnormal)
|
||||
assert predict(sensor, 3, 20., mapped=False).status["fn"] == "coordinate_unmapped"
|
||||
assert np.isnan(sensor.fnormal)
|
||||
|
||||
|
||||
def test_failed_frame_does_not_enter_newton_median_history():
|
||||
sensor = make_sensor(median=3)
|
||||
sensor.set_calibration_reference({"fn": 0.}, reference_id="zero")
|
||||
predict(sensor, 1, 10.)
|
||||
predict(sensor, 2, 1000., contact=False)
|
||||
predict(sensor, 3, 20.)
|
||||
predict(sensor, 4, 30.)
|
||||
assert sensor.fnormal == pytest.approx(2.)
|
||||
assert sensor._force_unfiltered[0] == 3.
|
||||
|
||||
|
||||
def test_new_reference_clears_cached_prediction_and_filter():
|
||||
sensor = make_sensor(median=3)
|
||||
sensor.set_calibration_reference({"fn": 0.}, reference_id="first")
|
||||
predict(sensor, 1, 10.)
|
||||
sensor.set_calibration_reference({"fn": 20.}, reference_id="second")
|
||||
assert sensor.calibration_result is None
|
||||
assert np.isnan(sensor.fnormal)
|
||||
predict(sensor, 1, 30.)
|
||||
assert sensor.fnormal == pytest.approx(1.)
|
||||
assert sensor.calibration.resets[-1] == "second"
|
||||
|
||||
|
||||
def test_requires_actual_pipeline_context():
|
||||
sensor = make_sensor()
|
||||
with pytest.raises(CalibrationError, match="pipeline mismatch"):
|
||||
sensor.bind_calibration_context({"signature": "different"})
|
||||
assert sensor._calibration_context == {"signature": "actual"}
|
||||
sensor._calibration_context = None
|
||||
with pytest.raises(CalibrationError, match="actual calibration context"):
|
||||
predict(sensor, 1, 10.)
|
||||
|
||||
|
||||
def test_contact_map_never_reuses_previous_coordinate():
|
||||
sensor = make_sensor()
|
||||
sensor.centroid_found = True
|
||||
sensor.centroid = (0, 1)
|
||||
assert sensor._calibration_contact_uv() == (0., 1.)
|
||||
sensor.centroid = (1, 0)
|
||||
assert sensor._calibration_contact_uv() is None
|
||||
sensor.centroid = (10, 10)
|
||||
assert sensor._calibration_contact_uv() is None
|
||||
sensor.centroid_found = False
|
||||
sensor.centroid = (0, 1)
|
||||
assert sensor._calibration_contact_uv() is None
|
||||
|
||||
|
||||
def test_hhd_infers_after_current_contact_only_once_and_clears_old_contact():
|
||||
sensor = make_sensor()
|
||||
sensor.set_calibration_reference({"fn": 10.}, reference_id="zero")
|
||||
field = np.zeros((2, 2, 2), dtype=np.float32)
|
||||
sensor._Sensor__hhd = SimpleNamespace(
|
||||
get_force=lambda *args, **kwargs: (None, (20., 0., 0., 0.)),
|
||||
get_normal=lambda: field, get_shear=lambda: field,
|
||||
)
|
||||
contact = {"centroid": (0, 1), "centroid_found": True,
|
||||
"divergence_map": np.zeros((2, 2))}
|
||||
sensor._Sensor__contact3d = SimpleNamespace(compute_foe=lambda *args, **kwargs: contact)
|
||||
sensor.cuda = sensor.verbose = False
|
||||
sensor._hhd_downsample = 1
|
||||
sensor._hhd_grid_shape = sensor._hhd_ref_shape = (2, 2)
|
||||
sensor.valid_mask = sensor.valid_mask_hhd = None
|
||||
sensor._model_frame_id = 0
|
||||
sensor.compute_hhd(field)
|
||||
assert sensor.calibration.calls[-1]["features"]["v"] == 1.
|
||||
assert sensor.fnormal == pytest.approx(1.)
|
||||
sensor.compute_contact()
|
||||
assert len(sensor.calibration.calls) == 1
|
||||
contact["centroid_found"] = False
|
||||
sensor.compute_hhd(field)
|
||||
assert len(sensor.calibration.calls) == 2
|
||||
assert sensor.calibration_status == {"fn": "no_contact"}
|
||||
assert sensor.calibration.calls[-1]["features"]["u"] is None
|
||||
assert np.isnan(sensor.fnormal)
|
||||
|
||||
|
||||
def test_real_model_adapter_uses_same_lifecycle_as_stateful_model():
|
||||
from orisys.calibration import parse
|
||||
|
||||
data = {
|
||||
"schema_version": 3, "unit": "N",
|
||||
"identity": {"sensor_type": "finger", "sn": "test"},
|
||||
"pipeline": {"stitch_resolution": [2, 2], "dis_mode": 4, "hhd_downsample": 1,
|
||||
"filters": {"measurement": {"flow_ema_alpha": 1.}},
|
||||
"feature_signature_sha256": "a" * 64},
|
||||
"model": {"type": "global_gain/1", "input_profile": "force_delta/1",
|
||||
"output_frame": "sdk_force/1", "outputs": ["fn"],
|
||||
"params": {"gains_N_per_raw": {"fn": .1}, "valid_force_range_N": [0., 3.]}},
|
||||
}
|
||||
sensor = make_sensor()
|
||||
sensor.calibration = parse(data)
|
||||
sensor.bind_calibration_context(data["pipeline"])
|
||||
sensor.set_calibration_reference({"fn": 10.}, reference_id="reference")
|
||||
result = sensor.apply_calibration_frame(frame_id=1, raw=(20., 30., 40.))
|
||||
assert result.values_N == {"fn": 1.}
|
||||
assert sensor.force_units == ("N", "raw", "raw")
|
||||
assert sensor.calibration_status == {"fn": "ok"}
|
||||
result = sensor.apply_calibration_frame(frame_id=2, raw=(20., 30., 40.), contact_found=False)
|
||||
assert result.status == {"fn": "no_contact"}
|
||||
|
||||
|
||||
def test_reversed_frame_requires_a_new_zero_reference():
|
||||
sensor = make_sensor()
|
||||
sensor.set_calibration_reference({"fn": 10.}, reference_id="reference")
|
||||
predict(sensor, 10, 20.)
|
||||
assert predict(sensor, 2, 20.).status == {"fn": "zero_unavailable"}
|
||||
assert sensor.calibration.resets[-1] is None
|
||||
|
||||
|
||||
def test_replacement_is_atomic_when_pipeline_cannot_be_validated():
|
||||
sensor = make_sensor()
|
||||
sensor.device_id = None
|
||||
original = sensor.calibration
|
||||
replacement = RecordingModel()
|
||||
with pytest.raises(CalibrationError, match="pipeline mismatch"):
|
||||
sensor.set_calibration(replacement, context={"signature": "different"})
|
||||
assert sensor.calibration is original
|
||||
sensor._calibration_context = None
|
||||
with pytest.raises(CalibrationError, match="actual pipeline context"):
|
||||
sensor.set_calibration(replacement)
|
||||
assert sensor.calibration is original
|
||||
sensor.set_calibration(replacement, context={"signature": "actual"})
|
||||
assert sensor.calibration is replacement
|
||||
assert sensor._calibration_reference == {}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("use_legacy", [False, True])
|
||||
def test_schema3_swap_cannot_relabel_cached_newtons_as_raw_or_legacy(use_legacy):
|
||||
from orisys.calibration import parse
|
||||
|
||||
sensor = make_sensor()
|
||||
sensor.device_id = None
|
||||
sensor.set_calibration_reference({"fn": 10.}, reference_id="reference")
|
||||
predict(sensor, 1, 25.)
|
||||
assert sensor.fnormal == 1.5
|
||||
legacy = parse({"schema_version": 2, "unit": "N", "channels": {
|
||||
"normal": {"method": "piecewise_linear", "points": [[0., 0.], [100., 2.]],
|
||||
"extrapolate": True},
|
||||
}}) if use_legacy else None
|
||||
sensor.set_calibration(legacy)
|
||||
expected = .5 if use_legacy else 25.
|
||||
assert sensor.fnormal == pytest.approx(expected)
|
||||
assert sensor.force_triplets()["unfiltered"]["newton"][0] == pytest.approx(expected)
|
||||
assert sensor.force_triplets()["filtered"]["raw"][0] == 25.
|
||||
assert sensor.calibration_status == {}
|
||||
|
||||
|
||||
def test_reset_invalidates_contact_cache_and_coordinate():
|
||||
sensor = make_sensor()
|
||||
sensor._Sensor__tracker = SimpleNamespace(reset=lambda: None)
|
||||
sensor.img = None
|
||||
sensor.compute_contact_valid = sensor.centroid_found = True
|
||||
sensor.centroid = (1, 1)
|
||||
sensor._model_contact_result = {"centroid": (1, 1)}
|
||||
sensor.compute_deformation_valid = True
|
||||
sensor.reset()
|
||||
assert sensor.compute_contact_valid is False
|
||||
assert sensor.compute_deformation_valid is False
|
||||
assert sensor.centroid_found is False
|
||||
assert sensor._model_contact_result is None
|
||||
assert sensor._calibration_contact_uv() is None
|
||||
with pytest.raises(CalibrationError, match="current deformation"):
|
||||
sensor.compute_contact()
|
||||
|
||||
|
||||
def test_schema3_rejects_measurement_filter_change_before_mutation():
|
||||
sensor = make_sensor()
|
||||
sensor.compute_deformation_valid = False
|
||||
with pytest.raises(CalibrationError, match="measurement identity"):
|
||||
sensor.configure(flow_ema_alpha=.5)
|
||||
assert sensor._force_readout.flow_ema_alpha == 1.
|
||||
assert sensor._calibration_context == {"signature": "actual"}
|
||||
|
||||
|
||||
def test_readout_configuration_applies_before_and_after_first_model_frame():
|
||||
sensor = make_sensor()
|
||||
sensor.compute_deformation_valid = False
|
||||
sensor.configure(force_median_window=3, lowpass_sample_rate_hz=20.)
|
||||
sensor.set_calibration_reference({"fn": 0.}, reference_id="reference")
|
||||
predict(sensor, 1, 10.)
|
||||
readout = sensor._model_readouts["fn"]
|
||||
assert readout.force_median_window == 3
|
||||
assert readout.lowpass_sample_rate_hz == 20.
|
||||
predict(sensor, 2, 20.)
|
||||
predict(sensor, 3, 30.)
|
||||
assert sensor.fnormal == 2.
|
||||
sensor.configure(force_median_window=1)
|
||||
assert readout.force_median_window == 1
|
||||
predict(sensor, 4, 40.)
|
||||
assert sensor.fnormal == 4.
|
||||
|
||||
|
||||
def test_generic_global_fallback_binds_actual_mirror_identity():
|
||||
from orisys.calibration import parse
|
||||
from orisys.calibration_context import build_context
|
||||
|
||||
sensor = make_sensor()
|
||||
sensor.device_id = None
|
||||
sensor.mirror = True
|
||||
sensor._config.update(camera={"resolution_stitch": [2, 2]}, tracker={"dis_mode": 1})
|
||||
context = build_context(sensor._config, sensor.filters_state(), preprocessing={"mirror": True})
|
||||
data = {"schema_version": 3, "unit": "N", "identity": {"sensor_type": "finger", "sn": "test"},
|
||||
"pipeline": context,
|
||||
"model": {"type": "global_gain/1", "input_profile": "force_delta/1",
|
||||
"output_frame": "sdk_force/1", "outputs": ["fn"],
|
||||
"params": {"gains_N_per_raw": {"fn": .1}, "valid_force_range_N": [0., 3.]}}}
|
||||
sensor._calibration_context = None
|
||||
sensor.set_calibration(parse(data))
|
||||
assert sensor._calibration_context == context
|
||||
sensor.mirror = False
|
||||
sensor._calibration_context = None
|
||||
with pytest.raises(CalibrationError, match="feature_signature"):
|
||||
sensor.set_calibration(parse(data))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("reference", [{"fn": float("nan")}, {"other": 0.}])
|
||||
def test_bad_zero_reference_is_rejected_without_reset(reference):
|
||||
sensor = make_sensor()
|
||||
sensor.set_calibration_reference({"fn": 10.}, reference_id="valid")
|
||||
with pytest.raises(ValueError):
|
||||
sensor.set_calibration_reference(reference, reference_id="bad")
|
||||
assert sensor._calibration_reference == {"fn": 10.}
|
||||
Reference in New Issue
Block a user