From 467651fc475df7a39855a9aa8dc67bcefe843a71 Mon Sep 17 00:00:00 2001 From: lxp <2770281812@qq.com> Date: Fri, 11 Sep 2026 17:45:44 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9C=AB=E8=8A=82=E9=87=87=E7=94=A8=E5=8E=9F?= =?UTF-8?q?=20CAD=20=E9=9B=B6=E4=BD=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../CALIBRATION_V3_IMPLEMENTATION.md | 51 +++- src/linkerhand_calibration/README.md | 65 ++++- .../config/g20_right_product.yaml | 2 +- .../config/l6_right_product.yaml | 2 +- .../config/o12_right_product.yaml | 2 +- .../config/o6_right_product.yaml | 2 +- .../config/profiles/g20_right_19.yaml | 12 + .../config/profiles/l6_right_8.yaml | 18 ++ .../config/profiles/o12_right_16.yaml | 15 + .../config/profiles/o6_right_8.yaml | 18 ++ .../models/l6/left_transfer.py | 3 + .../models/o6/left_transfer.py | 3 + .../core/domain/profile.py | 63 ++++ .../core/fitting/mimic_approximation.py | 1 + .../core/fitting/motion_transfer.py | 109 +++++++ .../core/fitting/session.py | 13 +- .../core/urdf/measured_acceptance.py | 20 +- .../core/urdf/result_plan.py | 5 +- .../linkerhand_calibration/profiles/loader.py | 2 + .../profiles/validator.py | 33 ++- .../runtime/artifacts/finalization.py | 10 +- .../runtime/artifacts/publisher.py | 41 ++- .../runtime/artifacts/reader.py | 23 +- .../artifacts/serializers/measured_v3.py | 109 ++++++- .../runtime/camera_preflight.py | 48 +++ .../runtime/coordinator.py | 25 ++ .../linkerhand_calibration/runtime/runner.py | 4 +- .../runtime/snapshot.py | 8 +- .../linkerhand_calibration/runtime/status.py | 19 ++ .../test/measured_capture_fixture.py | 11 +- .../test/test_camera_preflight.py | 222 ++++++++++++++ .../test_measured_transfer_finalization.py | 67 +++++ .../test/test_motion_transfer.py | 276 ++++++++++++++++++ .../test/test_observation_strategy.py | 166 +++++++++++ .../test/test_transfer_artifact_contract.py | 235 +++++++++++++++ 35 files changed, 1636 insertions(+), 67 deletions(-) create mode 100644 src/linkerhand_calibration/linkerhand_calibration/core/fitting/motion_transfer.py create mode 100644 src/linkerhand_calibration/linkerhand_calibration/runtime/camera_preflight.py create mode 100644 src/linkerhand_calibration/test/test_camera_preflight.py create mode 100644 src/linkerhand_calibration/test/test_measured_transfer_finalization.py create mode 100644 src/linkerhand_calibration/test/test_motion_transfer.py create mode 100644 src/linkerhand_calibration/test/test_observation_strategy.py create mode 100644 src/linkerhand_calibration/test/test_transfer_artifact_contract.py diff --git a/src/linkerhand_calibration/CALIBRATION_V3_IMPLEMENTATION.md b/src/linkerhand_calibration/CALIBRATION_V3_IMPLEMENTATION.md index 6bc9ffa..5183571 100644 --- a/src/linkerhand_calibration/CALIBRATION_V3_IMPLEMENTATION.md +++ b/src/linkerhand_calibration/CALIBRATION_V3_IMPLEMENTATION.md @@ -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、PIP、DIP | 部分末端/原 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 skipped，495.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` 跳过。 diff --git a/src/linkerhand_calibration/README.md b/src/linkerhand_calibration/README.md index d85f553..732af2f 100644 --- a/src/linkerhand_calibration/README.md +++ b/src/linkerhand_calibration/README.md @@ -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 v3，rad 输入节点查表 | 任务数、主动关节数和实测关节数不必相等。一条运动可观测多个主动/被动关节。 -**这些配置目前不能发布全关节独立实测产物。** 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 会话继续只读兼容，不自动覆盖或转换历史产物；新输出单表必须重新通过最终文件验收。 diff --git a/src/linkerhand_calibration/config/g20_right_product.yaml b/src/linkerhand_calibration/config/g20_right_product.yaml index ddc30ad..097cee9 100644 --- a/src/linkerhand_calibration/config/g20_right_product.yaml +++ b/src/linkerhand_calibration/config/g20_right_product.yaml @@ -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 diff --git a/src/linkerhand_calibration/config/l6_right_product.yaml b/src/linkerhand_calibration/config/l6_right_product.yaml index a41ee19..21d3b53 100644 --- a/src/linkerhand_calibration/config/l6_right_product.yaml +++ b/src/linkerhand_calibration/config/l6_right_product.yaml @@ -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 diff --git a/src/linkerhand_calibration/config/o12_right_product.yaml b/src/linkerhand_calibration/config/o12_right_product.yaml index d4911e9..67c1161 100644 --- a/src/linkerhand_calibration/config/o12_right_product.yaml +++ b/src/linkerhand_calibration/config/o12_right_product.yaml @@ -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 diff --git a/src/linkerhand_calibration/config/o6_right_product.yaml b/src/linkerhand_calibration/config/o6_right_product.yaml index 2cd95cf..ef7c46b 100644 --- a/src/linkerhand_calibration/config/o6_right_product.yaml +++ b/src/linkerhand_calibration/config/o6_right_product.yaml @@ -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 diff --git a/src/linkerhand_calibration/config/profiles/g20_right_19.yaml b/src/linkerhand_calibration/config/profiles/g20_right_19.yaml index a20b414..aad9158 100644 --- a/src/linkerhand_calibration/config/profiles/g20_right_19.yaml +++ b/src/linkerhand_calibration/config/profiles/g20_right_19.yaml @@ -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 diff --git a/src/linkerhand_calibration/config/profiles/l6_right_8.yaml b/src/linkerhand_calibration/config/profiles/l6_right_8.yaml index ecdaf46..8acb0e3 100644 --- a/src/linkerhand_calibration/config/profiles/l6_right_8.yaml +++ b/src/linkerhand_calibration/config/profiles/l6_right_8.yaml @@ -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 diff --git a/src/linkerhand_calibration/config/profiles/o12_right_16.yaml b/src/linkerhand_calibration/config/profiles/o12_right_16.yaml index 18b772a..15a2b94 100644 --- a/src/linkerhand_calibration/config/profiles/o12_right_16.yaml +++ b/src/linkerhand_calibration/config/profiles/o12_right_16.yaml @@ -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 diff --git a/src/linkerhand_calibration/config/profiles/o6_right_8.yaml b/src/linkerhand_calibration/config/profiles/o6_right_8.yaml index 380b1c0..e85eeed 100644 --- a/src/linkerhand_calibration/config/profiles/o6_right_8.yaml +++ b/src/linkerhand_calibration/config/profiles/o6_right_8.yaml @@ -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 diff --git a/src/linkerhand_calibration/linkerhand_calibration/compat/legacy_diagnostic_tools/models/l6/left_transfer.py b/src/linkerhand_calibration/linkerhand_calibration/compat/legacy_diagnostic_tools/models/l6/left_transfer.py index 0449c6c..31ec24c 100644 --- a/src/linkerhand_calibration/linkerhand_calibration/compat/legacy_diagnostic_tools/models/l6/left_transfer.py +++ b/src/linkerhand_calibration/linkerhand_calibration/compat/legacy_diagnostic_tools/models/l6/left_transfer.py @@ -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( diff --git a/src/linkerhand_calibration/linkerhand_calibration/compat/legacy_diagnostic_tools/models/o6/left_transfer.py b/src/linkerhand_calibration/linkerhand_calibration/compat/legacy_diagnostic_tools/models/o6/left_transfer.py index 5d921b2..13131b9 100644 --- a/src/linkerhand_calibration/linkerhand_calibration/compat/legacy_diagnostic_tools/models/o6/left_transfer.py +++ b/src/linkerhand_calibration/linkerhand_calibration/compat/legacy_diagnostic_tools/models/o6/left_transfer.py @@ -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( diff --git a/src/linkerhand_calibration/linkerhand_calibration/core/domain/profile.py b/src/linkerhand_calibration/linkerhand_calibration/core/domain/profile.py index 6df822e..035432d 100644 --- a/src/linkerhand_calibration/linkerhand_calibration/core/domain/profile.py +++ b/src/linkerhand_calibration/linkerhand_calibration/core/domain/profile.py @@ -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): diff --git a/src/linkerhand_calibration/linkerhand_calibration/core/fitting/mimic_approximation.py b/src/linkerhand_calibration/linkerhand_calibration/core/fitting/mimic_approximation.py index 7102745..b796171 100644 --- a/src/linkerhand_calibration/linkerhand_calibration/core/fitting/mimic_approximation.py +++ b/src/linkerhand_calibration/linkerhand_calibration/core/fitting/mimic_approximation.py @@ -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 diff --git a/src/linkerhand_calibration/linkerhand_calibration/core/fitting/motion_transfer.py b/src/linkerhand_calibration/linkerhand_calibration/core/fitting/motion_transfer.py new file mode 100644 index 0000000..600e64d --- /dev/null +++ b/src/linkerhand_calibration/linkerhand_calibration/core/fitting/motion_transfer.py @@ -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}") diff --git a/src/linkerhand_calibration/linkerhand_calibration/core/fitting/session.py b/src/linkerhand_calibration/linkerhand_calibration/core/fitting/session.py index 2ce4d28..cf21cbc 100644 --- a/src/linkerhand_calibration/linkerhand_calibration/core/fitting/session.py +++ b/src/linkerhand_calibration/linkerhand_calibration/core/fitting/session.py @@ -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()} diff --git a/src/linkerhand_calibration/linkerhand_calibration/core/urdf/measured_acceptance.py b/src/linkerhand_calibration/linkerhand_calibration/core/urdf/measured_acceptance.py index 27bf27a..929b316 100644 --- a/src/linkerhand_calibration/linkerhand_calibration/core/urdf/measured_acceptance.py +++ b/src/linkerhand_calibration/linkerhand_calibration/core/urdf/measured_acceptance.py @@ -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"] diff --git a/src/linkerhand_calibration/linkerhand_calibration/core/urdf/result_plan.py b/src/linkerhand_calibration/linkerhand_calibration/core/urdf/result_plan.py index fa1cf6b..c523a65 100644 --- a/src/linkerhand_calibration/linkerhand_calibration/core/urdf/result_plan.py +++ b/src/linkerhand_calibration/linkerhand_calibration/core/urdf/result_plan.py @@ -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") diff --git a/src/linkerhand_calibration/linkerhand_calibration/profiles/loader.py b/src/linkerhand_calibration/linkerhand_calibration/profiles/loader.py index 5bdc7bc..d331ca3 100644 --- a/src/linkerhand_calibration/linkerhand_calibration/profiles/loader.py +++ b/src/linkerhand_calibration/linkerhand_calibration/profiles/loader.py @@ -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", ())), diff --git a/src/linkerhand_calibration/linkerhand_calibration/profiles/validator.py b/src/linkerhand_calibration/linkerhand_calibration/profiles/validator.py index e1fb39f..5a6adb1 100644 --- a/src/linkerhand_calibration/linkerhand_calibration/profiles/validator.py +++ b/src/linkerhand_calibration/linkerhand_calibration/profiles/validator.py @@ -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): diff --git a/src/linkerhand_calibration/linkerhand_calibration/runtime/artifacts/finalization.py b/src/linkerhand_calibration/linkerhand_calibration/runtime/artifacts/finalization.py index e2c196e..b5fceac 100644 --- a/src/linkerhand_calibration/linkerhand_calibration/runtime/artifacts/finalization.py +++ b/src/linkerhand_calibration/linkerhand_calibration/runtime/artifacts/finalization.py @@ -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 diff --git a/src/linkerhand_calibration/linkerhand_calibration/runtime/artifacts/publisher.py b/src/linkerhand_calibration/linkerhand_calibration/runtime/artifacts/publisher.py index 4bdf6a8..8ec10af 100644 --- a/src/linkerhand_calibration/linkerhand_calibration/runtime/artifacts/publisher.py +++ b/src/linkerhand_calibration/linkerhand_calibration/runtime/artifacts/publisher.py @@ -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: diff --git a/src/linkerhand_calibration/linkerhand_calibration/runtime/artifacts/reader.py b/src/linkerhand_calibration/linkerhand_calibration/runtime/artifacts/reader.py index 6e1bf92..1c2f30a 100644 --- a/src/linkerhand_calibration/linkerhand_calibration/runtime/artifacts/reader.py +++ b/src/linkerhand_calibration/linkerhand_calibration/runtime/artifacts/reader.py @@ -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) diff --git a/src/linkerhand_calibration/linkerhand_calibration/runtime/artifacts/serializers/measured_v3.py b/src/linkerhand_calibration/linkerhand_calibration/runtime/artifacts/serializers/measured_v3.py index 3741543..89c54a1 100644 --- a/src/linkerhand_calibration/linkerhand_calibration/runtime/artifacts/serializers/measured_v3.py +++ b/src/linkerhand_calibration/linkerhand_calibration/runtime/artifacts/serializers/measured_v3.py @@ -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: diff --git a/src/linkerhand_calibration/linkerhand_calibration/runtime/camera_preflight.py b/src/linkerhand_calibration/linkerhand_calibration/runtime/camera_preflight.py new file mode 100644 index 0000000..74315b5 --- /dev/null +++ b/src/linkerhand_calibration/linkerhand_calibration/runtime/camera_preflight.py @@ -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) diff --git a/src/linkerhand_calibration/linkerhand_calibration/runtime/coordinator.py b/src/linkerhand_calibration/linkerhand_calibration/runtime/coordinator.py index def89ae..1d6bb84 100644 --- a/src/linkerhand_calibration/linkerhand_calibration/runtime/coordinator.py +++ b/src/linkerhand_calibration/linkerhand_calibration/runtime/coordinator.py @@ -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. diff --git a/src/linkerhand_calibration/linkerhand_calibration/runtime/runner.py b/src/linkerhand_calibration/linkerhand_calibration/runtime/runner.py index b7cee7b..14799b8 100644 --- a/src/linkerhand_calibration/linkerhand_calibration/runtime/runner.py +++ b/src/linkerhand_calibration/linkerhand_calibration/runtime/runner.py @@ -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) diff --git a/src/linkerhand_calibration/linkerhand_calibration/runtime/snapshot.py b/src/linkerhand_calibration/linkerhand_calibration/runtime/snapshot.py index fd4814d..26c00f0 100644 --- a/src/linkerhand_calibration/linkerhand_calibration/runtime/snapshot.py +++ b/src/linkerhand_calibration/linkerhand_calibration/runtime/snapshot.py @@ -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) diff --git a/src/linkerhand_calibration/linkerhand_calibration/runtime/status.py b/src/linkerhand_calibration/linkerhand_calibration/runtime/status.py index 0d292a6..f4dcb8b 100644 --- a/src/linkerhand_calibration/linkerhand_calibration/runtime/status.py +++ b/src/linkerhand_calibration/linkerhand_calibration/runtime/status.py @@ -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"): diff --git a/src/linkerhand_calibration/test/measured_capture_fixture.py b/src/linkerhand_calibration/test/measured_capture_fixture.py index 9cdca2e..c4ad5a3 100644 --- a/src/linkerhand_calibration/test/measured_capture_fixture.py +++ b/src/linkerhand_calibration/test/measured_capture_fixture.py @@ -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 = [] diff --git a/src/linkerhand_calibration/test/test_camera_preflight.py b/src/linkerhand_calibration/test/test_camera_preflight.py new file mode 100644 index 0000000..3c06c92 --- /dev/null +++ b/src/linkerhand_calibration/test/test_camera_preflight.py @@ -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) diff --git a/src/linkerhand_calibration/test/test_measured_transfer_finalization.py b/src/linkerhand_calibration/test/test_measured_transfer_finalization.py new file mode 100644 index 0000000..d7f929e --- /dev/null +++ b/src/linkerhand_calibration/test/test_measured_transfer_finalization.py @@ -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] diff --git a/src/linkerhand_calibration/test/test_motion_transfer.py b/src/linkerhand_calibration/test/test_motion_transfer.py new file mode 100644 index 0000000..c1e5644 --- /dev/null +++ b/src/linkerhand_calibration/test/test_motion_transfer.py @@ -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", {}) diff --git a/src/linkerhand_calibration/test/test_observation_strategy.py b/src/linkerhand_calibration/test/test_observation_strategy.py new file mode 100644 index 0000000..0726f40 --- /dev/null +++ b/src/linkerhand_calibration/test/test_observation_strategy.py @@ -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 diff --git a/src/linkerhand_calibration/test/test_transfer_artifact_contract.py b/src/linkerhand_calibration/test/test_transfer_artifact_contract.py new file mode 100644 index 0000000..bbc72b0 --- /dev/null +++ b/src/linkerhand_calibration/test/test_transfer_artifact_contract.py @@ -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 = [''] + 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'' + f'' + '' + + (f'' if mimic else '') + '') + source.write_text(''.join(parts) + '') + 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)