feat(tracking): 汇总 v0.1.3 回放与 ACT 准备进度

新增专家轨迹诊断、原视频估计相机入口和状态参考ACT数据门禁/训练链路;更新版本与进度文档。

验证:120项CPU/USD回归和4步synthetic CPU smoke通过;1333帧默认GUI历史回放PASS。原视频相机完整回放超时124,策略GPU E2E未执行,完整pre-commit工具缺失。

兼容性:HDF5、Cartpole、USD和控制阈值保持不变。本提交为实验进度快照,不宣称完整发布验收通过;数据、视频和权重不纳入。
This commit is contained in:
2026-09-15 16:29:17 +08:00
parent 8e7ab5fc76
commit f5d84e26f6
26 changed files with 2662 additions and 7 deletions
+11
View File
@@ -241,3 +241,14 @@ export PYTHONPATH="$PWD/source/dex_workbench${PYTHONPATH:+:$PYTHONPATH}"
给右手数据同事发送本文时,附上同版**右清单**与右模型坐标/映射说明。模型来源、转换设置、 给右手数据同事发送本文时,附上同版**右清单**与右模型坐标/映射说明。模型来源、转换设置、
本地使用及发布限制见 [`右手资产说明`](assets/robots/dex_hand/linkerhand_g20_right/README.md)。 本地使用及发布限制见 [`右手资产说明`](assets/robots/dex_hand/linkerhand_g20_right/README.md)。
## 11. 状态输入ACT训练的额外准备(不改变HDF5 schema
路线A的训练入口与详细契约见 [`L20_IMITATION.md`](L20_IMITATION.md)。
除本文的HDF5与模型清单,还需要仓库外的episode/capture-group划分JSON及绑定HDF5哈希的数据审查JSON。
至少两个独立来源组用于训练/验证,建议另留test组;同一录制的切片、慢放、重采样副本必须同组同分区。
单条样例不足以构造无泄漏训练/验证集,不能通过改名复制来补齐。
审查需明确坐标/尺度与同步依据、组来源,以及同意将未来参考状态作为实验目标代理。
不必为此伪造实测动作、速度、力矩或物体数据。初版ACT只做参考轨迹模仿,不证明已学会抓取。
示例模板默认不具备批准状态;收到独立核验材料前不得自动改为已审核。
+222
View File
@@ -0,0 +1,222 @@
# L20 状态输入 ACT:训练准备与运行
**v0.1.3 更新**:最新汇总见 [`PROGRESS_v0.1.3.md`](PROGRESS_v0.1.3.md)。本轮120项回归及新的4步CPU smoke通过。
已收到1333帧同源轨迹和估计相机;这不构成新的独立采集组或实测尺度审查。
默认展示相机参考回放已有PASS,原视频相机完整运行超时;ACT策略GPU E2E仍NOT_RUN。
下文环境安装、118项测试和“未升级/未提交”等描述是初次ACT准备阶段的历史记录,不是v0.1.3提交状态。
## 当前交付范围
路线A:**腕手参考轨迹模仿 → 仿真评估 → 后续PPO refinement**。
当前提供可执行的状态输入ACT-style CVAE Transformer、数据门禁、离线训练/保存/加载/留出评估,
以及已有Isaac Lab受限回放入口的可选闭环策略分支。
- **PASS**:本机CPU依赖、editable安装、synthetic短训练→保存→加载→离线预测链路及回归测试。
- **BLOCKED**:真实专家训练,目前只有单条示例,且独立坐标/尺度审查未闭合;不能自动批准。
- **NOT_RUN**:新策略分支的GPU仿真E2E、真实专家训练/质量评测、GUI策略回放。
历史75/80倍参考回放PASS不等于新策略回放PASS。
- 没有创建L20 Gym/RL任务、奖励、物体场景、PPO、视觉模型或硬件驱动;Cartpole任务及checkpoint保留。
本页的“训练可运行”指离线参考模仿,不是抓取训练就绪或Sim2Real验收。
## 1. 本机环境
继续使用已有Isaac Sim Python**不更换它的PyTorch/CUDA**。
本次实测:Python3.12.13、PyTorch2.10.0+cu128、numpy2.5.1、h5py3.16.0
Isaac Sim6.0.1-rc.7Isaac Lab可发现。本轮没有初始化CUDA训练或启动Kit。
从仓库根目录运行:
```bash
# 本机默认使用 ~/isaacsim/python.sh;不同安装位置由调用者显式设置。
# export ISAACSIM_PYTHON=/path/to/isaacsim/python.sh
bash scripts/imitation.sh doctor
# 本机已实际执行:仅安装项目,不访问索引、不安装/升级依赖。
~/isaacsim/python.sh -m pip install --no-index --no-deps --no-build-isolation -e source/dex_workbench
```
`setup.py`已收录`dex_workbench_imitation`包及`imitation`可选依赖。
本机旧installed metadata为0.1.0,本次重新安装后与源码0.1.2一致;**未修改源码版本号**。
不要在Isaac Python里盲目执行`pip install --upgrade torch`或复制另一机器的CUDA wheel。
新节点须先安装兼容的Isaac Sim/Lab和已核验依赖,再执行上述本地安装与doctor。
右手USD目前仅本地存在,存储/再分发许可尚未闭合;离线训练只消费HDF5与清单,
新节点仿真评估另需合法取得同哈希资产并重建prepared层,不保证干净克隆直接可回放。
可选 `doctor --cuda`仅检查CUDA可用性,不是GPU容量或仿真验收。
包装脚本设置本仓库`PYTHONPATH`,不改变调用者cwd;输入/输出相对路径仍相对于调用目录。
不要求激活系统Python或下载额外模型,也不会自动搜索“最新checkpoint”或自动开始训练。
## 2. HDF5补齐后还需要哪些输入
沿用 [`l20_tracking_v1`](HDF5_REQUIREMENTS.md),不新增伪造观测/力矩字段,不改变现有回放契约。
训练前需要三个实际文件及同版模型清单:
1. **HDF5**:经过同版模型绑定,全部使用一侧L20与同一校准;各episode全有效,单位/时间/顺序/限位/联动正确。
不能直接把原始交付的暂定资产哈希替换成当前清单哈希。
2. **划分文件**:参照 `configs/imitation/splits.example.json`,在仓库外填写实际episode与采集来源。
3. **数据审查文件**:参照 `configs/imitation/data_review.example.json`,在独立复核后由负责人填写。
### 划分与来源
- 至少 **两个独立采集组**才可进行训练+验证,建议另有独立test组;这只是最小接口门槛,不是数据充足标准。
- 每个episode必须且只能属于train/validation/test中的一个。不能随机按帧划分。
- 同一原始录制的切片、慢放、重采样或重复导出,必须使用同一`episode_groups`值且落在同一分区。
- 自动拒绝跨分区同group及几何数组逐值重复的轨迹(忽略时间,捕获改名/慢放副本)。
不能自动识别所有重采样、裁剪或旋转等价副本;仍需负责人核对真实采集来源。
- `episode_groups`不是自动生成的独立性证据,示例中的占位名称必须替换。
- HDF5中所有episode必须明确划分,不静默忽略文件里的额外轨迹。
### 审查文件不是绕过门禁的开关
填写**实际HDF5文件SHA256**和资产包SHA,记录`reviewer``evidence`,明确:
- `coordinate_and_scale_reviewed`:坐标变换、SLAM尺度与同步已经独立复核;
- `reference_state_targets_accepted`:同意把未来参考状态作为这一实验的目标代理,而非宣称实测动作;
- `capture_groups_reviewed`:数据来源与跨分区泄漏已人工审核。
示例默认false/空值,**刻意不能通过训练门禁**。程序只检查声明与文件绑定,不能替代人的真实审查。
当前已收到估计相机SE3,但尺度与独立数据审查仍未闭合,不能把标记自动改为true来启动。
### 速度和时间
- 配置控制频率30Hz;保留数据物理时间,按线性位置/关节插值及shortest-arc SLERP重采样。
- episode末时刻须落在控制频率网格上;只处理浮点舍入,不静默裁去尾帧或改变时长。
- 保留既有参考上限:腕平移0.05m/s、腕旋转0.5rad/s、所有状态关节0.5rad/s。
默认实验工作半径0.8m,与先前扩范围回放一致。
- **训练准备不自动慢放、缩放、平滑、重居中或跨无效帧插值**。超限时明确失败,由数据方另出可追溯版本。
- 单文件读取上限2GiB,重采样总帧数默认500000;大数据须明确分片/扩展,不自动放大资源预算。
## 3. 观测/动作契约:`l20_goal_reference_act_v1`
这是**状态参考模仿基线**,不是完整Canonical Skill Space或全项目DexSchema实现。
共享几何使用相对episode初始根link的腕部位姿;构型特有的关节顺序、主从映射与限幅封装在Adapter中。
不把21状态关节或16个目标硬编码为所有手型接口,也不把模型独立目标数当作硬件电机数。
| 项目 | 定义 |
| --- | --- |
| 当前状态 | 初始根坐标系中的当前腕位置3维、旋转6D(旋转矩阵前两列)、清单全部J个状态关节 |
| 条件目标 | 同坐标系下的episode终点腕位姿、终点M个模型独立关节姿态 |
| 时间条件 | 归一化phase及episode时长(秒);用于指定此参考的时间参数,不是奖励 |
| 观测维数 | `20 + J + M`(当前已核验右手为57 |
| 输出 | 未来chunk的腕位置/旋转6D与M个独立`q_target`代理,维数`9 + M`(当前右手25 |
| 训练输入来源 | HDF5的参考状态;没有声称这是执行后传感器观测 |
| 仿真输入来源 | 实际当前root-link pose/q,加预先给定终点目标、phase和时长;不喂未来中间参考帧 |
当前状态→未来状态目标不是已测量的动作因果对。训练中teacher-forced参考状态与部署时受扰状态存在分布差异;
多条路径也可能共享相同起终点。这是需要真实仿真评估和后续示教/残差策略研究的风险,不因loss降低而消失。
不向网络输入完整未来参考轨迹来假装闭环成功。
### 模型与loss
- 状态输入ACT-style CVAE:训练时Transformer posterior编码观测及未padding的未来动作块,
得到高斯latentdecoder根据观测+latent与chunk queries预测未来目标。
- masked normalized L1 + KLpadding既不参与reconstruction loss,也被posterior attention mask排除。
- 推理固定latent=0,不使用训练posterior的未来目标;无dropoutCPU推理确定。
- 不是官方ACT仓库/权重兼容实现;没有视觉ResNet或Diffusion分支。
- normalization仅由训练分区的非padding帧拟合,验证/test数据不参与统计。
- best checkpoint仅按validation选择;test不参与优化或checkpoint选择。
默认 `configs/imitation/l20_right_act.json`:chunk16、每次执行前4个目标再规划、30Hz目标频率,
Transformer宽128/2层/4heads/latent32、batch32、seed42、最多1000更新且600秒。
这是**有界起始配置,不是已调优或保证收敛的超参**。600秒是训练/验证循环的内部预算,
不包含此前的数据预检;命令另设外部进程超时。预算耗尽记FAIL/未完成,不伪装成完成训练。
## 4. 数据到位后的命令
下面变量由调用者指向审核后的文件;`OUT`必须是新的仓库外目录。
当前示例数据不会通过这些正式训练门禁,这属于预期BLOCKED,不是安装失败。
```bash
MANIFEST=assets/robots/dex_hand/linkerhand_g20_right/tracking_manifest.json
CONFIG=configs/imitation/l20_right_act.json
# DATA=/approved/path/demonstrations.hdf5
# SPLITS=/approved/path/splits.json
# REVIEW=/approved/path/data_review.json
# OUT=/external/artifacts/l20-act-run-001
# 先只做CPU数据预检;父目录须存在,输出JSON必须为新文件。
bash scripts/imitation.sh preflight --hdf5 "$DATA" --manifest "$MANIFEST" \
--config "$CONFIG" --splits "$SPLITS" --data-review "$REVIEW" --output "$OUT-preflight.json"
# 审核数据与资源预算后再由用户启动;不传--execute会拒绝。
# CUDA是显式选择;不可用时报错,不静默换CPU,也不安装替代CUDA依赖。
timeout --kill-after=15s 660s bash scripts/imitation.sh train --hdf5 "$DATA" --manifest "$MANIFEST" \
--config "$CONFIG" --splits "$SPLITS" --data-review "$REVIEW" \
--device cuda --execute --output "$OUT"
# 按完整test分区做离线预测评测(CPU);这不是仿真rollout或任务成功率。
bash scripts/imitation.sh evaluate --checkpoint "$OUT/best.pt" --hdf5 "$DATA" \
--manifest "$MANIFEST" --split test --output "$OUT-test.json"
```
`train`默认device为CPU;不要省略CUDA选择后误以为已用GPU。修改`max_updates/max_seconds`等资源预算应先确认。
训练输出为`run.json`、逐更新`metrics.jsonl``best.pt``last.pt``result.json`
受控异常输出`failure.json`,不写成功result;进程硬超时/中断可能只留下部分制品,缺少result不能算PASS。
已有目录拒绝使用,不自动覆盖历史实验。
同一新run内best/last通过临时文件原子发布;记录配置、数据/Adapter、split/review、源模块哈希与依赖版本。
checkpoint使用tensor-only加载,绑定contract、资产/Adapter及normalization。
本版本支持**推理加载,不提供优化器resume**;不要把另一次新训练冒充恢复同一优化状态。
同一checkpoint的评估必须使用同一HDF5哈希;新评测数据需要显式评估接口扩展,不靠改metadata冒充原评测集。
## 5. 仿真策略评估入口(已实现,GPU E2E为NOT_RUN
新增可选 `scripts/tracking/track_l20.py --policy-checkpoint`,默认不传时保留参考目标回放逻辑。
PASS日志增加`control_source``policy_checkpoint_sha256`字段;不改变HDF5 schema。
只接受训练数据中明确属于validation/test的episode,拒绝train episode;要求完整回合回放。
运行前仍检查source/prepared USD、PhysX后端、参考速度/范围及所有旧门禁。
策略在控制步读取实际状态,执行chunk前4个目标后重规划;每个240Hz物理步均经过Adapter
- 将6D旋转恢复为合法旋转;退化/非有限输出明确失败,不能静默给单位姿态。
- 限制起点相对工作半径、目标平移/转动速度;关节按主从约束交集限位。
- 独立关节限速还考虑mimic倍率,确保派生状态关节目标也满足参考速度上限。
- 只发master `q_target`及既有根部wrench PD;从动关节由物理耦合响应,不独立驱动。
- 两轮分别清空policy chunk、时间计数及限幅器状态;只有原有reset路径写物理状态。
- 保留原有动态误差/限位/速度/mimic/非零运动及reset、位姿/速度重复性检查。
策略未训练好时应该FAIL,不用参考轨迹替代网络输出来通过。
确认无其他GPU/Kit作业,并**另行批准单次仿真预算后**才运行:
```bash
export PYTHONPATH="$PWD/source/dex_workbench${PYTHONPATH:+:$PYTHONPATH}"
# EPISODE必须属于validation/testSTEPS必须等于该回合秒数*240,且100<=STEPS<=32000。
# TIMEOUT_S须覆盖已批准预算;外部终止宽限15秒,不自动重试。
timeout --kill-after=15s "$TIMEOUT_S" ~/isaacsim/python.sh scripts/tracking/track_l20.py \
assets/robots/dex_hand/linkerhand_g20_right/tracking.usda --manifest "$MANIFEST" \
--hdf5 "$DATA" --episode "$EPISODE" --steps "$STEPS" --full-episode \
--workspace-radius 0.8 --policy-checkpoint "$OUT/best.pt" --execute-experimental --headless
```
策略与运行时控制参数必须与checkpoint绑定值一致;不允许临时扩大范围或阈值来通过。
日志使用独立标记`bounded_reference_act_rollout`,记录checkpoint SHA、推理次数和目标限幅步数。
目标限幅计数**不是**真实wrench/力矩饱和统计。
完整轨迹、非零运动和误差门禁仍可能限制部分静止/局部手指episode的仿真验收;
这些数据可以用于离线训练,不得为了满足诊断而添加伪运动。
## 6. 本次测试证据与剩余工作
本机证据位于 `../dex_workbench-evidence/l20-act-setup-M9Y0OX/`,不随代码分发。
```bash
# 已执行:只生成明确synthetic的三条不同解析轨迹,4次小CPU更新;不启动Kit。
bash scripts/imitation.sh smoke --manifest assets/robots/dex_hand/linkerhand_g20_right/tracking_manifest.json \
--output "$NEW_EXTERNAL_SMOKE_DIRECTORY"
PYTHONPATH=source/dex_workbench ~/isaacsim/python.sh -m unittest discover \
-s source/dex_workbench/tests -p 'test_*.py' -v
```
- 依赖/安装PASS:项目editable安装成功,无依赖下载/升级;CPU doctor及`doctor --cuda`可用性探测成功。
另在仓库外、清除PYTHONPATH后验证安装包可导入;CUDA可用不代表本轮执行过GPU训练或仿真。
- CPU smoke PASS:真实反向传播4次、有限loss、保存非空checkpoint、tensor-only重新加载;
重新加载预测最大差0;独立test分区离线预测有限。仅为管线验证,不宣称收敛。
- 全部118项回归PASS,无跳过(包含20项新增模仿学习测试);覆盖坐标/半周旋转、coupling/rate限幅、
数据划分/分组/重复轨迹、训练统计隔离、padding mask、CVAE梯度、checkpoint身份、预算失败、
拒绝覆盖、held-out策略时钟/reset,以及旧tracking回归。
- 新仿真策略分支E2E、当前修改后的默认参考回放GPU回归、真实数据训练为NOT_RUN;
没有沿用上一轮已消耗的600秒GPU授权。完整pre-commit工具仍缺失,不宣称全量发布门禁通过。
- 后续:补齐数据审查与独立episode → 批准小规模真实训练预算 → 新checkpoint仿真闭环验收;
之后才设计PPO refinement或视觉/物体任务。尚无任务成功率、Sim2Real或SkillBundle交付声明。
- 本轮新增包/配置/文档,扩展可选策略入口;原始数据、USD、控制增益与物理阈值未改,
不升级包版本或改变Cartpole任务。未暂存、提交、推送。
+314
View File
@@ -0,0 +1,314 @@
# L20 专家参考轨迹:离线可执行性诊断
**最新进展(2026-09-14**:用户另行授权的一次75倍慢放、两轮各30000步动态回放已PASS,
见第7节。第1–6节保留此前CPU审计与当时的NOT_RUN边界,不将数据真实性标为通过。
## 结论
**80 倍慢放主要由腕部平移参考速度门禁决定,而不是已证实的动力学极限。**
对当前这条轨迹,保持现有速度阈值和几何路径时,统一慢放的速度必要下界约为 **69.37266 倍**
70/75/80 倍在 **0.8 m 对比工作半径**下通过现有 CPU `validate_reference()`
这不是动力学通过范围,也不授权仿真或真机动作。默认工作半径仍为 0.1 m,没有修改控制参数。
本次不启动 Kit/GPU、不运行训练、不改原始/派生 HDF5、不重绑定资产。当前代码基线
`8e7ab5f`(v0.1.2)加本次诊断模块;工作区原有其他未提交修改保留。
## 1. 输入身份与质量
- 原始输入:`Data/l20_linkerbot_urdf_delivery/demonstrations.hdf5`,右手,
`expert_retargeted``demo_000000`,51 帧全有效,50 个源采样区间。
- 本次实际清单校验输入:历史制品目录
`expert-replay-recovery-20260914/recovered-20260914T033624Z/bound.hdf5`
复核其历史绑定记录;原始文件与该文件的时间、腕位姿、关节、valid、关节顺序及坐标矩阵逐值相同。
80 倍派生文件全部几何样本也逐值相同,时间恰为原时间乘 80。
- 原始交付使用供应方暂定资产摘要,直接配当前清单仍 **FAIL: asset bundle hash mismatch**
本次没有添加任何 hash override;已有绑定文件通过清单校验不等于重新检查整个 USD 或硬件身份。
- `l20_tracking_v1` schema、有限值、时间严格递增、四元数及清单限位/联动检查:**PASS**。
长度 1.666666667 s30 Hzdt 最小/最大 0.033333333333333215 / 0.03333333333333344 s
没有可见重复时间或异常长采样间隔。四元数最大范数误差约 `3.47e-8`
- 关节最小限位余量约 `-7.43e-8 rad`,在已有 `1e-6 rad` 浮点容差内;
多个参考关节触及边界,不代表有动态余量。五组联动最大参考残差约 `2.56e-8 rad`
小于既有 `1e-3 rad` 容差;不能把这些参考等式当作本轮物理联动验收。
- 米制尺度来源明确为 **SLAM 估计**,不是测量控制网。字段正确不证明尺度、腕根转换或重定向正确;
不再次应用 `world_from_source``scale_to_meters`
| 输入 | SHA-256 |
| --- | --- |
| 原始 HDF5 | `bd9321096803fbb64f590b4fdeb9056edc11ac446a7e16458c509beedc8b9552` |
| 历史 bound HDF5 | `ad64968150a67f504a7a2a4d18667c5e69bb2ee47f251917f8f9477bd4014c35` |
| 历史 slow80 HDF5 | `8f584c7409698394429737d509924013b8b90700117f8af2fb22465223dc1815` |
| 右手清单文件 | `52419b16d8faedc299f36711b6109b90eeb908e10e1c1b9faec3e16b9eaf37c9` |
## 2. 哪些区间超限
帧号均为 **0-based**,时间为原始秒。速度是相邻参考样本的派生量,不是传感器或仿真速度。
平移/关节使用 float64 差分;旋转使用 shortest-arc 世界旋转向量。
表中百分位按区间等权,关节列先取各区间所有 21 个状态关节的最大绝对速度。
21 个状态关节不等于 21 个独立驱动;JSON 中显式标识模型 follower。
| 量 | 中位数 | P95 | 峰值 | 现有参考上限 | 超限区间数 | 单项慢放下界 |
| --- | ---: | ---: | ---: | ---: | ---: | ---: |
| 腕平移 m/s | 0.988018 | 2.203842 | **3.468633** | 0.05 | **50/50** | **69.372659** |
| 腕旋转 rad/s | 2.893062 | 6.756863 | **8.999576** | 0.5 | 49/50 | 17.999153 |
| 最大关节 rad/s | 1.263341 | 2.640378 | **2.935250** | 0.5 | 46/50 | 5.870501 |
优先核查的原始片段:
1. **帧 49→501.633333→1.666667 s**:平移峰值,约 0.11562 m / 帧。
帧 47→48、48→49 的平移速度分别为 1.967910、2.696393 m/s,末段速度连续上升。
应回看原视频/SLAM/腕根重定向;不能仅凭高速就判为离群并删除尾帧。
2. **帧 28→290.933333→0.966667 s**:旋转峰值 8.999576 rad/s。
3. **帧 29→300.966667→1.0 s**`ring_mcp_pitch` 速度峰值 2.935250 rad/s。
4. **帧 0→1** 已有平移 1.684308 m/s、旋转 4.293173 rad/s。
当前 replay reset 清零速度;参考并非从静止平滑起步,启动瞬态需运行时验证。
整条腕路径长度约 **1.872604 m**,相对起点最大距离 **0.651218 m(帧 15**
**250** 超过默认 0.1 m 半径;任何慢放都不改变这一点。
历史使用的 0.8 m 半径可包含该几何路径,但不是碰撞、可达性或硬件工作空间证明。
附加的中点差分加速度峰值:平移 33.683240 m/s²、旋转 99.073881 rad/s²、
各区间最大关节 62.699264 rad/s²。它们仅用于定位参考粗糙程度,**不是物理加速度上界**。
现有线性/SLERP 插值在节点速度可能跳变,不能据此估算真实力矩、认定饱和或保证平滑性。
独立的归一化四元数 dot/acos 算法与报告旋转速度的最大差约 `1.06e-12 rad/s`
## 3. 慢放候选:仅 CPU 门禁结果
对原始 bound 样本在内存中使用现有 `stretch_time()`,再实际调用 `validate_reference()`
没有生成新的慢放 HDF5。报告中的派生速度提升到 float64 计算,门禁仍使用既有实现,未改精度或容差。
| 慢放倍数 | 单轮时长 s | 平移峰值 m/s | 0.8 m 半径下参考门禁 | 本次动力学 |
| --- | ---: | ---: | --- | --- |
| 1 | 1.666667 | 3.468633 | FAIL:平移过快 | NOT_RUN |
| 60 | 100 | 0.057811 | FAIL:平移过快 | NOT_RUN |
| 69 | 115 | 0.050270 | FAIL:平移过快 | NOT_RUN |
| 70 | 116.666667 | 0.049552 | PASS | NOT_RUN |
| 75 | 125 | 0.046248 | PASS | NOT_RUN |
| 80 | 133.333333 | 0.043358 | PASS | NOT_RUN |
默认 0.1 m 半径下,这六档全部 FAIL(较慢档仍违反工作半径)。
70 倍的平移速度余量仅约 **0.90%**75 倍约 **7.50%**80 倍约 **13.28%**
因此不建议先盲试 40/20 倍,也不把 69.37266 倍数学边界直接当作可执行速度。
历史 80 倍动力学 PASS 及误差见 [`L20_TRACKING.md` 第5节](L20_TRACKING.md#5-显式扩范围与完整专家参考回放)。
本次重新核对历史原始日志 SHA,但没有重跑;不能把历史 PASS 扩展到70/75倍或当前GUI入口。
## 4. 控制器归因与下一步
源码核查 `control.py` / `scripts/tracking/track_l20.py`
- 腕 PD 为 `Kp*(p_ref-p)-Kd*v_measured`,角速度同理,**没有参考速度前馈**;
手指位置目标的 velocity target 也保持0。移动目标需要误差抵消阻尼,存在跟踪滞后机制。
忽略惯性、重力残差和耦合的恒速近似下,腕滞后时间 `Kd/Kp=0.1 s`;这只是定性诊断,
不能当作全轨迹误差预测或改变增益的依据。
- 当前目标在 step 开始时发送,step 后与下一参考点比较,另有离散采样因素。
- 20 N / 1 N·m 总 wrench 限制、0.2 N·m finger effort、0.5 rad/s finger velocity
都是未标定的仿真配置。仅凭参考数据无法知道实际执行是否饱和。
- 默认参考速度门禁、物理 drive cap、运行时速度安全断言及误差断言是不同层次;
不能拿较宽的实测速度安全界限替代较窄的参考门禁。
**建议顺序:**
1. 先请数据方核查上述关键帧、SLAM 米制尺度与腕根变换,保留原文件与版本;
当前证据不能区分真实高速动作与重建/重定向误差,不擅自平滑、裁剪、缩放或重居中。
2. 若几何确认无误且继续使用现有控制器,可另行授权一次 **75 倍、两轮各30000步、单环境、
240 Hz、工作半径0.8 m、外部上限600秒**的完整回放;保持阈值,先记录误差及饱和信息。
本轮未运行,也未增加运行时日志字段。75倍通过后才考虑70倍。
3. 若目标是接近原速,这类慢放提速收益很有限。需单独设计经确认的参考速度前馈/启动过渡
与控制带宽实验,而不是直接放宽门禁;随后再接入任务、示教基线与RL。
## 5. 复现与验证
新增入口 `source/dex_workbench/dex_workbench_tracking/diagnostic.py` 仅依赖已有 numpy/h5py
不导入 pxr、torch、isaaclab 或 Kit。JSON 包含每关节统计、全部区间、超限位置、mimic残差、
输入/清单哈希、未标定默认参数和候选实际门禁结果。拒绝无效帧间隙、不匹配身份、覆盖输出。
`status=PASS` 仅表示报告成功生成;候选的 `reference_gate` 单独报告,`dynamic_replay=NOT_RUN`
从仓库根目录执行,`BOUND` 必须指向已有独立核验的同清单文件;**不是原始交付的 hash 替换开关**。
`EVIDENCE` 指向已存在的仓库外制品目录,新输出路径不得已存在。
```bash
export PYTHONPATH="$PWD/source/dex_workbench${PYTHONPATH:+:$PYTHONPATH}"
MANIFEST=assets/robots/dex_hand/linkerhand_g20_right/tracking_manifest.json
~/isaacsim/python.sh -m dex_workbench_tracking.diagnostic "$BOUND" \
--manifest "$MANIFEST" --output "$EVIDENCE/default-workspace.json"
~/isaacsim/python.sh -m dex_workbench_tracking.diagnostic "$BOUND" \
--manifest "$MANIFEST" --workspace-radius 0.8 --output "$EVIDENCE/extended-workspace.json"
~/isaacsim/python.sh -m unittest discover -s source/dex_workbench/tests -p 'test_tracking_*.py' -v
~/isaacsim/python.sh -m ruff check source/dex_workbench/dex_workbench_tracking/diagnostic.py \
source/dex_workbench/tests/test_tracking_diagnostic.py
~/isaacsim/python.sh -m ruff format --check source/dex_workbench/dex_workbench_tracking/diagnostic.py \
source/dex_workbench/tests/test_tracking_diagnostic.py
git diff --check
```
实际 CPU 环境:Python 3.12.13、numpy 2.5.1、h5py 3.16.0;未安装/升级依赖。
本机制品目录(相对仓库根):`../dex_workbench-evidence/l20-reference-9Q2Cbs/`,不随源码分发。
含两份JSON、输入审计脚本/结果、测试日志及实际命令记录;干净克隆不自带 bound/右USD/专家数据。
- **PASS**:新诊断10项回归;全部tracking CPU/USD回归 **98项,无跳过**
首次诊断测试因合成测试清单缺少右手必需身份字段失败;补全测试夹具后全量重跑。
没有放宽生产身份门禁。初次Ruff行宽失败,经格式化后复检。
- **PASS**:原始/bound/slow80身份与样本保留审计,独立旋转数值交叉核对。
- **NOT_RUN**GPU仿真、GUI、70/75倍动力学、碰撞/接触、训练及硬件;本任务限定CPU诊断。
- 完整 pre-commit 工具不可用,不将局部Ruff与单测代替全量发布门禁。
- 新增只读诊断,不改变既有数据schema、运行控制器、资产、Cartpole任务、checkpoint或包版本。
没有提交/推送。
## 6. 继续核查:关键帧与导出链路
本节为后续 CPU 审计,不新增运行时验收。使用交付中的 `retarget/right_qpos.npz`
`demo_000000_triple_0_50.mp4`,没有重新拟合、修复或发布数据。
### 已排除:关节导出列错位与时间轴不一致
- NPZ 有121帧、21列,HDF5交付前51帧;两者关节列顺序不同。
**按名称显式重排**后,NPZ前51帧转为float32,与HDF5的全部关节样本逐值完全一致。
原float64到HDF5 float32的最大差仅 `5.88378966e-8 rad`
- `ring_mcp_pitch` 的帧29→30峰值在NPZ中已经存在,约 `2.93524959 rad/s`
因而该峰值不是本次HDF5序列化或关节列错位制造的;仍不证明重定向结果等于真实关节运动。
- HDF5时间逐值等于 `arange(51)/30`NPZ声明fps=30。
预览按顺序解码得到51帧、30fps;帧28/29/30/49/50的媒体时间与对应HDF5时间一致。
这只核对交付内部时间索引,不能证明与原始相机/SLAM真实同步。
### 尚未闭合:相机系腕位姿到世界系的变换
NPZ声明 `view=ego_cam`,其 `wrist_pos` 不能直接与世界系HDF5位置相减后当作误差。
NPZ本身也没有显式腕位置单位字段;交付说明暗示米制,但仍应由数据方确认。
| 帧49→50 | NPZ相机系(原单位) | HDF5世界系(m) |
| --- | ---: | ---: |
| 位移范数 | 0.013635702 | 0.115621099 |
| 位移范数 / dt | 0.409071067 | 3.468632970 |
**若双方长度单位确认为米**,此差异不能由一个固定刚体坐标变换单独解释:
固定旋转/平移不会改变相邻点间距离。交付声明另外应用了逐帧SLAM相机运动,
因此需要复算的是这一步的旋转、平移、方向、帧对齐与尺度,不能认定世界系高速就是错误。
不以逆向拟合出来的相机轨迹代替缺失的独立SLAM证据。
预览关键帧0、28、29、30、47、48、49、50已实际抽帧并查看:
末段桌椅在画面中明显移动,画面模糊,手/叠加模型接近下边缘;
这与视角变化相容,但单目叠加预览没有独立米制依据,既不能证实也不能否定3.47 m/s世界速度。
帧28–30的叠加显示也不足以验证8.9996 rad/s世界系腕角速度。
另有待确认字段:NPZ的 `hand_scale` 前51帧范围为 **0.6783040.845200**
帧49→50从0.720546降到0.681751(约 **−5.38%**)。它的估计方法及应用位置未随NPZ提供,
**不能把它等同于HDF5的 `scale_to_meters=1.0`**,也不能据此直接重新缩放轨迹。
它可能与手模型拟合有关,是否影响腕根平移需查看上游代码,而不是从字段名推断。
### 给数据方的最小补充清单
> 请补充同一 `video_0`、至少覆盖帧0–51(包含交付末帧后的一个邻帧)的以下材料;
> 最好提供已有完整121帧,不必重新采集:
>
> 1. `hawor_slam_w_scale_0_121.npz` 或等价逐帧相机SE3,附帧ID、秒时间戳、
> 变换方向(camera→world还是world→camera)、轴向、位置单位及四元数顺序。
> 2. 同版本 `export_l20_tracking_hdf5.py` 与 `retarget_mano_to_g20_mujoco.py`
> 或相关变换/尺度计算片段,附版本/提交及实际参数。
> 3. 明确 `right_qpos.npz` 的腕位置单位、四元数顺序、`hand_scale` 的含义、
> 每帧变化是否应用于腕根平移,以及SLAM米制尺度来自何种估计/标定。
> 4. 原始 `video_0.mp4` 的对应短片及原帧号/时间映射;说明重定向、视频、SLAM是否存在偏移或丢帧。
>
> 优先解释帧49→50的世界位移0.115621 m及帧28→29的世界旋转峰值。
> 请保留当前交付,任何修正另出版本与哈希,不原地替换样本或静默裁剪。
当前交付与本仓库 `Data/``scripts/` 中未找到上述SLAM文件、原始视频或两份上游脚本。
因此 **世界腕位姿导出复算为BLOCKED(缺独立输入)**,不是仿真或训练失败。
先补齐此接口,再判断是否需要修正数据或开展75倍受限回放。
### 本轮证据与边界
本机制品目录:`../dex_workbench-evidence/l20-keyframes-ANfXqd/`,不随代码提交。
包含 `audit_keyframes.py``keyframe-audit.json``audit.log`
`keyframes-video-mano.jpg` 和完整三联帧49/50图片。
审计使用 `np.load(..., allow_pickle=False)`OpenCV 4.13.0软件解码既有预览,不调用Kit或GPU仿真。
环境PATH无ffmpeg/ffprobe,本轮使用已有OpenCV,未安装依赖。
实际命令(仓库根;`KEYFRAME_EVIDENCE`指上述仓库外目录):
```bash
PYTHONPATH=source/dex_workbench ~/isaacsim/python.sh "$KEYFRAME_EVIDENCE/audit_keyframes.py"
git diff --check
```
- **PASS**:脚本退出0;命名关节映射/float32样本一致、时间一致、51帧预览解码断言,
对交付目录全部文件的运行前后SHA-256相等;原始数据没有改变。
- **BLOCKED**:世界系腕位姿独立复算,缺逐帧相机SE3/导出源码/尺度说明。
- **NOT_RUN**:新GPU回放、GUI、训练、硬件;本轮没有启动,也未授权扩大默认资源预算。
- 本轮仅追加本文及仓库外诊断制品;未改变应用代码、控制参数、资产、schema或包版本。
不重复把上一轮98项回归当作本轮新运行;本轮文档变更的仿真E2E为N/A。
未暂存、提交或推送。
## 7. 单次授权75倍慢放动态回放:PASS2026-09-14
用户明确确认GPU预算后,只执行一次:单环境、seed42、240Hz、两轮各30000步、
每轮125秒完整参考,工作半径0.8mheadless,外部timeout600秒(终止宽限15秒)。
未重试、未启动70倍或其他实验;未调增益、限速、力/力矩上限或验收断言。
### 准备与真实验收结果
- 以历史已核验的 `bound.hdf5` 为输入,使用已有 `stretch-time --factor 75` 生成仓库外新文件。
原始交付不改写、不重新绑定;原始/bound样本完全一致,slow75几何/valid/坐标元数据保持,
时间恰为原始时间乘75。派生文件通过清单及现有参考速度/工作半径检查。
- 注册现有PhysX/Newton schema后的prepared检查通过,覆盖层包SHA保持
`3e2071a60496b23bed252f4e2fc6e1d4540eecc8098578506419b5eb594f7da0`
- 本轮新运行全部tracking CPU/USD回归:**98项PASS,无跳过**。
- GPU运行 **364秒、退出0**,日志唯一的 `bounded_experimental_dynamic_tracking` 结果为PASS
确认HDF5哈希、全覆盖125秒、两轮各30000步、浮动根、1个有效articulation root、21个状态关节。
原入口的有限值、速度安全界限、限位、非零被动联动、reset、逐步位姿及速度重复性断言全部执行通过。
- 每轮的max/RMS汇总相同,但不承诺跨运行GPU位级确定性。下列关节/联动统计沿用运行入口语义:
每步先取各关节最大绝对误差/残差,再对时间汇总。
| 误差 | 最大值 | RMS |
| --- | ---: | ---: |
| 腕位置 m | **0.004526687** | 0.001701794 |
| 腕姿态 rad | **0.014233187** | 0.007083100 |
| 各步最大关节误差 rad | **0.011136770** | 0.007015317 |
| 各步最大mimic残差 rad | **0.000231415** | 0.000100018 |
历史80倍的最大腕位置/姿态误差分别为0.004234530m / 0.013369173rad
本次75倍略大,但均在既有门禁内。这是两个指定轨迹速度的结果对照,不是收敛、性能或全速度区间保证。
### 执行环境、命令与证据
Isaac Sim安装版本 `6.0.1-rc.7+release.42383.32955d8d.gl`,运行时PhysX `110.1.13`
RTX5080,驱动595.84Python3.12.13、numpy2.5.1、h5py3.16.0。
Isaac Lab基线 `6a7acb0320a0bdc15b13e44e83b575e00797faf4`,其本机工作区有既有
`source/isaaclab/isaaclab/cloner/cloner_utils.py` 修改:本任务未改该文件,已另存patch,
不能将本次结果当作干净上游Lab环境验收。项目运行源码仍为 `8e7ab5f` 对应控制入口,
另有前几轮新增的独立CPU诊断和文档;本次未改应用代码。
实际运行命令(仓库根目录;`EVIDENCE`指下方本机制品目录):
```bash
export PYTHONPATH="$PWD/source/dex_workbench${PYTHONPATH:+:$PYTHONPATH}"
RIGHT=assets/robots/dex_hand/linkerhand_g20_right
# 仅复现记录,不是再次启动授权。
timeout --kill-after=15s 600s ~/isaacsim/python.sh scripts/tracking/track_l20.py \
"$RIGHT/tracking.usda" --manifest "$RIGHT/tracking_manifest.json" \
--hdf5 "$EVIDENCE/slow75.hdf5" --steps 30000 --full-episode \
--workspace-radius 0.8 --execute-experimental --headless
```
本机制品目录:`../dex_workbench-evidence/l20-slow75-DeXIEX/`,不随源码提交。
包含精确展开命令 `exact-command.txt`、单次启动脚本 `run-once.sh``LAUNCH-CONSUMED`
`replay.log``result.txt``metrics.json``acceptance.json`、输入预检/样本保留检查、
`source-before.json`、CPU测试日志、前后GPU/进程快照及既有Lab patch。
- slow75 HDF5 SHA256`6133e58499e3eb4739bed4b70f3147c9361c445dc97d32ba4f03a60c74afe8f9`
- replay.log SHA256`794c5a95b3176285ca2852febacd295a701c4ea846e916008d582397271cf75d`
- 接受脚本另外核验退出0、预算、唯一PASS、步数/轮数/身份、参数、有限指标及阈值;
在更新本文前,93个被快照的源码/文档/资产/交付文件SHA与运行前相同。
- 启动前后无Kit或GPU计算进程;运行后显存633MiB,仍有正常桌面图形进程。
没有停止其他用户进程,不据此保证所有资源位级恢复。
### 剩余边界
- 保留了现有headless参数弃用/重复提示、protobuf重复注册诊断、visualizer配置缺失、
MaterialX/usdrt及TGS外力迭代警告。16/21 actuator提示对应5个有意被动关节,未关闭警告或放宽检查。
- 当前入口只保存汇总误差,未导出逐步wrench/力矩饱和统计;**不能声称无控制饱和**。
要做带宽或前馈归因,应另行增加可审计的遥测,而不是仅靠本次PASS推断。
- 上游相机SE3、尺度和原视频的独立核查依旧BLOCKED;本次只是把给定参考在仿真中受限跟踪。
- 原速、70倍、GUI整合、接触/物体任务、多环境、训练与硬件本轮均NOT_RUN。
不自动启动下一档;不把75/80倍两个离散PASS泛化为整个区间可执行。
- 本轮仅更新验收文档、变更记录及仓库外制品;资产/控制器/schema/版本不变。
完整pre-commit仍未具备,测试通过不等于完整Release就绪。未暂存、提交或推送。
+7
View File
@@ -1,5 +1,12 @@
# L20 左手轨迹跟踪:数据交付与准备入口 # L20 左手轨迹跟踪:数据交付与准备入口
**最新 v0.1.3 进度**见 [`PROGRESS_v0.1.3.md`](PROGRESS_v0.1.3.md):1333帧右手3倍慢放完整GUI回放PASS;
原视频相机入口已接入但完整运行600秒超时(FAIL),不能视为同视角验收通过。
当前专家参考的 CPU 速度/限位/联动与慢放下界诊断见
[`L20_REFERENCE_DIAGNOSTIC.md`](L20_REFERENCE_DIAGNOSTIC.md);第7节另记录用户授权的
单次75倍慢放、两轮各30000步动态PASS,不代表原速、数据真实性或训练验收。
本页保留左手验证历史。新增右手入口、实际转换/回放证据与发布限制见 本页保留左手验证历史。新增右手入口、实际转换/回放证据与发布限制见
[`右手资产说明`](assets/robots/dex_hand/linkerhand_g20_right/README.md);左右共用校验器与控制入口,显式绑定hand_side。 [`右手资产说明`](assets/robots/dex_hand/linkerhand_g20_right/README.md);左右共用校验器与控制入口,显式绑定hand_side。
HDF5右手差异见独立契约第10节(右thumb联动1.03,不能套用左1.02)。 HDF5右手差异见独立契约第10节(右thumb联动1.03,不能套用左1.02)。
+69
View File
@@ -0,0 +1,69 @@
# v0.1.3 进度快照(2026-09-15
本版汇总 L20 参考回放、相机对齐入口及状态输入 ACT 准备。**不是完整发布门禁通过、抓取任务完成或 Sim2Real 验收。**
版本唯一来源为 `source/dex_workbench/config/extension.toml`Cartpole 任务 ID、HDF5 schema、USD 和控制增益/动态误差阈值不变。
## 已实现与验证
| 工作 | 结果及边界 |
| --- | --- |
| L20 左/右模型与受限参考跟踪 | 延续 v0.1.2;右手 21 状态关节、16 模型独立目标、5 组联动,不代表硬件电机数 |
| 专家参考 CPU 诊断 | 速度、限位、联动、工作半径及慢放下界;旧 51 帧参考的 75 倍回放历史 PASS 见 `L20_REFERENCE_DIAGNOSTIC.md` |
| 新 1333 帧参考 | 原始 44.4 s / 30 Hz23 个 URDF/网格文件哈希、关节顺序/限位/联动核验通过;仅仓库外绑定副本和时间拉伸 |
| 3 倍慢放默认展示相机 GUI | **PASS**0.4 m 半径、240 Hz、单环境 seed42,两轮各31968步,完整133.2 s/轮,退出0,约531秒关闭 |
| 原视频相机数据 | 包内全部SHA校验通过;与上述HDF5字节一致;1333帧相机/时间映射齐全,矩阵互逆误差约4.4e-16 |
| 原视频相机接入 | 已实现轴转换、1280×720中心内参、逐源帧保持、显式慢放同步;相机姿态是估计值,不是实测标定 |
| 原视频相机完整 GUI | **FAIL(超时)**:到达GUI_READY,600秒外部终止、退出124;没有两轮PASS或可靠完成步数,不推断具体完成比例;退出后无残留GPU计算进程 |
| 状态输入ACT | 数据审查/来源分组门禁、Adapter、CVAE Transformer、训练/保存/加载/留出离线评估及可选策略回放分支;不是视觉ACT或完整抓取策略 |
| 本轮CPU/USD回归 | **PASS**:120项,无跳过;包含2项相机CPU回归 |
| 本轮ACT synthetic smoke | **PASS**:4次真实CPU更新、有限loss、保存/重载预测最大差0、独立test预测有限;不证明策略质量 |
新参考默认展示相机两轮最大误差一致:腕位置 **0.003582456 m**、腕姿态 **0.017285813 rad**、关节 **0.009641975 rad**、联动残差 **0.000128251 rad**
这些结果属于相机改动前的实际运行;不将历史默认回放PASS算作相机分支PASS。
## 复现入口与证据
以下命令在仓库根目录执行,Isaac Python 启动器由本机环境提供。数据、USD及实验制品不随本次代码提交分发。
```bash
export PYTHONPATH="$PWD/source/dex_workbench${PYTHONPATH:+:$PYTHONPATH}"
~/isaacsim/python.sh -m unittest discover -s source/dex_workbench/tests -p 'test_*.py' -v
bash scripts/imitation.sh doctor
# OUT为新的仓库外目录;本轮实际执行了下面的CPU smoke。
timeout 120s bash scripts/imitation.sh smoke \
--manifest assets/robots/dex_hand/linkerhand_g20_right/tracking_manifest.json --output "$OUT"
git diff --check
```
相机运行的关键参数(实际完整命令在证据目录 `run.sh` 中):
```bash
# SLOW3必须是独立核验模型绑定后的副本,不能直接替换原始资产hash。
# 本命令历史结果为超时;再次运行需另行确认GPU预算,不自动重试。
timeout --signal=TERM --kill-after=20s 600s ~/isaacsim/python.sh scripts/tracking/track_l20.py \
assets/robots/dex_hand/linkerhand_g20_right/tracking.usda \
--manifest assets/robots/dex_hand/linkerhand_g20_right/tracking_manifest.json \
--hdf5 "$SLOW3" --episode demo_000000 --full-episode --steps 31968 \
--workspace-radius 0.4 --execute-experimental --gui \
--source-camera Data/camera_delivery_1333/source_camera.npz --camera-time-factor 3
```
本机工作区外制品目录名(供负责人定位,不是可移植依赖):
- `replay-evidence/right1333-RnTtzC/`:绑定脚本/哈希、默认展示相机命令、退出码0、`replay-metrics.json`
- `replay-evidence/right1333-source-camera/`:相机命令、`replay.log`、退出码124。
- `replay-evidence/v0.1.3/`:本轮120项测试日志和ACT smoke。
原始HDF5 SHA256`41abdcfd6bf0512ebbab4f09f5e7353ad2ca56dd8138bb939d109ddb7133acef`
3倍派生HDF5 SHA256`991fc634feb4e68e947ee1ed784afb3435f358f3fab571a39d1dc05e5ab92994`
本輪CPU环境:Python3.12.13 / PyTorch2.10.0+cu128 / NumPy2.5.1;历史GUI为Isaac Sim6 / PhysX110.1.13 / RTX5080。
## 未完成与下一步
1. 相机完整E2E未通过;先加可追踪进度和检查渲染耗时,在新预算内完成有限回放。不能通过扩大控制阈值或把超时算成功来验收。
2. 摄像机投影还需截图/像素几何验证;实际动态渲染帧、画幅和原视频并排对照尚未验收。当前相机按源帧保持,不是插值相机;1280×720中心内参是本交付专用约束。
3. 新相机包虽然包含瓶子场景/物体轨迹,但尚未接入Isaac;本次手部回放无物体接触,不证明抓瓶成功。
4. 原视频相机/米制尺度是上游估计,逐帧接触修正改变手部动作;缺可信畸变参数,不保证像素级重合或实测正确性。
5. 1333帧与旧短片同属视频2047635068,不能自动当作独立训练/验证采集组。真实训练仍需独立来源、数据审查和预算;ACT策略GPU E2E、PPO及硬件均NOT_RUN。
6. 完整pre-commit **BLOCKED**:本机缺工具且许可证hook引用文件缺失;不关闭hook或使用`--no-verify`。数据/视频/USD再分发和LFS许可未闭合,干净克隆不保证可运行仿真。
本次提交范围为以上功能源码、配置、测试、进度文档与版本。原有`.gitignore`、README中的治理/资产说明、未授权资产及其他脚本改动保持原样,不顺带提交。Git提交/推送结果由实际操作后交付说明,本文不预先声称已发布Release。
@@ -0,0 +1,10 @@
{
"schema": "l20_training_review_v1",
"hdf5_sha256": null,
"asset_sha256": null,
"coordinate_and_scale_reviewed": false,
"reference_state_targets_accepted": false,
"capture_groups_reviewed": false,
"reviewer": "",
"evidence": ""
}
+20
View File
@@ -0,0 +1,20 @@
{
"contract": "l20_goal_reference_act_v1",
"hand_side": "right",
"control_hz": 30,
"chunk_size": 16,
"execute_steps": 4,
"hidden_dim": 128,
"heads": 4,
"layers": 2,
"latent_dim": 32,
"batch_size": 32,
"learning_rate": 0.0001,
"kl_weight": 10.0,
"max_updates": 1000,
"evaluate_every": 100,
"max_seconds": 600.0,
"max_total_frames": 500000,
"workspace_radius": 0.8,
"seed": 42
}
+11
View File
@@ -0,0 +1,11 @@
{
"schema": "l20_episode_splits_v1",
"train": ["demo_000000"],
"validation": ["demo_000001"],
"test": ["demo_000002"],
"episode_groups": {
"demo_000000": "REPLACE_WITH_SOURCE_RECORDING_A",
"demo_000001": "REPLACE_WITH_SOURCE_RECORDING_B",
"demo_000002": "REPLACE_WITH_SOURCE_RECORDING_C"
}
}
+11
View File
@@ -0,0 +1,11 @@
#!/usr/bin/env bash
# Reuse Isaac's pinned Python/Torch rather than replacing its CUDA dependencies.
set -euo pipefail
ROOT=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." && pwd)
LAUNCHER=${ISAACSIM_PYTHON:-"$HOME/isaacsim/python.sh"}
if [[ ! -x "$LAUNCHER" ]]; then
printf 'BLOCKED: set ISAACSIM_PYTHON to an executable Isaac Sim python.sh launcher.\n' >&2
exit 2
fi
export PYTHONPATH="$ROOT/source/dex_workbench${PYTHONPATH:+:$PYTHONPATH}"
exec "$LAUNCHER" -m dex_workbench_imitation.cli "$@"
+52 -4
View File
@@ -56,6 +56,11 @@ def build_parser(add_app_launcher_args):
parser.add_argument("--manifest", type=Path, required=True, help="Original data/source manifest") parser.add_argument("--manifest", type=Path, required=True, help="Original data/source manifest")
parser.add_argument("--hdf5", type=Path, help="Omit for explicitly synthetic diagnostic reference") parser.add_argument("--hdf5", type=Path, help="Omit for explicitly synthetic diagnostic reference")
parser.add_argument("--episode", default="demo_000000") parser.add_argument("--episode", default="demo_000000")
parser.add_argument(
"--policy-checkpoint",
type=Path,
help="Optional reference ACT policy; requires held-out HDF5 and --full-episode",
)
parser.add_argument("--steps", type=int, default=480, help="Per repetition at 240Hz; two repetitions") parser.add_argument("--steps", type=int, default=480, help="Per repetition at 240Hz; two repetitions")
parser.add_argument( parser.add_argument(
"--full-episode", action="store_true", help="Require complete HDF5 duration; allow up to 32000 steps" "--full-episode", action="store_true", help="Require complete HDF5 duration; allow up to 32000 steps"
@@ -66,6 +71,8 @@ def build_parser(add_app_launcher_args):
parser.add_argument("--execute-experimental", action="store_true", help="Acknowledge uncalibrated controller") parser.add_argument("--execute-experimental", action="store_true", help="Acknowledge uncalibrated controller")
parser.add_argument("--limits", type=Path, help="JSON fields of control.Limits, SI units; UNCALIBRATED") parser.add_argument("--limits", type=Path, help="JSON fields of control.Limits, SI units; UNCALIBRATED")
parser.add_argument("--gui", action="store_true", help="Opt-in bright-hand/dark-backdrop Kit preview") parser.add_argument("--gui", action="store_true", help="Opt-in bright-hand/dark-backdrop Kit preview")
parser.add_argument("--source-camera", type=Path, help="Estimated source-camera NPZ; requires GUI and HDF5")
parser.add_argument("--camera-time-factor", type=float, default=1.0, help="Must match HDF5 time stretch")
add_app_launcher_args(parser) add_app_launcher_args(parser)
# None distinguishes omitted flags from explicit --headless / --viz none. # None distinguishes omitted flags from explicit --headless / --viz none.
parser.set_defaults(headless=None, visualizer=None) parser.set_defaults(headless=None, visualizer=None)
@@ -79,6 +86,10 @@ def main():
args = parser.parse_args() args = parser.parse_args()
if not args.execute_experimental: if not args.execute_experimental:
parser.error("Require --execute-experimental") parser.error("Require --execute-experimental")
if args.source_camera and (not args.gui or args.hdf5 is None):
parser.error("--source-camera requires --gui and --hdf5")
if args.policy_checkpoint and (args.hdf5 is None or not args.full_episode):
parser.error("--policy-checkpoint requires --hdf5 and --full-episode")
try: try:
configure_presentation(args) configure_presentation(args)
validate_replay_request(args.steps, args.full_episode, args.hdf5 is not None, args.workspace_radius) validate_replay_request(args.steps, args.full_episode, args.hdf5 is not None, args.workspace_radius)
@@ -113,6 +124,11 @@ def main():
data = load(args.hdf5, manifest) if args.hdf5 else synthetic(manifest) data = load(args.hdf5, manifest) if args.hdf5 else synthetic(manifest)
episode = data.episodes[args.episode] episode = data.episodes[args.episode]
validate_reference(episode, limits) validate_reference(episode, limits)
policy = None
if args.policy_checkpoint:
from dex_workbench_imitation.policy import ClosedLoopReferenceACT
policy = ClosedLoopReferenceACT(args.policy_checkpoint, manifest, args.hdf5, args.episode, episode, limits)
dt = 1 / 240 dt = 1 / 240
validate_replay_duration(args.steps, dt, episode.time[-1], args.full_episode) validate_replay_duration(args.steps, dt, episode.time[-1], args.full_episode)
reference = sample(episode, np.arange(args.steps + 1) * dt) reference = sample(episode, np.arange(args.steps + 1) * dt)
@@ -192,6 +208,13 @@ def main():
eye, center = create_preview(sim.stage, episode.wrist_position) eye, center = create_preview(sim.stage, episode.wrist_position)
sim.set_camera_view(eye, center) sim.set_camera_view(eye, center)
source_camera = None
if args.source_camera:
from dex_workbench_tracking.source_camera import SourceCamera
# The old backdrop is oriented for the default showcase camera, not this moving camera.
sim.stage.RemovePrim("/World/PreviewBackdrop")
source_camera = SourceCamera(sim.stage, args.source_camera, episode.time, args.camera_time_factor)
sim.reset() sim.reset()
if args.gui: if args.gui:
sim.render() sim.render()
@@ -231,6 +254,8 @@ def main():
traces, reset_states, summaries = [], [], [] traces, reset_states, summaries = [], [], []
for repetition in range(2): for repetition in range(2):
failure_context = {"phase": "reset", "completed_repetitions": repetition, "repetition": repetition} failure_context = {"phase": "reset", "completed_repetitions": repetition, "repetition": repetition}
if policy is not None:
policy.reset()
hand.reset() hand.reset()
hand.permanent_wrench_composer.reset() hand.permanent_wrench_composer.reset()
hand.instantaneous_wrench_composer.reset() hand.instantaneous_wrench_composer.reset()
@@ -266,9 +291,17 @@ def main():
} }
require(app.is_running(), "Application stopped before finite test completed") require(app.is_running(), "Application stopped before finite test completed")
pose, velocity, q = state() pose, velocity, q = state()
if policy is None:
target_position, target_quaternion, target_joints = (
reference.wrist_position[step],
reference.wrist_quaternion[step],
reference.joint_position[step],
)
else:
target_position, target_quaternion, target_joints = policy.target(step, pose, q)
force, torque = wrench( force, torque = wrench(
reference.wrist_position[step], target_position,
reference.wrist_quaternion[step], target_quaternion,
pose, pose,
velocity, velocity,
array(hand.data.root_com_pose_w)[0, :3], array(hand.data.root_com_pose_w)[0, :3],
@@ -283,9 +316,11 @@ def main():
is_global=True, is_global=True,
) )
hand.set_joint_position_target_index( hand.set_joint_position_target_index(
target=tensor(reference.joint_position[step : step + 1, master_columns]), joint_ids=master_ids target=tensor(target_joints[None, master_columns]), joint_ids=master_ids
) )
hand.write_data_to_sim() hand.write_data_to_sim()
if source_camera is not None and step % 8 == 0:
source_camera.update((step + 1) * dt)
sim.step(render=args.gui and step % 8 == 0) sim.step(render=args.gui and step % 8 == 0)
hand.update(dt) hand.update(dt)
pose, velocity, q = state() pose, velocity, q = state()
@@ -347,6 +382,11 @@ def main():
"rms_errors_m_rad_rad_rad": np.sqrt((errors**2).mean(axis=0)).tolist(), "rms_errors_m_rad_rad_rad": np.sqrt((errors**2).mean(axis=0)).tolist(),
} }
) )
if policy is not None:
summaries[-1]["policy"] = policy.summary()
if source_camera is not None:
source_camera.update(args.steps * dt)
sim.render()
failure_context = {"phase": "repeatability", "completed_repetitions": 2, "steps_per_repetition": args.steps} failure_context = {"phase": "repeatability", "completed_repetitions": 2, "steps_per_repetition": args.steps}
np.testing.assert_allclose(reset_states[0], reset_states[1], atol=1e-6, rtol=0) np.testing.assert_allclose(reset_states[0], reset_states[1], atol=1e-6, rtol=0)
position_end = 7 + len(names) position_end = 7 + len(names)
@@ -358,7 +398,15 @@ def main():
json.dumps( json.dumps(
{ {
"status": "PASS", "status": "PASS",
"check": "bounded_experimental_dynamic_tracking", "check": "bounded_reference_act_rollout"
if policy is not None
else "bounded_experimental_dynamic_tracking",
"source_camera_sha256": hashlib.sha256(args.source_camera.read_bytes()).hexdigest()
if args.source_camera else None,
"camera_time_factor": args.camera_time_factor if args.source_camera else None,
"camera_sampling": "source_frame_hold" if args.source_camera else None,
"control_source": "reference_act" if policy is not None else "reference_targets",
"policy_checkpoint_sha256": policy.checkpoint_hash if policy is not None else None,
"runtime_verified_for_this_run_only": True, "runtime_verified_for_this_run_only": True,
"provenance": data.metadata["provenance"], "provenance": data.metadata["provenance"],
"reference_source": "hdf5" if args.hdf5 else "analytic_in_memory", "reference_source": "hdf5" if args.hdf5 else "analytic_in_memory",
+1 -1
View File
@@ -1,7 +1,7 @@
[package] [package]
# Semantic Versioning is used: https://semver.org/ # Semantic Versioning is used: https://semver.org/
version = "0.1.2" version = "0.1.3"
# Description # Description
category = "isaaclab" category = "isaaclab"
@@ -0,0 +1,5 @@
"""State-only reference imitation, separate from Isaac Lab task registration.
Importing this package does not launch Kit or initialize CUDA. It is not a robot
hardware controller, grasping task, or implementation of the full DexSchema.
"""
@@ -0,0 +1,182 @@
"""Manifest-bound geometry/posture encoding and bounded reference target adapter.
Shared geometry is root-link pose relative to the episode's initial root frame.
Joint layout/coupling lives ONLY in this adapter. Targets are state-derived
proxies, not measured actuator commands. No hardware or collision guarantees.
"""
import numpy as np
from dex_workbench_tracking.control import bounded, rotation_error
from dex_workbench_tracking.identity import manifest_side
from dex_workbench_tracking.trajectory import require
def qmul(a, b):
return np.r_[a[0] * b[0] - np.dot(a[1:], b[1:]), a[0] * b[1:] + b[0] * a[1:] + np.cross(a[1:], b[1:])]
def qmatrix(q):
q = np.asarray(q, dtype=float)
require(q.shape == (4,) and np.isfinite(q).all() and abs(np.linalg.norm(q) - 1) < 1e-4, "Unit wxyz required")
w, x, y, z = q / np.linalg.norm(q)
return np.array(
[
[1 - 2 * (y * y + z * z), 2 * (x * y - z * w), 2 * (x * z + y * w)],
[2 * (x * y + z * w), 1 - 2 * (x * x + z * z), 2 * (y * z - x * w)],
[2 * (x * z - y * w), 2 * (y * z + x * w), 1 - 2 * (x * x + y * y)],
]
)
def matrixq(matrix):
"""Rotation matrix -> unit wxyz, stable at half turns (largest component branch)."""
m = np.asarray(matrix, dtype=float)
candidates = np.array(
[
1 + np.trace(m),
1 + m[0, 0] - m[1, 1] - m[2, 2],
1 - m[0, 0] + m[1, 1] - m[2, 2],
1 - m[0, 0] - m[1, 1] + m[2, 2],
]
)
i = int(candidates.argmax())
s = 2 * np.sqrt(max(candidates[i], 0))
require(s > 1e-8, "Degenerate rotation")
if i == 0:
q = [s / 4, (m[2, 1] - m[1, 2]) / s, (m[0, 2] - m[2, 0]) / s, (m[1, 0] - m[0, 1]) / s]
elif i == 1:
q = [(m[2, 1] - m[1, 2]) / s, s / 4, (m[0, 1] + m[1, 0]) / s, (m[0, 2] + m[2, 0]) / s]
elif i == 2:
q = [(m[0, 2] - m[2, 0]) / s, (m[0, 1] + m[1, 0]) / s, s / 4, (m[1, 2] + m[2, 1]) / s]
else:
q = [(m[1, 0] - m[0, 1]) / s, (m[0, 2] + m[2, 0]) / s, (m[1, 2] + m[2, 1]) / s, s / 4]
q = np.asarray(q)
return q / np.linalg.norm(q)
def rotation6d(values):
a, b = np.asarray(values, dtype=float).reshape(2, 3)
require(np.isfinite(a).all() and np.isfinite(b).all(), "Nonfinite policy rotation")
require(np.linalg.norm(a) > 1e-6, "Degenerate policy rotation first axis")
a = a / np.linalg.norm(a)
b = b - np.dot(a, b) * a
require(np.linalg.norm(b) > 1e-6, "Degenerate policy rotation second axis")
b = b / np.linalg.norm(b)
return np.column_stack((a, b, np.cross(a, b)))
class Adapter:
def __init__(self, manifest):
self.side = manifest_side(manifest)
self.names = [j["name"] for j in manifest["joints"]]
require(len(set(self.names)) == len(self.names) and len(self.names) > 0, "Unique joint names required")
self.lower = np.array([j["lower_rad"] for j in manifest["joints"]], dtype=float)
self.upper = np.array([j["upper_rad"] for j in manifest["joints"]], dtype=float)
require(
np.isfinite(self.lower).all() and np.isfinite(self.upper).all() and (self.lower <= self.upper).all(),
"Invalid limits",
)
self.mimic = manifest.get("source_urdf", {}).get("mimic", [])
followers = [eq["joint"] for eq in self.mimic]
require(len(followers) == len(set(followers)), "Duplicate follower")
self.master_names = [n for n in self.names if n not in followers]
self.columns = [self.names.index(n) for n in self.master_names]
self.target_lower = self.lower[self.columns].copy()
self.target_upper = self.upper[self.columns].copy()
self.rate_multiplier = np.ones(len(self.columns))
for eq in self.mimic:
require(
eq["joint"] in self.names and eq["reference"] in self.master_names,
"Only inspected direct master/follower equations supported",
)
m, o = eq["multiplier"], eq["offset_rad"]
require(np.isfinite(m) and np.isfinite(o) and m != 0, "Invalid mimic coefficients")
child, master = self.names.index(eq["joint"]), self.master_names.index(eq["reference"])
bounds = sorted([(self.lower[child] - o) / m, (self.upper[child] - o) / m])
self.target_lower[master] = max(self.target_lower[master], bounds[0])
self.target_upper[master] = min(self.target_upper[master], bounds[1])
self.rate_multiplier[master] = max(self.rate_multiplier[master], abs(m))
require((self.target_lower <= self.target_upper).all(), "Infeasible coupled limits")
self.spec = {
"hand_side": self.side,
"asset_sha256": manifest["asset_sha256"],
"root_link": manifest["root_link"],
"joint_names": self.names,
"master_names": self.master_names,
"lower_rad": self.lower.tolist(),
"upper_rad": self.upper.tolist(),
"mimic": self.mimic,
}
self.action_dim = 9 + len(self.columns)
# Current root pose/all state joints + terminal geometric/posture goal + phase/duration.
self.observation_dim = 9 + len(self.names) + self.action_dim + 2
def encode_pose(self, position, quaternion, origin):
p0, q0 = origin
r0 = qmatrix(q0)
relative = r0.T @ qmatrix(quaternion)
return np.r_[r0.T @ (np.asarray(position) - p0), relative[:, 0], relative[:, 1]]
def action(self, position, quaternion, joints, origin):
return np.r_[self.encode_pose(position, quaternion, origin), np.asarray(joints)[self.columns]].astype(
np.float32
)
def observation(self, position, quaternion, joints, origin, goal, elapsed, duration):
require(duration > 0 and 0 <= elapsed <= duration + 1e-8, "Invalid trajectory phase")
result = np.r_[
self.encode_pose(position, quaternion, origin), joints, goal, min(elapsed / duration, 1), duration
]
require(result.shape == (self.observation_dim,) and np.isfinite(result).all(), "Invalid policy observation")
return result.astype(np.float32)
def expand(self, masters):
result = np.zeros(len(self.names), dtype=float)
result[self.columns] = masters
for eq in self.mimic:
result[self.names.index(eq["joint"])] = (
eq["multiplier"] * result[self.names.index(eq["reference"])] + eq["offset_rad"]
)
return result
def decode(self, action, origin):
a = np.asarray(action, dtype=float)
require(a.shape == (self.action_dim,) and np.isfinite(a).all(), "Invalid policy action")
p0, q0 = origin
r0 = qmatrix(q0)
return p0 + r0 @ a[:3], matrixq(r0 @ rotation6d(a[3:9])), a[9:]
class TargetLimiter:
"""Bound commands at each physics step; never teleport simulated state."""
def __init__(self, adapter, limits, position, quaternion, joints):
self.adapter, self.limits = adapter, limits
self.origin = np.array(position, dtype=float)
self.position, self.quaternion = np.array(position, dtype=float), np.array(quaternion, dtype=float)
self.masters = np.clip(np.asarray(joints)[adapter.columns], adapter.target_lower, adapter.target_upper)
self.limited_steps = 0
self.steps = 0
def step(self, desired, dt):
require(np.isfinite(dt) and dt > 0, "Positive command dt required")
position, quaternion, masters = desired
limits, adapter = self.limits, self.adapter
require(all(np.isfinite(v).all() for v in desired), "Nonfinite desired command")
bounded_position = self.origin + bounded(np.asarray(position) - self.origin, limits.workspace_radius)
p = self.position + bounded(bounded_position - self.position, limits.reference_speed * dt)
error = rotation_error(quaternion, self.quaternion)
angle = np.linalg.norm(error)
limited = min(angle, limits.reference_angular_speed * dt)
increment = np.r_[np.cos(limited / 2), error * (np.sin(limited / 2) / angle if angle > 1e-12 else 0.5)]
q = qmul(increment, self.quaternion)
q /= np.linalg.norm(q)
clipped = np.clip(masters, adapter.target_lower, adapter.target_upper)
maximum = min(limits.reference_joint_speed, limits.finger_velocity) * dt / adapter.rate_multiplier
m = self.masters + np.clip(clipped - self.masters, -maximum, maximum)
self.limited_steps += int(
np.linalg.norm(p - position) > 1e-8 or angle - limited > 1e-8 or np.max(np.abs(m - masters)) > 1e-8
)
self.steps += 1
self.position, self.quaternion, self.masters = p, q, m
return p.copy(), q.copy(), adapter.expand(m)
@@ -0,0 +1,193 @@
"""Offline ACT preparation/training commands. No command here starts Isaac Sim."""
import argparse
import importlib.util
import json
import platform
import sys
from dataclasses import replace
from pathlib import Path
import numpy as np
import torch
from dex_workbench_tracking.cli import publish_validated, synthetic
from dex_workbench_tracking.trajectory import ContractError, Demonstrations, Episode, require
from .config import Config
from .data import digest, prepare
from .engine import json_write, load_checkpoint, score, train
def read_json(path):
return json.loads(Path(path).read_text(encoding="utf-8"))
def provenance(metadata):
package = Path(__file__).parent
metadata["implementation_sha256"] = {p.name: digest(p) for p in sorted(package.glob("*.py"))}
metadata["python"] = platform.python_version()
metadata["numpy"] = np.__version__
metadata["torch"] = str(torch.__version__)
return metadata
def evaluate(checkpoint, hdf5, manifest, partition):
model, config, adapter, normalization, metadata = load_checkpoint(checkpoint, manifest)
require(
digest(hdf5) == metadata["hdf5_sha256"], "Evaluation HDF5 must match checkpoint; re-review new data explicitly"
)
_, datasets, fitted, _ = prepare(
hdf5, manifest, metadata["splits"], config, metadata["data_review"], synthetic_smoke=metadata["synthetic_smoke"]
)
require(partition in ("validation", "test") and partition in datasets, "Select a nonempty held-out split")
for key in normalization:
np.testing.assert_array_equal(normalization[key], fitted[key])
torch.set_num_threads(1)
metric = score(model, datasets[partition], config.batch_size, "cpu")
return {
"status": "PASS",
"scope": "offline_heldout_reference_prediction_not_dynamic_success",
"split": partition,
"windows": len(datasets[partition]),
"l1_normalized_z0": metric,
"checkpoint_sha256": digest(checkpoint),
"hdf5_sha256": digest(hdf5),
"synthetic_smoke": metadata["synthetic_smoke"],
"simulation_e2e": "NOT_RUN",
}
def smoke(manifest, output):
output = Path(output)
output.mkdir(parents=True, exist_ok=False)
side = manifest.get("hand_side", "left")
config = replace(
Config(),
hand_side=side,
chunk_size=4,
execute_steps=2,
hidden_dim=32,
heads=4,
layers=1,
latent_dim=4,
batch_size=4,
max_updates=4,
evaluate_every=2,
max_seconds=120,
)
original = synthetic(manifest)
episode = original.episodes["demo_000000"]
episodes = {}
for index, factor in enumerate((1.0, 1.3, 1.6)):
position = episode.wrist_position.copy()
position[:, 0] *= factor
episodes[f"demo_{index:06d}"] = Episode(
episode.time.copy(),
position,
episode.wrist_quaternion.copy(),
episode.joint_position.copy() * factor,
episode.valid.copy(),
)
data = Demonstrations(dict(original.metadata), original.joint_names, original.world_from_source.copy(), episodes)
data.metadata["source_description"] += "; THREE ANALYTIC CPU SMOKE FIXTURES, not independent expert recordings"
split = {
"schema": "l20_episode_splits_v1",
"train": ["demo_000000"],
"validation": ["demo_000001"],
"test": ["demo_000002"],
"episode_groups": {name: f"analytic_fixture_{i}" for i, name in enumerate(episodes)},
}
hdf5 = output / "synthetic.hdf5"
publish_validated(hdf5, data, manifest)
json_write(output / "splits.json", split)
json_write(output / "config.json", config.to_dict())
adapter, datasets, normalization, metadata = prepare(hdf5, manifest, split, config, synthetic_smoke=True)
result = train(adapter, datasets, normalization, provenance(metadata), config, output / "train", device="cpu")
evaluation = evaluate(output / "train/last.pt", hdf5, manifest, "test")
json_write(output / "offline-evaluation.json", evaluation)
return {
"status": "PASS",
"scope": "synthetic_CPU_training_save_reload_evaluate_only",
"training": result,
"evaluation": evaluation,
"simulation_e2e": "NOT_RUN",
"output": str(output),
}
def main():
parser = argparse.ArgumentParser(description=__doc__)
commands = parser.add_subparsers(dest="command", required=True)
doctor = commands.add_parser("doctor", help="CPU dependency check; CUDA probing is opt-in")
doctor.add_argument("--cuda", action="store_true")
for command in ("preflight", "train"):
item = commands.add_parser(command)
item.add_argument("--hdf5", type=Path, required=True)
item.add_argument("--manifest", type=Path, required=True)
item.add_argument("--config", type=Path, required=True)
item.add_argument("--splits", type=Path, required=True)
item.add_argument("--data-review", type=Path, required=True)
item.add_argument("--output", type=Path, required=True)
if command == "train":
item.add_argument("--device", choices=("cpu", "cuda"), default="cpu")
item.add_argument(
"--execute", action="store_true", help="Acknowledge configured updates/time/device budget"
)
evaluation = commands.add_parser("evaluate", help="CPU held-out prediction, not simulation rollout")
evaluation.add_argument("--checkpoint", type=Path, required=True)
evaluation.add_argument("--hdf5", type=Path, required=True)
evaluation.add_argument("--manifest", type=Path, required=True)
evaluation.add_argument("--split", choices=("validation", "test"), default="test")
evaluation.add_argument("--output", type=Path, required=True)
test = commands.add_parser("smoke", help="Four tiny CPU updates on labeled synthetic fixtures; no GPU/Kit")
test.add_argument("--manifest", type=Path, required=True)
test.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
try:
if args.command == "doctor":
require(sys.version_info >= (3, 12), "Python >=3.12 required")
major_minor = tuple(int(v) for v in torch.__version__.split("+")[0].split(".")[:2])
require(major_minor >= (2, 6), "Torch >=2.6 required for tensor-only checkpoint loading")
if args.cuda:
require(torch.cuda.is_available(), "CUDA unavailable; do not install/replace Isaac Torch automatically")
report = {
"status": "PASS",
"scope": "dependency_imports_only",
"python": platform.python_version(),
"torch": str(torch.__version__),
"numpy": np.__version__,
"isaaclab_discoverable": importlib.util.find_spec("isaaclab") is not None,
"cuda": "PASS" if args.cuda else "NOT_RUN",
"simulation_e2e": "NOT_RUN",
}
elif args.command == "smoke":
report = smoke(read_json(args.manifest), args.output)
elif args.command == "evaluate":
report = evaluate(args.checkpoint, args.hdf5, read_json(args.manifest), args.split)
json_write(args.output, report)
else:
if args.command == "train":
require(args.execute, "Training requires --execute; review configured compute budget first")
config = Config.load(args.config)
manifest, split, review = (read_json(p) for p in (args.manifest, args.splits, args.data_review))
adapter, datasets, normalization, metadata = prepare(args.hdf5, manifest, split, config, review)
metadata = provenance(metadata)
if args.command == "preflight":
report = {
"status": "PASS",
"scope": "data_and_training_configuration_only",
"config": config.to_dict(),
"metadata": metadata,
"windows": {k: len(v) for k, v in datasets.items()},
"simulation_e2e": "NOT_RUN",
}
json_write(args.output, report)
else:
report = train(adapter, datasets, normalization, metadata, config, args.output, args.device)
print(json.dumps(report, allow_nan=False))
except (ContractError, OSError, KeyError, TypeError, ValueError, RuntimeError, AssertionError) as error:
parser.exit(1, f"FAIL: {error}\n")
if __name__ == "__main__":
main()
@@ -0,0 +1,69 @@
"""Versioned, bounded configuration for goal-conditioned reference ACT."""
import json
from dataclasses import asdict, dataclass
from pathlib import Path
import numpy as np
from dex_workbench_tracking.trajectory import require
CONTRACT = "l20_goal_reference_act_v1"
CHECKPOINT = "l20_reference_act_checkpoint_v1"
@dataclass(frozen=True)
class Config:
contract: str = CONTRACT
hand_side: str = "right"
control_hz: int = 30
chunk_size: int = 16
execute_steps: int = 4
hidden_dim: int = 128
heads: int = 4
layers: int = 2
latent_dim: int = 32
batch_size: int = 32
learning_rate: float = 0.0001
kl_weight: float = 10.0
max_updates: int = 1000
evaluate_every: int = 100
max_seconds: float = 600.0
max_total_frames: int = 500000
workspace_radius: float = 0.8
seed: int = 42
def __post_init__(self):
require(self.contract == CONTRACT, "Unsupported observation/action contract")
require(self.hand_side in ("left", "right"), "Explicit left/right side required")
for key in (
"control_hz",
"chunk_size",
"execute_steps",
"hidden_dim",
"heads",
"layers",
"latent_dim",
"batch_size",
"max_updates",
"evaluate_every",
"max_total_frames",
"seed",
):
value = getattr(self, key)
require(type(value) is int and value > 0, f"{key} must be a positive integer")
require(240 % self.control_hz == 0, "control_hz must divide the validated 240Hz physics rate")
require(self.execute_steps <= self.chunk_size <= 256, "Require execute_steps <= chunk_size <= 256")
require(self.hidden_dim % self.heads == 0 and self.hidden_dim <= 512, "Invalid transformer width/heads")
require(self.layers <= 6 and self.latent_dim <= 256 and self.batch_size <= 512, "Model/batch resource cap")
require(self.max_updates <= 500000 and self.max_total_frames <= 2000000, "Training/data resource cap")
for key in ("learning_rate", "kl_weight", "max_seconds", "workspace_radius"):
value = getattr(self, key)
require(type(value) in (int, float) and np.isfinite(value) and value > 0, f"Invalid {key}")
require(self.max_seconds <= 86400, "Maximum one-day process budget; choose bounded experiments")
def to_dict(self):
return asdict(self)
@classmethod
def load(cls, path):
return cls(**json.loads(Path(path).read_text(encoding="utf-8")))
@@ -0,0 +1,168 @@
"""Episode/group-isolated reference windows; training-only normalization."""
import hashlib
from dataclasses import replace
from pathlib import Path
import numpy as np
import torch
from dex_workbench_tracking.control import Limits, validate_reference
from dex_workbench_tracking.trajectory import load, require, sample
from torch.utils.data import Dataset
from .adapter import Adapter
def digest(path):
value = hashlib.sha256()
with Path(path).open("rb") as stream:
for block in iter(lambda: stream.read(1024 * 1024), b""):
value.update(block)
return value.hexdigest()
def validate_splits(data, split):
require(split.get("schema") == "l20_episode_splits_v1", "Unsupported split schema")
require(
set(split) == {"schema", "train", "validation", "test", "episode_groups"}, "Unexpected/missing split fields"
)
groups = split["episode_groups"]
require(isinstance(groups, dict) and set(groups) == set(data.episodes), "Provide capture group for every episode")
require(all(isinstance(v, str) and v.strip() for v in groups.values()), "Nonempty source capture groups required")
require(split["train"] and split["validation"], "At least one independent train and validation episode required")
seen, group_owner, geometry_owner = set(), {}, {}
for partition in ("train", "validation", "test"):
require(isinstance(split[partition], list), "Split entries must be episode lists")
for name in split[partition]:
require(
isinstance(name, str) and name in data.episodes and name not in seen,
"Unknown/overlapping episode split",
)
seen.add(name)
group = groups[name]
require(group_owner.get(group, partition) == partition, "Capture group leakage across splits")
group_owner[group] = partition
# Catch exact renamed/slowdown copies even if the caller supplies false group IDs.
ep = data.episodes[name]
fingerprint = hashlib.sha256()
for field in (ep.wrist_position, ep.wrist_quaternion, ep.joint_position, ep.valid):
fingerprint.update(np.ascontiguousarray(field).tobytes())
fingerprint = fingerprint.hexdigest()
require(
geometry_owner.get(fingerprint, partition) == partition, "Duplicate trajectory geometry across splits"
)
geometry_owner[fingerprint] = partition
require(seen == set(data.episodes), "Every episode must have an explicit split; no silently ignored data")
def validate_review(review, data_hash, manifest):
require(review.get("schema") == "l20_training_review_v1", "Training review sidecar required")
require(review.get("hdf5_sha256") == data_hash, "Review must bind exact HDF5 bytes")
require(review.get("asset_sha256") == manifest["asset_sha256"], "Review asset identity mismatch")
for flag in ("coordinate_and_scale_reviewed", "reference_state_targets_accepted", "capture_groups_reviewed"):
require(review.get(flag) is True, f"BLOCKED: explicit data-owner approval missing: {flag}")
for key in ("reviewer", "evidence"):
require(isinstance(review.get(key), str) and review[key].strip(), f"Review {key} required")
class Windows(Dataset):
def __init__(self, series, names, chunk_size, normalization=None):
self.series, self.names, self.chunk_size = series, list(names), chunk_size
self.normalization = normalization
self.ends = np.cumsum([len(series[n][0]) - 1 for n in names])
require(len(self.ends) > 0 and self.ends[-1] > 0, "Empty partition")
def __len__(self):
return int(self.ends[-1])
def __getitem__(self, index):
require(0 <= index < len(self), "Window index out of range")
i = int(np.searchsorted(self.ends, index, side="right"))
start = index - (int(self.ends[i - 1]) if i else 0)
obs, actions = self.series[self.names[i]]
stop = min(start + 1 + self.chunk_size, len(actions))
length = stop - start - 1
target = np.repeat(actions[stop - 1 : stop], self.chunk_size, axis=0)
target[:length] = actions[start + 1 : stop]
mask = np.arange(self.chunk_size) < length
x = obs[start].copy()
if self.normalization is not None:
n = self.normalization
x = (x - n["observation_mean"]) / n["observation_std"]
target = (target - n["action_mean"]) / n["action_std"]
return (
torch.from_numpy(x.astype(np.float32)),
torch.from_numpy(target.astype(np.float32)),
torch.from_numpy(mask),
)
def fit_normalization(series, train_names):
result = {}
for label, column in (("observation", 0), ("action", 1)):
# Fit each TRAIN frame once; not padded windows and never validation/test.
values = np.concatenate([series[n][column][:-1] if column == 0 else series[n][column][1:] for n in train_names])
require(np.isfinite(values).all(), "Nonfinite training features")
result[label + "_mean"] = values.mean(axis=0, dtype=np.float64).astype(np.float32)
result[label + "_std"] = np.maximum(values.std(axis=0, dtype=np.float64), 1e-4).astype(np.float32)
return result
def prepare(hdf5, manifest, split, config, review=None, *, synthetic_smoke=False):
require(Path(hdf5).stat().st_size <= 2 * 1024**3, "Single HDF5 exceeds 2GiB preflight cap; shard explicitly")
data = load(hdf5, manifest)
require(data.metadata["hand_side"] == config.hand_side, "Config/data hand side mismatch")
data_hash = digest(hdf5)
if synthetic_smoke:
require(data.metadata["provenance"] == "synthetic", "Smoke cannot relabel real expert data")
else:
require(data.metadata["provenance"] == "expert_retargeted", "Real training requires expert_retargeted data")
validate_review(review or {}, data_hash, manifest)
validate_splits(data, split)
adapter = Adapter(manifest)
limits = replace(Limits(), workspace_radius=config.workspace_radius)
total, series, counts = 0, {}, {}
for name, episode in data.episodes.items():
validate_reference(episode, limits) # No implicit slowdown, clipping or invalid-gap bridging.
steps = int(round(episode.time[-1] * config.control_hz))
require(
steps >= 1 and abs(steps / config.control_hz - episode.time[-1]) < 1e-8,
f"{name}: duration must end on the configured control grid; no silent tail trimming",
)
total += steps + 1
require(total <= config.max_total_frames, "Resampled dataset exceeds configured frame budget")
query = np.arange(steps + 1, dtype=float) / config.control_hz
query[-1] = episode.time[-1] # Only the checked floating-point grid roundoff, not time warping.
ep = sample(episode, query)
origin = ep.wrist_position[0], ep.wrist_quaternion[0]
goal = adapter.action(ep.wrist_position[-1], ep.wrist_quaternion[-1], ep.joint_position[-1], origin)
observations, actions = [], []
for t, p, q, joints in zip(ep.time, ep.wrist_position, ep.wrist_quaternion, ep.joint_position, strict=True):
observations.append(adapter.observation(p, q, joints, origin, goal, t, ep.time[-1]))
actions.append(adapter.action(p, q, joints, origin))
series[name] = np.asarray(observations), np.asarray(actions)
counts[name] = {
"source_frames": len(episode.time),
"control_frames": len(ep.time),
"duration_s": float(ep.time[-1]),
}
normalization = fit_normalization(series, split["train"])
datasets = {
key: Windows(series, split[key], config.chunk_size, normalization)
for key in ("train", "validation", "test")
if split[key]
}
metadata = {
"hdf5_sha256": data_hash,
"provenance": data.metadata["provenance"],
"splits": split,
"data_review": review if not synthetic_smoke else None,
"synthetic_smoke": synthetic_smoke,
"episodes": counts,
"adapter": adapter.spec,
"limits_uncalibrated": vars(limits),
"target_semantics": "next_reference_root_pose_and_independent_q_target_proxy_not_measured_commands",
"observation_semantics": "current_start_relative_pose_all_q_terminal_goal_phase_duration",
"normalization_source": "train_frames_only",
}
return adapter, datasets, normalization, metadata
@@ -0,0 +1,183 @@
"""Bounded offline training, trusted tensor-only checkpoints and deterministic inference."""
import json
import os
import random
import time
from pathlib import Path
import numpy as np
import torch
from dex_workbench_tracking.trajectory import require
from torch.utils.data import DataLoader
from .adapter import Adapter
from .config import CHECKPOINT, Config
from .data import digest
from .model import ReferenceACT, objective
def json_write(path, value):
with Path(path).open("x", encoding="utf-8") as stream:
json.dump(value, stream, indent=2, allow_nan=False)
def score(model, dataset, batch_size, device, deadline=None):
model.eval()
total, count = 0.0, 0
with torch.inference_mode():
for obs, target, valid in DataLoader(dataset, batch_size=batch_size, shuffle=False, num_workers=0):
require(deadline is None or time.monotonic() < deadline, "Training/evaluation wall-time budget exhausted")
prediction = model(obs.to(device))[0]
error = (prediction - target.to(device)).abs() * valid.to(device).unsqueeze(-1)
require(torch.isfinite(error).all().item(), "Nonfinite held-out prediction")
total += error.sum().item()
count += valid.sum().item() * target.shape[-1]
return total / count
def checkpoint_payload(model, config, normalization, metadata, update):
return {
"format": CHECKPOINT,
"config": config.to_dict(),
"metadata": metadata,
"update": update,
"normalization": {k: torch.tensor(v.copy()) for k, v in normalization.items()},
"state_dict": {k: v.detach().cpu().clone() for k, v in model.state_dict().items()},
"torch_version": str(torch.__version__),
}
def save_checkpoint(path, payload):
"""Owned new run directory: atomic replace of this run's best/last checkpoint only."""
path = Path(path)
temporary = path.with_suffix(".pending")
with temporary.open("xb") as stream:
torch.save(payload, stream)
stream.flush()
os.fsync(stream.fileno())
os.replace(temporary, path)
def load_checkpoint(path, manifest):
require(Path(path).stat().st_size <= 512 * 1024**2, "Checkpoint exceeds 512MiB local loading cap")
# Caller-supplied local file only; no unsafe custom pickle globals.
payload = torch.load(path, map_location="cpu", weights_only=True)
require(payload.get("format") == CHECKPOINT, "Unsupported checkpoint")
config = Config(**payload["config"])
adapter = Adapter(manifest)
require(payload["metadata"]["adapter"] == adapter.spec, "Checkpoint embodiment/asset/adapter mismatch")
require(config.hand_side == adapter.side, "Checkpoint config side mismatch")
normalization = {k: v.cpu().numpy() for k, v in payload["normalization"].items()}
require(
set(normalization) == {"observation_mean", "observation_std", "action_mean", "action_std"},
"Normalization fields mismatch",
)
for key, value in normalization.items():
dim = adapter.observation_dim if key.startswith("observation") else adapter.action_dim
require(value.shape == (dim,) and np.isfinite(value).all(), "Invalid checkpoint normalization")
if key.endswith("std"):
require((value > 0).all(), "Normalization scales must be positive")
model = ReferenceACT(adapter.observation_dim, adapter.action_dim, config)
model.load_state_dict(payload["state_dict"], strict=True)
require(all(torch.isfinite(p).all().item() for p in model.parameters()), "Nonfinite checkpoint parameters")
model.eval()
return model, config, adapter, normalization, payload["metadata"]
def predict(model, normalization, observation):
obs = np.asarray(observation, dtype=np.float32)
require(
obs.shape == normalization["observation_mean"].shape and np.isfinite(obs).all(), "Invalid inference observation"
)
obs = (obs - normalization["observation_mean"]) / normalization["observation_std"]
with torch.inference_mode():
result = model(torch.from_numpy(obs[None]))[0][0].cpu().numpy()
result = result * normalization["action_std"] + normalization["action_mean"]
require(np.isfinite(result).all(), "Nonfinite policy prediction")
return result
def train(adapter, datasets, normalization, metadata, config, output, device="cpu"):
require(device in ("cpu", "cuda"), "Choose explicit cpu or cuda device")
require(device != "cuda" or torch.cuda.is_available(), "CUDA unavailable; no silent CPU fallback")
if metadata["synthetic_smoke"]:
require(
device == "cpu" and config.max_updates <= 10 and config.hidden_dim <= 64 and config.max_seconds <= 120,
"Synthetic smoke is CPU-only and tightly bounded",
)
output = Path(output)
output.mkdir(parents=True, exist_ok=False) # Never mix new data/config/checkpoints with an old run.
json_write(output / "run.json", {"config": config.to_dict(), "metadata": metadata, "device": device})
random.seed(config.seed)
np.random.seed(config.seed)
torch.manual_seed(config.seed)
# Keep CPU smoke inexpensive; this only affects this dedicated CLI/test process.
torch.set_num_threads(1)
model = ReferenceACT(adapter.observation_dim, adapter.action_dim, config).to(device)
optimizer = torch.optim.AdamW(model.parameters(), lr=config.learning_rate, weight_decay=1e-4)
generator = torch.Generator().manual_seed(config.seed)
started = time.monotonic()
deadline = started + config.max_seconds
best, completed = float("inf"), 0
last_loss = None
try:
with (output / "metrics.jsonl").open("x", encoding="utf-8") as log:
for update in range(1, config.max_updates + 1):
require(time.monotonic() < deadline, "Training wall-time budget exhausted; incomplete run")
indices = torch.randint(len(datasets["train"]), (config.batch_size,), generator=generator).tolist()
batch = [datasets["train"][index] for index in indices]
obs, target, valid = (torch.stack(values).to(device) for values in zip(*batch, strict=True))
model.train()
optimizer.zero_grad(set_to_none=True)
prediction, mu, logvar = model(obs, target, valid)
loss, l1, kl = objective(prediction, target, valid, mu, logvar, config.kl_weight)
require(torch.isfinite(loss).item(), "Nonfinite training loss")
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0, error_if_nonfinite=True)
optimizer.step()
completed = update
last_loss = {"update": update, "loss": loss.item(), "l1": l1.item(), "kl": kl.item()}
if update % config.evaluate_every == 0 or update == config.max_updates:
validation = score(model, datasets["validation"], config.batch_size, device, deadline)
last_loss["validation_l1_normalized_z0"] = validation
if validation < best:
best = validation
save_checkpoint(
output / "best.pt", checkpoint_payload(model, config, normalization, metadata, update)
)
log.write(json.dumps(last_loss, allow_nan=False) + "\n")
log.flush()
model.eval()
payload = checkpoint_payload(model, config, normalization, metadata, completed)
save_checkpoint(output / "last.pt", payload)
reloaded = ReferenceACT(adapter.observation_dim, adapter.action_dim, config)
reloaded.load_state_dict(
torch.load(output / "last.pt", map_location="cpu", weights_only=True)["state_dict"], strict=True
)
reloaded.eval()
probe = datasets["validation"][0][0].unsqueeze(0)
with torch.inference_mode():
expected = model.cpu()(probe)[0]
actual = reloaded(probe)[0]
torch.testing.assert_close(actual, expected, rtol=0, atol=1e-6)
require(time.monotonic() <= deadline, "Training exceeded budget during finalization")
result = {
"status": "PASS",
"scope": "bounded_offline_reference_act_training_not_policy_quality",
"synthetic_smoke": metadata["synthetic_smoke"],
"completed_updates": completed,
"elapsed_seconds": time.monotonic() - started,
"last_metrics": last_loss,
"best_validation_l1_normalized": best,
"best_checkpoint_sha256": digest(output / "best.pt"),
"last_checkpoint_sha256": digest(output / "last.pt"),
"checkpoint_reload_prediction_max_abs_diff": float((actual - expected).abs().max()),
"simulation_e2e": "NOT_RUN",
"hardware_and_training_quality_validated": False,
}
json_write(output / "result.json", result)
return result
except BaseException as error:
json_write(output / "failure.json", {"status": "FAIL", "completed_updates": completed, "error": str(error)})
raise
@@ -0,0 +1,51 @@
"""State-only ACT-style CVAE Transformer (no vision backbone or official-weight compatibility)."""
import torch
from torch import nn
class ReferenceACT(nn.Module):
def __init__(self, observation_dim, action_dim, config):
super().__init__()
self.config = config
width = config.hidden_dim
self.observation = nn.Linear(observation_dim, width)
self.action = nn.Linear(action_dim, width)
self.cls = nn.Parameter(torch.zeros(1, 1, width))
self.posterior_position = nn.Parameter(torch.randn(1, config.chunk_size + 2, width) * 0.02)
encoder = nn.TransformerEncoderLayer(
width, config.heads, width * 4, dropout=0, batch_first=True, norm_first=True
)
self.posterior = nn.TransformerEncoder(encoder, config.layers, enable_nested_tensor=False)
self.distribution = nn.Linear(width, config.latent_dim * 2)
self.latent = nn.Linear(config.latent_dim, width)
self.queries = nn.Parameter(torch.randn(1, config.chunk_size, width) * 0.02)
decoder = nn.TransformerDecoderLayer(
width, config.heads, width * 4, dropout=0, batch_first=True, norm_first=True
)
self.decoder = nn.TransformerDecoder(decoder, config.layers)
self.output = nn.Linear(width, action_dim)
def forward(self, observation, actions=None, valid=None):
batch = len(observation)
obs = self.observation(observation).unsqueeze(1)
mu = logvar = None
if actions is not None:
tokens = torch.cat((self.cls.expand(batch, -1, -1), obs, self.action(actions)), dim=1)
padding = torch.cat((torch.zeros((batch, 2), dtype=torch.bool, device=obs.device), ~valid), dim=1)
encoded = self.posterior(tokens + self.posterior_position, src_key_padding_mask=padding)[:, 0]
mu, logvar = self.distribution(encoded).chunk(2, dim=-1)
logvar = logvar.clamp(-10, 10)
z = mu + torch.exp(0.5 * logvar) * torch.randn_like(mu)
else:
# Deterministic ACT inference: posterior is never supplied reference future actions.
z = observation.new_zeros((batch, self.config.latent_dim))
memory = torch.cat((obs, self.latent(z).unsqueeze(1)), dim=1)
prediction = self.output(self.decoder(self.queries.expand(batch, -1, -1), memory))
return prediction, mu, logvar
def objective(prediction, actions, valid, mu, logvar, kl_weight):
reconstruction = ((prediction - actions).abs() * valid.unsqueeze(-1)).sum() / (valid.sum() * actions.shape[-1])
kl = -0.5 * (1 + logvar - mu.square() - logvar.exp()).sum(dim=-1).mean()
return reconstruction + kl_weight * kl, reconstruction, kl
@@ -0,0 +1,77 @@
"""Receding action-chunk policy for the existing bounded Isaac Lab replay runner."""
import torch
from dex_workbench_tracking.trajectory import require
from .adapter import TargetLimiter
from .data import digest
from .engine import load_checkpoint, predict
class ClosedLoopReferenceACT:
def __init__(self, checkpoint, manifest, hdf5, episode_name, episode, limits):
# This optional branch owns the dedicated evaluation process; avoid a large
# CPU thread pool for tiny, latency-sensitive chunk inference.
torch.set_num_threads(1)
self.model, self.config, self.adapter, self.normalization, self.metadata = load_checkpoint(checkpoint, manifest)
require(
hdf5 is not None and digest(hdf5) == self.metadata["hdf5_sha256"],
"Policy evaluation must bind exact trained dataset",
)
split = self.metadata["splits"]
require(
episode_name in split["validation"] + split["test"] and episode_name not in split["train"],
"Policy evaluation requires a held-out episode, never train replay",
)
require(vars(limits) == self.metadata["limits_uncalibrated"], "Policy/runtime controller limits mismatch")
self.episode, self.limits = episode, limits
self.checkpoint_hash = digest(checkpoint)
self.physics_per_control = 240 // self.config.control_hz
self.duration = float(episode.time[-1])
self.origin = episode.wrist_position[0], episode.wrist_quaternion[0]
self.goal = self.adapter.action(
episode.wrist_position[-1], episode.wrist_quaternion[-1], episode.joint_position[-1], self.origin
)
self.reset()
def reset(self):
e = self.episode
self.limiter = TargetLimiter(
self.adapter, self.limits, e.wrist_position[0], e.wrist_quaternion[0], e.joint_position[0]
)
self.chunk = None
self.next_step = 0
self.inference_calls = 0
self.desired = None
def target(self, step, pose, joints):
require(step == self.next_step, "Policy clock must advance one physics step at a time; reset explicitly")
if step % self.physics_per_control == 0:
control_step = step // self.physics_per_control
slot = control_step % self.config.execute_steps
if slot == 0:
observation = self.adapter.observation(
pose[:3], pose[3:], joints, self.origin, self.goal, step / 240, self.duration
)
self.chunk = predict(self.model, self.normalization, observation)
self.inference_calls += 1
self.desired = self.adapter.decode(self.chunk[slot], self.origin)
result = self.limiter.step(self.desired, 1 / 240)
self.next_step += 1
return result
def summary(self):
return {
"checkpoint_sha256": self.checkpoint_hash,
"contract": self.config.contract,
"control_hz": self.config.control_hz,
"chunk_size": self.config.chunk_size,
"execute_steps": self.config.execute_steps,
"inference_calls": self.inference_calls,
"limited_target_steps": self.limiter.limited_steps,
"physics_steps": self.limiter.steps,
"synthetic_smoke_checkpoint": self.metadata["synthetic_smoke"],
"observation_source": "measured_current_state_plus_terminal_goal_phase_duration",
"future_reference_used_for_policy": "terminal_goal_only; no teacher forcing",
"limiting_is_not_force_saturation_telemetry": True,
}
@@ -0,0 +1,216 @@
"""CPU-only reference diagnostics; no dynamics, identity overrides or data rewriting.
Requires a schema/manifest-validated, fully valid episode. Finite differences are
reference estimates, not measured velocities. Slowdown bounds cover reference
speed gates only, never force/torque, tracking quality, workspace or hardware.
"""
import argparse
import hashlib
import json
from dataclasses import replace
from pathlib import Path
import numpy as np
from .cli import stretch_time
from .control import Limits, rotation_error, validate_reference
from .trajectory import ContractError, load, require, validate_against_manifest
def stats(values):
values = np.asarray(values, dtype=np.float64)
require(values.size > 0 and np.isfinite(values).all(), "Finite nonempty diagnostic values required")
return {
"min": float(values.min()),
"median": float(np.median(values)),
"p95": float(np.percentile(values, 95)),
"max": float(values.max()),
}
def speed_summary(values, time, limit):
"""Unweighted interval statistics, with zero-based source frame locations."""
index = int(np.argmax(values))
return {
**stats(values),
"reference_limit": limit,
"peak_interval_frames": [index, index + 1],
"peak_interval_time_s": [float(time[index]), float(time[index + 1])],
"over_limit_interval_starts": np.flatnonzero(values > limit).tolist(),
"speed_only_slowdown_lower_bound": max(1.0, float(np.max(values)) / limit),
}
def diagnose(data, manifest, episode_name, limits=None, factors=(1, 60, 69, 70, 75, 80)):
"""No implicit name reorder, invalid-gap bridging, clipping or asset rebinding."""
limits = Limits() if limits is None else limits
validate_against_manifest(data, manifest)
episode = data.episodes[episode_name]
require(episode.valid.all(), "Offline diagnostic needs one fully valid episode; segment explicitly")
time = episode.time
dt = np.diff(time)
require(len(dt) > 0 and np.isfinite(dt).all() and (dt > 0).all(), "Increasing finite time required")
# Promote float32 payload before subtraction to avoid extra cancellation rounding.
position = episode.wrist_position.astype(np.float64)
joints = episode.joint_position.astype(np.float64)
linear = np.diff(position, axis=0) / dt[:, None]
angular = (
np.array(
[
rotation_error(b, a)
for a, b in zip(episode.wrist_quaternion[:-1], episode.wrist_quaternion[1:], strict=True)
]
)
/ dt[:, None]
)
joint_velocity = np.diff(joints, axis=0) / dt[:, None]
speed = np.linalg.norm(linear, axis=1)
omega = np.linalg.norm(angular, axis=1)
qspeed = np.abs(joint_velocity)
displacement = np.linalg.norm(position - position[0], axis=1)
lower = np.array([j["lower_rad"] for j in manifest["joints"]])
upper = np.array([j["upper_rad"] for j in manifest["joints"]])
followers = {eq["joint"] for eq in manifest.get("source_urdf", {}).get("mimic", [])}
per_joint = {}
for column, name in enumerate(data.joint_names):
per_joint[name] = {
"role": "model_follower" if name in followers else "model_independent_target_not_hardware_mapping",
"position_range_rad": [float(joints[:, column].min()), float(joints[:, column].max())],
"minimum_limit_margin_rad": float(
np.minimum(joints[:, column] - lower[column], upper[column] - joints[:, column]).min()
),
"speed_rad_s": speed_summary(qspeed[:, column], time, limits.reference_joint_speed),
}
mimic = {}
for eq in manifest.get("source_urdf", {}).get("mimic", []):
child, parent = (data.joint_names.index(eq[key]) for key in ("joint", "reference"))
residual = joints[:, child] - eq["multiplier"] * joints[:, parent] - eq["offset_rad"]
mimic[eq["joint"]] = {
"max_abs_residual_rad": float(np.abs(residual).max()),
"peak_frame": int(np.argmax(np.abs(residual))),
"reference_tolerance_rad": 1e-3,
"leader_range_rad": float(np.ptp(joints[:, parent])),
}
summaries = {
"translation_m_s": speed_summary(speed, time, limits.reference_speed),
"rotation_rad_s": speed_summary(omega, time, limits.reference_angular_speed),
"all_joint_max_rad_s": speed_summary(qspeed.max(axis=1), time, limits.reference_joint_speed),
}
speed_bound = max(s["speed_only_slowdown_lower_bound"] for s in summaries.values())
candidates = []
for factor in factors:
# Reuse the real float32 reference gate, rather than treating our double
# precision estimates as an exact reproduction of boundary comparisons.
stretched = stretch_time(data, factor).episodes[episode_name]
try:
validate_reference(stretched, limits)
gate_status, reason = "PASS", None
except ContractError as error:
gate_status, reason = "FAIL", str(error)
candidates.append(
{
"factor": float(factor),
"duration_s": float(stretched.time[-1]),
"estimated_peak_translation_m_s": float(speed.max() / factor),
"estimated_peak_rotation_rad_s": float(omega.max() / factor),
"estimated_peak_joint_rad_s": float(qspeed.max() / factor),
"reference_gate": gate_status,
"first_gate_failure": reason,
"dynamics": "NOT_RUN",
}
)
# Midpoint finite differences describe reference roughness only. The replay
# interpolator is piecewise linear/SLERP: knot acceleration is not bounded by
# these estimates, and no dynamics/force inference is justified from them.
roughness = None
if len(dt) > 1:
midpoint_dt = (dt[:-1] + dt[1:]) / 2
roughness = {
"translation_m_s2": stats(np.linalg.norm(np.diff(linear, axis=0), axis=1) / midpoint_dt),
"rotation_rad_s2": stats(np.linalg.norm(np.diff(angular, axis=0), axis=1) / midpoint_dt),
"all_joint_max_rad_s2": stats((np.abs(np.diff(joint_velocity, axis=0)) / midpoint_dt[:, None]).max(axis=1)),
}
return {
"report_version": "l20_reference_diagnostic_v1",
"status": "PASS",
"status_scope": "offline report generated; NOT dynamic or replay approval",
"asset_compatibility": "PASS_MANIFEST_ONLY_NOT_USD_REINSPECTION",
"episode": episode_name,
"hand_side": data.metadata["hand_side"],
"asset_sha256": data.metadata["asset_sha256"],
"provenance": data.metadata["provenance"],
"frames": len(time),
"valid_frames": int(episode.valid.sum()),
"duration_s": float(time[-1]),
"sample_dt_s": stats(dt),
"effective_sample_hz": float(1 / np.median(dt)),
"quaternion_max_norm_error": float(
np.abs(np.linalg.norm(episode.wrist_quaternion.astype(float), axis=1) - 1).max()
),
"limits_uncalibrated": vars(limits),
"speed_estimates": summaries,
"speed_only_slowdown_lower_bound": speed_bound,
"workspace": {
"max_displacement_from_start_m": float(displacement.max()),
"peak_frame": int(displacement.argmax()),
"radius_m": limits.workspace_radius,
"outside_frames": np.flatnonzero(displacement > limits.workspace_radius).tolist(),
"fixable_by_time_stretch": False,
"path_length_m": float(np.linalg.norm(np.diff(position, axis=0), axis=1).sum()),
},
"joints": per_joint,
"mimic": mimic,
"midpoint_acceleration_estimates_not_physical_bounds": roughness,
"intervals": [
{
"frames": [i, i + 1],
"time_s": [float(time[i]), float(time[i + 1])],
"translation_m_s": float(speed[i]),
"rotation_rad_s": float(omega[i]),
"max_joint_rad_s": float(qspeed[i].max()),
"fastest_joint": data.joint_names[int(qspeed[i].argmax())],
}
for i in range(len(dt))
],
"candidates": candidates,
"dynamic_replay": "NOT_RUN",
"limitations": [
"Derived reference differences, not measured hardware or simulator velocities.",
"No position/rotation rescaling, recentering, smoothing or invalid-gap interpolation.",
"Speed-only lower bound is necessary for source intervals, not a dynamics guarantee.",
"Piecewise interpolation has velocity jumps; midpoint accelerations are not knot bounds.",
"No force/torque saturation, collision, calibration or tracking stability measured.",
"Candidate reference gate is the existing validate_reference, not full runtime acceptance.",
],
}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("input", type=Path)
parser.add_argument("--manifest", type=Path, required=True)
parser.add_argument("--episode", default="demo_000000")
parser.add_argument("--output", type=Path, required=True, help="New JSON path, never overwrite")
parser.add_argument("--workspace-radius", type=float, help="Comparison only; does not change runtime defaults")
parser.add_argument("--factors", nargs="+", type=float, default=[1, 60, 69, 70, 75, 80])
args = parser.parse_args()
try:
manifest = json.loads(args.manifest.read_text())
data = load(args.input, manifest)
limits = Limits()
if args.workspace_radius is not None:
limits = replace(limits, workspace_radius=args.workspace_radius)
report = diagnose(data, manifest, args.episode, limits, args.factors)
report["input_hdf5_sha256"] = hashlib.sha256(args.input.read_bytes()).hexdigest()
report["manifest_file_sha256"] = hashlib.sha256(args.manifest.read_bytes()).hexdigest()
encoded = json.dumps(report, indent=2, allow_nan=False) + "\n"
with args.output.open("x", encoding="utf-8") as stream:
stream.write(encoded)
print(json.dumps({"status": "PASS", "scope": report["status_scope"], "output": str(args.output)}))
except (ContractError, OSError, KeyError, TypeError, ValueError) as error:
parser.exit(1, f"FAIL: {error}\n")
if __name__ == "__main__":
main()
@@ -0,0 +1,65 @@
"""Presentation-only estimated source camera; no physics or reference changes."""
import numpy as np
def load_camera(path, reference_time, time_factor):
"""Validate source timestamps against the replay's explicitly stretched time."""
if not np.isfinite(time_factor) or time_factor < 1:
raise ValueError("Camera time factor must be finite and >= 1")
with np.load(path, allow_pickle=False) as data:
time = data["time"].copy()
poses = data["final_world_from_camera"].copy()
inverse = data["camera_from_final_world"].copy()
k = data["K"].copy()
if time.ndim != 1 or len(time) < 2 or poses.shape != (len(time), 4, 4):
raise ValueError("Invalid camera array shapes")
if not all(np.isfinite(a).all() for a in (time, poses, inverse, k)):
raise ValueError("Nonfinite camera data")
if not (np.diff(time) > 0).all() or time[0] != 0:
raise ValueError("Invalid camera timestamps")
np.testing.assert_allclose(time * time_factor, reference_time, atol=1e-9, rtol=0)
np.testing.assert_allclose(poses @ inverse, np.broadcast_to(np.eye(4), poses.shape), atol=2e-6, rtol=0)
rotation = poses[:, :3, :3]
np.testing.assert_allclose(rotation.transpose(0, 2, 1) @ rotation,
np.broadcast_to(np.eye(3), rotation.shape), atol=3e-6, rtol=0)
np.testing.assert_allclose(np.linalg.det(rotation), 1, atol=3e-6, rtol=0)
np.testing.assert_allclose(poses[:, 3], np.tile([0, 0, 0, 1], (len(time), 1)), atol=1e-9)
# This delivery uses centered 1280x720 full-image intrinsics, no crop/distortion.
if k.shape != (3, 3) or k[0, 0] <= 0 or k[1, 1] <= 0:
raise ValueError("Invalid camera intrinsics")
np.testing.assert_allclose(k, [[k[0, 0], 0, 640], [0, k[1, 1], 360], [0, 0, 1]], atol=1e-9)
return time * time_factor, poses, k
def usd_pose(cv_pose):
"""OpenCV +Y down/+Z forward to USD +Y up/-Z forward, column-vector SE3."""
return cv_pose @ np.diag([1.0, -1.0, -1.0, 1.0])
class SourceCamera:
def __init__(self, stage, path, reference_time, time_factor):
from pxr import Gf, UsdGeom
from omni.kit.viewport.utility import get_active_viewport
self.time, self.poses, k = load_camera(path, reference_time, time_factor)
self.Gf = Gf
camera = UsdGeom.Camera.Define(stage, "/World/SourceVideoCamera")
camera.CreateProjectionAttr("perspective")
camera.CreateFocalLengthAttr(20.0)
camera.CreateHorizontalApertureAttr(20.0 * 1280 / k[0, 0])
camera.CreateVerticalApertureAttr(20.0 * 720 / k[1, 1])
camera.CreateClippingRangeAttr(Gf.Vec2f(0.01, 100.0))
self.op = UsdGeom.Xformable(camera.GetPrim()).AddTransformOp()
viewport = get_active_viewport()
if viewport is None:
raise RuntimeError("Source camera requires an active Kit viewport")
viewport.set_texture_resolution((1280, 720))
viewport.camera_path = str(camera.GetPath())
self.update(0)
def update(self, replay_time):
# Hold each exported source frame until the next timestamp (no guessed poses).
index = int(np.clip(np.searchsorted(self.time, replay_time, side="right") - 1, 0, len(self.time) - 1))
matrix = usd_pose(self.poses[index])
self.op.Set(self.Gf.Matrix4d(*matrix.T.reshape(-1).tolist()))
+41
View File
@@ -4,6 +4,47 @@ Changelog
Unreleased Unreleased
~~~~~~~~~~ ~~~~~~~~~~
0.1.3 (2026-09-15)
~~~~~~~~~~~~~~~~~~
Experimental progress snapshot; not a full simulation release-gate PASS.
Added
^^^^^
* Add bounded state-only reference ACT (CVAE Transformer) training, train-only
normalization, capture-group/episode splits, hash-bound data-review requirements,
tensor-only checkpoint loading and offline held-out evaluation. Provide explicit
configuration templates, an Isaac-Python wrapper and synthetic CPU smoke.
* Add an opt-in checkpoint branch to the existing bounded tracking runner with
measured-state feedback, terminal goal/phase conditioning, receding action chunks
and manifest-bound coupled position/rate limits. Existing runtime assertions remain;
the new policy GPU E2E and real-data training are NOT_RUN. No new Gym/PPO task or
hardware mapping is claimed; existing tracking HDF5 schema and Cartpole are unchanged.
* Add a CPU-only, identity-validated reference diagnostic with source-interval
speeds, workspace/limit/mimic metrics and actual existing reference-gate checks
for in-memory slowdown candidates. Preserve inputs and refuse output overwrites.
Document the right expert trajectory's translation-dominated ~69.37x speed-only
slowdown bound; this is not dynamic feasibility or original-speed acceptance.
Runtime controls, assets and data schema remain unchanged.
* Record one separately authorized right expert-reference replay at 75x slowdown:
full 2x30000 steps at 240Hz, exit0/PASS in 364 seconds with unchanged controls
and assertions; 98 CPU/USD regressions pass. Preserve the upstream world-frame
and metric-scale verification blocker; no original-speed, GUI, task or hardware claim.
* Add opt-in estimated source-video camera replay with explicit OpenCV-to-USD
axes, centered 1280x720 intrinsics, source-frame hold and validated time stretch.
Camera data is external; no calibration or pixel-alignment claim. The first
full GUI run reached GUI_READY but timed out at 600 seconds (exit 124); full
camera replay acceptance is FAIL, not PASS.
* Record the new 1333-frame right reference: verified source model binding and
unchanged geometry, 3x slowdown / 0.4 m workspace, full default-camera GUI
replay 2x31968 steps at 240 Hz, exit 0/PASS. Maximum wrist position error is
3.582 mm; this does not validate original speed, object contact or grasp success.
* Verify 120 CPU/USD regression tests without skips and a fresh four-update
synthetic CPU ACT save/reload/held-out smoke. Full pre-commit is BLOCKED by
missing tooling; policy GPU E2E and real-data training remain NOT_RUN.
0.1.2 (2026-09-14) 0.1.2 (2026-09-14)
~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~
+5 -2
View File
@@ -24,7 +24,7 @@ INSTALL_REQUIRES = [
# Installation operation # Installation operation
setup( setup(
name="dex_workbench", name="dex_workbench",
packages=["dex_workbench", "dex_workbench_tracking"], packages=["dex_workbench", "dex_workbench_tracking", "dex_workbench_imitation"],
author=EXTENSION_TOML_DATA["package"]["author"], author=EXTENSION_TOML_DATA["package"]["author"],
maintainer=EXTENSION_TOML_DATA["package"]["maintainer"], maintainer=EXTENSION_TOML_DATA["package"]["maintainer"],
url=EXTENSION_TOML_DATA["package"]["repository"], url=EXTENSION_TOML_DATA["package"]["repository"],
@@ -32,7 +32,10 @@ setup(
description=EXTENSION_TOML_DATA["package"]["description"], description=EXTENSION_TOML_DATA["package"]["description"],
keywords=EXTENSION_TOML_DATA["package"]["keywords"], keywords=EXTENSION_TOML_DATA["package"]["keywords"],
install_requires=INSTALL_REQUIRES, install_requires=INSTALL_REQUIRES,
extras_require={"tracking": ["numpy>=1.26", "h5py>=3.10"]}, extras_require={
"tracking": ["numpy>=1.26", "h5py>=3.10"],
"imitation": ["numpy>=1.26", "h5py>=3.10", "torch>=2.6"],
},
license="Apache-2.0", license="Apache-2.0",
include_package_data=True, include_package_data=True,
python_requires=">=3.12", python_requires=">=3.12",
@@ -0,0 +1,427 @@
"""Analytic CPU tests; no Kit, GPU, real demonstrations or policy-quality claims."""
import copy
import importlib.util
import json
import subprocess
import sys
import tempfile
import unittest
from dataclasses import replace
from pathlib import Path
import numpy as np
import torch
from dex_workbench_imitation.adapter import Adapter, TargetLimiter, matrixq, qmatrix, qmul, rotation6d
from dex_workbench_imitation.cli import evaluate, smoke
from dex_workbench_imitation.config import Config
from dex_workbench_imitation.data import digest, fit_normalization, prepare, validate_review, validate_splits
from dex_workbench_imitation.engine import load_checkpoint, predict, train
from dex_workbench_imitation.model import ReferenceACT, objective
from dex_workbench_imitation.policy import ClosedLoopReferenceACT
from dex_workbench_tracking.cli import synthetic, write
from dex_workbench_tracking.control import Limits, rotation_error
from dex_workbench_tracking.trajectory import ContractError, Demonstrations, Episode, load
ROOT = Path(__file__).resolve().parents[3]
def manifest_fixture():
return {
"manifest_version": "l20_asset_manifest_v1",
"hand_side": "left",
"asset_sha256": "a" * 64,
"root_link": "analytic_base",
"joints": [
{"name": "follower", "lower_rad": 0.0, "upper_rad": 0.8},
{"name": "leader", "lower_rad": 0.0, "upper_rad": 1.0},
],
"source_urdf": {"mimic": [{"joint": "follower", "reference": "leader", "multiplier": 2.0, "offset_rad": 0.0}]},
}
def fixtures(manifest):
data = synthetic(manifest)
ep = data.episodes["demo_000000"]
episodes = {}
for i, factor in enumerate((1.0, 1.3, 1.6)):
p = ep.wrist_position.copy()
p[:, 0] *= factor
episodes[f"demo_{i:06d}"] = Episode(
ep.time.copy(), p, ep.wrist_quaternion.copy(), ep.joint_position * factor, ep.valid.copy()
)
data = Demonstrations(dict(data.metadata), data.joint_names, data.world_from_source.copy(), episodes)
split = {
"schema": "l20_episode_splits_v1",
"train": ["demo_000000"],
"validation": ["demo_000001"],
"test": ["demo_000002"],
"episode_groups": {n: f"analytic_{i}" for i, n in enumerate(episodes)},
}
return data, split
class AdapterTests(unittest.TestCase):
def setUp(self):
self.adapter = Adapter(manifest_fixture())
def test_half_turn_and_sign_equivalent_geometry_round_trip(self):
for q in ([1.0, 0, 0, 0], [0.0, 1, 0, 0], [0.0, 0, 1, 0], [0.0, 0, 0, 1], [0.5, 0.5, 0.5, 0.5]):
with self.subTest(q=q):
np.testing.assert_allclose(qmatrix(matrixq(qmatrix(q))), qmatrix(q), atol=1e-12)
np.testing.assert_allclose(qmatrix(-np.array(q)), qmatrix(q), atol=1e-12)
r = qmatrix(q)
np.testing.assert_allclose(rotation6d(np.r_[r[:, 0], r[:, 1]]), r, atol=1e-12)
def test_start_relative_encoding_invariant_to_global_rigid_transform(self):
origin = np.array([1.0, 2, 3]), np.array([1.0, 0, 0, 0])
p, q, joints = np.array([1.1, 2.2, 3.3]), np.array([0.5, 0.5, 0.5, 0.5]), np.array([0.4, 0.2])
a = self.adapter.action(p, q, joints, origin)
global_q = np.array([0.5, -0.5, 0.5, 0.5])
r, shift = qmatrix(global_q), np.array([4.0, 5, 6])
transformed = self.adapter.action(
r @ p + shift, qmul(global_q, q), joints, (r @ origin[0] + shift, qmul(global_q, origin[1]))
)
np.testing.assert_allclose(a, transformed, atol=1e-6)
p2, q2, m = self.adapter.decode(a, origin)
np.testing.assert_allclose(p2, p, atol=1e-6)
np.testing.assert_allclose(qmatrix(q2), qmatrix(q), atol=1e-6)
np.testing.assert_allclose(self.adapter.expand(m), joints, atol=1e-6)
def test_degenerate_action_is_rejected_not_silently_repaired(self):
for values in ([0] * 6, [1, 0, 0, 2, 0, 0], [float("nan")] * 6):
with self.assertRaises(ContractError):
rotation6d(values)
def test_coupled_limit_and_all_state_joint_rate_envelope(self):
self.assertEqual(self.adapter.master_names, ["leader"])
np.testing.assert_array_equal(self.adapter.target_upper, [0.4])
limiter = TargetLimiter(self.adapter, Limits(), [0, 0, 0], [1, 0, 0, 0], [0, 0])
desired = np.array([10.0, 0, 0]), np.array([0.0, 0, 0, 1]), np.array([10.0])
p0, q0, j0 = np.zeros(3), np.array([1.0, 0, 0, 0]), np.zeros(2)
for _ in range(250):
p, q, j = limiter.step(desired, 1 / 240)
self.assertLessEqual(np.linalg.norm(p - p0), 0.05 / 240 + 1e-12)
self.assertLessEqual(np.linalg.norm(rotation_error(q, q0)), 0.5 / 240 + 1e-12)
self.assertLessEqual(np.max(np.abs(j - j0)), 0.5 / 240 + 1e-12)
self.assertLessEqual(np.linalg.norm(p), 0.1 + 1e-12)
self.assertAlmostEqual(j[0], 2 * j[1])
p0, q0, j0 = p, q, j
self.assertEqual(limiter.limited_steps, 250)
def test_negative_mimic_intersection_and_unsupported_cascade(self):
manifest = manifest_fixture()
manifest["source_urdf"]["mimic"][0].update(multiplier=-2.0, offset_rad=0.8)
adapter = Adapter(manifest)
np.testing.assert_allclose(adapter.target_lower, [0])
np.testing.assert_allclose(adapter.target_upper, [0.4])
manifest["source_urdf"]["mimic"][0]["reference"] = "follower"
with self.assertRaises(ContractError):
Adapter(manifest)
class DatasetTests(unittest.TestCase):
def setUp(self):
self.directory = tempfile.TemporaryDirectory()
self.addCleanup(self.directory.cleanup)
self.path = Path(self.directory.name) / "data.hdf5"
self.manifest = manifest_fixture()
self.data, self.split = fixtures(self.manifest)
self.config = replace(Config(), hand_side="left", chunk_size=4)
write(self.path, self.data)
def prepared(self, **kwargs):
return prepare(self.path, self.manifest, self.split, self.config, synthetic_smoke=True, **kwargs)
def test_future_windows_padding_and_split_isolation(self):
_, datasets, norm, metadata = self.prepared()
training = datasets["train"]
self.assertEqual(training.names, ["demo_000000"])
obs, actions, valid = training[len(training) - 1]
self.assertEqual(valid.tolist(), [True, False, False, False])
n = norm
actual = actions[0].numpy() * n["action_std"] + n["action_mean"]
np.testing.assert_allclose(actual, training.series["demo_000000"][1][-1], atol=1e-6)
self.assertEqual(metadata["episodes"]["demo_000000"]["control_frames"], 61)
self.assertTrue(torch.isfinite(obs).all())
def test_normalization_cannot_see_heldout_values(self):
_, datasets, norm, _ = self.prepared()
series = copy.deepcopy(datasets["train"].series)
for name in self.split["validation"] + self.split["test"]:
series[name] = (series[name][0] * 10000, series[name][1] * 10000)
new = fit_normalization(series, self.split["train"])
for k in norm:
np.testing.assert_array_equal(norm[k], new[k])
def test_cross_partition_episode_group_and_duplicate_geometry_rejected(self):
for kind in ("episode", "group", "geometry", "unassigned"):
split, data = copy.deepcopy(self.split), copy.deepcopy(self.data)
if kind == "episode":
split["test"] = split["train"]
elif kind == "group":
split["episode_groups"]["demo_000001"] = "analytic_0"
elif kind == "geometry":
data.episodes["demo_000001"] = copy.deepcopy(data.episodes["demo_000000"])
else:
split["test"] = []
with self.subTest(kind=kind), self.assertRaises(ContractError):
validate_splits(data, split)
def test_review_requires_hash_owner_and_all_flags(self):
review = {
"schema": "l20_training_review_v1",
"hdf5_sha256": digest(self.path),
"asset_sha256": "a" * 64,
"coordinate_and_scale_reviewed": True,
"reference_state_targets_accepted": True,
"capture_groups_reviewed": True,
"reviewer": "analytic test fixture only",
"evidence": "test",
}
validate_review(review, digest(self.path), self.manifest)
for key in (
"hdf5_sha256",
"coordinate_and_scale_reviewed",
"reference_state_targets_accepted",
"capture_groups_reviewed",
"reviewer",
):
bad = dict(review)
bad[key] = False
with self.subTest(key=key), self.assertRaises(ContractError):
validate_review(bad, digest(self.path), self.manifest)
def test_real_training_cannot_accept_synthetic_or_unreviewed_expert_flags(self):
with self.assertRaisesRegex(ContractError, "expert_retargeted"):
prepare(self.path, self.manifest, self.split, self.config, {})
# Deliberately forged expert flag in an analytic contract test, never real training data.
self.data.metadata["provenance"] = "expert_retargeted"
expert = self.path.with_name("expert-flag-contract-fixture.hdf5")
write(expert, self.data)
with self.assertRaisesRegex(ContractError, "review"):
prepare(expert, self.manifest, self.split, self.config, {})
with self.assertRaisesRegex(ContractError, "relabel"):
prepare(expert, self.manifest, self.split, self.config, synthetic_smoke=True)
def test_speed_invalid_gap_off_grid_and_frame_budget_fail_closed(self):
for kind in ("speed", "gap", "offgrid", "budget"):
data = copy.deepcopy(self.data)
config = self.config
e = data.episodes["demo_000000"]
if kind == "speed":
e.wrist_position[1, 0] = 0.09
elif kind == "gap":
e.valid[1] = False
elif kind == "offgrid":
e.time[-1] += 0.001
else:
config = replace(config, max_total_frames=10)
path = self.path.with_name(kind + ".hdf5")
write(path, data)
with self.subTest(kind=kind), self.assertRaises(ContractError):
prepare(path, self.manifest, self.split, config, synthetic_smoke=True)
def test_prepare_does_not_modify_file_or_manifest(self):
before, manifest = self.path.read_bytes(), copy.deepcopy(self.manifest)
self.prepared()
self.assertEqual(before, self.path.read_bytes())
self.assertEqual(manifest, self.manifest)
def test_config_rejects_invalid_dimensions_and_resource_values(self):
for fields in (
{"control_hz": 29},
{"hidden_dim": 31},
{"execute_steps": 17},
{"max_updates": True},
{"max_seconds": float("nan")},
{"max_total_frames": 2000001},
):
with self.subTest(fields=fields), self.assertRaises(ContractError):
Config(**fields)
class ModelTests(unittest.TestCase):
def test_real_cvae_gradient_padding_and_deterministic_zero_latent_inference(self):
torch.manual_seed(42)
torch.set_num_threads(1)
config = replace(Config(), hidden_dim=32, layers=1, latent_dim=4, chunk_size=4, execute_steps=2)
model = ReferenceACT(6, 3, config)
obs, target = torch.randn(2, 6), torch.randn(2, 4, 3)
valid = torch.tensor([[True, True, False, False], [True, False, False, False]])
pred, mu, logvar = model(obs, target, valid)
changed = target.clone()
changed[~valid] = 10000
_, mu2, lv2 = model(obs, changed, valid)
torch.testing.assert_close(mu, mu2, rtol=0, atol=1e-6)
torch.testing.assert_close(logvar, lv2, rtol=0, atol=1e-6)
loss, _, _ = objective(pred, target, valid, mu, logvar, config.kl_weight)
loss.backward()
self.assertTrue(all(p.grad is None or torch.isfinite(p.grad).all() for p in model.parameters()))
model.eval()
with torch.inference_mode():
a = model(obs)[0]
b = model(obs)[0]
torch.testing.assert_close(a, b, rtol=0, atol=0)
class TrainingAndPolicyTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.directory = tempfile.TemporaryDirectory()
cls.path = Path(cls.directory.name) / "smoke"
cls.manifest = manifest_fixture()
cls.result = smoke(cls.manifest, cls.path)
cls.checkpoint = cls.path / "train/last.pt"
@classmethod
def tearDownClass(cls):
cls.directory.cleanup()
def test_four_update_save_reload_and_heldout_evaluation(self):
self.assertEqual(self.result["training"]["completed_updates"], 4)
self.assertEqual(self.result["training"]["checkpoint_reload_prediction_max_abs_diff"], 0)
self.assertEqual(self.result["evaluation"]["split"], "test")
self.assertEqual(self.result["simulation_e2e"], "NOT_RUN")
self.assertTrue(self.checkpoint.stat().st_size > 0)
model, config, adapter, norm, _ = load_checkpoint(self.checkpoint, self.manifest)
values = predict(model, norm, norm["observation_mean"])
self.assertEqual(values.shape, (config.chunk_size, adapter.action_dim))
with self.assertRaises(ContractError):
evaluate(self.checkpoint, self.path / "synthetic.hdf5", self.manifest, "train")
def test_checkpoint_side_identity_and_bad_normalization_rejected(self):
manifest = copy.deepcopy(self.manifest)
manifest["asset_sha256"] = "b" * 64
with self.assertRaisesRegex(ContractError, "mismatch"):
load_checkpoint(self.checkpoint, manifest)
payload = torch.load(self.checkpoint, weights_only=True)
payload["normalization"]["action_std"][0] = 0
path = self.path / "invalid-norm.pt"
torch.save(payload, path)
with self.assertRaisesRegex(ContractError, "positive"):
load_checkpoint(path, self.manifest)
def test_policy_reset_clock_rate_limits_and_no_future_teacher_forcing(self):
hdf5 = self.path / "synthetic.hdf5"
ep = load(hdf5, self.manifest).episodes["demo_000001"]
limits = replace(Limits(), workspace_radius=0.8)
policy = ClosedLoopReferenceACT(self.checkpoint, self.manifest, hdf5, "demo_000001", ep, limits)
pose = np.r_[ep.wrist_position[0], ep.wrist_quaternion[0]]
traces = []
for repetition in range(2):
policy.reset()
trace = []
for step in range(128):
# Analytic observations only: not a simulated physics trace.
p, q, j = policy.target(step, pose, ep.joint_position[0])
self.assertAlmostEqual(j[0], 2 * j[1])
self.assertTrue((j >= policy.adapter.lower - 1e-8).all())
self.assertTrue((j <= policy.adapter.upper + 1e-8).all())
trace.append(np.r_[p, q, j])
traces.append(trace)
np.testing.assert_array_equal(traces[0], traces[1])
self.assertEqual(policy.summary()["inference_calls"], 8)
self.assertEqual(policy.summary()["physics_steps"], 128)
with self.assertRaisesRegex(ContractError, "clock"):
policy.target(130, pose, ep.joint_position[0])
with self.assertRaisesRegex(ContractError, "held-out"):
ClosedLoopReferenceACT(self.checkpoint, self.manifest, hdf5, "demo_000000", ep, limits)
with self.assertRaisesRegex(ContractError, "limits"):
ClosedLoopReferenceACT(self.checkpoint, self.manifest, hdf5, "demo_000001", ep, Limits())
def test_run_directory_is_never_overwritten_and_budget_failure_is_not_pass(self):
with self.assertRaises(FileExistsError):
smoke(self.manifest, self.path)
config = replace(
Config(),
hand_side="left",
hidden_dim=32,
layers=1,
latent_dim=4,
chunk_size=4,
execute_steps=2,
batch_size=2,
max_updates=1,
max_seconds=1e-12,
)
split = json.loads((self.path / "splits.json").read_text())
adapter, datasets, norm, metadata = prepare(
self.path / "synthetic.hdf5", self.manifest, split, config, synthetic_smoke=True
)
output = self.path / "deadline-failure"
with self.assertRaisesRegex(ContractError, "budget"):
train(adapter, datasets, norm, metadata, config, output)
self.assertFalse((output / "result.json").exists())
self.assertEqual(json.loads((output / "failure.json").read_text())["status"], "FAIL")
def test_cli_heldout_eval_and_missing_inputs_fail_without_kit(self):
manifest = self.path / "manifest.json"
manifest.write_text(json.dumps(self.manifest))
output = self.path / "cli-evaluation.json"
base = [sys.executable, "-m", "dex_workbench_imitation.cli"]
result = subprocess.run(
base
+ [
"evaluate",
"--checkpoint",
str(self.checkpoint),
"--hdf5",
str(self.path / "synthetic.hdf5"),
"--manifest",
str(manifest),
"--output",
str(output),
],
capture_output=True,
text=True,
)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(json.loads(output.read_text())["simulation_e2e"], "NOT_RUN")
result = subprocess.run(
base
+ [
"preflight",
"--hdf5",
"missing.hdf5",
"--manifest",
str(manifest),
"--config",
str(ROOT / "configs/imitation/l20_right_act.json"),
"--splits",
str(self.path / "splits.json"),
"--data-review",
str(ROOT / "configs/imitation/data_review.example.json"),
"--output",
str(self.path / "missing.json"),
],
capture_output=True,
text=True,
)
self.assertNotEqual(result.returncode, 0)
self.assertFalse((self.path / "missing.json").exists())
class EntryTests(unittest.TestCase):
def test_optional_policy_argument_preserves_default_and_requires_full_hdf5_before_kit(self):
script = ROOT / "scripts/tracking/track_l20.py"
spec = importlib.util.spec_from_file_location("act_entry_test", script)
entry = importlib.util.module_from_spec(spec)
spec.loader.exec_module(entry)
parser = entry.build_parser(lambda parser: None)
args = parser.parse_args(["asset.usda", "--manifest", "manifest.json"])
self.assertIsNone(args.policy_checkpoint)
args = parser.parse_args(["asset.usda", "--manifest", "manifest.json", "--policy-checkpoint", "checkpoint.pt"])
self.assertEqual(args.policy_checkpoint, Path("checkpoint.pt"))
# AST/source invariant: original runtime assertion thresholds have not been disabled.
source = script.read_text()
for assertion in (
"residual < 0.002",
"position_error < 0.05 and angle_error < 0.5 and joint_error < 0.2",
"np.testing.assert_allclose(reset_states[0], reset_states[1], atol=1e-6, rtol=0)",
):
self.assertIn(assertion, source)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,40 @@
"""CPU regression for camera axes and explicit slow-replay synchronization."""
import tempfile
import unittest
from pathlib import Path
import numpy as np
from dex_workbench_tracking.source_camera import load_camera, usd_pose
class SourceCameraTests(unittest.TestCase):
def test_axes_and_projection(self):
cv = np.eye(4)
cv[:3, 3] = [1, 2, 3]
usd = usd_pose(cv)
np.testing.assert_array_equal(usd[:3, 3], cv[:3, 3])
point = cv @ np.array([.1, .2, 1, 1])
np.testing.assert_allclose(np.linalg.inv(usd) @ point, [.1, -.2, -1, 1])
def test_time_binding_and_invalid_rotation(self):
with tempfile.TemporaryDirectory() as folder:
path = Path(folder) / 'camera.npz'
poses = np.tile(np.eye(4), (2, 1, 1))
k = np.array([[1000, 0, 640], [0, 1000, 360], [0, 0, 1]])
np.savez(path, time=[0, 1], final_world_from_camera=poses,
camera_from_final_world=poses, K=k)
time, _, _ = load_camera(path, np.array([0, 3]), 3)
np.testing.assert_array_equal(time, [0, 3])
with self.assertRaises(AssertionError):
load_camera(path, np.array([0, 3]), 1)
with self.assertRaises(ValueError):
load_camera(path, np.array([0, 3]), float('nan'))
poses[:, 0, 0] = -1
np.savez(path, time=[0, 1], final_world_from_camera=poses,
camera_from_final_world=poses, K=k)
with self.assertRaises(AssertionError):
load_camera(path, np.array([0, 3]), 3)
if __name__ == '__main__':
unittest.main()
@@ -0,0 +1,212 @@
"""CPU analytic tests: offline reference metrics never authorize physical replay."""
import copy
import json
import subprocess
import sys
import tempfile
import unittest
from dataclasses import replace
from pathlib import Path
import numpy as np
from dex_workbench_tracking.cli import stretch_time, write
from dex_workbench_tracking.control import Limits
from dex_workbench_tracking.diagnostic import diagnose
from dex_workbench_tracking.identity import RIGHT_URDF_SHA
from dex_workbench_tracking.trajectory import ContractError, Demonstrations, Episode, load
class DiagnosticTests(unittest.TestCase):
def setUp(self):
self.manifest = {
"manifest_version": "l20_asset_manifest_v1",
"hand_side": "right",
"asset_sha256": "a" * 64,
"root_link": "hand_base_link",
"joints": [
{"name": "follower", "lower_rad": 0, "upper_rad": 2},
{"name": "leader", "lower_rad": 0, "upper_rad": 1},
],
"source_urdf": {
# Identity token required by right-side schema; geometry below is synthetic.
"sha256": RIGHT_URDF_SHA,
"mimic": [
{"joint": "follower", "reference": "leader", "multiplier": 2, "offset_rad": 0},
],
},
}
time = np.array([0, 0.25, 1.0], dtype=np.float64)
position = np.zeros((3, 3), dtype=np.float32)
position[:, 0] = 0.2 * time
quaternion = np.zeros((3, 4), dtype=np.float32)
quaternion[:, 0], quaternion[:, 3] = np.cos(time / 2), np.sin(time / 2)
joints = np.array([time * 0.8, time * 0.4], dtype=np.float32).T
self.data = Demonstrations(
{
"schema_version": "l20_tracking_v1",
"embodiment": "L20",
"hand_side": "right",
"asset_sha256": "a" * 64,
"root_link": "hand_base_link",
"provenance": "synthetic",
"metric_scale_provenance": "Analytic test, not measured",
"scale_to_meters": 1.0,
"source_description": "Analytic nonuniform samples",
},
("follower", "leader"),
np.eye(4),
{"demo_000000": Episode(time, position, quaternion, joints, np.ones(3, dtype=bool))},
)
def report(self, **kwargs):
return diagnose(self.data, self.manifest, "demo_000000", **kwargs)
def test_nonuniform_si_derivatives_and_passive_joint_roles(self):
report = self.report()
self.assertAlmostEqual(report["speed_estimates"]["translation_m_s"]["max"], 0.2, places=6)
self.assertAlmostEqual(report["speed_estimates"]["rotation_rad_s"]["max"], 1.0, places=6)
self.assertAlmostEqual(report["speed_estimates"]["all_joint_max_rad_s"]["max"], 0.8, places=6)
self.assertAlmostEqual(report["speed_only_slowdown_lower_bound"], 4.0, places=6)
self.assertEqual(report["joints"]["follower"]["role"], "model_follower")
self.assertEqual(report["mimic"]["follower"]["max_abs_residual_rad"], 0)
self.assertEqual(report["sample_dt_s"]["min"], 0.25)
def test_workspace_cannot_be_fixed_by_slowdown(self):
report = self.report(factors=(1, 80))
self.assertEqual(report["workspace"]["outside_frames"], [2])
self.assertFalse(report["workspace"]["fixable_by_time_stretch"])
self.assertEqual(report["candidates"][1]["reference_gate"], "FAIL")
self.assertIn("workspace", report["candidates"][1]["first_gate_failure"])
extended = self.report(limits=replace(Limits(), workspace_radius=0.8), factors=(80,))
self.assertEqual(extended["candidates"][0]["reference_gate"], "PASS")
self.assertEqual(extended["dynamic_replay"], "NOT_RUN")
self.assertEqual(extended["candidates"][0]["dynamics"], "NOT_RUN")
def test_slowdown_scaling_and_no_input_mutation(self):
before = copy.deepcopy(self.data)
report = self.report()
slow = diagnose(stretch_time(self.data, 10), self.manifest, "demo_000000")
for metric in report["speed_estimates"]:
self.assertAlmostEqual(
slow["speed_estimates"][metric]["max"] * 10, report["speed_estimates"][metric]["max"]
)
self.assertEqual(self.data.metadata, before.metadata)
for field in vars(self.data.episodes["demo_000000"]):
np.testing.assert_array_equal(
getattr(self.data.episodes["demo_000000"], field), getattr(before.episodes["demo_000000"], field)
)
def test_peak_interval_locations_and_limit_margin(self):
episode = self.data.episodes["demo_000000"]
episode.wrist_position[1, 0] = 0.15
report = self.report()
metric = report["speed_estimates"]["translation_m_s"]
self.assertEqual(metric["peak_interval_frames"], [0, 1])
self.assertEqual(metric["peak_interval_time_s"], [0, 0.25])
self.assertEqual(metric["over_limit_interval_starts"], [0, 1])
self.assertEqual(report["joints"]["leader"]["minimum_limit_margin_rad"], 0)
def test_identity_side_order_limits_mimic_fail_closed(self):
for mutation in ("hash", "side", "order", "limits", "mimic"):
manifest = copy.deepcopy(self.manifest)
if mutation == "hash":
manifest["asset_sha256"] = "b" * 64
elif mutation == "side":
manifest["hand_side"] = "left"
manifest["source_urdf"].pop("sha256")
elif mutation == "order":
manifest["joints"].reverse()
elif mutation == "limits":
manifest["joints"][1]["upper_rad"] = 0.1
else:
manifest["source_urdf"]["mimic"][0]["multiplier"] = 1.5
with self.subTest(mutation=mutation), self.assertRaises(ContractError):
diagnose(self.data, manifest, "demo_000000")
def test_invalid_gap_and_invalid_factors_rejected(self):
for factor in (0, 0.5, float("inf"), float("nan")):
with self.subTest(factor=factor), self.assertRaises(ContractError):
self.report(factors=(factor,))
self.data.episodes["demo_000000"].valid[1] = False
with self.assertRaisesRegex(ContractError, "fully valid"):
self.report()
def test_two_frame_and_stationary_episode(self):
episode = self.data.episodes["demo_000000"]
episode = Episode(*(getattr(episode, field)[:2].copy() for field in vars(episode)))
episode.wrist_position[:] = 0
episode.wrist_quaternion[:] = [1, 0, 0, 0]
episode.joint_position[:] = 0
self.data.episodes["demo_000000"] = episode
report = self.report()
self.assertEqual(report["speed_only_slowdown_lower_bound"], 1)
self.assertIsNone(report["midpoint_acceleration_estimates_not_physical_bounds"])
# Stationary data can satisfy the reference gate, not the runtime motion checks.
self.assertEqual(report["candidates"][0]["reference_gate"], "PASS")
self.assertEqual(report["dynamic_replay"], "NOT_RUN")
def test_cli_no_overwrite_and_rejects_hash_without_output(self):
with tempfile.TemporaryDirectory() as temporary:
directory = Path(temporary)
source, manifest, output = (directory / name for name in ("input.hdf5", "manifest.json", "report.json"))
write(source, self.data)
manifest.write_text(json.dumps(self.manifest))
before = source.read_bytes()
command = [
sys.executable,
"-m",
"dex_workbench_tracking.diagnostic",
str(source),
"--manifest",
str(manifest),
"--output",
str(output),
]
first = subprocess.run(command, capture_output=True, text=True)
self.assertEqual(first.returncode, 0, first.stderr)
self.assertEqual(json.loads(output.read_text())["dynamic_replay"], "NOT_RUN")
report_bytes = output.read_bytes()
second = subprocess.run(command, capture_output=True, text=True)
self.assertNotEqual(second.returncode, 0)
self.assertEqual(output.read_bytes(), report_bytes)
self.assertEqual(source.read_bytes(), before)
self.manifest["asset_sha256"] = "b" * 64
manifest.write_text(json.dumps(self.manifest))
rejected = directory / "rejected.json"
command[-1] = str(rejected)
third = subprocess.run(command, capture_output=True, text=True)
self.assertNotEqual(third.returncode, 0)
self.assertIn("hash mismatch", third.stderr)
self.assertFalse(rejected.exists())
def test_loader_rejects_bad_time_and_nan_before_diagnostic(self):
with tempfile.TemporaryDirectory() as temporary:
for kind in ("duplicate_time", "nan"):
data = copy.deepcopy(self.data)
episode = data.episodes["demo_000000"]
if kind == "duplicate_time":
episode.time[1] = 0
else:
episode.wrist_position[1, 0] = np.nan
path = Path(temporary) / (kind + ".hdf5")
write(path, data)
with self.subTest(kind=kind), self.assertRaises(ContractError):
load(path, self.manifest)
def test_import_does_not_load_simulator_or_torch(self):
result = subprocess.run(
[
sys.executable,
"-c",
"import sys; import dex_workbench_tracking.diagnostic; "
"assert not any(n in sys.modules for n in ('pxr', 'torch', 'isaacsim', 'isaaclab', 'omni'))",
],
capture_output=True,
text=True,
)
self.assertEqual(result.returncode, 0, result.stderr)
if __name__ == "__main__":
unittest.main()