末节采用原 CAD 零位

This commit is contained in:
lxp
2026-09-11 17:45:44 +08:00
parent 69c2da6808
commit 467651fc47
35 changed files with 1636 additions and 67 deletions
@@ -1,4 +1,4 @@
# 按关节零位与独立实测 JSON v3 实施记录
# 按关节零位与实测/复制 JSON v3 实施记录
日期:2026-09-11。范围:`linkerhand_calibration`
@@ -18,6 +18,7 @@
| 会话/运动版本核对及冻结提交 | `runtime/coordinator.py` |
| 原零位核验后逐任务恢复数据 | `runtime/joint_resume.py` |
| 冻结参考下的相对运动与指令映射 | `core/fitting/motion_fit.py``command_mapping.py``anchored_mapping.py` |
| 明确声明的运动复制、零位迁移及近似来源 | `core/fitting/motion_transfer.py` |
| 独立线性 mimic 近似 | `core/fitting/mimic_approximation.py` |
| 空间零偏及授权 URDF 修正 | `core/fitting/session.py``spatial_solver/``core/urdf/result_plan.py` |
| v3 序列化、读取和最终文件验收 | `runtime/artifacts/``core/urdf/measured_acceptance.py``tag_acceptance.py` |
@@ -34,25 +35,32 @@ Profile 必须声明完整零位指令和非零的最终到达动作;扫描起
零位稳定性判断使用各机位最近的足量独立帧,短暂异常帧会随新帧退出判断窗口;
原始记录保留,稳定性门限不变。
JSON 每个运动关节来自自己的视觉样本,同一 SDK 通道允许绑定多个关节。
JSON 关节来自自己的视觉样本,或 Profile 明确声明的对应实测来源;同一 SDK 通道允许绑定多个关节。
接收关节保留自己的 SDK 通道,复制来源的指令/反馈节点及角度,不再拟合。
接收关节没有独立零位帧和第四轮指标,报告标记 `independently_measured=false` 并引用来源证据。
正式单条指令曲线只用前三轮,约束 baseline=0;两条方向诊断曲线保留回差,共用同一参考。
第四轮分别检验两方向,不能用于重新拟合曲线、零位、Tag 安装或基座。
序列化只导出冻结结果;u8 必须实测支持 0–255 才能生成 256 项,rad 只在 `input_values` 范围内插值。
URDF mimic 使用前三轮成对实测角度拟合经过零点的直线,输出 offset 精确为 0。
非线性 JSON 与 mimic 近似分别保存。近似误差不否决完整映射;机械范围、拓扑、授权字段及标准加载仍必须合法。
缺少独立运动观测或绝对几何依据时保留相对运动诊断拒绝完整发布,不能使用旧 mimic、迁移或默认零偏补全
既无运动观测也无显式复制声明,或缺少绝对零位来源时,保留相对运动诊断拒绝完整发布。
旧 mimic 及旧迁移字段不能自动授权 v3 复制,不用默认零偏补全未知几何。
接受原 CAD 末节零位的前提单独记录在 `cad_zero_assumptions`,不作为独立实测证据;
仅这些明确列出的末节取 δ=0,保持其自身 `origin`。O12 的被动 PIP 不在该假设范围。
## 验收与消费
发布器从最终文件回读并分别检验:
1. 最终 JSON 对第四轮原始视觉角度;逐关节、逐方向检查。
1. 最终 JSON 对第四轮原始视觉角度;逐实测关节、逐方向检查;复制表另核验来源与一致性
2. 最终 JSON+修正 URDF 的完整 FK 对第四轮 Tag 位姿;显式使用所有被动角。
3. 相同主动映射配合标准 mimic 的损失;使用同一份训练冻结安装关系,只报告近似误差。
前两项精度门限仍为角度 MAE ≤1°、P95 ≤2°、最大 ≤3°,Tag 位置 P95 ≤3 mm。
三文件哈希、保护输入、观测覆盖和通道绑定一并检查;拟合失败不自动运动,取消后不能提交。
混合产物的 manifest acceptance 为 `measured_and_transferred_json_urdf_v3`
明确记录观测关节、复制关节和 CAD 零位假设。未观测接收关节不具有独立精度结论。
公共读取器和 `calibrated_joint_state_bridge` 支持 v3 的完整关节映射及历史 v1/v2 原语义。
完整 FK 使用 `independent_mimic_angles=True`。标准 mimic 加载器用于结构兼容检查,不能覆盖完整非线性关节状态。
@@ -60,19 +68,20 @@ URDF mimic 使用前三轮成对实测角度拟合经过零点的直线,输出
## 当前型号的发布边界
| 型号 | 独立运动观测 | 仍缺少的运动观测 | 绝对几何边界 |
| 型号 | 独立运动观测 | 声明的运动复制 | 绝对几何边界 |
| --- | --- | --- | --- |
| G20 | 21 / 21 个运动关节 | 无 | 五个末端被动关节仍缺 CAD baseline 依据 |
| L6 | 5 / 11 | 食指、中指、无名指的 MCP、DIP | 旧 CAD 保留及迁移关系不能充当独立依据 |
| O6 | 5 / 11 | 食指、中指、无名指的 MCP、DIP | 同上 |
| O12 | 16 / 19 | 无名指 MCP pitch、PIPDIP | 部分末端/原 CAD 保留关节仍缺依据 |
| G20 | 21 / 21 个运动关节 | 无 | 五个末节的原 CAD 零位为明确假设 |
| L6 | 5 / 11 | 其余三指 MCP、DIP ← 小指对应关节 | MCP 零偏明确迁移;五个末节 CAD 零位为假设 |
| O6 | 5 / 11 | 其余三指 MCP、DIP ← 小指对应关节 | 同上 |
| O12 | 16 / 19 | 无名指 MCP/PIP/DIP ← 小指对应关节 | pinky_pip、ring_pip、thumb_mcp 零位仍须另行确定 |
`--validate-only``independent_measurement_coverage` 返回具体关节列表。
配置检查通过表示合同可执行,不能解释为全关节精度通过。
这四份现有配置目前均不能发布完整 v3;补齐 Tag、任务、避让与几何依据需要实机确认
上述运动策略已经覆盖各型号全部运动关节;完整发布仍要求实测来源验收通过、零位有来源且范围合法
O12 当前仍被三项几何缺口阻止完整发布;L6/O12/G20 的现有 CAD mimic/限位冲突保留诊断,不能放宽限位掩盖。
现有 Tag 布局没有被补写成未经验证的新布局,新增参考动作也没有标为实机已验证。
## 软件验证记录
## 原按关节零位改造的软件验证记录
修改前在原工作区测试路径上加载备份代码:**582 passed、1 skipped**。
固定基线保存在 `/tmp/calibration_v3_baseline_ck_q_c99/`,包含原包、输入哈希和运行日志。
@@ -103,9 +112,25 @@ URDF mimic 使用前三轮成对实测角度拟合经过零点的直线,输出
| 更换通道顺序、机位名、任务顺序的虚拟型号走同一完整发布链 | `test_measured_finalization.py``measured_capture_fixture.py` |
| 旧格式读取、标准 URDF 授权和结构、READY 后断流、迟到观测、后台反馈、中止与提交竞争 | 原有相关回归及扩展的 `test_calibration_coordinator.py` |
## 本轮观测策略扩展
全量回归:**700 passed、1 skipped495.34 秒**,日志 `/tmp/calibration_transfer_full_verified.log`
全量收集后补入的两项反馈范围一致性用例另行通过(`/tmp/calibration_transfer_feedback_domain_verified.log`),
本轮共覆盖 **702 项通过、1 项缺少实测数据而跳过**。未放宽任何空间求解或最终精度门限。
构建成功;安装后 12 个 console entry point 均能导入,四型号 `--validate-only` 均以 0 退出。
对照当前提交检查 142 个原始 URDF/mesh 文件,内容全部不变;四份产品配置仅更新对应 Profile 哈希。
新增回归见 `test_observation_strategy.py``test_motion_transfer.py``test_transfer_artifact_contract.py`
覆盖三种观测策略、虚拟型号改名/重排、接收通道、非线性被动表、来源与假设篡改拒绝、
O12 PIP 不补零、末节自身 origin 和各指 CAD 几何保持、机械范围及混合产物读回。
历史全实测合成夹具显式清空新的产品 CAD 假设,继续使用原非零合成真值;未修改其数值期望。
`test_measured_transfer_finalization.py` 进一步覆盖 O6 及重排后的虚拟 O6:
由 5 个关节的原始合成记录实际拟合,展开为 11 个关节,经过最终文件验收、发布及 manifest 读回;
6 个复制关节保留自己的通道,且没有被制造为独立观测。完整组合通过时仍报告 mimic 近似损失。
合成数据的已知 baseline 几何明确标注 `synthetic_fixture_truth`,不写入产品 Profile。
完整 v3 非线性回放覆盖合成 G20 拓扑及重排后的虚拟型号;四型号执行器测试不等于四型号都有完整实测回放。
两种布局的合成回放均得到:最终关节映射最大误差约 0.079°,完整 FK 的 Tag 姿态最大误差约 0.078°、
完整 v3 非线性回放覆盖合成 G20、O6 拓扑及重排后的虚拟型号;四型号执行器测试不等于四型号都有完整实测回放。
原 G20 两种布局的合成回放均得到:最终关节映射最大误差约 0.079°,完整 FK 的 Tag 姿态最大误差约 0.078°、
最差 Tag 位置 P95 约 0.025 mm;标准 mimic 的姿态最大误差约 5.465°,仍被如实报告。
这些数字用于检验软件是否分开评价非线性映射与线性近似,不是相机或实机测量结果。
O12 实测回放因未提供 `O12_REPLAY_RAW``O12_REPLAY_REFERENCE` 跳过。
+54 -11
View File
@@ -3,9 +3,10 @@
G20、L6、O6、O12 使用同一个产品启动器、在线状态机、采集器、拟合器、标准 URDF
验收和发布器。型号差异来自 `config/profiles/*.yaml` 和 SDK Adapter,不再调用型号节点。
本包按关节冻结实机 baseline 零位,分别输出独立实测非线性 JSON v3 和线性 mimic 近似 URDF
本包按观测关节冻结实机 baseline 零位,输出实测曲线及明确声明的复制曲线组成的 JSON v3,
以及线性 mimic 近似 URDF;报告区分独立实测、迁移和 CAD 零位假设。
保留 CAD 连杆尺寸、轴位置、mesh、惯量。完整运动由 JSON 的全部关节角度配合修正 URDF 计算。
软件仿真通过不等于实机精度通过;当前原始 G20/O12 存在需要确认的 mimic/限位冲突。
软件仿真通过不等于实机精度通过;当前原始 G20/L6/O12 存在需要确认的 mimic/限位冲突。
本轮实现与验证见 [v3 实施记录](CALIBRATION_V3_IMPLEMENTATION.md)。
## 启动
@@ -38,10 +39,22 @@ ros2 run linkerhand_calibration calibrate_hand --config \
| O12/right/o12_right_16/v1 | 12 | 11 | 16 | unified v3rad 输入节点查表 |
任务数、主动关节数和实测关节数不必相等。一条运动可观测多个主动/被动关节。
**这些配置目前不能发布全关节独立实测产物。** L6/O6 的食指、中指、无名指缺少独立观测;
O12 无名指缺少独立观测。G20 虽有完整运动观测,末端被动关节仍缺少绝对 CAD 零位依据。
其他型号保留 CAD 零位的关节也须补充几何依据;启动检查的 `independent_measurement_coverage` 列出确切关节。
历史迁移关系保留供旧格式读取与诊断,v3 不用迁移、mimic 或补零填满 JSON。
观测策略由 `measurement.transferred_motion_sources` 声明,目标关节指向直接实测的来源关节:
| 型号 | 独立采集 | 复制的运动映射 |
| --- | --- | --- |
| O6/L6 | 拇指、小指 | 食指/中指/无名指的 MCP、DIP 分别复制小指对应关节 |
| O12 | 拇指、小指、中指、食指 | 无名指 MCP/PIP/DIP 分别复制小指对应关节 |
| G20 / 全关节模式 | 每个关节 | 不声明复制关系 |
复制保留接收关节自己的 SDK 通道、URDF 名称与 CAD 几何,不要求接收关节另贴 Tag 或采集零位帧。
复制参数不等于接收关节独立实测;接收关节仍须有明确的绝对零位来源及合法机械范围。
`zero.cad_zero_assumptions` 显式记录“实机 baseline 与原 CAD 末节零位一致”的使用前提,
相应 DIP(及拇指末节 IP)δ=0、自身 `origin` 不改。这是假设,不是视觉测得的绝对 CAD 朝向。
**O12 的 PIP 不采用此假设。** 当前 `pinky_pip``ring_pip` 和主动 `thumb_mcp` 仍缺绝对零位依据,
可以采集诊断数据,但这些缺口解决前不能发布完整 O12 v3 JSON/URDF。
启动检查分别列出 `unmeasured_joints`(未独立观测)、`unresolved_motion_joints`(也未声明复制)
`missing_baseline_geometry`。旧迁移字段单独存在不会自动授权新的 v3 曲线复制。
本次沿用现有 Tag 和避让布局声明延后零位动作,没有把新增动作或缺失 Tag 布局标为实机验证通过。
每任务四个往返周期:前三轮训练,第四轮独立验证。
每个方向先连续扫描,再单通道访问稳态点。训练每方向默认 9 点,并包含该通道 baseline,第四轮为交错点加端点;
@@ -49,6 +62,27 @@ O12 无名指缺少独立观测。G20 虽有完整运动观测,末端被动关
稳态点缺失与连续扫描不足统一在方向结束处理,最多同速重扫一次。
字节拟合不再改写原始 command/feedback,也不将接近端点的读数伪装成 0/255。
## 启动前相机检查
`--validate-only` 检查相机配置文件、序列号、内外参指纹和已有外参质量记录,不采集现场图像。
正式启动还检查实时 `CameraInfo` 的宽高及 K/D/R/P 是否与受保护外参中的记录一致;
参数不匹配、无效或消息过期时不能开始,并显示具体机位和原因。
**配置匹配不等于现场内外参有效。** 当前只有手上的 Tag,各机位使用不同 Tag,
没有声明跨机位共视或已知的 Tag 间几何关系;手掌和贴纸又允许在开始前移动、重贴。
因此光学内参与相对外参的现场有效性记录为 `unverified`,不能用重新锁定掌心 Tag、
较小的 PnP 残差或未变化的配置哈希代替实测,也不能由 Tag 位置变化断定相机被碰。
`READY` 只表示设备与消息满足现有开始条件。显示上述能力提示后沿用现有启动行为,
不要求各手指 Tag 可见,不增加人工确认或新的暂停条件;`READY` 不表示物理相机参数已验证通过。
会话目录中的 `camera_preflight.json``raw_samples.jsonl` 保存检查结果与依据,
开始前仅在结果变化时记录,接受 Start 前再检查并冻结一次;后续状态中的检查结果指这次启动检查。
单凭这些记录不能判定是否需要重新标定。`expected_cameras` 是配置期望身份,不是现场实测结果。
镜头、焦距或成像设置变化后应重新核验内参;相机之间相对位置变化后应重新核验外参。
未来同一个物理 Tag 被多个机位共同看到时,可以检查相对外参一致性,但当前尚未实现该能力。
本次检查不修改 Profile、SDK 或原始 URDF。
## 基准、避让与恢复
- 开始前可以调整手掌、相机和 Tag;预览数据不会成为正式参考。
@@ -107,7 +141,7 @@ ROS 和离线采集使用相同默认值,型号 YAML 不重复声明该值。`
有效数据、基准、断点来源,以及中文原因、建议和原始诊断。耗时拟合在后台执行,不阻塞反馈与状态。
标题明确显示型号、左右手和序列号。开始前等待状态列出缺少或过期的相机内参、检测消息和 SDK 条件;
空 Tag 检测消息可以证明视觉链路工作,不要求开始前已经识别全部 Tag。
收到开始请求时重新检查设备状态;相机有效内参必须与受保护外参中的指纹一致。
收到开始请求时重新检查设备状态;实时 CameraInfo 参数必须与受保护外参中的指纹一致。
开始前相机消息的新鲜度窗口为 2 秒,该窗口不用于扫描中的实时暂停。
## 唯一生产代码链
@@ -130,6 +164,7 @@ linkerhand_calibration/
├── runtime/parameters.py 已解析的运行参数;无 ROS 类型
├── runtime/inputs.py 相机/检测输入、时钟与发布接口
├── runtime/cameras.py 受保护相机身份与消息新鲜度
├── runtime/camera_preflight.py 启动前配置检查结果与内外参实测能力边界
├── runtime/session.py 唯一在线状态机
├── runtime/execution.py Profile 任务与运动效果调度
├── runtime/motion_execution.py 平滑轨迹与位移到位判定
@@ -186,7 +221,7 @@ Profile 声明:
1. Adapter、SDK 通道顺序、单位、正负方向、命令/反馈范围、基准值、独立速度槽布局。
2. Tag ID/尺寸/机位/安装 link,以及每次运动的父子观测关系;不明确的 link 不能靠名称猜。
3. 扫描任务、速度、准备/避让/收尾 waypoint;每关节首次零位任务、完整保持指令及固定到达姿态。
4. 声明空间可观测约束和有依据的已知 baseline 几何;完整 v3 每个运动关节须独立观测。
4. 声明空间可观测约束已知 baseline 几何或明确接受的 CAD 零位假设;每个运动关节须独立观测或显式指定对应实测来源
5. 每关节允许修正的 `origin.rpy``limit.lower/upper``mimic.multiplier/offset`
这仍需要正确的可观测性设计:单轴加一对未知安装角的 Tag 不能自动辨识绝对 CAD 零位。
@@ -211,19 +246,27 @@ manifest 记录 JSON、URDF、标定报告三文件 SHA256、受保护输入和
O12 等 rad 输入还包含一一对应的 `input_values`(在有效节点间线性插值,不外推)。
字节型号固定 256 项:SDK 指令 `v` 对应 `angle_rad[v]`,只能在训练指令实际覆盖 0–255 时导出。
输出均是修正 URDF 的 q_output,不再加零偏。顶层记录身份、单位、完整 `baseline_command` 和共同零位约定。
主动和被动关节直接导出各自视觉实测曲线,同一驱动通道可绑定多个关节。
独立观测的主动和被动关节直接导出各自视觉曲线,同一驱动通道可绑定多个关节。
显式复制的关节使用来源关节的同一组节点与角度,不重新拟合、归零或由 mimic 推导;接收通道保持不变。
最终单条曲线在前三轮拟合中约束 baseline=0,正反向诊断分支共用冻结参考、保留回差。
第四轮分别检验两个方向;超差时禁止发布,不隐藏切换分支。
URDF mimic 单独用前三轮成对视觉角度拟合 `q_child≈a*q_parent`,共同零位下 offset=0
URDF mimic 单独用前三轮成对视觉角度拟合 `q_child≈a*q_parent`,共同零位下 offset=0
复制的被动关节使用对应来源的线性近似,证据仍归属于来源关节。
其近似误差不会否决已达标的完整映射,但结构、授权字段与机械限位始终强制检查。
验收分别记录:最终 JSON 对第四轮独立视觉角度;完整 JSON+URDF FK 对第四轮 Tag 位姿;
同一主动映射配合标准 mimic 的近似损失。前两项决定精度通过,第三项仅报告。
门限保持角度 MAE≤1°、P95≤2°、最大≤3°,Tag 位置 P95≤3 mm。
独立精度结论只覆盖实际观测的关节和 Tag;复制关节另检查来源、查表一致性、通道与机械范围,
不把来源关节的第四轮误差称为接收关节精度。混合产物 manifest 使用
`measured_and_transferred_json_urdf_v3`,列出实测和复制覆盖。
两种 FK 使用相同的训练冻结安装与基座;第四轮不拟合零位、曲线、Tag 安装或基座。
`calibration_report.json` 保存完整的 `command_to_rad` / `feedback_to_rad`、输入域、方向分支、
零位原始参考帧、完整指令/反馈/到达方向、来源和训练/holdout 指标。报告中的 `applicability` 记录保持姿态;
实测关节的零位原始参考帧、完整指令/反馈/到达方向、来源和训练/holdout 指标。
复制关节记录 `independently_measured: false``transferred_from_joint` 与来源参考链接,
不生成自己的伪零位样本或独立精度指标。绝对零位来源另由 `zero_method`
`cad_zero_assumption``zero_transferred_from_joint` 表达。报告中的 `applicability` 记录保持姿态;
未采集多姿态证据不能宣称任意多轴工况已验证。主 JSON 用于指令查表,不能误作反馈转换。
旧 unified v1 会话继续只读兼容,不自动覆盖或转换历史产物;新输出单表必须重新通过最终文件验收。
@@ -1,7 +1,7 @@
schema_version: 3
profile_id: G20/right/g20_right_19/v1
profile_config: package://linkerhand_calibration/config/profiles/g20_right_19.yaml
profile_config_sha256: 049a8d7b75d3f7881d6990115416718792e775f513154d59f5843191df03d068
profile_config_sha256: f73c8a391a8feb6cb3028fbd8278d3b18d6e2e875980235f7797fd8de2efbac1
model: G20
side: right
tag_layout: g20_right_19
@@ -1,7 +1,7 @@
schema_version: 3
profile_id: L6/right/l6_right_8/v1
profile_config: package://linkerhand_calibration/config/profiles/l6_right_8.yaml
profile_config_sha256: 71cfe506126ffc485de878ee620f2b453b4afaccde66bcf585231edb6daf6dd2
profile_config_sha256: 733094679e5477bbe97a633f4fb1ce25242804b667cd435ef9132f491e6ef247
model: L6
side: right
tag_layout: l6_right_8
@@ -1,7 +1,7 @@
schema_version: 3
profile_id: O12/right/o12_right_16/v1
profile_config: package://linkerhand_calibration/config/profiles/o12_right_16.yaml
profile_config_sha256: a1faa89632a9e3ce38f5083abf080c194f19bd906f6ec40ae377f3399cc3c00a
profile_config_sha256: d0c1e2ac62f96162612f0ec45c25c6f467d045c1c1030b5be34ccc115dca4cf9
model: O12
side: right
tag_layout: o12_right_16
@@ -1,7 +1,7 @@
schema_version: 3
profile_id: O6/right/o6_right_8/v1
profile_config: package://linkerhand_calibration/config/profiles/o6_right_8.yaml
profile_config_sha256: 4872b6eee080eb8bb3596a53f4b731d0dc317881f64589ea2e55676ad2a0a2cf
profile_config_sha256: df6c4abb3c9d48b186c1bd6ab304ea9c0bbab54f9381ff222d5d8d3a3f65babe
model: O6
side: right
tag_layout: o6_right_8
@@ -1034,6 +1034,7 @@ motion:
approach_commands:
- [255, 255, 0, 0, 0, 255, 127, 127, 127, 127, 255, 255, 255, 255, 255, 255, 0, 0, 0, 0]
measurement:
transferred_motion_sources: {}
measurements:
thumb_cmc_pitch:
joint: thumb_cmc_pitch
@@ -1258,6 +1259,17 @@ measurement:
stable_cross_view_cone_bias: true
input_domain: feedback_u8
zero:
cad_zero_assumptions:
index_dip: User accepts original CAD joint zero as the physical baseline; not independently
measured.
middle_dip: User accepts original CAD joint zero as the physical baseline; not independently
measured.
pinky_dip: User accepts original CAD joint zero as the physical baseline; not independently
measured.
ring_dip: User accepts original CAD joint zero as the physical baseline; not independently
measured.
thumb_ip: User accepts original CAD joint zero as the physical baseline; not independently
measured.
active_joints:
- index_mcp_pitch
- index_mcp_roll
@@ -182,6 +182,13 @@ motion:
approach_commands:
- [255, 255, 255, 255, 255, 0]
measurement:
transferred_motion_sources:
rh_index_mcp_pitch: rh_pinky_mcp_pitch
rh_index_dip: rh_pinky_dip
rh_middle_mcp_pitch: rh_pinky_mcp_pitch
rh_middle_dip: rh_pinky_dip
rh_ring_mcp_pitch: rh_pinky_mcp_pitch
rh_ring_dip: rh_pinky_dip
measurements:
rh_thumb_cmc_roll:
joint: rh_thumb_cmc_roll
@@ -229,6 +236,17 @@ measurement:
cross_view_roll_curve: false
stable_cross_view_cone_bias: false
zero:
cad_zero_assumptions:
rh_index_dip: User accepts original CAD joint zero as the physical baseline; not
independently measured.
rh_middle_dip: User accepts original CAD joint zero as the physical baseline; not
independently measured.
rh_pinky_dip: User accepts original CAD joint zero as the physical baseline; not
independently measured.
rh_ring_dip: User accepts original CAD joint zero as the physical baseline; not
independently measured.
rh_thumb_dip: User accepts original CAD joint zero as the physical baseline; not
independently measured.
active_joints:
- rh_index_mcp_pitch
- rh_middle_mcp_pitch
@@ -645,6 +645,10 @@ motion:
approach_commands:
- [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 1.530653753999027, 0.0, 1.3578661580515883, 1.8151424220741028, 1.53588974175501, 1.53588974175501]
measurement:
transferred_motion_sources:
ring_mcp_pitch: pinky_mcp_pitch
ring_pip: pinky_pip
ring_dip: pinky_dip
candidate_selection_tasks:
- thumb_mcp_dip_front
directional_zero: true
@@ -739,6 +743,17 @@ measurement:
child_role: index_dip
pose_axis_line_required: false
zero:
cad_zero_assumptions:
index_dip: User accepts original CAD joint zero as the physical baseline; not independently
measured.
middle_dip: User accepts original CAD joint zero as the physical baseline; not independently
measured.
pinky_dip: User accepts original CAD joint zero as the physical baseline; not independently
measured.
ring_dip: User accepts original CAD joint zero as the physical baseline; not independently
measured.
thumb_dip: User accepts original CAD joint zero as the physical baseline; not independently
measured.
active_joints:
- thumb_cmc_roll
- thumb_cmc_yaw
@@ -178,6 +178,13 @@ motion:
approach_commands:
- [255, 255, 255, 255, 255, 0]
measurement:
transferred_motion_sources:
rh_index_mcp_pitch: rh_pinky_mcp_pitch
rh_index_dip: rh_pinky_dip
rh_middle_mcp_pitch: rh_pinky_mcp_pitch
rh_middle_dip: rh_pinky_dip
rh_ring_mcp_pitch: rh_pinky_mcp_pitch
rh_ring_dip: rh_pinky_dip
measurements:
rh_thumb_cmc_yaw:
joint: rh_thumb_cmc_yaw
@@ -225,6 +232,17 @@ measurement:
cross_view_roll_curve: false
stable_cross_view_cone_bias: false
zero:
cad_zero_assumptions:
rh_index_dip: User accepts original CAD joint zero as the physical baseline; not
independently measured.
rh_middle_dip: User accepts original CAD joint zero as the physical baseline; not
independently measured.
rh_pinky_dip: User accepts original CAD joint zero as the physical baseline; not
independently measured.
rh_ring_dip: User accepts original CAD joint zero as the physical baseline; not
independently measured.
rh_thumb_ip: User accepts original CAD joint zero as the physical baseline; not
independently measured.
active_joints:
- rh_index_mcp_pitch
- rh_middle_mcp_pitch
@@ -48,6 +48,8 @@ def build_typed_profile():
measurement=replace(
right.measurement,
measurements=measurements,
# Legacy mirroring does not inherit v3 acquisition authorizations.
transferred_motion_sources={},
cross_view_sources={
left_joint_name(name): left_joint_name(source)
for name, source in right.measurement.cross_view_sources.items()
@@ -66,6 +68,7 @@ def build_typed_profile():
# Runtime-only mirrored products do not acquire/solve new geometry.
spatial={},
known_baseline_geometry={},
cad_zero_assumptions={},
active_joints=active,
passive_joints=passive,
direct_zero_joints=tuple(
@@ -47,6 +47,8 @@ def build_typed_profile():
motion=replace(right.motion, tasks=tasks, joint_zero_references={}),
measurement=replace(
right.measurement,
# Legacy mirroring does not inherit v3 acquisition authorizations.
transferred_motion_sources={},
measurements=measurements,
cross_view_sources={
left_joint_name(name): left_joint_name(source)
@@ -66,6 +68,7 @@ def build_typed_profile():
# Runtime-only mirrored products do not acquire/solve new geometry.
spatial={},
known_baseline_geometry={},
cad_zero_assumptions={},
active_joints=active,
passive_joints=passive,
direct_zero_joints=tuple(
@@ -274,6 +274,9 @@ class MeasurementPolicy:
stable_cross_view_cone_bias: bool = False
candidate_selection_tasks: frozenset[str] = frozenset()
input_domain: str = ""
# Explicit motion reuse, independent of legacy zero/mimic transfer policy.
# Recipients keep their own SDK channel and cannot become measured donors.
transferred_motion_sources: Mapping[str, str] = field(default_factory=dict)
@dataclass(frozen=True)
@@ -300,6 +303,8 @@ class ZeroSolvePolicy:
# Geometry policy only; no Python callback/import strings are permitted.
spatial: Mapping[str, object] = field(default_factory=dict)
known_baseline_geometry: Mapping[str, KnownBaselineGeometry] = field(default_factory=dict)
# User-accepted CAD q=0 at physical baseline, not a measured geometry prior.
cad_zero_assumptions: Mapping[str, str] = field(default_factory=dict)
@dataclass(frozen=True)
@@ -393,6 +398,63 @@ class ProfileValidationError(ValueError):
"""Raised before hardware startup when a profile is internally unsafe."""
def _transfer_channel(profile: CalibrationProfile, joint: str) -> int | None:
"""Resolve a declared channel without inventing a recipient binding."""
seen = set()
while joint not in profile.command.command_index_by_joint:
if joint in seen or joint not in profile.zero.mimic_source_by_joint:
return None
seen.add(joint)
joint = profile.zero.mimic_source_by_joint[joint]
return profile.command.command_index_by_joint[joint]
def _validate_observation_strategy(profile: CalibrationProfile, errors: list[str]) -> None:
zero, command = profile.zero, profile.command
transfers = profile.measurement.transferred_motion_sources
observed = {joint for task in profile.motion.tasks if not task.validation_only
for joint in task.joints if joint in profile.measurement.measurements
and profile.measurement.measurements[joint].view is not None
and profile.measurement.measurements[joint].kind != "axis_cross_view_validation"}
all_joints = zero.active_joints | zero.passive_joints
for target, source in transfers.items():
if target not in all_joints or source not in all_joints or source not in observed or target == source:
errors.append(f"motion transfer requires a known recipient and directly observed donor:{target}")
continue
if target in observed or source in transfers:
errors.append(f"motion transfer cannot replace observations or chain donors:{target}")
if (target in zero.active_joints) != (source in zero.active_joints):
errors.append(f"motion transfer must preserve active/passive joint type:{target}")
if target in zero.passive_joints:
parent = zero.mimic_source_by_joint.get(target)
donor_parent = zero.mimic_source_by_joint.get(source)
if parent is None or donor_parent is None or transfers.get(parent, parent) != donor_parent:
errors.append(f"motion transfer must preserve mimic parent correspondence:{target}")
target_channel, source_channel = _transfer_channel(profile, target), _transfer_channel(profile, source)
channels = (target_channel, source_channel)
values = (command.baseline_values, command.minimum_values,
command.maximum_values, command.joint_directions,
command.minimum_feedback_values, command.maximum_feedback_values)
if any(index is None or index < 0 or index in command.disabled_indices
or any(index >= len(row) for row in values) for index in channels):
errors.append(f"motion transfer requires valid enabled SDK bindings:{target}")
elif any(row[target_channel] != row[source_channel] for row in values):
errors.append(f"motion transfer requires equal SDK ranges, baseline and sign:{target}")
if target in zero.direct_zero_joints:
errors.append(f"motion recipient cannot claim a directly observed zero:{target}")
if target in zero.transferred_zero_sources and zero.transferred_zero_sources[target] != source:
errors.append(f"motion and zero transfer donors differ:{target}")
for joint, reason in zero.cad_zero_assumptions.items():
if joint not in all_joints or not isinstance(reason, str) or not reason.strip():
errors.append(f"CAD zero assumption requires a known joint and explicit reason:{joint}")
if joint in zero.direct_zero_joints or joint in zero.known_baseline_geometry \
or joint in zero.transferred_zero_sources:
errors.append(f"CAD zero assumption conflicts with another zero source:{joint}")
for joint in zero.transferred_zero_sources:
if joint in zero.known_baseline_geometry or joint in zero.direct_zero_joints:
errors.append(f"transferred zero conflicts with another zero source:{joint}")
def validate_profile(profile: CalibrationProfile) -> None:
"""Hard-check all cross-policy references before hardware is enabled."""
errors: list[str] = []
@@ -597,6 +659,7 @@ def validate_profile(profile: CalibrationProfile) -> None:
if zero.active_joints & zero.passive_joints:
errors.append("active and passive joints must be disjoint")
all_joints = zero.active_joints | zero.passive_joints
_validate_observation_strategy(profile, errors)
if not zero.active_joints.issubset(command.command_index_by_joint):
errors.append("every active joint must bind to a command channel")
if not set(zero.direct_zero_joints).issubset(zero.active_joints):
@@ -13,6 +13,7 @@ class MimicApproximation:
training_sample_ids: tuple[str, ...]
holdout_sample_ids: tuple[str, ...]
holdout_errors_rad: tuple[float, ...]
transferred_from_joint: str | None = None
def in_cad_coordinates(self, offsets):
from dataclasses import replace
@@ -0,0 +1,109 @@
"""Apply explicit donor policies without creating recipient observations.
Motion curves, absolute zeros and URDF approximations have separate provenance.
Only the numerical mappings are expanded; training and holdout evidence remain
attached to the joints that were actually observed.
"""
from dataclasses import replace
from ..domain.profile import CalibrationProfile, validate_profile
from ..domain.result import CalibrationResult, JointMapping
from .motion_fit import channel_for_joint
def transferred_mapping(profile: CalibrationProfile, mapping: JointMapping,
target: str) -> JointMapping:
return replace(mapping, joint=target, motor_index=channel_for_joint(profile, target),
transferred_from=mapping.joint)
def expand_profile_transfers(profile: CalibrationProfile, fit: CalibrationResult) -> CalibrationResult:
"""Copy frozen fits once, after all measured command holdouts have passed."""
transfers = profile.measurement.transferred_motion_sources
if not transfers:
return fit
validate_profile(profile)
outputs, commands = dict(fit.output_mappings), dict(fit.command_mappings)
offsets, methods = dict(fit.zero_offsets_rad), dict(fit.zero_method_by_joint)
approximations = dict(fit.mimic_approximations)
for target, source in transfers.items():
if target in outputs or target in commands or target in fit.curves or target in fit.joint_zero_references:
raise ValueError(f"motion_transfer_cannot_replace_existing_result:{target}")
if source not in fit.curves or source not in outputs or source not in commands:
raise ValueError(f"motion_transfer_requires_fitted_donor:{target}:{source}")
for mappings in (outputs, commands):
donor = mappings[source]
if donor.joint != source or donor.transferred_from is not None \
or donor.motor_index != channel_for_joint(profile, source):
raise ValueError(f"motion_transfer_requires_direct_donor_binding:{source}")
mappings[target] = transferred_mapping(profile, donor, target)
# Dynamic similarity does not authorize copying an absolute CAD zero.
if profile.zero.transferred_zero_sources.get(target) == source and source in offsets:
offsets[target] = offsets[source]
methods[target] = "transferred_static_zero_on_own_cad"
if target in profile.zero.passive_joints:
if source not in fit.mimic_approximations:
raise ValueError(f"motion_transfer_requires_paired_donor_mimic:{target}:{source}")
donor = fit.mimic_approximations[source]
approximations[target] = replace(donor,
output_fit=replace(donor.output_fit, target_joint=target,
source_joint=profile.zero.mimic_source_by_joint[target]),
transferred_from_joint=source)
return replace(fit, output_mappings=outputs, command_mappings=commands,
zero_offsets_rad=offsets, zero_method_by_joint=methods, mimic_approximations=approximations)
def validate_mapping_provenance(profile: CalibrationProfile, fit: CalibrationResult) -> None:
"""Require every published table to be measured or an exact authorized copy."""
validate_profile(profile)
transfers = profile.measurement.transferred_motion_sources
observed = set(fit.curves)
if observed & transfers.keys():
raise ValueError("independent_release_contains_transferred_or_unknown_joints")
expected = observed | transfers.keys()
for mappings in (fit.output_mappings, fit.command_mappings):
if set(mappings) != expected:
raise ValueError("independent_release_contains_transferred_or_unknown_joints")
for name, mapping in mappings.items():
if mapping.joint != name or mapping.motor_index != channel_for_joint(profile, name):
raise ValueError(f"mapping_sdk_binding_differs:{name}")
source = transfers.get(name)
if source is None:
if mapping.transferred_from is not None:
raise ValueError("independent_release_contains_transferred_or_unknown_joints")
elif source not in observed or mappings[source].transferred_from is not None \
or mapping != transferred_mapping(profile, mappings[source], name):
raise ValueError(f"mapping_differs_from_declared_donor:{name}")
for name, approximation in fit.mimic_approximations.items():
source = transfers.get(name)
if source is None:
if approximation.transferred_from_joint is not None:
raise ValueError(f"undeclared_mimic_transfer:{name}")
elif source not in fit.mimic_approximations:
raise ValueError(f"missing_donor_mimic_approximation:{name}")
else:
donor = fit.mimic_approximations[source]
expected_approximation = replace(donor, output_fit=replace(donor.output_fit,
target_joint=name, source_joint=profile.zero.mimic_source_by_joint[name]),
transferred_from_joint=source)
if approximation != expected_approximation:
raise ValueError(f"mimic_differs_from_declared_donor:{name}")
def validate_zero_provenance(profile: CalibrationProfile, fit: CalibrationResult) -> None:
"""Keep accepted premises and explicit zero copies true in the final plan."""
offsets, methods = fit.zero_offsets_rad, fit.zero_method_by_joint
for name in profile.zero.cad_zero_assumptions:
if offsets.get(name) != 0.0 or methods.get(name) != "assumed_source_cad_zero":
raise ValueError(f"assumed_cad_zero_differs_from_profile:{name}")
for name, geometry in profile.zero.known_baseline_geometry.items():
if offsets.get(name) != geometry.cad_angle_rad \
or methods.get(name) != "independent_known_baseline_geometry":
raise ValueError(f"known_baseline_geometry_differs_from_profile:{name}")
for target, source in profile.zero.transferred_zero_sources.items():
if profile.measurement.transferred_motion_sources.get(target) != source:
continue
if source not in offsets or offsets.get(target) != offsets[source] \
or methods.get(target) != "transferred_static_zero_on_own_cad":
raise ValueError(f"transferred_zero_differs_from_declared_donor:{target}")
@@ -7,7 +7,7 @@ import math
from pathlib import Path
from typing import Mapping
from ..domain.profile import CalibrationProfile
from ..domain.profile import CalibrationProfile, validate_profile
from ..domain.result import CalibrationResult, JointMapping
from ..urdf.kinematics import UrdfKinematicModel
from .motion_fit import channel_for_joint, fit_profile_motion, joint_input_direction
@@ -72,6 +72,8 @@ def fit_profile_calibration(profile: CalibrationProfile, source_urdf: Path, reco
No mechanical endpoint or range-centre fallback can replace a requested
but unobservable spatial zero. The fourth cycle never updates the fit.
"""
if profile.artifacts.output_schema_version >= 3:
validate_profile(profile)
if profile.zero.endpoint_anchor_by_joint:
raise ValueError("unverified mechanical endpoints cannot define CAD zeros")
required_pose_fields = {"parent_pose_common", "child_pose_common",
@@ -88,9 +90,12 @@ def fit_profile_calibration(profile: CalibrationProfile, source_urdf: Path, reco
motion_fitted(motion)
curves = motion.coordinate_curves
measured = profile.artifacts.output_schema_version >= 3
baseline_geometry = {name: value.cad_angle_rad for name, value in profile.zero.known_baseline_geometry.items()}
if measured:
baseline_geometry.update({name: 0.0 for name in profile.zero.cad_zero_assumptions})
spatial_curves = {name: replace(curve, circle={**curve.circle,
"known_cad_baseline_rad": profile.zero.known_baseline_geometry[name].cad_angle_rad})
if measured and name in profile.zero.known_baseline_geometry and name not in zero_profile.direct_zero_joints
"known_cad_baseline_rad": baseline_geometry[name]})
if measured and name in baseline_geometry and name not in zero_profile.direct_zero_joints
else curve for name, curve in curves.items()}
observations = []
for cycle in range(4):
@@ -134,6 +139,8 @@ def fit_profile_calibration(profile: CalibrationProfile, source_urdf: Path, reco
from .mimic_approximation import fit_mimic_approximations
for name, datum in profile.zero.known_baseline_geometry.items():
offsets[name], methods[name] = datum.cad_angle_rad, "independent_known_baseline_geometry"
for name in profile.zero.cad_zero_assumptions:
offsets[name], methods[name] = 0.0, "assumed_source_cad_zero"
mappings = {name: JointMapping(name, channel_for_joint(profile, name), profile.curve_input_domain,
tuple(curve.circle["input_knots"]), curve.angle_rad, curve.increasing_rad, curve.decreasing_rad)
for name, curve in curves.items()}
@@ -6,10 +6,26 @@ from .acceptance import SerializedJointMapping, angular_metrics
from ..fitting.spatial import measure_joint_curve_observation
def validate_measured_joint_holdout(payload, curves, observations):
def validate_transferred_joint_tables(payload, measured_joints, transferred_motion_sources):
"""Check declared copies without inventing measurements for recipients."""
mapping = SerializedJointMapping(payload, input_kind="command")
if not mapping.measured or not mapping.compact or set(mapping.joints) != set(curves):
observed, transfers = set(measured_joints), dict(transferred_motion_sources)
if (not mapping.measured or not mapping.compact or not observed
or set(mapping.joints) != observed | transfers.keys() or observed & transfers.keys()
or any(source not in observed for source in transfers.values())):
raise ValueError("v3 final mapping differs from independently measured joints")
baseline = payload["baseline_command"]
for target, source in transfers.items():
recipient, donor = mapping.joints[target], mapping.joints[source]
if (recipient["angle_rad"] != donor["angle_rad"]
or recipient["curve_input_knots_rad"] != donor["curve_input_knots_rad"]
or baseline[recipient["motor_index"]] != baseline[donor["motor_index"]]):
raise ValueError(f"transferred_final_table_differs_from_source:{target}")
def validate_measured_joint_holdout(payload, curves, observations, *, transferred_motion_sources=None):
validate_transferred_joint_tables(payload, curves, transferred_motion_sources or {})
mapping = SerializedJointMapping(payload, input_kind="command")
errors = {name: {d: [] for d in ("increasing", "decreasing")} for name in curves}
seen = set()
unit = payload["input_unit"]
@@ -65,8 +65,11 @@ def prepare_measured_result(profile, model, source_sha256, fit):
geometry = moving - fit.zero_offsets_rad.keys()
if missing or geometry:
raise ValueError(f"independent_release_incomplete:unmeasured={sorted(missing)}:absolute_geometry={sorted(geometry)}")
if set(fit.output_mappings) != moving or any(m.transferred_from for m in fit.output_mappings.values()):
if set(fit.output_mappings) != moving:
raise ValueError("independent_release_contains_transferred_or_unknown_joints")
from ..fitting.motion_transfer import validate_mapping_provenance, validate_zero_provenance
validate_mapping_provenance(profile, fit)
validate_zero_provenance(profile, fit)
passive = {name for name, joint in model.joints.items() if joint.mimic_joint}
if set(fit.mimic_approximations) != passive:
raise ValueError("independent_release_missing_paired_mimic_observations")
@@ -270,6 +270,7 @@ def load_hand_profile(path: str | Path) -> CalibrationProfile:
bool(measurement.get("stable_cross_view_cone_bias", False)),
_set(measurement.get("candidate_selection_tasks", ())),
str(measurement.get("input_domain", "")),
dict(measurement.get("transferred_motion_sources", {})),
),
zero=ZeroSolvePolicy(
_set(zero.get("active_joints")),
@@ -289,6 +290,7 @@ def load_hand_profile(path: str | Path) -> CalibrationProfile:
{str(name): KnownBaselineGeometry(float(row["cad_angle_rad"]),
float(row["uncertainty_rad"]), str(row["evidence_source"]))
for name, row in zero.get("known_baseline_geometry", {}).items()},
dict(zero.get("cad_zero_assumptions", {})),
),
quality=QualityPolicy(
tuple(int(value) for value in quality.get("training_cycles", ())),
@@ -19,7 +19,7 @@ def validate_executable_profile(profile, source_urdf):
if joint == observer:
raise ValueError(f"absolute_zero_unobservable:{joint}:a free Tag on its own hinge is not a CAD datum")
measured = {joint for task in profile.motion.tasks for joint in task.joints}
if not measured <= model.joints.keys() or not profile.zero.active_joints <= model.joints.keys():
if not measured <= model.joints.keys() or not (profile.zero.active_joints | profile.zero.passive_joints) <= model.joints.keys():
raise ValueError("Profile joint keys must name URDF joints; SDK names belong in command.names")
for name, spec in profile.measurement.measurements.items():
if spec.kind not in {"relative_rotation", "urdf_axis_chain", "curve", "axis_cross_view_validation"}:
@@ -36,6 +36,7 @@ def validate_executable_profile(profile, source_urdf):
for target, source in profile.zero.mimic_source_by_joint.items():
if target not in model.joints or model.joints[target].mimic_joint != source:
raise ValueError(f"Profile mimic topology differs from CAD:{target}")
_validate_transfer_geometry(profile, model)
if not set(profile.urdf_authorized_fields) <= model.joints.keys():
raise ValueError("URDF authorization references a missing joint")
conflicts = []
@@ -51,20 +52,44 @@ def validate_executable_profile(profile, source_urdf):
conflicts.append({"joint": name, "source_joint": joint.mimic_joint,
"reachable_rad": [lo, hi], "cad_limits_rad": [joint.lower, joint.upper],
"measured_mimic_authorized": name in profile.zero.fitted_mimic_joints})
coverage = {"unmeasured_joints": sorted({n for n, j in model.joints.items() if j.kind != "fixed"}-measured),
transfers = profile.measurement.transferred_motion_sources
moving = {name for name, joint in model.joints.items() if joint.kind != "fixed"}
zero_transfers = {target for target, source in profile.zero.transferred_zero_sources.items()
if transfers.get(target) == source}
coverage = {"unmeasured_joints": sorted(moving-measured),
"unresolved_motion_joints": sorted(moving-measured-set(transfers)),
"transferred_motion_sources": dict(transfers),
"missing_baseline_geometry": sorted((profile.zero.active_joints | profile.zero.passive_joints)
-set(profile.zero.direct_zero_joints)-set(profile.zero.known_baseline_geometry)),
"transferred_joints": sorted(set(profile.zero.transferred_zero_sources)|set(profile.zero.transferred_mimic_sources))}
-set(profile.zero.direct_zero_joints)-set(profile.zero.known_baseline_geometry)
-set(profile.zero.cad_zero_assumptions)-zero_transfers),
"transferred_joints": sorted(set(transfers)|set(profile.zero.transferred_zero_sources)
|set(profile.zero.transferred_mimic_sources))}
return {"profile_id": profile.key.profile_id, "tag_links": links,
"independent_measurement_coverage": coverage,
"zero_observers": dict(spatial.offset_observer_joint),
"cad_retained": sorted(profile.zero.cad_frozen_joints),
"cad_zero_assumptions": dict(profile.zero.cad_zero_assumptions),
"transferred": dict(profile.zero.transferred_zero_sources),
"cad_constraint_conflicts": conflicts,
"observability": "structural_checks_only; numerical_rank_and_holdout_required",
"arbitrary_multiaxis_validated": False}
def _validate_transfer_geometry(profile, model):
"""Check the declared strategy against CAD signs and actual joint types."""
from ..core.fitting.motion_fit import joint_input_direction
for target, source in profile.measurement.transferred_motion_sources.items():
recipient, donor = model.joints[target], model.joints[source]
if recipient.kind == "fixed" or recipient.kind != donor.kind:
raise ValueError(f"motion transfer requires matching movable CAD joint types:{target}")
for name, joint in ((target, recipient), (source, donor)):
if (joint.mimic_joint is not None) != (name in profile.zero.passive_joints):
raise ValueError(f"motion transfer active/passive type differs from CAD:{name}")
if joint_input_direction(profile, model, target) != joint_input_direction(profile, model, source):
raise ValueError(f"motion transfer requires equal CAD output signs:{target}")
def _validate_separable_observations(profile, model, links):
from ..core.fitting.motion_fit import channel_for_joint
def chain(link):
@@ -9,6 +9,7 @@ from typing import Callable
from ...core.artifacts.storage import atomic_write_json
from ...core.fitting.session import fit_profile_calibration
from ...core.fitting.command_mapping import fit_command_mappings
from ...core.fitting.motion_fit import channel_for_joint
from ...core.geometry.candidate_selection import resolve_chain_observations
from ...core.geometry.extrinsics import transform_matrix
from ...core.urdf.kinematics import UrdfKinematicModel
@@ -80,6 +81,9 @@ def finalize_profile_session(*, profile, session_dir, serial_number, source_urdf
steady = accepted_joint_records(profile, records, sample_phase="steady",
directions_are_task_relative=directions_are_task_relative)
fit = fit_command_mappings(profile, fit, steady)
if profile.artifacts.output_schema_version >= 3:
from ...core.fitting.motion_transfer import expand_profile_transfers
fit = expand_profile_transfers(profile, fit)
except ValueError as error:
atomic_write_json(directory / "fit_diagnostics.json", {
"passed": False, "reason": str(error), "publication_allowed": False})
@@ -114,7 +118,11 @@ def finalize_profile_session(*, profile, session_dir, serial_number, source_urdf
command_observations=command_holdout,
joint_observations=tuple(r for rows in steady.values() for r in rows if r["cycle"] == 3),
joint_curves=fit.curves,
joint_channels={name: mapping.motor_index for name, mapping in fit.output_mappings.items()})
joint_channels={name: channel_for_joint(profile, name) for name in fit.output_mappings},
transferred_motion_sources=profile.measurement.transferred_motion_sources,
cad_zero_assumptions=profile.zero.cad_zero_assumptions,
transferred_zero_sources=profile.zero.transferred_zero_sources,
joint_zero_methods=fit.zero_method_by_joint)
report_writer, command_writer = serialize_report, from_report
if profile.artifacts.output_schema_version >= 3:
from .serializers.measured_v3 import serialize_report as report_writer, from_report as command_writer
@@ -105,6 +105,10 @@ class FrozenTagArtifactValidator:
joint_observations: tuple[Mapping, ...] = ()
joint_curves: Mapping = field(default_factory=dict)
joint_channels: Mapping[str, int] = field(default_factory=dict)
transferred_motion_sources: Mapping[str, str] = field(default_factory=dict)
cad_zero_assumptions: Mapping[str, str] = field(default_factory=dict)
transferred_zero_sources: Mapping[str, str] = field(default_factory=dict)
joint_zero_methods: Mapping[str, str] = field(default_factory=dict)
def __call__(self, json_path: Path, urdf_path: Path) -> Mapping[str, Any]:
payload = json.loads(json_path.read_text(encoding="utf-8"))
@@ -121,6 +125,7 @@ class FrozenTagArtifactValidator:
raise ValueError("v3 report protected inputs differ from the session")
if {name: row["sdk_channel"] for name, row in payload["joints"].items()} != dict(self.joint_channels):
raise ValueError("v3 SDK channel bindings differ from measured Profile bindings")
self._validate_report_provenance(feedback_payload)
validate_artifact_structure(source_urdf=self.source_urdf,
corrected_urdf=urdf_path, source_sha256=self.source_sha256,
authorized_fields=self.authorized_fields, payload=feedback_payload,
@@ -153,15 +158,25 @@ class FrozenTagArtifactValidator:
if measured:
from ...core.urdf.measured_acceptance import validate_measured_joint_holdout
extra = {"independent_mapping_holdout": validate_measured_joint_holdout(payload,
self.joint_curves, self.joint_observations),
self.joint_curves, self.joint_observations,
transferred_motion_sources=self.transferred_motion_sources),
"full_json_urdf_holdout": command_metrics,
"independently_measured_joints": sorted(self.joint_curves),
"transferred_joints": sorted(self.transferred_motion_sources),
"transferred_motion_sources": dict(self.transferred_motion_sources),
"cad_zero_assumptions": dict(self.cad_zero_assumptions),
"precision_scope": "observed_joints_travel_directions_and_held_poses",
"urdf_mimic_approximation": validate_serialized_tag_holdout(corrected_urdf=urdf_path,
payload=payload, common_from_base=self.common_from_base, installations=self.installations,
observations=self.command_observations, required_roles=self.required_roles, minimum_samples=3,
input_kind="command", mimic_approximation=True, enforce_precision=False)}
evidence["joint_observations"] = self.joint_observations
evidence["frozen_joint_curves"] = {key: asdict(value) for key, value in self.joint_curves.items()}
return {"acceptance": "independently_measured_json_urdf_v3" if measured else "serialized_standard_urdf_tag_replay_v1",
evidence["transferred_motion_sources"] = dict(self.transferred_motion_sources)
evidence["cad_zero_assumptions"] = dict(self.cad_zero_assumptions)
acceptance_kind = ("measured_and_transferred_json_urdf_v3" if self.transferred_motion_sources
else "independently_measured_json_urdf_v3") if measured else "serialized_standard_urdf_tag_replay_v1"
return {"acceptance": acceptance_kind,
**extra,
"profile_id": self.profile_id, "source_urdf_sha256": self.source_sha256,
"protected_inputs": dict(self.protected_inputs),
@@ -172,6 +187,28 @@ class FrozenTagArtifactValidator:
sort_keys=True, separators=(",", ":"), allow_nan=False).encode()).hexdigest(),
"structured_changes": list(changes)}
def _validate_report_provenance(self, report):
"""The report cannot grant itself new transfer or CAD authorizations."""
if report.get("transferred_motion_sources", {}) != dict(self.transferred_motion_sources):
raise ValueError("v3 transfer sources differ from the measured Profile")
rows = report["joints"]
observed = {name for name, row in rows.items() if row.get("independently_measured") is True}
if (observed != set(self.joint_curves)
or set(rows) != observed | self.transferred_motion_sources.keys()):
raise ValueError("v3 reported measurement coverage differs from frozen observations")
assumptions = {name: row["cad_zero_assumption"] for name, row in rows.items() if "cad_zero_assumption" in row}
if assumptions != dict(self.cad_zero_assumptions):
raise ValueError("v3 CAD zero assumptions differ from the Profile")
zero_sources = {name: row["zero_transferred_from_joint"] for name, row in rows.items()
if "zero_transferred_from_joint" in row}
if zero_sources != dict(self.transferred_zero_sources):
raise ValueError("v3 static zero transfer sources differ from the Profile")
if self.joint_zero_methods and {name: row["zero_method"] for name, row in rows.items()} != dict(self.joint_zero_methods):
raise ValueError("v3 zero provenance differs from the frozen fit")
for name in assumptions:
if rows[name]["zero_method"] != "assumed_source_cad_zero" or rows[name]["zero_offset_rad"] != 0:
raise ValueError(f"v3 assumed CAD zero is mislabeled:{name}")
@dataclass(frozen=True)
class ArtifactRelease:
@@ -22,6 +22,9 @@ class UnifiedCommandMapper:
if compact and (report is None or (measured_from_report if self.measured else from_report)(report) != payload):
raise ValueError("compact JSON requires its matching calibration report")
metadata = report if compact else payload
self.transferred_motion_sources = dict(metadata.get("transferred_motion_sources", {})) if self.measured else {}
self.independently_measured_joints = tuple(sorted(name for name, row in metadata["joints"].items()
if row.get("independently_measured") is True)) if self.measured else ()
self.mapping = SerializedJointMapping(metadata if compact and input_kind == "feedback" else payload, input_kind=input_kind)
self.urdf = UrdfKinematicModel(corrected_urdf)
self.active = tuple(name for name, j in self.urdf.joints.items() if j.kind != "fixed" and j.mimic_joint is None)
@@ -88,9 +91,23 @@ def load_unified_mapper(calibration_file, *, expected_side=None, input_kind="com
if validation.get("final_file_spatial_replay") != "passed" or not validation.get("steady_command_tag_holdout"):
raise ValueError("release lacks independent command and feedback URDF evidence")
payload = json.loads(json_path.read_text())
if payload.get("format") == MEASURED_FORMAT and (validation.get("acceptance") != "independently_measured_json_urdf_v3"
or not validation.get("independent_mapping_holdout") or not validation.get("full_json_urdf_holdout")):
raise ValueError("v3 release lacks independent measured acceptance")
report = json.loads(artifact("calibration_report").read_text()) if payload.get("format") in {COMPACT_FORMAT, MEASURED_FORMAT} else None
if payload.get("format") == MEASURED_FORMAT:
transfers = report.get("transferred_motion_sources", {})
expected_kind = "measured_and_transferred_json_urdf_v3" if transfers else "independently_measured_json_urdf_v3"
observed = {name for name, row in report["joints"].items() if row.get("independently_measured") is True}
if (validation.get("acceptance") != expected_kind or not observed
or set(validation.get("independent_mapping_holdout", {})) != observed
or not validation.get("full_json_urdf_holdout")
or validation.get("transferred_motion_sources", {}) != transfers):
raise ValueError("v3 release lacks independent measured acceptance")
if transfers or "independently_measured_joints" in validation:
if (validation.get("independently_measured_joints") != sorted(observed)
or validation.get("transferred_joints") != sorted(transfers)):
raise ValueError("v3 release measurement coverage differs from its report")
assumptions = {name: row["cad_zero_assumption"] for name, row in report["joints"].items()
if "cad_zero_assumption" in row}
if validation.get("cad_zero_assumptions", {}) != assumptions:
raise ValueError("v3 release CAD assumptions differ from its report")
return UnifiedCommandMapper(payload, urdf_path,
input_kind=input_kind, expected_side=expected_side, report=report)
@@ -1,4 +1,4 @@
"""Independent measured command tables. Serialization never fits or rebases."""
"""Measured command tables and explicit copies; no fitting or rebasing here."""
from dataclasses import asdict
import numpy as np
@@ -12,27 +12,47 @@ ZERO_CONVENTION = "q_CAD=q_output+delta; joint_baseline:q_output=0"
def serialize_report(profile, fit, prepared, *, serial_number, protected_inputs):
expected = set(prepared.mappings)
transfers = dict(profile.measurement.transferred_motion_sources)
observed = expected - transfers.keys()
if not expected or any(set(values) != expected for values in
(fit.command_mappings, fit.output_mappings, fit.zero_offsets_rad, fit.joint_zero_references)):
(fit.command_mappings, fit.output_mappings, fit.zero_offsets_rad)):
raise ValueError("independent_measurement_or_geometry_incomplete")
if set(fit.joint_zero_references) != observed or set(fit.curves) != observed:
raise ValueError("independent_measurement_or_reference_coverage_differs")
rows = {}
for name in sorted(expected):
if fit.command_mappings[name].transferred_from or fit.output_mappings[name].transferred_from:
raise ValueError(f"transferred_mapping_is_not_independent:{name}")
row = {"urdf_joint": name, "independently_measured": True,
donor = transfers.get(name)
if any(mapping.transferred_from != donor for mapping in
(fit.command_mappings[name], fit.output_mappings[name])):
raise ValueError(f"mapping_transfer_differs_from_profile:{name}")
row = {"urdf_joint": name, "independently_measured": donor is None,
"zero_offset_rad": fit.zero_offsets_rad[name], "zero_method": fit.zero_method_by_joint[name],
"zero_reference": fit.joint_zero_references[name].as_record(),
"command_to_rad": _mapping(fit.command_mappings[name]),
"feedback_to_rad": _mapping(fit.output_mappings[name]),
"applicability": fit.command_applicability[name],
"command_holdout": fit.command_holdout_metrics[name],
"passive": name in prepared.plan.mimic_output}
if donor is None:
row.update(zero_reference=fit.joint_zero_references[name].as_record(),
applicability=fit.command_applicability[name],
command_holdout=fit.command_holdout_metrics[name])
else:
row.update(transferred_from_joint=donor, zero_reference_source_joint=donor,
command_holdout_source_joint=donor,
baseline_command=profile.command.baseline_values[fit.command_mappings[name].motor_index],
applicability={"scope": "transferred_from_declared_joint", "source_joint": donor,
"recipient_independently_validated": False, "arbitrary_multiaxis_validated": False})
if row["passive"]:
approximation = fit.mimic_approximations[name]
if approximation.transferred_from_joint != donor:
raise ValueError(f"mimic_transfer_differs_from_profile:{name}")
parent, multiplier, offset = prepared.plan.mimic_output[name]
row["mimic_approximation"] = {"joint": parent, "multiplier": multiplier, "offset_rad": offset,
"evidence": asdict(fit.mimic_approximations[name])}
"evidence": asdict(approximation)}
if name in profile.zero.known_baseline_geometry:
row["known_baseline_geometry"] = asdict(profile.zero.known_baseline_geometry[name])
if name in profile.zero.cad_zero_assumptions:
row["cad_zero_assumption"] = profile.zero.cad_zero_assumptions[name]
if name in profile.zero.transferred_zero_sources:
row["zero_transferred_from_joint"] = profile.zero.transferred_zero_sources[name]
rows[name] = row
return {"format": REPORT_FORMAT, "schema_version": 3,
"profile_id": profile.key.profile_id, "model": profile.key.model, "side": profile.key.side,
@@ -41,8 +61,10 @@ def serialize_report(profile, fit, prepared, *, serial_number, protected_inputs)
"feedback_by_index": profile.command.feedback_by_index,
"feedback_name_aliases": dict(profile.command.feedback_name_aliases),
"coordinate_convention": ZERO_CONVENTION, "protected_inputs": dict(protected_inputs),
"joints": rows, "quality": {"training_cycles": [0, 1, 2], "holdout_cycle": 3,
"joints": rows, **({"transferred_motion_sources": transfers} if transfers else {}),
"quality": {"training_cycles": [0, 1, 2], "holdout_cycle": 3,
"precision_scope": "observed_travel_directions_and_held_poses",
"independently_measured_joints": sorted(observed), "transferred_joints": sorted(transfers),
"final_file_acceptance": "release_manifest.json", "arbitrary_multiaxis_validated": False}}
@@ -51,16 +73,71 @@ def from_report(report):
if report.get("format") != REPORT_FORMAT or report.get("coordinate_convention") != ZERO_CONVENTION:
raise ValueError("v3 requires its original measured report; legacy promotion is forbidden")
mapping = SerializedJointMapping(report, input_kind="command")
feedback = SerializedJointMapping(report, input_kind="feedback")
unit, baseline, rows = report["command_unit"], report["baseline_command"], {}
if len(baseline) != len(report["command_names"]) or not np.all(np.isfinite(baseline)):
raise ValueError("joint_baseline_vector_differs_from_sdk_layout")
transfers = report.get("transferred_motion_sources", {})
if (not isinstance(transfers, dict) or not set(transfers) <= mapping.joints.keys()
or any(not isinstance(source, str) or source not in mapping.joints or source in transfers
for source in transfers.values())):
raise ValueError("invalid_direct_motion_transfer_contract")
quality = report.get("quality", {})
if transfers or "independently_measured_joints" in quality:
if (quality.get("independently_measured_joints") != sorted(mapping.joints.keys() - transfers.keys())
or quality.get("transferred_joints") != sorted(transfers)):
raise ValueError("reported_motion_coverage_differs_from_provenance")
for name, data in mapping.joints.items():
evidence = report["joints"][name]
if evidence.get("independently_measured") is not True or evidence.get("transferred_from_joint"):
raise ValueError(f"joint_is_not_independently_measured:{name}")
reference = evidence.get("zero_reference", {})
if "cad_zero_assumption" in evidence and (
not isinstance(evidence["cad_zero_assumption"], str) or not evidence["cad_zero_assumption"].strip()
or evidence.get("zero_method") != "assumed_source_cad_zero" or evidence.get("zero_offset_rad") != 0
or "known_baseline_geometry" in evidence or "zero_transferred_from_joint" in evidence):
raise ValueError(f"assumed_cad_zero_provenance_differs:{name}")
if "zero_transferred_from_joint" in evidence:
zero_donor = report["joints"].get(evidence["zero_transferred_from_joint"], {})
if (not zero_donor or "zero_transferred_from_joint" in zero_donor
or evidence.get("zero_method") != "transferred_static_zero_on_own_cad"
or evidence.get("zero_offset_rad") != zero_donor.get("zero_offset_rad")):
raise ValueError(f"transferred_zero_provenance_differs:{name}")
index = data["motor_index"]
if (reference.get("joint") != name or reference.get("channel") != index
or reference.get("baseline_command") != baseline[index] or not reference.get("samples")):
raise ValueError(f"joint_zero_evidence_differs:{name}")
if index >= len(baseline) or feedback.joints[name]["motor_index"] != index:
raise ValueError(f"joint_input_channel_differs:{name}")
donor = transfers.get(name)
if donor is None:
if evidence.get("independently_measured") is not True or evidence.get("transferred_from_joint"):
raise ValueError(f"joint_is_not_independently_measured:{name}")
reference = evidence.get("zero_reference", {})
if (reference.get("joint") != name or reference.get("channel") != index
or reference.get("baseline_command") != baseline[index] or not reference.get("samples")):
raise ValueError(f"joint_zero_evidence_differs:{name}")
else:
source = report["joints"][donor]
if (evidence.get("independently_measured") is not False
or evidence.get("transferred_from_joint") != donor
or evidence.get("zero_reference_source_joint") != donor
or evidence.get("command_holdout_source_joint") != donor
or "zero_reference" in evidence or "command_holdout" in evidence
or evidence.get("baseline_command") != baseline[index]
or source.get("independently_measured") is not True
or baseline[index] != baseline[mapping.joints[donor]["motor_index"]]):
raise ValueError(f"transferred_joint_evidence_differs:{name}")
for kind in ("command_to_rad", "feedback_to_rad"):
# Copies preserve all knots, branches and coordinate metadata.
# Only the recipient's real SDK channel is allowed to differ.
left = {key: value for key, value in evidence[kind].items() if key != "channel_index"}
right = {key: value for key, value in source[kind].items() if key != "channel_index"}
if left != right:
raise ValueError(f"transferred_curve_differs_from_source:{name}:{kind}")
if evidence.get("passive"):
approximation = evidence.get("mimic_approximation", {})
source_approximation = source.get("mimic_approximation", {}).get("evidence", {})
expected_approximation = {**source_approximation,
"output_fit": {**source_approximation.get("output_fit", {}),
"target_joint": name, "source_joint": approximation.get("joint")},
"transferred_from_joint": donor}
if not source_approximation or approximation.get("evidence") != expected_approximation:
raise ValueError(f"transferred_mimic_evidence_differs_from_source:{name}")
knots = np.asarray(data["curve_input_knots_rad"])
angles = np.asarray(data["angle_rad"])
if not knots[0] <= baseline[index] <= knots[-1] or abs(np.interp(baseline[index], knots, angles)) > 1e-10:
@@ -0,0 +1,48 @@
"""Describe the evidence available before motion without claiming optical validation."""
from __future__ import annotations
from dataclasses import asdict, dataclass
from typing import TYPE_CHECKING, Literal, Mapping
if TYPE_CHECKING:
from .snapshot import CameraReadiness
@dataclass(frozen=True)
class CameraCheck:
status: Literal["passed", "failed", "unverified"]
reason: str
suggestion: str = ""
@dataclass(frozen=True)
class CameraPreflight:
configuration: Mapping[str, CameraCheck]
intrinsics: CameraCheck
extrinsics: CameraCheck
@classmethod
def from_readiness(cls, cameras: Mapping[str, CameraReadiness]) -> CameraPreflight:
configuration = {}
for view, camera in cameras.items():
if camera.error:
check = CameraCheck("failed", camera.error,
"检查相机绑定、分辨率和内参文件;参数不匹配时不能开始,不能只更新哈希绕过检查。")
elif camera.camera_info_ready:
check = CameraCheck("passed", "实时 CameraInfo 参数与受保护外参记录一致")
else:
check = CameraCheck("unverified", "未收到有效、新鲜的 CameraInfo",
"检查相机出图和 CameraInfo 发布;收到匹配参数后重新检查。")
configuration[view] = check
# Hand/Tag placement is allowed to change before Start. Neither old
# palm pixels nor the CameraInfo hash establishes physical validity.
# No live geometric verification is implemented in this path yet.
return cls(configuration,
CameraCheck("unverified", "仅核对加载参数,未实测验证当前焦距、畸变等光学内参",
"镜头或成像设置变化后需重新核验内参;现有手上 Tag 的单次姿态不足以证明内参有效。"),
CameraCheck("unverified", "缺少已验证的跨机位共同观测,无法确认当前相机相对外参",
"相机相对位置变化后需重新核验外参;重新锁定手掌 Tag 不能替代外参检查。"))
def as_dict(self) -> dict:
return asdict(self)
@@ -8,6 +8,7 @@ and pose estimation never hold the state lock while doing expensive work.
from __future__ import annotations
from collections import deque
from dataclasses import asdict
import hashlib
import json
import threading
@@ -24,6 +25,7 @@ from .reference_lock import ReferenceLock
from .acquisition import load_capture
from .artifacts.controller import FinalizationController, FinalizationInputs, FinalizationProtocolError
from .cameras import CameraObservations
from .camera_preflight import CameraPreflight
from .capture import CaptureFrame, ObservationCapture
from .execution import SessionExecution
from .inputs import AdapterFactory, CameraModelInput, DetectionInput, RuntimePorts, ServiceResult
@@ -64,6 +66,7 @@ class CalibrationCoordinator:
self.step_speed_ready_at = 0.0
self.reason = "waiting_for_feedback_and_camera_info"
self._started_at: float | None = None
self._camera_preflight: CameraPreflight | None = None
self._observation_epoch = 0
self._pause_effects_applied = False
self.segment = None
@@ -116,6 +119,7 @@ class CalibrationCoordinator:
if self.execution.session.phase != Phase.READY or not self.parameters.commands_enabled:
return ServiceResult(False, self.reason)
devices = self._device_status(self.ports.monotonic())
self._record_camera_preflight(devices, at_start=devices.ready)
if not devices.ready:
self.execution.session.devices_unavailable()
self.reason = "; ".join(devices.waiting_for)
@@ -237,6 +241,7 @@ class CalibrationCoordinator:
fresh = self.sdk_adapter.health()
if not self.started:
devices = self._device_status(now)
self._record_camera_preflight(devices)
if devices.ready:
if session.phase == Phase.WAIT_DEVICE:
session.device_ready()
@@ -418,6 +423,26 @@ class CalibrationCoordinator:
self._unit_rows = []
self._capture_unit_key = key
def _record_camera_preflight(self, devices: DeviceReadiness, *, at_start: bool = False) -> None:
"""Journal changed checks and freeze the last check before any motion."""
report = CameraPreflight.from_readiness(devices.cameras)
if not at_start and report == self._camera_preflight:
return
record = {
"kind": "camera_preflight", "schema_version": 1,
"stage": "at_start" if at_start else "before_start",
"stamp_ns": self.ports.clock_ns(),
"protected_inputs": dict(self.parameters.protected_inputs),
# These are the expected configuration identities, not live
# measurements of serial numbers or physical camera geometry.
"expected_cameras": {view: asdict(identity)
for view, identity in self.parameters.extrinsics.cameras.items()},
**report.as_dict(),
}
atomic_write_json(self.parameters.session_dir / "camera_preflight.json", record)
append_jsonl(self.raw_path, record)
self._camera_preflight = report
def _device_status(self, now):
"""Readiness is live, and independent of whether a Tag is visible.
@@ -125,8 +125,8 @@ def run_online(
contract_report = validate_executable_profile(profile, config.source_urdf)
atomic_write_json(session/"measurement_contract.json", contract_report)
coverage = contract_report["independent_measurement_coverage"]
if profile.artifacts.output_schema_version >= 3 and (coverage["unmeasured_joints"] or coverage["missing_baseline_geometry"]):
print("当前 Profile 的全关节实测/绝对几何依据不完整,本次可采集诊断数据,但不能发布完整 v3 JSON/URDF"
if profile.artifacts.output_schema_version >= 3 and (coverage["unresolved_motion_joints"] or coverage["missing_baseline_geometry"]):
print("当前 Profile 仍有未实测且未声明迁移的关节,或缺少绝对零位来源;本次可采集诊断数据,但不能发布完整 v3 JSON/URDF"
"缺失关节见 measurement_contract.json。", flush=True)
if contract_report["cad_constraint_conflicts"]:
print("原始 CAD 存在限位/mimic 冲突,未修改原文件;详见 measurement_contract.json。最终范围验证不通过时禁止发布。", flush=True)
@@ -6,6 +6,7 @@ from dataclasses import asdict, dataclass, replace
from typing import TYPE_CHECKING, Mapping
from ..core.domain.status import CalibrationStatus, MotionStatus
from .camera_preflight import CameraPreflight
from .scan_quality import evaluate_capture_unit
from .session import CalibrationPhase as Phase
@@ -55,6 +56,7 @@ class RuntimeSnapshot:
devices: DeviceReadiness | None
steady: bool = False
steady_node: int | None = None
camera_preflight: CameraPreflight | None = None
def as_dict(self) -> dict:
value = self.calibration.as_dict()
@@ -63,6 +65,8 @@ class RuntimeSnapshot:
session_dir=self.session_dir, latest_base_drift_px=dict(self.latest_base_drift_px))
if self.devices is not None:
value["devices"] = asdict(self.devices)
if self.camera_preflight is not None:
value["camera_preflight"] = self.camera_preflight.as_dict()
if self.steady:
value["acquisition"]["steady_node"] = self.steady_node
return value
@@ -134,7 +138,9 @@ def build_snapshot(host: CalibrationCoordinator) -> RuntimeSnapshot:
outputs={"json": host.final_json, "urdf": host.final_urdf,
"publication_pointer": str(parameters.session_dir.parent/profile.artifacts.publication_pointer)
if session.phase == Phase.COMPLETE else ""})
devices = None if host.started else host._device_status(host.ports.monotonic())
preflight = host._camera_preflight if devices is None else CameraPreflight.from_readiness(devices.cameras)
return RuntimeSnapshot(status, profile.key.model, profile.key.side, profile.key.profile_id,
parameters.serial_number, host.reason, str(parameters.session_dir),
dict(host.reference_lock.latest_drift_px),
None if host.started else host._device_status(host.ports.monotonic()), steady, steady_node)
devices, steady, steady_node, preflight)
@@ -137,6 +137,25 @@ def render_status_zh(value: Mapping[str, Any], estimator=None) -> str:
waiting = value.get("devices", {}).get("waiting_for", ())
if waiting:
lines.append("等待项:" + "".join(waiting))
preflight = value.get("camera_preflight")
if preflight:
labels = {"passed": "通过", "failed": "失败", "unverified": "未验证"}
configuration = preflight.get("configuration", {})
lines.append("启动前相机配置:" + "".join(
f"{view}={labels.get(check['status'], check['status'])}"
for view, check in configuration.items()))
lines.append("启动前实测:" + "".join(
f"{label}={labels.get(preflight[key]['status'], preflight[key]['status'])}"
for key, label in (("intrinsics", "光学内参"), ("extrinsics", "相对外参"))))
if value.get("state") in {"READY", "WAIT_DEVICE", "WAIT_DEVICES"}:
for view, check in configuration.items():
if check["status"] != "passed":
lines.append(f"{view}{check['reason']}{check['suggestion']}")
for key in ("intrinsics", "extrinsics"):
check = preflight[key]
if check["status"] != "passed":
lines.append(f"检查依据:{check['reason']}{check['suggestion']}")
lines.append("提示:设备就绪不代表内外参实测通过;缺少依据时无法判断是否需重新标定。")
if motion.get("speed") is not None:
lines.append(f"运动速度:{motion['speed']} {'rad/s' if unit == 'rad' else 'SDK速度档'}")
if value.get("reference_locked"):
@@ -44,8 +44,8 @@ def reordered_profile(profile):
measurements={name: replace(spec, view=cameras.get(spec.view)) for name, spec in profile.measurement.measurements.items()}))
def measured_capture(tmp_path, *, nonlinear=True, reordered=False):
profile, source, records, offsets = virtual_capture(tmp_path, "g20_right_19", include_baseline=True,
def measured_capture(tmp_path, *, nonlinear=True, reordered=False, layout="g20_right_19"):
profile, source, records, offsets = virtual_capture(tmp_path, layout, include_baseline=True,
profile_transform=reordered_profile if reordered else None)
model = UrdfKinematicModel(source)
for name, joint in model.joints.items():
@@ -64,8 +64,11 @@ def measured_capture(tmp_path, *, nonlinear=True, reordered=False):
motion=replace(profile.motion, joint_zero_references=specs),
urdf_authorized_fields={name: tuple(sorted(set(profile.urdf_authorized_fields.get(name, ()))
| {"origin.rpy", "limit.lower", "limit.upper"})) for name, joint in model.joints.items() if joint.kind != "fixed"},
zero=replace(profile.zero, known_baseline_geometry={name: KnownBaselineGeometry(value, 0, "synthetic_fixture_truth")
for name, value in offsets.items() if name not in profile.zero.direct_zero_joints}))
# Synthetic datums replace the production CAD-zero premise; preserve
# the original nonzero truth and numerical regression expectations.
zero=replace(profile.zero, cad_zero_assumptions={}, known_baseline_geometry={name: KnownBaselineGeometry(value, 0, "synthetic_fixture_truth")
for name, value in offsets.items() if name not in profile.zero.direct_zero_joints
and name not in profile.zero.transferred_zero_sources}))
mount = rigid((.12, -.09, .2), (.009, -.025, .016))
identities = {identity: (i+1)*1000 for i, identity in enumerate(dict.fromkeys(r["sample_id"] for r in records))}
updated = []
@@ -0,0 +1,222 @@
"""Pre-Start evidence must not confuse protected files with physical calibration."""
import json
from dataclasses import FrozenInstanceError, asdict, replace
import numpy as np
import pytest
from linkerhand_calibration.core.geometry.extrinsics import (
CameraCalibrationIdentity,
CameraExtrinsics,
camera_info_fingerprint,
)
from linkerhand_calibration.runtime.camera_preflight import CameraPreflight
from linkerhand_calibration.runtime.cameras import CameraObservations
from linkerhand_calibration.runtime.inputs import CameraModelInput
from linkerhand_calibration.runtime.snapshot import CameraReadiness
from linkerhand_calibration.runtime.status import render_status_zh
from runtime_host_fixture import coordinator_fixture, ready
def preflight_rows(host):
path = host.parameters.session_dir / "raw_samples.jsonl"
return [row for line in path.read_text().splitlines()
if (row := json.loads(line)).get("kind") == "camera_preflight"]
def saved_preflight(host):
return json.loads((host.parameters.session_dir / "camera_preflight.json").read_text())
def assert_physical_calibration_unverified(report):
for name in ("intrinsics", "extrinsics"):
assert report[name]["status"] == "unverified"
assert report[name]["reason"]
assert report[name]["suggestion"]
@pytest.mark.parametrize("detections_ready", [False, True])
def test_valid_camera_info_only_proves_configuration(detections_ready):
report = CameraPreflight.from_readiness({
"arbitrary_camera_name": CameraReadiness(True, detections_ready, ""),
})
payload = report.as_dict()
assert payload["configuration"]["arbitrary_camera_name"]["status"] == "passed"
assert_physical_calibration_unverified(payload)
with pytest.raises(FrozenInstanceError):
report.intrinsics = None
def test_configuration_failure_takes_precedence_over_old_ready_flags():
payload = CameraPreflight.from_readiness({
"front": CameraReadiness(True, True, "camera_info_does_not_match_extrinsics"),
"side": CameraReadiness(False, True, ""),
"top": CameraReadiness(True, False, ""),
}).as_dict()
assert payload["configuration"]["front"]["status"] == "failed"
assert "camera_info_does_not_match_extrinsics" in payload["configuration"]["front"]["reason"]
assert payload["configuration"]["front"]["suggestion"]
assert payload["configuration"]["side"]["status"] == "unverified"
assert payload["configuration"]["top"]["status"] == "passed"
assert_physical_calibration_unverified(payload)
@pytest.mark.parametrize("model", ["o6", "l6", "o12", "g20"])
def test_all_models_record_start_evidence_without_requiring_visible_finger_tags(tmp_path, model):
host, clock = coordinator_fixture(tmp_path, model=model)
try:
# The detector stream is alive, but no hand Tag or historical pose is supplied.
ready(host, clock)
report = host.snapshot().as_dict()["camera_preflight"]
assert all(item["status"] == "passed" for item in report["configuration"].values())
assert_physical_calibration_unverified(report)
before_start = saved_preflight(host)
assert before_start["kind"] == "camera_preflight"
assert before_start["stage"] == "before_start"
assert before_start["stamp_ns"] == host.ports.clock_ns()
assert before_start["protected_inputs"] == dict(host.parameters.protected_inputs)
assert before_start["expected_cameras"] == {
view: asdict(identity) for view, identity in host.parameters.extrinsics.cameras.items()
}
for key, value in report.items():
assert before_start[key] == value
assert not clock.positions
assert host.start().success
at_start = saved_preflight(host)
assert at_start["stage"] == "at_start"
assert_physical_calibration_unverified(at_start)
assert preflight_rows(host)[-1] == at_start
# Later freshness changes cannot rewrite the historical Start evidence.
clock.now += 3.0
assert host.snapshot().as_dict()["camera_preflight"] == report
assert saved_preflight(host) == at_start
finally:
host.close()
def test_unchanged_preflight_does_not_fill_the_sample_journal(tmp_path):
host, clock = coordinator_fixture(tmp_path)
try:
ready(host, clock)
original = preflight_rows(host)
for _ in range(5):
clock.now += 0.05
host.tick()
host.snapshot()
assert preflight_rows(host) == original
view = host.profile.vision.view_names[0]
host.cameras.errors[view] = "camera_info_does_not_match_extrinsics"
host.tick()
assert len(preflight_rows(host)) == len(original) + 1
assert saved_preflight(host)["configuration"][view]["status"] == "failed"
host.tick()
assert len(preflight_rows(host)) == len(original) + 1
finally:
host.close()
@pytest.mark.parametrize("fault", ["camera_info_stale", "camera_info_mismatch", "detector_stale"])
def test_start_rechecks_live_devices_and_records_rejection_before_motion(tmp_path, fault):
host, clock = coordinator_fixture(tmp_path)
try:
ready(host, clock)
view = host.profile.vision.view_names[0]
if fault == "camera_info_stale":
host.cameras.info_received_at[view] = clock.now - 2.01
expected = "unverified"
elif fault == "camera_info_mismatch":
host.cameras.errors[view] = "camera_info_does_not_match_extrinsics"
expected = "failed"
else:
host.cameras.detections_received_at[view] = clock.now - 2.01
expected = "passed"
positions, settings = list(clock.positions), list(clock.settings)
assert not host.start().success
assert not host.started
report = saved_preflight(host)
assert report["stage"] == "before_start"
assert report["configuration"][view]["status"] == expected
assert_physical_calibration_unverified(report)
assert preflight_rows(host)[-1] == report
assert clock.positions == positions
assert clock.settings == settings
finally:
host.close()
@pytest.mark.parametrize("state", ["READY", "WAIT_DEVICE", "BASELINE"])
def test_progress_distinguishes_configuration_from_unverified_physical_calibration(state):
report = CameraPreflight.from_readiness({"side": CameraReadiness(True, True, "")}).as_dict()
rendered = render_status_zh({"state": state, "phase": state, "camera_preflight": report})
assert "启动前相机配置:side=通过" in rendered
assert "启动前实测:光学内参=未验证;相对外参=未验证" in rendered
if state in {"READY", "WAIT_DEVICE"}:
assert "设备就绪不代表内外参实测通过" in rendered
assert "缺少依据时无法判断是否需重新标定" in rendered
assert report["intrinsics"]["suggestion"] in rendered
assert report["extrinsics"]["suggestion"] in rendered
else:
assert "检查依据:" not in rendered
@pytest.mark.parametrize("writer", ["atomic_write_json", "append_jsonl"])
def test_failed_start_evidence_write_cannot_start_motion(tmp_path, monkeypatch, writer):
host, clock = coordinator_fixture(tmp_path)
try:
ready(host, clock)
positions, settings = list(clock.positions), list(clock.settings)
def failed_write(*args, **kwargs):
raise OSError("synthetic_preflight_write_failure")
monkeypatch.setattr(f"linkerhand_calibration.runtime.coordinator.{writer}", failed_write)
with pytest.raises(OSError, match="synthetic_preflight_write_failure"):
host.start()
assert not host.started
assert clock.positions == positions
assert clock.settings == settings
finally:
host.close()
@pytest.mark.parametrize("changed_field", ["k", "d", "r", "p", "width"])
def test_real_camera_model_fingerprint_check_is_reported_without_physical_pass(tmp_path, changed_field):
message = CameraModelInput(
width=640, height=480,
k=(600., 0., 300., 0., 620., 220., 0., 0., 1.),
d=(.1, .01, 0., 0., 0.),
r=tuple(np.eye(3).flat),
p=(500., 0., 320., 0., 0., 510., 240., 0., 0., 0., 1., 0.),
)
identity = CameraCalibrationIdentity("synthetic_camera", message.width, message.height,
camera_info_fingerprint(width=message.width, height=message.height,
camera_matrix=message.k, distortion=message.d, rectification=message.r, projection=message.p))
extrinsics = CameraExtrinsics({"test_view": identity}, "test_view",
{"test_view": np.eye(4)}, {})
observations = CameraObservations(extrinsics, tmp_path / "raw.jsonl", lambda: 10.)
pauses = []
def accept_and_report(model):
observations.accept("test_view", model, started=False, pause=pauses.append, trackers={})
return CameraPreflight.from_readiness({"test_view": CameraReadiness(
"test_view" in observations.info_received_at, True,
observations.errors.get("test_view", ""))}).as_dict()
valid = accept_and_report(message)
assert valid["configuration"]["test_view"]["status"] == "passed"
assert_physical_calibration_unverified(valid)
old_value = getattr(message, changed_field)
new_value = old_value + 1 if changed_field == "width" else (old_value[0] + .01, *old_value[1:])
invalid = accept_and_report(replace(message, **{changed_field: new_value}))
assert invalid["configuration"]["test_view"]["status"] == "failed"
assert_physical_calibration_unverified(invalid)
assert not pauses # Before Start this is a device gate, not a running-session pause.
assert "test_view" not in observations.matrices
restored = accept_and_report(message)
assert restored["configuration"]["test_view"]["status"] == "passed"
assert_physical_calibration_unverified(restored)
@@ -0,0 +1,67 @@
"""O6 synthetic records traverse the actual mixed v3 production finalizer.
The elementary-FK fixture supplies synthetic geometry datums and its existing
virtual limits. This is a software integration check, not O6 hardware evidence.
"""
import hashlib
import json
import pytest
from measured_capture_fixture import measured_capture
from linkerhand_calibration.core.fitting.motion_fit import channel_for_joint
from linkerhand_calibration.core.urdf.kinematics import UrdfKinematicModel
from linkerhand_calibration.runtime.artifacts.finalization import finalize_profile_session
from linkerhand_calibration.runtime.artifacts.reader import load_unified_mapper
@pytest.mark.parametrize("reordered", [False, True], ids=["o6", "reordered_virtual_o6"])
def test_o6_records_fit_expand_publish_and_read_back_without_recipient_observations(tmp_path, reordered):
profile, source, records, offsets = measured_capture(tmp_path, layout="o6_right_8", reordered=reordered)
source_bytes = source.read_bytes()
transfers = dict(profile.measurement.transferred_motion_sources)
assert len(transfers) == 6
observed = {row["joint"] for row in records if row.get("kind") == "joint_sample"}
assert len(observed) == 5 and not observed & transfers.keys()
assert {row["joint"] for row in records if row.get("kind") == "joint_zero_reference"} == observed
hashes = {name: "a"*64 for name in profile.artifacts.protected_input_fields}
hashes["source_urdf_sha256"] = hashlib.sha256(source_bytes).hexdigest()
phases = []
payload, fit, artifact = finalize_profile_session(profile=profile, session_dir=tmp_path/"session",
serial_number="SYNTHETIC_O6_ONLY", source_urdf=source, protected_inputs=hashes,
records=records, standard_loader=lambda _: None, phase_changed=phases.append)
assert source.read_bytes() == source_bytes
assert phases == ["fit_complete", "holdout_complete", "artifacts_built", "urdf_validated"]
moving = {name for name, joint in UrdfKinematicModel(source).joints.items() if joint.kind != "fixed"}
assert len(moving) == 11 and set(payload["joints"]) == moving
assert set(fit.curves) == set(fit.joint_zero_references) == set(fit.command_holdout_metrics) == observed
assert set(fit.command_mappings) == set(fit.output_mappings) == moving
report = json.loads((artifact.path.parent/"calibration_report.json").read_text())
acceptance = json.loads(artifact.staged_release.validation_json)
assert acceptance["acceptance"] == "measured_and_transferred_json_urdf_v3"
assert set(acceptance["independent_mapping_holdout"]) == observed
assert acceptance["transferred_motion_sources"] == transfers
observed_roles = {profile.measurement.measurements[name].child_role for name in observed}
assert set(acceptance["full_json_urdf_holdout"]) == observed_roles
assert max(row["maximum_deg"] for row in acceptance["urdf_mimic_approximation"].values()) > 3
for target, donor in transfers.items():
assert payload["joints"][target]["angle_rad"] == payload["joints"][donor]["angle_rad"]
assert payload["joints"][target]["sdk_channel"] == channel_for_joint(profile, target)
assert fit.command_mappings[target].transferred_from == donor
row = report["joints"][target]
assert row["independently_measured"] is False and "zero_reference" not in row
assert "command_holdout" not in row and row["command_holdout_source_joint"] == donor
for name, angle in offsets.items():
assert fit.zero_offsets_rad[name] == pytest.approx(angle, abs=.004)
mapper = load_unified_mapper(artifact.path.parent/"release_manifest.json")
assert mapper.transferred_motion_sources == transfers
baseline = profile.command.baseline_values
assert max(abs(value) for value in mapper.map_positions(baseline)) < 1e-10
target = "rh_index_mcp_pitch"
command = list(baseline)
command[channel_for_joint(profile, target)] = 102
angles = dict(zip(mapper.urdf_joint_names, mapper.map_positions(command)))
assert angles["rh_pinky_mcp_pitch"] == pytest.approx(0, abs=1e-10)
assert angles[target] == payload["joints"][target]["angle_rad"][102]
assert angles["rh_index_dip"] == payload["joints"]["rh_index_dip"]["angle_rad"][102]
@@ -0,0 +1,276 @@
"""Declared finger transfers preserve frozen motion and recipient CAD geometry.
These synthetic frozen results exercise integration contracts, not hardware
accuracy or the acquisition/optimization stages covered by capture tests.
"""
from dataclasses import replace
import hashlib
from pathlib import Path
import xml.etree.ElementTree as ET
import numpy as np
import pytest
from measured_capture_fixture import reordered_profile
from linkerhand_calibration.core.domain.measurement import JointCurveFit
from linkerhand_calibration.core.domain.profile import KnownBaselineGeometry, ProfileValidationError
from linkerhand_calibration.core.domain.reference import JointZeroReference
from linkerhand_calibration.core.domain.result import CalibrationResult, JointMapping, evaluate_mappings
from linkerhand_calibration.core.fitting.coupling import LinearMimicFit
from linkerhand_calibration.core.fitting.mimic_approximation import MimicApproximation
from linkerhand_calibration.core.fitting.motion_fit import channel_for_joint
from linkerhand_calibration.core.fitting.motion_transfer import expand_profile_transfers
from linkerhand_calibration.core.fitting.session import fit_profile_calibration
from linkerhand_calibration.core.fitting.spatial import ZeroSolveResult
from linkerhand_calibration.core.urdf.result_plan import prepare_standard_result
from linkerhand_calibration.profiles.loader import load_bundled_hand_profile
PACKAGE = Path(__file__).resolve().parents[1]
O6_SOURCE = PACKAGE / "urdf/o6_right/linkerhand_o6_right.urdf"
O12_SOURCE = PACKAGE / "urdf/o12_right/linkerhand_o12_t3_right-0703.urdf"
def _digest(path):
return hashlib.sha256(path.read_bytes()).hexdigest()
def _frozen_fit(profile):
observed = {name for name, spec in profile.measurement.measurements.items()
if spec.kind != "validation_only" and name in profile.motion.joint_zero_references}
outputs, commands, curves, references = {}, {}, {}, {}
for name in sorted(observed):
channel = channel_for_joint(profile, name)
lo, hi = profile.command.minimum_values[channel], profile.command.maximum_values[channel]
baseline = profile.command.baseline_values[channel]
knots = np.arange(256.) if profile.command.unit == "u8" else np.unique([lo, (lo+hi)/2, baseline, hi])
travel = (knots-baseline)/(hi-lo)*profile.command.joint_directions[channel]
# The passive table is deliberately nonlinear and not equal to mimic.
y = (.18*travel+.07*travel**2 if name in profile.zero.passive_joints
else .35*travel+.05*travel**2)
increasing, decreasing = tuple(y+.001*np.abs(travel)), tuple(y-.001*np.abs(travel))
outputs[name] = JointMapping(name, channel, f"feedback_{profile.command.unit}",
tuple(knots), tuple(y), increasing, decreasing)
command_y = tuple(y+.002*travel**2)
commands[name] = replace(outputs[name], input_domain=f"command_{profile.command.unit}",
angle_rad=command_y)
curves[name] = JointCurveFit(tuple(y), decreasing, increasing, {}, 0., .002, {})
spec = profile.motion.joint_zero_references[name]
# Samples are intentionally absent: these references test ownership,
# and this helper is never used as a publishable capture fixture.
references[name] = JointZeroReference(name, spec.task_key, baseline, channel, 1, 3, ())
offsets = {name: .02 for name in profile.zero.direct_zero_joints}
methods = {name: "urdf_serial_axis_geometry" for name in offsets}
for name, geometry in profile.zero.known_baseline_geometry.items():
offsets[name] = geometry.cad_angle_rad
methods[name] = "independent_known_baseline_geometry"
for name in profile.zero.cad_zero_assumptions:
offsets[name] = 0.
methods[name] = "assumed_source_cad_zero"
spatial = ZeroSolveResult(direct_offsets_rad={}, all_active_offsets_rad={},
base_translation_xyz_m=(0., 0., 0.), base_quaternion_xyzw=(0., 0., 0., 1.),
validation_errors_rad=(), validation_error_by_joint_rad={}, validation_line_error_by_joint_m={},
axis_line_rms_m=0., passed=True, cycle_offsets_rad={}, offset_uncertainty_rad={},
offset_confidence_half_width_rad={}, training_cycles=(0, 1, 2), validation_cycle=3,
validation_original_error_by_joint_rad={}, validation_improvement_by_joint_rad={},
validation_improvement_confidence_lower_rad={}, observability_rank=0,
observability_parameter_count=0, observability_condition_number=1., offset_covariance_rad2={},
axis_cone_mismatch_by_joint_rad={}, axis_cone_bias_classification_by_joint={}, failure_reasons={})
mimics = {name: MimicApproximation(
LinearMimicFit(parent, name, .6, 0., False, (0, 1, 2), (.001, -.002)),
(f"{name}:train",), (f"{name}:holdout",), (.003,))
for name, parent in profile.zero.mimic_source_by_joint.items()
if name in observed and parent in observed}
return CalibrationResult(curves, outputs, offsets, methods, {},
{name: (.003,) for name in observed}, spatial,
reference_inputs={name: ref.baseline_command for name, ref in references.items()},
command_mappings=commands,
command_holdout_metrics={name: {"independently_measured": True} for name in observed},
command_applicability={name: {"scope": "single_channel_at_declared_pose"} for name in observed},
joint_zero_references=references, mimic_approximations=mimics)
@pytest.mark.parametrize("layout,count", [
("o6_right_8", 6), ("l6_right_8", 6), ("o12_right_16", 3), ("g20_right_19", 0),
])
@pytest.mark.parametrize("reordered", [False, True], ids=["product", "virtual_reordered"])
def test_three_observation_strategies_use_recipient_channels_without_refitting(layout, count, reordered):
profile = load_bundled_hand_profile(layout)
if reordered:
profile = reordered_profile(profile)
original = _frozen_fit(profile)
expanded = expand_profile_transfers(profile, original)
transfers = profile.measurement.transferred_motion_sources
assert len(transfers) == count
assert set(expanded.output_mappings) == set(profile.zero.active_joints) | set(profile.zero.passive_joints)
for field in ("curves", "joint_zero_references", "holdout_errors_rad", "command_holdout_metrics"):
assert getattr(expanded, field) == getattr(original, field)
assert not set(transfers) & getattr(expanded, field).keys()
for target, source in transfers.items():
recipient_channel = channel_for_joint(profile, target)
assert recipient_channel != channel_for_joint(profile, source)
for field in ("output_mappings", "command_mappings"):
donor = getattr(original, field)[source]
mapping = getattr(expanded, field)[target]
assert mapping == replace(donor, joint=target, motor_index=recipient_channel, transferred_from=source)
assert mapping.evaluate(profile.command.baseline_values[recipient_channel]) == 0.
positions = list(profile.command.baseline_values)
positions[recipient_channel] = donor.knots[len(donor.knots)//2]
actual = evaluate_mappings(getattr(expanded, field), positions, {})
assert actual[target] == donor.evaluate(positions[recipient_channel])
assert actual[source] == 0.
assert target not in original.output_mappings
if target in profile.zero.passive_joints:
approximate = expanded.mimic_approximations[target]
source_fit = original.mimic_approximations[source]
assert approximate.transferred_from_joint == source
assert approximate.training_sample_ids == source_fit.training_sample_ids
assert approximate.holdout_sample_ids == source_fit.holdout_sample_ids
assert approximate.holdout_errors_rad == source_fit.holdout_errors_rad
assert approximate.output_fit.source_joint == profile.zero.mimic_source_by_joint[target]
parent = expanded.command_mappings[approximate.output_fit.source_joint]
child = expanded.command_mappings[target]
assert np.max(np.abs(np.asarray(child.angle_rad)-.6*np.asarray(parent.angle_rad))) > .005
def test_old_zero_and_mimic_transfer_declarations_do_not_authorize_v3_motion_copy():
profile = load_bundled_hand_profile("o6_right_8")
profile = replace(profile, measurement=replace(profile.measurement, transferred_motion_sources={}))
original = _frozen_fit(profile)
expanded = expand_profile_transfers(profile, original)
assert expanded == original
assert "rh_index_mcp_pitch" not in expanded.output_mappings
with pytest.raises(ValueError, match="unmeasured"):
prepare_standard_result(profile, O6_SOURCE, _digest(O6_SOURCE), expanded)
def test_o12_motion_transfer_does_not_invent_passive_pip_zero():
profile = load_bundled_hand_profile("o12_right_16")
expanded = expand_profile_transfers(profile, _frozen_fit(profile))
for name in ("pinky_pip", "ring_pip"):
assert name in expanded.output_mappings
assert name not in expanded.zero_offsets_rad
assert name not in profile.zero.cad_zero_assumptions
with pytest.raises(ValueError, match="absolute_geometry.*pinky_pip.*ring_pip"):
prepare_standard_result(profile, O12_SOURCE, _digest(O12_SOURCE), expanded)
def test_passive_pip_zero_uses_separate_datums_while_motion_is_copied():
profile = load_bundled_hand_profile("o12_right_16")
geometry = dict(profile.zero.known_baseline_geometry,
pinky_pip=KnownBaselineGeometry(.07, .001, "synthetic_pinky_datum"))
profile = replace(profile, zero=replace(profile.zero, known_baseline_geometry=geometry))
expanded = expand_profile_transfers(profile, _frozen_fit(profile))
assert expanded.zero_offsets_rad["pinky_pip"] == .07
assert "ring_pip" not in expanded.zero_offsets_rad
geometry = dict(geometry, ring_pip=KnownBaselineGeometry(.11, .001, "synthetic_ring_datum"))
profile = replace(profile, zero=replace(profile.zero, known_baseline_geometry=geometry))
expanded = expand_profile_transfers(profile, _frozen_fit(profile))
assert expanded.zero_offsets_rad["pinky_pip"] == .07
assert expanded.zero_offsets_rad["ring_pip"] == .11
assert expanded.zero_method_by_joint["ring_pip"] == "independent_known_baseline_geometry"
assert "ring_pip" not in profile.zero.transferred_zero_sources
assert expanded.command_mappings["ring_pip"].angle_rad == expanded.command_mappings["pinky_pip"].angle_rad
assert "ring_pip" not in expanded.joint_zero_references
def test_o6_urdf_keeps_terminal_origins_and_each_recipient_link_geometry(tmp_path):
profile = load_bundled_hand_profile("o6_right_8")
expanded = expand_profile_transfers(profile, _frozen_fit(profile))
source_hash = _digest(O6_SOURCE)
prepared = prepare_standard_result(profile, O6_SOURCE, source_hash, expanded)
destination = prepared.plan.write(O6_SOURCE, tmp_path/"corrected.urdf")
before, after = ET.parse(O6_SOURCE).getroot(), ET.parse(destination).getroot()
before_joints = {j.get("name"): j for j in before.findall("joint")}
after_joints = {j.get("name"): j for j in after.findall("joint")}
for name in profile.zero.cad_zero_assumptions:
assert expanded.zero_offsets_rad[name] == 0.
assert after_joints[name].find("origin").attrib == before_joints[name].find("origin").attrib
assert prepared.plan.mimic_output[name][2] == 0.
for name, original in before_joints.items():
current = after_joints[name]
for tag in ("parent", "child", "axis"):
assert current.find(tag).attrib == original.find(tag).attrib
assert current.find("origin").get("xyz") == original.find("origin").get("xyz")
assert [ET.tostring(link) for link in before.findall("link")] == [ET.tostring(link) for link in after.findall("link")]
assert after_joints["rh_index_mcp_pitch"].find("origin").get("xyz") != after_joints["rh_pinky_mcp_pitch"].find("origin").get("xyz")
assert _digest(O6_SOURCE) == source_hash
@pytest.mark.parametrize("field", ["output_mappings", "command_mappings"])
def test_changed_recipient_curve_cannot_pass_as_exact_transfer(field):
profile = load_bundled_hand_profile("o6_right_8")
expanded = expand_profile_transfers(profile, _frozen_fit(profile))
mappings = dict(getattr(expanded, field))
target = "rh_index_dip"
mappings[target] = replace(mappings[target], angle_rad=tuple(v*1.01 for v in mappings[target].angle_rad))
with pytest.raises(ValueError, match="differs_from_declared_donor"):
prepare_standard_result(profile, O6_SOURCE, _digest(O6_SOURCE), replace(expanded, **{field: mappings}))
def test_transferred_curve_cannot_expand_recipient_mechanical_range(tmp_path):
profile = load_bundled_hand_profile("o6_right_8")
expanded = expand_profile_transfers(profile, _frozen_fit(profile))
source = tmp_path/"narrow_recipient.urdf"
tree = ET.parse(O6_SOURCE)
tree.getroot().find("joint[@name='rh_index_mcp_pitch']/limit").set("upper", "0.2")
tree.write(source, encoding="utf-8", xml_declaration=True)
with pytest.raises(ValueError, match="range.*(mechanical|physical)|mapping exceeds"):
prepare_standard_result(profile, source, _digest(source), expanded)
@pytest.mark.parametrize("problem", ["missing_donor", "wrong_channel", "already_expanded"])
def test_invalid_donor_result_cannot_be_transferred(problem):
profile = load_bundled_hand_profile("o6_right_8")
fit = _frozen_fit(profile)
mappings = dict(fit.command_mappings)
if problem == "missing_donor":
mappings.pop("rh_pinky_dip")
fit = replace(fit, command_mappings=mappings)
elif problem == "wrong_channel":
mappings["rh_pinky_dip"] = replace(mappings["rh_pinky_dip"], motor_index=0)
fit = replace(fit, command_mappings=mappings)
else:
fit = expand_profile_transfers(profile, fit)
with pytest.raises(ValueError, match="motion_transfer"):
expand_profile_transfers(profile, fit)
@pytest.mark.parametrize("policy,joint,reason", [
("transferred", "rh_index_mcp_pitch", "transferred_zero_differs_from_declared_donor"),
("assumed", "rh_pinky_dip", "assumed_cad_zero_differs_from_profile"),
("known", "rh_thumb_ip", "known_baseline_geometry_differs_from_profile"),
])
@pytest.mark.parametrize("changed", ["offset", "method"])
def test_result_plan_rejects_zero_parameters_that_change_the_declared_evidence(policy, joint, reason, changed):
profile = load_bundled_hand_profile("o6_right_8")
if policy == "known":
assumptions = {name: value for name, value in profile.zero.cad_zero_assumptions.items() if name != joint}
geometry = dict(profile.zero.known_baseline_geometry,
**{joint: KnownBaselineGeometry(0., .001, "synthetic_independent_datum")})
profile = replace(profile, zero=replace(profile.zero,
known_baseline_geometry=geometry, cad_zero_assumptions=assumptions))
fit = expand_profile_transfers(profile, _frozen_fit(profile))
# Verify the valid starting point, then change one field only. A copied
# dynamic curve must not authorize a new geometric zero or evidence label.
prepare_standard_result(profile, O6_SOURCE, _digest(O6_SOURCE), fit)
if changed == "offset":
offsets = dict(fit.zero_offsets_rad)
offsets[joint] += .04
fit = replace(fit, zero_offsets_rad=offsets)
else:
methods = dict(fit.zero_method_by_joint)
methods[joint] = "urdf_serial_axis_geometry"
fit = replace(fit, zero_method_by_joint=methods)
with pytest.raises(ValueError, match=reason):
prepare_standard_result(profile, O6_SOURCE, _digest(O6_SOURCE), fit)
def test_public_fitting_rejects_conflicting_zero_sources_before_reading_inputs(tmp_path):
profile = load_bundled_hand_profile("o6_right_8")
geometry = dict(profile.zero.known_baseline_geometry,
rh_pinky_dip=KnownBaselineGeometry(.03, .001, "synthetic_independent_datum"))
profile = replace(profile, zero=replace(profile.zero, known_baseline_geometry=geometry))
with pytest.raises(ProfileValidationError, match="CAD zero assumption conflicts.*rh_pinky_dip"):
fit_profile_calibration(profile, tmp_path/"nonexistent.urdf", {})
@@ -0,0 +1,166 @@
"""Measured and explicitly reused motions share one profile contract."""
from dataclasses import replace
import hashlib
from pathlib import Path
import xml.etree.ElementTree as ET
import pytest
import yaml
from linkerhand_calibration.core.domain.profile import ProfileKey, validate_profile
from linkerhand_calibration.profiles import dump_hand_profile, load_bundled_hand_profile, load_hand_profile
from linkerhand_calibration.profiles.validator import validate_executable_profile
PACKAGE = Path(__file__).resolve().parents[1]
PROFILES = (
("o6_right_8", "o6_right/linkerhand_o6_right.urdf", 6, ()),
("l6_right_8", "l6_right/linkerhand_l6v3.1_right.urdf", 6, ()),
("o12_right_16", "o12_right/linkerhand_o12_t3_right-0703.urdf", 3,
("pinky_pip", "ring_pip", "thumb_mcp")),
("g20_right_19", "g20_right/linkerhand_g20_right.urdf", 0, ()),
)
@pytest.mark.parametrize("layout,urdf,transfer_count,geometry_gaps", PROFILES)
def test_common_strategy_retains_measured_coverage_and_explicit_gaps(layout, urdf, transfer_count, geometry_gaps):
profile = load_bundled_hand_profile(layout)
report = validate_executable_profile(profile, PACKAGE / "urdf" / urdf)
coverage = report["independent_measurement_coverage"]
transfers = profile.measurement.transferred_motion_sources
assert len(transfers) == transfer_count
assert coverage["unmeasured_joints"] == sorted(transfers)
assert coverage["unresolved_motion_joints"] == []
assert coverage["transferred_motion_sources"] == transfers
assert coverage["missing_baseline_geometry"] == list(geometry_gaps)
observed = {name for task in profile.motion.tasks for name in task.joints}
assert set(profile.motion.joint_zero_references) == observed
assert not set(transfers) & observed
assert set(transfers.values()) <= observed
assert not set(profile.zero.cad_zero_assumptions) & set(profile.zero.known_baseline_geometry)
product = yaml.safe_load((PACKAGE / "config" / f"{profile.key.model.lower()}_right_product.yaml").read_text())
assert product["profile_config_sha256"] == hashlib.sha256(
(PACKAGE / "config" / "profiles" / f"{layout}.yaml").read_bytes()).hexdigest()
def test_old_geometry_transfer_does_not_authorize_copying_motion():
profile = load_bundled_hand_profile("o6_right_8")
profile = replace(profile, measurement=replace(profile.measurement, transferred_motion_sources={}))
coverage = validate_executable_profile(profile,
PACKAGE / "urdf/o6_right/linkerhand_o6_right.urdf")["independent_measurement_coverage"]
assert len(coverage["unresolved_motion_joints"]) == 6
assert coverage["transferred_motion_sources"] == {}
assert set(coverage["missing_baseline_geometry"]) == set(profile.zero.transferred_zero_sources)
@pytest.mark.parametrize("target,source", [
("missing_joint", "rh_pinky_mcp_pitch"),
("rh_index_mcp_pitch", "missing_joint"),
("rh_index_mcp_pitch", "rh_middle_mcp_pitch"),
("rh_pinky_mcp_pitch", "rh_pinky_mcp_pitch"),
("rh_thumb_cmc_pitch", "rh_pinky_mcp_pitch"),
("rh_index_mcp_pitch", "rh_pinky_dip"),
("rh_index_dip", "rh_thumb_ip"),
])
def test_invalid_donors_types_and_chains_are_rejected(target, source):
profile = load_bundled_hand_profile("o6_right_8")
profile = replace(profile, measurement=replace(profile.measurement,
transferred_motion_sources={**profile.measurement.transferred_motion_sources, target: source}))
with pytest.raises(ValueError, match="motion"):
validate_profile(profile)
@pytest.mark.parametrize("field,value", [
("baseline_u8", 254), ("lower_bounds", 1), ("upper_bounds", 254),
("sdk_to_joint_direction", 1),
])
def test_recipient_domain_baseline_and_sign_must_match_donor(field, value):
profile = load_bundled_hand_profile("o6_right_8")
original = getattr(profile.command, field)
if not original:
original = profile.command.minimum_values if field == "lower_bounds" else profile.command.maximum_values
changed = list(original)
changed[2] = value
profile = replace(profile, command=replace(profile.command, **{field: tuple(changed)}))
with pytest.raises(ValueError, match="motion transfer requires equal SDK"):
validate_profile(profile)
def test_native_recipient_range_cannot_silently_rescale_donor_curve():
profile = load_bundled_hand_profile("o12_right_16")
upper = list(profile.command.upper_bounds)
upper[10] -= 0.01
profile = replace(profile, command=replace(profile.command, upper_bounds=tuple(upper)))
with pytest.raises(ValueError, match="motion transfer requires equal SDK"):
validate_profile(profile)
@pytest.mark.parametrize("field,delta", [
("feedback_lower_bounds", -0.01), ("feedback_upper_bounds", 0.01),
])
def test_recipient_feedback_domain_must_match_donor_even_when_command_domain_matches(field, delta):
profile = load_bundled_hand_profile("o12_right_16")
original = (profile.command.minimum_feedback_values if field == "feedback_lower_bounds"
else profile.command.maximum_feedback_values)
changed = list(original)
changed[10] += delta
command = replace(profile.command, **{field: tuple(changed)})
assert command.minimum_values[10] == command.minimum_values[11]
assert command.maximum_values[10] == command.maximum_values[11]
with pytest.raises(ValueError, match="motion transfer requires equal SDK"):
validate_profile(replace(profile, command=command))
def test_recipient_cad_mimic_sign_is_checked(tmp_path):
profile = load_bundled_hand_profile("o6_right_8")
tree = ET.parse(PACKAGE / "urdf/o6_right/linkerhand_o6_right.urdf")
mimic = tree.getroot().find("joint[@name='rh_index_dip']/mimic")
mimic.set("multiplier", "-1")
source = tmp_path / "reversed_mimic.urdf"
tree.write(source)
with pytest.raises(ValueError, match="equal CAD output signs"):
validate_executable_profile(profile, source)
@pytest.mark.parametrize("joint,reason", [
("missing", "accepted"), ("rh_pinky_dip", " "),
("rh_thumb_cmc_pitch", "accepted"), ("rh_index_mcp_pitch", "accepted"),
])
def test_cad_assumptions_cannot_hide_invalid_or_conflicting_zero_sources(joint, reason):
profile = load_bundled_hand_profile("o6_right_8")
profile = replace(profile, zero=replace(profile.zero,
cad_zero_assumptions={**profile.zero.cad_zero_assumptions, joint: reason}))
with pytest.raises(ValueError, match="CAD zero assumption"):
validate_profile(profile)
def test_virtual_model_can_reorder_channels_views_and_tasks(tmp_path):
profile = load_bundled_hand_profile("o6_right_8")
order = (5, 3, 1, 4, 0, 2)
index = {old: new for new, old in enumerate(order)}
reorder = lambda row: tuple(row[i] for i in order)
views = {"front": "inspection_a", "side": "inspection_b", "top": "inspection_c"}
command = replace(profile.command, names=reorder(profile.command.names),
baseline_u8=reorder(profile.command.baseline_u8),
sdk_to_joint_direction=reorder(profile.command.sdk_to_joint_direction),
command_index_by_joint={name: index[channel] for name, channel in profile.command.command_index_by_joint.items()})
motion = replace(profile.motion,
tasks=tuple(replace(task, view=views[task.view], command_index=index[task.command_index])
for task in reversed(profile.motion.tasks)),
joint_zero_references={name: replace(reference, command=reorder(reference.command),
approach_commands=tuple(reorder(row) for row in reference.approach_commands))
for name, reference in profile.motion.joint_zero_references.items()})
vision = replace(profile.vision, views=tuple(replace(view, name=views[view.name])
for view in reversed(profile.vision.views)), extrinsic_reference_view=views[profile.vision.extrinsic_reference_view])
measurement = replace(profile.measurement, measurements={name: replace(spec, view=views[spec.view])
for name, spec in profile.measurement.measurements.items()})
virtual = replace(profile, key=ProfileKey("virtual", "right", "mixed_strategy"),
command=command, motion=motion, vision=vision, measurement=measurement)
path = tmp_path / "virtual.yaml"
path.write_text(dump_hand_profile(virtual))
virtual = load_hand_profile(path)
report = validate_executable_profile(virtual, PACKAGE / "urdf/o6_right/linkerhand_o6_right.urdf")
assert not report["independent_measurement_coverage"]["unresolved_motion_joints"]
assert virtual.command.command_index_by_joint["rh_index_mcp_pitch"] == 5
assert virtual.command.command_index_by_joint["rh_pinky_mcp_pitch"] == 0
@@ -0,0 +1,235 @@
"""Synthetic transfer artifacts retain provenance without recipient evidence."""
from copy import deepcopy
from dataclasses import replace
import hashlib
import json
from types import SimpleNamespace
import numpy as np
import pytest
from scipy.spatial.transform import Rotation
from linkerhand_calibration.core.domain.measurement import JointCurveFit
from linkerhand_calibration.core.domain.reference import JointZeroPose, JointZeroReference, JointZeroSample
from linkerhand_calibration.core.domain.result import JointMapping
from linkerhand_calibration.core.fitting.coupling import LinearMimicFit
from linkerhand_calibration.core.fitting.mimic_approximation import MimicApproximation
from linkerhand_calibration.core.fitting.tag_installation import FrozenTagInstallation, matrix_tuple
from linkerhand_calibration.core.urdf.kinematics import UrdfKinematicModel
from linkerhand_calibration.core.urdf.measured_acceptance import validate_measured_joint_holdout
from linkerhand_calibration.core.urdf.tag_acceptance import TagHoldout
from linkerhand_calibration.runtime.artifacts.publisher import ArtifactPublisher, FrozenTagArtifactValidator
from linkerhand_calibration.runtime.artifacts.reader import load_unified_mapper
from linkerhand_calibration.runtime.artifacts.serializers.measured_v3 import from_report, serialize_report
@pytest.fixture
def transfer_artifacts(tmp_path):
"""Two synthetic chains; only source/source_tip have observation evidence."""
session = tmp_path / "session"
session.mkdir()
source = tmp_path / "source.urdf"
parts = ['<robot name="synthetic_transfer"><link name="base"/>']
for name, parent, xyz, mimic in (
("source", "base", "0 0 0", None), ("source_tip", "source_link", ".04 0 0", "source"),
("target", "base", "0 .1 0", None), ("target_tip", "target_link", ".04 0 0", "target")):
parts.append(f'<link name="{name}_link"/><joint name="{name}" type="revolute">'
f'<parent link="{parent}"/><child link="{name}_link"/><origin xyz="{xyz}" rpy="0 0 0"/>'
'<axis xyz="0 0 1"/><limit lower="0" upper="1.2" effort="1" velocity="1"/>'
+ (f'<mimic joint="{mimic}" multiplier="0.6" offset="0"/>' if mimic else '') + '</joint>')
source.write_text(''.join(parts) + '</robot>')
urdf = session / "corrected.urdf"
urdf.write_bytes(source.read_bytes())
model = UrdfKinematicModel(source)
transfers = {"target": "source", "target_tip": "source_tip"}
assumptions = {name: "synthetic accepted CAD baseline" for name in ("source_tip", "target_tip")}
profile = SimpleNamespace(key=SimpleNamespace(profile_id="VIRTUAL/right/transfers/v1", model="VIRTUAL", side="right"),
command=SimpleNamespace(unit="u8", names=("source", "target"), baseline_values=(0, 0),
feedback_by_index=True, feedback_name_aliases={}),
measurement=SimpleNamespace(transferred_motion_sources=transfers),
zero=SimpleNamespace(known_baseline_geometry={}, cad_zero_assumptions=assumptions,
transferred_zero_sources={"target": "source"}))
knots = tuple(float(i) for i in range(0, 256, 51))
curves, mappings, commands, references = {}, {}, {}, {}
identity = JointZeroPose((0, 0, 0, 1), (0, 0, 0))
for name in ("source", "source_tip"):
x = np.asarray(knots) / 255
angles = tuple(x if name == "source" else .2*x + .4*x*x)
mappings[name] = JointMapping(name, 0, "feedback_u8", knots, angles, angles, angles)
commands[name] = replace(mappings[name], input_domain="command_u8")
curves[name] = JointCurveFit(angles, angles, angles, {
"space": "relative_rotation_3d", "reference_quaternion_xyzw": (0, 0, 0, 1),
"axis_xyz": (0, 0, 1), "training_sample_ids": (f"training:{name}",)}, 0, 0, {})
sample = JointZeroSample("side:1", "side", 1, (0, 0, 0, 1), (0, 0, 0),
(0, 0), (0, 0), ("decreasing", "decreasing"), identity, identity)
references[name] = JointZeroReference(name, "source_scan", 0, 0, 1, 1, (sample,))
for target, donor in transfers.items():
mappings[target] = replace(mappings[donor], joint=target, motor_index=1, transferred_from=donor)
commands[target] = replace(commands[donor], joint=target, motor_index=1, transferred_from=donor)
methods = {"source": "urdf_serial_axis_geometry", "source_tip": "assumed_source_cad_zero",
"target": "transferred_static_zero_on_own_cad", "target_tip": "assumed_source_cad_zero"}
mimic_output = {"source_tip": ("source", .6, 0), "target_tip": ("target", .6, 0)}
approximations = {}
for child, (parent, multiplier, offset) in mimic_output.items():
approximate = LinearMimicFit(parent, child, multiplier, offset, True, (0, 1, 2), (0, -.1, 0))
approximations[child] = MimicApproximation(approximate, ("training:source_tip",), ("held:source_tip",),
(0, -.1, 0), transferred_from_joint=transfers.get(child))
fit = SimpleNamespace(curves=curves, output_mappings=mappings, command_mappings=commands,
zero_offsets_rad={name: 0 for name in mappings}, zero_method_by_joint=methods,
joint_zero_references=references, mimic_approximations=approximations,
command_applicability={name: {"scope": "single_channel_at_declared_pose"} for name in curves},
command_holdout_metrics={name: {"independently_measured": True} for name in curves})
prepared = SimpleNamespace(mappings=mappings, plan=SimpleNamespace(mimic_output=mimic_output))
hashes = {"source_urdf_sha256": hashlib.sha256(source.read_bytes()).hexdigest()}
report = serialize_report(profile, fit, prepared, serial_number="SYNTHETIC_ONLY", protected_inputs=hashes)
payload = from_report(report)
observations, poses = [], []
for name in curves:
for direction in ("increasing", "decreasing"):
for index, value in enumerate(knots):
angles = {joint: mapping.evaluate(value if mapping.motor_index == 0 else 0)
for joint, mapping in commands.items()}
image = f"holdout:{direction}:{index}"
observations.append({"joint": name, "sample_id": image, "cycle": 3,
"sample_phase": "steady", "direction": direction, "command_vector_u8": (value, 0),
"relative_quaternion_xyzw": Rotation.from_rotvec([0, 0, angles[name]]).as_quat().tolist()})
transform = model.link_transform(name, zero_offsets={}, joint_angles=angles, independent_mimic_angles=True)
poses.append(TagHoldout(image, name, 3, (value, 0), {name: direction}, matrix_tuple(transform)))
mounts = {name: FrozenTagInstallation(name, name+"_link", matrix_tuple(np.eye(4)),
(f"training:{name}",), 0, 0) for name in curves}
validator = FrozenTagArtifactValidator(profile.key.profile_id, source, hashes["source_urdf_sha256"],
{}, fit.zero_offsets_rad, matrix_tuple(np.eye(4)), mounts, tuple(poses), tuple(curves), lambda _: None,
minimum_samples=3, protected_inputs=hashes, command_observations=tuple(poses),
joint_observations=tuple(observations), joint_curves=curves,
joint_channels={name: mapping.motor_index for name, mapping in mappings.items()},
transferred_motion_sources=transfers, cad_zero_assumptions=assumptions,
transferred_zero_sources={"target": "source"}, joint_zero_methods=methods)
json_path = session / "calibration.json"
json_path.write_text(json.dumps(payload))
(session / "calibration_report.json").write_text(json.dumps(report))
return SimpleNamespace(profile=profile, report=report, payload=payload, fit=fit, validator=validator,
json_path=json_path, urdf=urdf, observations=observations, transfers=transfers)
def test_mixed_release_preserves_recipient_channels_without_claiming_recipient_precision(transfer_artifacts):
data = transfer_artifacts
release = ArtifactPublisher(data.json_path.parent.parent, "latest").publish(
session_directory=data.json_path.parent, calibration_json=data.json_path,
corrected_urdf=data.urdf, validate=data.validator)
validation = json.loads(release.manifest.read_text())["validation"]
assert validation["acceptance"] == "measured_and_transferred_json_urdf_v3"
assert set(validation["independent_mapping_holdout"]) == {"source", "source_tip"}
assert set(validation["full_json_urdf_holdout"]) == {"source", "source_tip"}
assert validation["transferred_joints"] == ["target", "target_tip"]
assert max(row["maximum_deg"] for row in validation["urdf_mimic_approximation"].values()) > 3
mapper = load_unified_mapper(release.manifest)
positions = dict(zip(mapper.urdf_joint_names, mapper.map_positions((0, 102))))
assert positions["source"] == 0
assert positions["target"] == pytest.approx(.4)
assert positions["target_tip"] == pytest.approx(.144)
assert mapper.transferred_motion_sources == data.transfers
for target, donor in data.transfers.items():
assert data.payload["joints"][target]["angle_rad"] == data.payload["joints"][donor]["angle_rad"]
row = data.report["joints"][target]
assert row["independently_measured"] is False
assert "zero_reference" not in row and "command_holdout" not in row
assert row["zero_reference_source_joint"] == donor
@pytest.mark.parametrize("corruption", ["undeclared", "pretend_measured", "invent_reference", "changed_curve", "feedback_channel", "wrong_baseline", "chained_source", "invent_mimic_samples"])
def test_reader_rejects_unfaithful_transfer_reports(transfer_artifacts, corruption):
report = deepcopy(transfer_artifacts.report)
target = report["joints"]["target"]
if corruption == "undeclared":
report.pop("transferred_motion_sources")
elif corruption == "pretend_measured":
target["independently_measured"] = True
elif corruption == "invent_reference":
target["zero_reference"] = deepcopy(report["joints"]["source"]["zero_reference"])
elif corruption == "changed_curve":
target["command_to_rad"]["angle_rad"][1] += .001
elif corruption == "feedback_channel":
target["feedback_to_rad"]["channel_index"] = 0
elif corruption == "wrong_baseline":
report["baseline_command"][1] = 1
elif corruption == "chained_source":
report["transferred_motion_sources"]["target"] = "target_tip"
else:
report["joints"]["target_tip"]["mimic_approximation"]["evidence"]["training_sample_ids"] = ["invented:recipient"]
with pytest.raises(ValueError):
from_report(report)
def test_publisher_binds_report_transfers_to_authoritative_profile(transfer_artifacts):
data = transfer_artifacts
with pytest.raises(ValueError, match="transfer sources differ"):
replace(data.validator, transferred_motion_sources={})(data.json_path, data.urdf)
with pytest.raises(ValueError, match="CAD zero assumptions differ"):
replace(data.validator, cad_zero_assumptions={})(data.json_path, data.urdf)
def test_publisher_rejects_consistently_rewritten_recipient_channel(transfer_artifacts):
data = transfer_artifacts
report = deepcopy(data.report)
for kind in ("command_to_rad", "feedback_to_rad"):
report["joints"]["target"][kind]["channel_index"] = 0
# The file pair is internally consistent, but its channel is not the one
# declared by the actual Profile supplied independently to the publisher.
data.json_path.write_text(json.dumps(from_report(report)))
(data.json_path.parent / "calibration_report.json").write_text(json.dumps(report))
with pytest.raises(ValueError, match="SDK channel bindings differ"):
data.validator(data.json_path, data.urdf)
def test_reader_rejects_cad_assumption_relabelled_as_independent_geometry(transfer_artifacts):
report = deepcopy(transfer_artifacts.report)
report["joints"]["target_tip"]["zero_method"] = "independent_known_baseline_geometry"
with pytest.raises(ValueError, match="assumed_cad_zero_provenance"):
from_report(report)
def test_transferred_native_rad_tables_preserve_target_channel_and_support(transfer_artifacts):
from linkerhand_calibration.runtime.artifacts.reader import UnifiedCommandMapper
data = transfer_artifacts
report = deepcopy(data.report)
report["command_unit"] = "rad"
for row in report["joints"].values():
for kind in ("command", "feedback"):
mapping = row[kind+"_to_rad"]
mapping.update(input_domain=kind+"_rad", input_unit="rad",
knots=[value/255 for value in mapping["knots"]], valid_input_range=[0, 1])
payload = from_report(report)
mapper = UnifiedCommandMapper(payload, data.urdf, report=report)
values = dict(zip(mapper.urdf_joint_names, mapper.map_positions((0, .4))))
assert values["source"] == 0
assert values["target"] == pytest.approx(.4)
assert values["target_tip"] == pytest.approx(.144)
assert payload["joints"]["target"]["input_values"] == payload["joints"]["source"]["input_values"]
with pytest.raises(ValueError, match="outside"):
mapper.map_positions((0, 1.01))
def test_holdout_never_credits_copied_or_fabricated_recipient_observations(transfer_artifacts):
data = transfer_artifacts
result = validate_measured_joint_holdout(data.payload, data.fit.curves, data.observations,
transferred_motion_sources=data.transfers)
assert set(result) == {"source", "source_tip"}
with pytest.raises(ValueError, match="wrong partition or joint"):
validate_measured_joint_holdout(data.payload, data.fit.curves,
data.observations+[dict(data.observations[0], joint="target")], transferred_motion_sources=data.transfers)
wrong = deepcopy(data.payload)
wrong["joints"]["target"]["angle_rad"][102] += .001
with pytest.raises(ValueError, match="transferred_final_table"):
validate_measured_joint_holdout(wrong, data.fit.curves, data.observations, transferred_motion_sources=data.transfers)
def test_recipient_limit_violation_is_rejected_instead_of_clipping(transfer_artifacts):
data = transfer_artifacts
import xml.etree.ElementTree as ET
root = ET.fromstring(data.urdf.read_text())
root.find("joint[@name='target']/limit").set("upper", ".3")
data.urdf.write_text(ET.tostring(root, encoding="unicode"))
validator = replace(data.validator, authorized_fields={"target": ("limit.upper",)})
with pytest.raises(ValueError, match="range|limit"):
validator(data.json_path, data.urdf)