feat(lekiwi): release V0.10.1 初步集成 LeKiwi,优化碰撞模型
web-platform-ci / TypeScript, lint, unit, build (push) Has been cancelled
web-platform-ci / Playwright E2E (push) Has been cancelled
lekiwi-compatibility / cpu-compatibility (push) Has been cancelled

集成通用机器人数值接口、本机控制桥、LeRobot 插件和统一键盘遥操作。采用离线 CoACD 全臂碰撞配方 revision 4、局部装配区切分与结构自接触,限制直接关节位姿写入并保留安全看门狗。同步版本号、变更记录、来源许可证和兼容性验证。
This commit is contained in:
2026-09-20 14:42:30 +08:00
parent da4d59c31a
commit 3ad29356c9
103 changed files with 23055 additions and 280 deletions
+143
View File
@@ -0,0 +1,143 @@
# LeKiwi 仿真示例
目标:通过统一机器人接口连接 LeRobot 0.6.1 的上层控制逻辑。此目录不包含实体硬件驱动、相机或 RL 训练任务。
## 可重建模型输入
```bash
source .venv/bin/activate
python examples/lekiwi/prepare_assets.py --source /path/to/LeKiwi
# 或明确允许下载固定版本所需文件(不会克隆整个仓库)
python examples/lekiwi/prepare_assets.py --download
```
生成 `build/lekiwi/lekiwi-v1.zip`、原始 URDF/资源副本、SHA-256 清单及 Apache-2.0 许可证。原参考仓库保持只读;常规前端构建不下载任何资产。
适用源:SIGRobotics-UIUC/LeKiwi 的 `efa608d7ee5a495a4803b1d28cd0c955b4f1e033`;URDF SHA-256 见 `robot_profiles/lekiwi-v1.json`。未知变体拒绝套用。
## 动力学边界
- CAD 轮 STL 单个超过 300,000 个三角形,超出 MuJoCo 的 STL 上限。因此显式 profile 在中间编译前替换这些几何,之后生成简化轮毂与每轮 12 个被动滚子;不是保留高多边形轮子的视觉模型。
- 九个主执行器:5 个臂位置伺服、夹爪位置伺服、3 个轮速度伺服;另外 36 个滚子 hinge **不加电机**。仅通过物理接触驱动车体,不强写底盘位姿/速度。
- CAD +Y 朝前,在 canonical 基座中旋转 -90°。简化三轮按理想 0.125 m 轮距、0.05 m 半径布置;并非沿用 CAD 不等距的轮轴位置。
- 原始 CAD 的 17.324 kg 惯性估计不用于动力学。仿真采用 2.2 kg 基座、按 profile 给出的臂段/轮质量、简化碰撞和保守限位。惯性、摩擦、增益均为**仿真估计**,不代表实体标定。
- 当前碰撞配方为 **revision 4**:从整臂 18 个原始视觉网格(含固定舵机/附件)离线 CoACD 分解成 538 个凸包,再按轴承配合区切分成 1,220 个独立碰撞凸包,替代旧臂胶囊和手工分段。保持视觉坐标变换、夹爪空隙、零附加碰撞质量与显式惯性。工具隔离在 `build/venvs/collision`,浏览器仅使用生成数据。来源、许可证、缓存/参数和重建方法见 [碰撞数据说明](../../robot_profiles/NOTICE.md)。
- 不再整对排除相邻连杆:显式启用 5,581 对结构凸包接触,包括 `SO_ARM100_08k_Mirror-v1` / `SO_ARM100_08k_116_Square-v1`。跨区域凸包先切分,只对完全落入关节轴承/舵机配合区域的部分保留局部例外,以免装配重叠锁死关节;非相邻碰撞保持启用。细化为 1 ms 物理步长,未放宽 500 ms 安全阈值。凸分解仍有误差,不等于逐三角形精确碰撞或所有姿态无穿入保证。
- 自碰撞会阻挡不可能到达的目标,动作确认值不等于实测关节角;遇阻时目标/实测不一致是正常反馈。这不是自动避障或无碰撞路径规划,也没有关闭全臂自碰撞、缩小肩关节范围来隐藏问题。
- CAD 夹爪轴已反向,使开度增大确实张开;内部关节范围为 -0.18…0.9 rad,LeRobot 的 0–100 开度接口不变。当前配方闭合端测试间隙约 1.47 mm,另有指面间接触保护;不会通过关闭自碰撞来允许两指交叉。
- 平地低速控制、静态障碍阻挡和夹爪指尖/物体接触经过真实浏览器 WASM 验证;不保证崎岖地形、可靠抓取、相机、训练或 sim-to-real 策略迁移。MuJoCo 使用软接触,受力时允许小量穿入,瞬态大小取决于速度/载荷;这与缺少碰撞体导致整体穿过不同,高速/薄物体也不属于已验证范围。
### 更新碰撞体后必须重新转换模型
1. 退出 Python 控制脚本,停止外控;刷新前端页面加载新代码。
2. **重新导入 `build/lekiwi/lekiwi-v1.zip` 中的原始 URDF**,显式选择 **LeKiwi v1(仿真专用)**,重新“转换并加载”。源 ZIP 没变,不必重新下载;不要继续使用旧的已转换 MJCF/缓存模型。旧 MJCF 会提示从原始 URDF 重新转换。
3. 在检查器勾选“显示碰撞几何”,可查看橙色凸包,确认覆盖两侧指尖、臂座、`Square`、`Mirror`、其他臂段和固定附件;此开关仅影响显示,不开关物理碰撞。
4. 重新连接桥接、播放并允许外部控制,再启动键盘脚本。`v` 张开、`b` 闭合;可以先用 `--gripper-speed 25` 低速检查。
### 网页关节滑条不是物理运动
旧关节滑条是暂停后直接写 `qpos` 的姿态编辑器,会绕过物理积分,碰撞体不能阻止这种“瞬移”。现在启用机器人 profile 时,**关节角只读、禁用关节拖动/单独重置关节、禁止直接位姿写入和忽略关节限位**;要运动请播放后调整“执行器”目标,或使用键盘/外控。普通非 profile 模型仍保留原姿态编辑功能。“重置仿真”仍可用,并保留撤销外控和 epoch 更新。外控期间继续禁止手动执行器写入。
## 完整演示
从仓库根目录执行:
```bash
source .venv/bin/activate # Python 3.12
python -m pip install -e ./control_bridge
python examples/lekiwi/setup_lerobot.py # 固定 CPU 依赖,仅写 build/venvs/lerobot/
npm ci
npm run dev
```
1. 将 `build/lekiwi/lekiwi-v1.zip` 导入工作台,在 URDF 对话框显式选择 **LeKiwi v1(仿真专用)**,点击“转换并加载”。若已导入可兼容 MJCF,在“控制台 → 开源项目 / 外部控制”选择 profile 并重新编译。
2. 另一终端用 `.venv/bin/python -m mujoco_control_bridge` 启动本机服务。复制启动时显示的 token,在工作台填入 `http://127.0.0.1:8766` 和 token,点击“连接桥接”。
3. 点击“播放”,再点击“允许外部控制”。外控锁定 1×;手动关节/执行器写入不可用,原 Python/ONNX 停止。
4. 第三个终端设置同一 token,运行真正的 LeRobot 插件示例:
```bash
read -rsp '控制 token: ' MUJOCO_CONTROL_TOKEN; echo
export MUJOCO_CONTROL_TOKEN
export MUJOCO_CONTROL_ENDPOINT=http://127.0.0.1:8766
env -u PYTHONPATH PYTHONNOUSERSITE=1 build/venvs/lerobot/bin/python \
examples/lekiwi/demo_control.py --duration 60
```
脚本保留“读状态 → 合成动作 → 发送 → 节拍等待”的上游 30 Hz 循环形状,九个通道一起运动,输出确认目标、实测反馈、时间和延迟统计。不运行硬件构造函数,不初始化串口/相机/GPU。
脚本退出会自动暂停。重新演示必须再次播放并授权;重置、模型重载、页面隐藏/退出、控制进程崩溃和超时也不会自动恢复。
键盘演示(包括从 `demo_control.py` 切换过来)必须再次完成 **播放 → 允许外部控制 → 启动脚本**。授权不会自动播放;确认按钮已变为“暂停”、外控状态为“等待 Python 控制者”,再执行:
```bash
env -u PYTHONPATH PYTHONNOUSERSITE=1 build/venvs/lerobot/bin/python \
examples/lekiwi/teleoperate_sim.py
```
### 底盘 + 机械臂键盘控制
原 LeKiwi 仓库主要提供硬件/CAD/URDF,其 README 的遥操作方案是 **WASD + leader arm**:键盘控制底盘,SO-ARM leader 的六路位置控制从臂。当前仿真已经实现五个臂关节和夹爪的执行器、单位换算与反馈;这里用键盘关节点动替代实体 leader,无需安装硬件驱动或更换桥接;但更新碰撞配方仍须按上文重新转换模型。
在运行脚本的 Linux 交互终端使用小写按键:
| 按键 | 功能 | LeRobot 动作通道 |
| -------------- | ---------------------------------------------- | ----------------------- |
| `w / s` | 底盘前进 / 后退 | `x.vel` |
| `a / d` | 底盘左移 / 右移 | `y.vel` |
| `z / x` | 底盘左转 / 右转 | `theta.vel` |
| `r / f` | 底盘速度升档 / 降档(不改变臂速度) | — |
| `u / j` | 肩部旋转角增大 / 减小 | `arm_shoulder_pan.pos` |
| `i / k` | 肩部俯仰角增大 / 减小 | `arm_shoulder_lift.pos` |
| `o / l` | 肘部角度增大 / 减小 | `arm_elbow_flex.pos` |
| `t / g` | 腕部俯仰角增大 / 减小 | `arm_wrist_flex.pos` |
| `y / h` | 腕部旋转角增大 / 减小 | `arm_wrist_roll.pos` |
| `v / b` | 夹爪张开 / 闭合 | `arm_gripper.pos` |
| 空格 | 清除按键脉冲,底盘停止、机械臂保持最后确认目标 | — |
| `q` / `Ctrl+C` | 退出、释放控制权并暂停仿真 | — |
- 臂按键是**关节空间点动**,不是末端 XYZ/逆运动学控制。五个关节使用度,夹爪使用 0–100 开度;角度正负不代表相机画面中的上下左右。
- 初始目标来自实测姿态,不会启动即归零。每轮 30 Hz 将底盘速度与臂增量合成一条动作;关节/夹爪限幅由现有 profile 执行,下一轮从**确认后的目标**累加,避免到限位后积累不可见的超限目标。
- 默认臂速度为 20 度/秒、夹爪速度为 50 百分点/秒,可用下列参数调慢(上限分别为 90 和 100,必须大于零)。控制循环卡顿不会补发大幅角度跳变。
```bash
env -u PYTHONPATH PYTHONNOUSERSITE=1 build/venvs/lerobot/bin/python \
examples/lekiwi/teleoperate_sim.py --arm-speed 5 --gripper-speed 25
```
终端输入是 **180 ms 脉冲**,不是系统级按下/松开监听;点按为小步调整,长按依赖系统键盘重复,组合键只是重叠脉冲。脉冲过期后底盘速度归零、机械臂保持最后确认目标,脚本仍持续发送动作。空格不会退出授权,也不会将机械臂归零;真正停止仿真请按 `q` 或在页面停止外控。请并排显示浏览器和终端,保持仿真页面可见;隐藏/最小化页面会撤销授权。
底盘和机械臂必须由**同一个控制循环**合成动作,不能分别启动两个 Python 控制者;外控期间也不能同时用网页执行器滑条写入。碰到其他臂段/安装板后不会继续到达目标;完整 CAD 下可用运动范围可能比声明的关节上下限窄,遇阻应反向点动,而不是扩大限位或禁用碰撞。
### 后续接实体 leader 的接入点
若要复用上游 SO100/SO101 leader,保留 `lekiwi_sim` 作为被控机器人,只替换机械臂输入源:在 leader 已连接、完成硬件校准且 `use_degrees=True` 的前提下,将 `leader.get_action()` 的 `shoulder_pan.pos` 等六路键名加 `arm_` 前缀,再与底盘动作合并后调用同一个 `robot.send_action()`。不要启动实体 LeKiwi follower/ZMQ 服务,也不要把归一化的 -100…100 臂角当成度。
这只是接口接入说明,当前终端脚本未实现串口 leader 模式,也未做硬件验证;接入前还需核对真实 leader 与仿真 profile 的零位/方向、限位和每步速度限制。当前模型不保证可靠抓取或 sim-to-real 一致性。
## 连接时提示“机器人观测超过 500ms 未更新”
这表示本机桥接可访问,但未收到浏览器端新的机器人观测;不是 LeRobot 安装或 Python 导入错误。桥接连接成功不等于仿真正在运行。
1. 将浏览器和终端并排显示,保持仿真页面可见,不要切换到其他浏览器标签、最小化或完全遮住仿真窗口。页面隐藏会撤销授权;若显示“未连接”,先重新连接桥接。
2. 查看“机器人观测状态”。若显示“已暂停(非实时观测)”或播放按钮仍为“播放”,先点击 **播放**,再点击 **允许外部控制**。前一个脚本退出、停止外控或超时后都会暂停,不能只重新运行 Python。
3. 确认仿真时间持续增加、观测年龄低于 500 ms,状态显示“等待 Python 控制者”,然后在终端重跑脚本。等待 Python 连接本身没有 500 ms 倒计时;这个阈值检查的是观测新鲜度,取得租约后还会检查动作是否持续更新。
4. 若正在播放且页面可见时观测仍持续过期,检查页面错误和主线程卡顿;必要时重新加载模型并连接、播放、授权。不要通过增大 SDK HTTP 超时来绕过:HTTP 超时与 500 ms 观测/动作安全看门狗是两回事。
旧版桥接会将暂停超过 500 ms 也报成“观测过期”;修复后会明确提示“仿真已暂停;请先在浏览器点击‘播放’,再点击‘允许外部控制’”。更新代码后需重启 `python -m mujoco_control_bridge`,并在页面重新连接、播放、授权;若服务生成了新 token,请同时更新浏览器和终端中的 token。
## 回归和证据
```bash
source .venv/bin/activate
npm run test:control-bridge
build/venvs/lerobot/bin/python -m unittest discover -s integrations/lerobot/tests -v
LEROBOT_PYTHON="$PWD/build/venvs/lerobot/bin/python" npm run test:e2e:lekiwi
# 仅物理测试(不需 LeRobot)
npx playwright test -c web_platform/playwright.lekiwi.config.ts lekiwi.physics.spec.ts lekiwi.gripper.spec.ts lekiwi.armCollision.spec.ts lekiwi.fullCollision.spec.ts
```
专用套件不是 mock:整臂 18 个视觉网格的 108 次独立 CAD 表面接触探针,六关节各两个有界目标扫掠,`Mirror/Square` 主体接触阻挡/脱离及禁止直接 `qpos` 写入;真实 WASM 3.11.0 的 30 仿真秒站稳、正负三轴运动、臂/夹爪、墙体阻挡、两侧 CAD 指尖探针接触、开度与实际指间距同向、张开空隙不误碰撞、闭合被物体阻挡以及指面自碰撞、上臂下压时与自身臂座组件的真实接触/穿入量及反向脱离(完整模型中安装板或肩部结构夹片先于底座本体接触);真实 Python/LeRobot、PTY 终端底盘 + 六路臂/夹爪正反点动、无输入保持与退出撤权、完整工作台文件导入→编译→授权→60 秒控制;另含旧控制回调、reset、新 epoch、崩溃、重载/回滚、隐藏和 JS 冻结恢复。
`build/e2e/lekiwi/` 保存 XML、物理轨迹、60 秒 JSON(RTT、新鲜度、FPS/步进预算、JS/WASM 堆容量)和截图。普通 E2E 不下载资产或安装 LeRobot;专用 CI 缺依赖即失败,不静默跳过。
软件 WebGL 比物理计算慢:外控物理/传输独立调度;检测到 SwiftShader 等软件渲染时关闭阴影、3D 显示上限 5 FPS,观测仍约 30 Hz。不能把此模式当作锁步 RL 或硬实时系统。性能实测、兼容矩阵和限制见 [机器人接口](../../docs/robot-interface.md)。
+98
View File
@@ -0,0 +1,98 @@
"""Hardware-free LeRobot loop: observe -> compose action -> send, at 30 Hz.
Only the robot implementation and input source differ from the upstream example.
This is real-time control, NOT lockstep RL or dataset recording.
"""
import argparse
import json
import math
import os
import time
from importlib.metadata import version
from lerobot.robots.config import RobotConfig
from lerobot.robots.utils import make_robot_from_config
from lerobot.utils.import_utils import register_third_party_plugins
def make_robot(endpoint):
register_third_party_plugins()
config_type = RobotConfig.get_choice_class("lekiwi_sim")
return make_robot_from_config(config_type(endpoint=endpoint, id="lekiwi-sim-demo"))
def run_demo(endpoint, duration=10.0):
if not math.isfinite(duration) or not 1 <= duration <= 3600:
raise ValueError("演示时长必须为1–3600秒")
robot = make_robot(endpoint)
robot.connect()
try:
start = time.monotonic()
first = robot.get_sim_observation()
latencies, gaps, max_motion = [], [], 0.0
steps = math.ceil(duration * 30)
previous_sim_time = first["simTime"]
for i in range(steps):
tick = time.monotonic()
observation = robot.get_observation()
t = i / 30
action = {
"arm_shoulder_pan.pos": 10 * math.sin(t),
"arm_shoulder_lift.pos": 6 * math.sin(t + 0.2),
"arm_elbow_flex.pos": 8 * math.sin(t + 0.4),
"arm_wrist_flex.pos": 6 * math.sin(t + 0.6),
"arm_wrist_roll.pos": 12 * math.sin(t + 0.8),
"arm_gripper.pos": 50 + 25 * math.sin(t),
"x.vel": 0.05 * math.cos(t),
"y.vel": 0.05 * math.sin(t),
"theta.vel": 10 * math.sin(t / 2),
}
accepted = robot.send_action(action)
platform = robot.get_sim_observation()
latencies.append((time.monotonic() - tick) * 1000)
gaps.append(platform["simTime"] - previous_sim_time)
previous_sim_time = platform["simTime"]
max_motion = max(
max_motion,
math.hypot(
platform["values"]["base.x"] - first["values"]["base.x"],
platform["values"]["base.y"] - first["values"]["base.y"],
),
)
time.sleep(max(0, start + (i + 1) / 30 - time.monotonic()))
ordered = sorted(latencies)
return {
"lerobot": version("lerobot"),
"torch": version("torch"),
"steps": steps,
"elapsedSeconds": time.monotonic() - start,
"simSeconds": platform["simTime"] - first["simTime"],
"rttMeanMs": sum(latencies) / len(latencies),
"rttP95Ms": ordered[int(0.95 * (len(ordered) - 1))],
"rttMaxMs": max(latencies),
"maxObservationSimGap": max(gaps),
"maxTranslationM": max_motion,
"observation": observation,
"accepted": accepted,
"sequence": platform["sequence"],
"appliedActionSeq": platform["appliedActionSeq"],
"timeouts": 0,
"droppedRequests": 0,
}
finally:
robot.disconnect()
def main():
parser = argparse.ArgumentParser(description="LeKiwi 仿真:先播放并在浏览器显式授权")
parser.add_argument(
"--endpoint", default=os.environ.get("MUJOCO_CONTROL_ENDPOINT", "http://127.0.0.1:8766")
)
parser.add_argument("--duration", type=float, default=10)
args = parser.parse_args()
print(json.dumps(run_demo(args.endpoint, args.duration), ensure_ascii=False))
if __name__ == "__main__":
main()
@@ -0,0 +1,51 @@
"""Offline CAD collisions for the fixed arm pedestal and shoulder-lift link.
Use the same source-bound triangle clipping as the jaw recipe. Separate convex
slices avoid spanning the pedestal's stepped silhouette or the link's full length
with one broad hull. These are conservative parts, not triangle-mesh collisions.
"""
import argparse
import json
from pathlib import Path
from generate_gripper_collisions import ROOT, generate
# Original STL coordinates, millimeters; seams overlap by 1 mm.
PARTS = [
("base_foot", "Base_08q-v1", [(1, 1, 16.0)]),
("base_column", "Base_08q-v1", [(1, -1, -15.0), (1, 1, 59.0)]),
("base_crown", "Base_08q-v1", [(1, -1, -58.0)]),
("upper_arm_end", "SO_ARM100_08k_116_Square-v1", [(0, 1, -28.0)]),
("upper_arm_beam", "SO_ARM100_08k_116_Square-v1", [(0, -1, 29.0), (0, 1, 50.0)]),
("upper_arm_hinge", "SO_ARM100_08k_116_Square-v1", [(0, -1, -49.0)]),
]
SOURCE_HASHES = {
"Base_08q-v1.stl": "a05be37db52657615ac69423fe577979efa0cd0634a5e9546bbc85537f003172",
"SO_ARM100_08k_116_Square-v1.stl": (
"64fdc308759ff58756e3c39a0ea8018d6b37b551bc3d29238b6e86bc5acae666"
),
}
def main():
parser = argparse.ArgumentParser(description="从固定版本 STL 重建 LeKiwi 底座/上臂碰撞体")
parser.add_argument("--source", type=Path, default=ROOT / "build/lekiwi/URDF/meshes")
parser.add_argument(
"--output", type=Path, default=ROOT / "robot_profiles/lekiwi-arm-collision.json"
)
parser.add_argument("--check", action="store_true", help="只验证已签入的碰撞数据可重建")
args = parser.parse_args()
result = generate(args.source, definitions=PARTS, source_hashes=SOURCE_HASHES, revision=3)
if args.check:
if json.loads(args.output.read_text()) != result:
raise ValueError("底座/上臂碰撞数据与固定 CAD/生成器不一致")
else:
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(result, indent=2) + "\n")
status = "验证通过" if args.check else args.output
print(f"LeKiwi 底座/上臂碰撞体:{len(result['parts'])} 个分段凸包,{status}")
if __name__ == "__main__":
main()
+494
View File
@@ -0,0 +1,494 @@
"""Offline collision-from-visuals cooking for the complete LeKiwi arm subtree.
Run in build/venvs/collision, NOT the training or LeRobot environment. CoACD is
never imported by the browser/bridge. Each convex part is a separate MJCF mesh.
"""
import argparse
import hashlib
import importlib.metadata
import json
import os
import xml.etree.ElementTree as ET
from pathlib import Path
# Fix the numerical execution environment before importing numpy/CoACD.
os.environ["OMP_NUM_THREADS"] = "4"
os.environ["OPENBLAS_NUM_THREADS"] = "1"
ROOT = Path(__file__).resolve().parents[2]
CONFIG = ROOT / "robot_profiles/lekiwi-collision-source.json"
VERSIONS = {"coacd": "1.0.14", "trimesh": "4.12.2", "numpy": "2.1.3", "scipy": "1.17.0"}
PARAMETERS = {
"threshold": 0.003,
"real_metric": True,
"preprocess_mode": "auto",
"preprocess_resolution": 60,
"resolution": 1000,
"mcts_nodes": 12,
"mcts_iterations": 60,
"mcts_max_depth": 3,
"decimate": True,
"max_ch_vertex": 96,
"seed": 42,
}
def digest(data):
return hashlib.sha256(data).hexdigest()
def visuals(source, config):
"""Select every visual in the subtree, including welded motors/accessories."""
urdf = (source / "LeKiwi.urdf").read_bytes()
if digest(urdf) != config["urdfSha256"]:
raise ValueError("URDF 与固定版本不符")
robot = ET.fromstring(urdf)
parents = {
j.find("child").get("link"): j.find("parent").get("link") for j in robot.findall("joint")
}
result = []
for link in robot.findall("link"):
ancestor = link.get("name")
visited = set()
while ancestor in parents and ancestor != config["rootLink"]:
if ancestor in visited:
raise ValueError("URDF 运动链存在环")
visited.add(ancestor)
ancestor = parents[ancestor]
if ancestor != config["rootLink"]:
continue
for visual in link.findall("visual"):
mesh = visual.find("geometry/mesh")
if mesh is None:
raise ValueError("当前固定资产只支持 STL 视觉网格,不能静默遗漏几何")
filename = mesh.get("filename")
path = (source / filename).resolve()
if not path.is_relative_to(source.resolve()) or filename not in config["meshes"]:
raise ValueError(f"未批准的源网格:{filename}")
sha = digest(path.read_bytes())
if sha != config["meshes"][filename]:
raise ValueError(f"源网格 SHA-256 不匹配:{filename}")
scale = [float(x) for x in mesh.get("scale", "1 1 1").split()]
if len(scale) != 3 or not all(0 < x < float("inf") for x in scale):
raise ValueError("无效的视觉网格缩放")
result.append(
{
"link": link.get("name"),
"visual": visual.get("name"),
"mesh": filename,
"sha256": sha,
"scale": scale,
}
)
if not result or {v["mesh"] for v in result} != set(config["meshes"]):
raise ValueError("整臂视觉覆盖清单不完整")
return result
def swept_bounds(vertices, axis):
"""Conservative axial/radial intervals for a full revolution, not pose sampling."""
import numpy as np
from scipy.spatial import ConvexHull
axial = vertices @ axis
reference = np.eye(3)[int(np.argmin(np.abs(axis)))]
x = np.cross(axis, reference)
x /= np.linalg.norm(x)
y = np.cross(axis, x)
projected = vertices @ np.column_stack([x, y])
hull = ConvexHull(projected)
radial_min = 0.0
if hull.equations[:, 2].max() > 0:
a = projected[hull.vertices]
b = np.roll(a, -1, axis=0)
edge = b - a
t = np.clip(-(a * edge).sum(axis=1) / (edge * edge).sum(axis=1), 0, 1)
radial_min = float(np.linalg.norm(a + t[:, None] * edge, axis=1).min())
return (
float(axial.min()),
float(axial.max()),
radial_min,
float(np.linalg.norm(projected, axis=1).max()),
)
def intervals_overlap(a, b, margin=0.0001):
# 0.1 mm conservative allowance for serialized/compiled frame precision.
return (
max(a[0], b[0]) <= min(a[1], b[1]) + margin and max(a[2], b[2]) <= min(a[3], b[3]) + margin
)
def assembly_frames(source, config, coverage):
"""URDF joint/visual frames for both geometric partitioning and pair policy."""
import numpy as np
from scipy.spatial.transform import Rotation
robot = ET.parse(source / "LeKiwi.urdf").getroot()
joints = {j.get("name"): j for j in robot.findall("joint")}
parents = {j.find("child").get("link"): j for j in joints.values()}
links = {link.get("name"): link for link in robot.findall("link")}
poses = {}
def origin(node):
result = np.eye(4)
if node is not None:
result[:3, 3] = [float(x) for x in node.get("xyz", "0 0 0").split()]
result[:3, :3] = Rotation.from_euler(
"xyz", [float(x) for x in node.get("rpy", "0 0 0").split()]
).as_matrix()
return result
def pose(link):
if link not in poses:
j = parents.get(link)
poses[link] = (
np.eye(4)
if j is None
else pose(j.find("parent").get("link")) @ origin(j.find("origin"))
)
return poses[link]
def owner(link):
if link == config["rootLink"]:
return "base"
j = parents[link]
return owner(j.find("parent").get("link")) if j.get("type") == "fixed" else j.get("name")
by_visual = {}
for item in coverage:
item["weldJoint"] = owner(item["link"])
visual = next(
v for v in links[item["link"]].findall("visual") if v.get("name") == item["visual"]
)
by_visual[item["visual"]] = (
item["weldJoint"],
pose(item["link"]) @ origin(visual.find("origin")),
)
result = []
for name, core in config["assemblyCores"].items():
if not (0 < core["radiusM"] <= 0.025 and 0 < core["axialHalfExtentM"] <= 0.06):
raise ValueError("装配配合区域超出已审核尺寸预算")
j = joints[name]
parent = owner(j.find("parent").get("link"))
transform = pose(j.find("child").get("link"))
axis = transform[:3, :3] @ np.array([float(x) for x in j.find("axis").get("xyz").split()])
axis /= np.linalg.norm(axis)
result.append((name, parent, transform, axis, core))
return by_visual, result
def joint_policy(source, config, parts, coverage):
"""Bound exceptions by joint-local geometry, never by a whole body pair."""
import numpy as np
by_visual, frames = assembly_frames(source, config, coverage)
result = []
for name, parent, transform, axis, core in frames:
members = []
groups = {parent: {}, name: {}}
for part in parts:
group, visual_pose = by_visual[part["visual"]]
if group not in (parent, name):
continue
vertices = np.fromstring(part["vertices"], sep=" ").reshape(-1, 3)
offsets = vertices @ visual_pose[:3, :3].T + visual_pose[:3, 3] - transform[:3, 3]
bounds = swept_bounds(offsets, axis)
groups[group][part["name"]] = bounds
if (
bounds[3] <= core["radiusM"]
and max(abs(bounds[0]), abs(bounds[1])) <= core["axialHalfExtentM"]
):
members.append(part["name"])
core_set = set(members)
pairs = []
filtered = pruned = 0
for a, bound_a in groups[parent].items():
for b, bound_b in groups[name].items():
if a in core_set or b in core_set:
filtered += 1
elif not intervals_overlap(bound_a, bound_b):
pruned += 1
else:
pairs.append([a, b])
if not pairs:
raise ValueError(f"不能关闭整对相邻连杆碰撞:{name}")
result.append(
{
"joint": name,
"parent": parent,
**core,
"coreParts": members,
"pairs": pairs,
"coreFilteredPairs": filtered,
"sweptPrunedPairs": pruned,
}
)
return result
def clip_convex(vertices, normal, offset):
"""Intersect a convex hull with a half-space, retaining all crossing edges."""
import numpy as np
from scipy.spatial import ConvexHull, QhullError
signed = vertices @ normal - offset
if signed.max() <= 1e-10:
return vertices
if signed.min() >= -1e-10:
return None
hull = ConvexHull(vertices)
result = list(vertices[signed <= 0])
for triangle in hull.simplices:
for i in range(3):
a, b = triangle[i], triangle[(i + 1) % 3]
if signed[a] * signed[b] < 0:
result.append(
vertices[a] + (vertices[b] - vertices[a]) * signed[a] / (signed[a] - signed[b])
)
points = np.unique(np.asarray(result), axis=0)
if len(points) < 4:
return None
try:
# Remove edge/triangulation interpolation points BEFORE rounding;
# rounding collinear points first invents tiny zigzag facet vertices.
points = np.unique(np.round(points[ConvexHull(points).vertices], 6), axis=0)
if len(points) < 4:
return None
hull = ConvexHull(points)
except QhullError:
return None # sub-micrometre cutting sliver collapsed by serialization
if hull.volume <= 1e-15:
return None
return points[hull.vertices]
def split_core(vertices, planes):
"""Partition, not delete: core and every outside fragment retain external contact."""
result = []
pending = vertices
for normal, offset in planes:
outside = clip_convex(pending, -normal, -offset)
if outside is not None:
result.append(outside)
pending = clip_convex(pending, normal, offset)
if pending is None:
return result
result.append(pending)
return result
def partition_assembly_cores(source, config, parts, coverage):
"""Split straddling hulls so a bearing contact cannot lock a structural hull.
A 16-sided inscribed prism stays inside each reviewed cylinder. The 2um
inset keeps rounding from reclassifying its core as structural. No region
is removed; only whole resulting core parts receive local mating exceptions.
"""
import numpy as np
from scipy.spatial import ConvexHull
by_visual, frames = assembly_frames(source, config, coverage)
shapes = [
(part["visual"], np.fromstring(part["vertices"], sep=" ").reshape(-1, 3)) for part in parts
]
for name, parent, transform, axis, core in frames:
x = np.cross(axis, np.eye(3)[int(np.argmin(np.abs(axis)))])
x /= np.linalg.norm(x)
y = np.cross(axis, x)
normals = [np.cos(t) * x + np.sin(t) * y for t in np.arange(16) * (2 * np.pi / 16)] + [
axis,
-axis,
]
distances = [(core["radiusM"] - 0.000002) * np.cos(np.pi / 16)] * 16 + [
core["axialHalfExtentM"] - 0.000002
] * 2
divided = []
for visual, vertices in shapes:
group, pose = by_visual[visual]
if group not in (parent, name):
divided.append((visual, vertices))
continue
origin = pose[:3, 3] - transform[:3, 3]
bounds = swept_bounds(vertices @ pose[:3, :3].T + origin, axis)
if (
(
bounds[3] <= core["radiusM"]
and max(abs(bounds[0]), abs(bounds[1])) <= core["axialHalfExtentM"]
)
or bounds[2] > core["radiusM"]
or bounds[0] > core["axialHalfExtentM"]
or bounds[1] < -core["axialHalfExtentM"]
):
divided.append((visual, vertices))
continue
planes = [
(pose[:3, :3].T @ n, distance - n @ origin)
for n, distance in zip(normals, distances, strict=True)
]
pieces = split_core(vertices, planes)
original = ConvexHull(vertices)
volume = sum(ConvexHull(piece).volume for piece in pieces)
if abs(volume - original.volume) > max(original.volume * 0.01, original.area * 0.00001):
raise ValueError(f"配合区切分体积误差超出微米量化预算:{visual}")
divided.extend((visual, piece) for piece in pieces)
shapes = divided
result = []
for item in coverage:
pieces = [v for visual, v in shapes if visual == item["visual"]]
pieces.sort(key=lambda v: tuple(v.mean(axis=0)))
item["coacdParts"] = item["parts"]
item["parts"] = len(pieces)
if not 1 <= len(pieces) <= 256:
raise ValueError("装配区域分割超出预算")
for i, vertices in enumerate(pieces):
if len(vertices) > 256:
raise ValueError(f"装配分割凸包顶点超出预算:{item['visual']} / {len(vertices)}")
result.append(
{
"name": f"{item['link']}__h{i:03d}",
"visual": item["visual"],
"vertices": " ".join(
f"{value:.6f}" for point in sorted(map(tuple, vertices)) for value in point
),
}
)
return result
def cook(source, config, cache):
import coacd
import numpy as np
import trimesh
from scipy.spatial import ConvexHull
for name, version in VERSIONS.items():
if importlib.metadata.version(name) != version:
raise ValueError(f"请使用离线固定依赖:{name}=={version}")
coacd.set_log_level("info")
parts = []
coverage = []
for visual in visuals(source, config):
key = digest(
json.dumps(
{
"source": visual["sha256"],
"scale": visual["scale"],
"parameters": PARAMETERS,
"versions": VERSIONS,
"threads": 4,
},
sort_keys=True,
).encode()
)
cached = cache / f"{key}.json"
mesh = trimesh.load(source / visual["mesh"], force="mesh")
mesh.apply_scale(visual["scale"])
if cached.exists():
hulls = json.loads(cached.read_text())
else:
print(f"生成 {visual['mesh']}", flush=True)
raw = coacd.run_coacd(coacd.Mesh(mesh.vertices, mesh.faces), **PARAMETERS)
hulls = [sorted(np.round(v, 6).tolist()) for v, _ in raw]
hulls.sort(key=lambda vertices: tuple(np.mean(vertices, axis=0)))
cache.mkdir(parents=True, exist_ok=True)
temporary = cached.with_suffix(".tmp")
temporary.write_text(json.dumps(hulls))
temporary.replace(cached)
if not 1 <= len(hulls) <= 128:
raise ValueError(f"凸包数量超出预算:{visual['mesh']}")
# An explicit sampled coverage diagnostic, not a certified Hausdorff bound.
samples = np.concatenate([mesh.vertices, mesh.triangles_center])[::3]
outside = np.full(len(samples), np.inf)
for index, vertices in enumerate(hulls):
v = np.asarray(vertices)
if (
v.ndim != 2
or v.shape[1] != 3
or not 4 <= len(v) <= PARAMETERS["max_ch_vertex"]
or not np.isfinite(v).all()
):
raise ValueError("凸包顶点损坏或超出预算")
if (v.min(axis=0) < mesh.bounds[0] - 0.002).any() or (
v.max(axis=0) > mesh.bounds[1] + 0.002
).any():
raise ValueError("凸包超出源网格边界预算")
hull = ConvexHull(v)
if hull.volume <= 1e-15:
raise ValueError("退化凸包")
signed_planes = (samples @ hull.equations[:, :3].T + hull.equations[:, 3]).max(axis=1)
outside = np.minimum(outside, signed_planes)
parts.append(
{
"name": f"{visual['link']}__h{index:03d}",
"visual": visual["visual"],
"vertices": " ".join(f"{x:.6f}" for x in v.ravel()),
}
)
if outside.max() > 0.0015:
raise ValueError(f"源表面采样覆盖不合格:{visual['mesh']}")
coverage.append(
{
**visual,
"parts": len(hulls),
"samples": len(samples),
"coacdSampleOutsidePlanesMaxM": float(max(0, outside.max())),
"sourceWatertight": bool(mesh.is_watertight),
"cacheKey": key,
}
)
print(
f"完成 {visual['link']}: {len(hulls)} hulls; sampled plane gap={outside.max():.6g} m",
flush=True,
)
parts = partition_assembly_cores(source, config, parts, coverage)
policy = joint_policy(source, config, parts, coverage)
geometry_sha = digest(json.dumps(parts, sort_keys=True, separators=(",", ":")).encode())
physics = {"timestep": 0.001, "solref": [0.002, 1], "solimp": [0.95, 0.99, 0.001]}
recipe_sha = digest(
json.dumps(
{"geometry": geometry_sha, "policy": policy, "physics": physics},
sort_keys=True,
separators=(",", ":"),
).encode()
)
return {
"revision": 4,
"geometrySha256": geometry_sha,
"recipeSha256": recipe_sha,
"physics": physics,
"jointPolicy": policy,
"source": config,
"generator": {
"versions": VERSIONS,
"parameters": PARAMETERS,
"threads": 4,
"corePartition": {"sides": 16, "insetM": 0.000002},
},
"coverage": coverage,
"parts": parts,
}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--source", type=Path, default=ROOT / "build/lekiwi/URDF")
parser.add_argument("--cache", type=Path, default=ROOT / "build/collision-cache")
parser.add_argument(
"--output", type=Path, default=ROOT / "robot_profiles/lekiwi-full-collision.json"
)
parser.add_argument("--check", action="store_true")
args = parser.parse_args()
result = cook(args.source, json.loads(CONFIG.read_text()), args.cache)
if args.check:
if json.loads(args.output.read_text()) != result:
raise ValueError("已生成碰撞数据与源文件/算法参数不一致")
else:
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(result, indent=2) + "\n")
print(f"整臂碰撞覆盖 {len(result['coverage'])} 个视觉网格 / {len(result['parts'])} 个凸包")
if __name__ == "__main__":
main()
@@ -0,0 +1,117 @@
"""Offline, deterministic convex parts from the pinned CAD (requires numpy/scipy).
Runtime imports the checked-in JSON only; no extra browser/Python teleop dependency.
Do not use one hull for the entire gripper: that would fill its open jaw space.
"""
import argparse
import hashlib
import json
import struct
from pathlib import Path
import numpy as np
from scipy.spatial import ConvexHull
ROOT = Path(__file__).resolve().parents[2]
# Cuts in the original STL's millimeters, with 1 mm overlap at each seam.
# Each plane keeps sign * coordinate[axis] <= limit.
PARTS = [
("fixed_palm", "Wrist_Roll_08c-v1", [(2, 1, 39.0)]),
("fixed_finger", "Wrist_Roll_08c-v1", [(2, -1, -38.0)]),
("moving_hinge", "Moving_Jaw_08d-v1", [(1, -1, 22.0)]),
("moving_finger", "Moving_Jaw_08d-v1", [(1, 1, -21.0), (1, -1, 60.0)]),
("moving_tip", "Moving_Jaw_08d-v1", [(1, 1, -59.0)]),
]
# Bound to prepare_assets.py's pinned source; never silently regenerate for another CAD.
SOURCE_HASHES = {
"Wrist_Roll_08c-v1.stl": "87507f73f485c2cacb3dc83924a712069fbd68f72b82d4a31a2cfe7f58c9e4c9",
"Moving_Jaw_08d-v1.stl": "71caabee267376791210950b3b4e2f7968d9b57b92ed3f57711c03f2b5666912",
}
def clip_polygon(polygon, axis, sign, limit):
result = []
for previous, current in zip(polygon[-1:] + polygon[:-1], polygon, strict=True):
a = sign * previous[axis] - limit
b = sign * current[axis] - limit
if (a <= 0) != (b <= 0):
result.append(previous + (current - previous) * (a / (a - b)))
if b <= 0:
result.append(current)
return result
def generate(source, *, definitions=PARTS, source_hashes=SOURCE_HASHES, revision=2):
"""Shared CAD-frame convex clipping; callers bind their own source hashes/cuts."""
parts = []
for name, mesh, planes in definitions:
raw = (source / (mesh + ".stl")).read_bytes()
digest = hashlib.sha256(raw).hexdigest()
if digest != source_hashes[mesh + ".stl"]:
raise ValueError(f"不支持的碰撞体 CAD:{mesh}.stl ({digest})")
count = struct.unpack_from("<I", raw, 80)[0]
if len(raw) != 84 + count * 50:
raise ValueError(f"无效的二进制 STL:{mesh}")
records = np.frombuffer(
raw,
offset=84,
dtype=np.dtype(
[
("normal", "<f4", (3,)),
("vertex", "<f4", (3, 3)),
("attribute", "<u2"),
]
),
)
points = []
for triangle in records["vertex"]:
polygon = [point.astype(float) for point in triangle]
for axis, sign, limit in planes:
if polygon:
polygon = clip_polygon(polygon, axis, sign, limit)
points.extend(polygon)
# Clip triangles, not just vertices, so long faces crossing cuts leave no holes.
points = np.unique(np.round(points, 3), axis=0) # 1 micrometer grid in CAD mm.
vertices = points[ConvexHull(points).vertices] / 1000 # MJCF meters.
vertices = vertices[np.lexsort(vertices.T[::-1])]
parts.append(
{
"name": name,
"visual": mesh + "_visual",
"vertices": " ".join(f"{value:.6f}" for value in vertices.flat),
}
)
return {
"revision": revision,
"source": {
"repository": "https://github.com/SIGRobotics-UIUC/LeKiwi",
"revision": "efa608d7ee5a495a4803b1d28cd0c955b4f1e033",
"license": "Apache-2.0",
},
"sources": source_hashes,
"parts": parts,
}
def main():
parser = argparse.ArgumentParser(description="从固定版本 STL 重建 LeKiwi 分段夹爪碰撞体")
parser.add_argument("--source", type=Path, default=ROOT / "build/lekiwi/URDF/meshes")
parser.add_argument(
"--output", type=Path, default=ROOT / "robot_profiles/lekiwi-gripper-collision.json"
)
parser.add_argument("--check", action="store_true", help="只验证已签入的碰撞数据可重建")
args = parser.parse_args()
result = generate(args.source)
if args.check:
if json.loads(args.output.read_text()) != result:
raise ValueError("夹爪碰撞数据与固定 CAD/生成器不一致")
else:
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(result, indent=2) + "\n")
status = "验证通过" if args.check else args.output
print(f"LeKiwi 夹爪碰撞体:{len(result['parts'])} 个分段凸包,{status}")
if __name__ == "__main__":
main()
+116
View File
@@ -0,0 +1,116 @@
#!/usr/bin/env python3
"""Rebuild a provenance-bound LeKiwi input ZIP without modifying the source repo."""
import argparse
import hashlib
import json
import urllib.request
import xml.etree.ElementTree as ET
import zipfile
from pathlib import Path, PurePosixPath
ROOT = Path(__file__).resolve().parents[2]
PROFILE = json.loads((ROOT / "robot_profiles/lekiwi-v1.json").read_text())
SOURCE = PROFILE["source"]
MAX_TOTAL = 128 * 1024 * 1024
def digest(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def prepare(source: Path | None, output: Path, download: bool = False) -> Path:
if source is not None:
source = source.resolve()
if source.name == "URDF":
source = source.parent
if not (source / "URDF/LeKiwi.urdf").is_file():
raise ValueError("Source must be the LeKiwi repository or its URDF directory")
if output.resolve().is_relative_to(source):
raise ValueError("Output must not modify the reference repository")
elif not download:
raise ValueError("Pass --source or explicitly opt in with --download")
files: dict[str, bytes] = {}
total = 0
def read(relative: str) -> bytes:
nonlocal total
path = PurePosixPath(relative)
if path.is_absolute() or ".." in path.parts or "\\" in relative:
raise ValueError(f"Unsafe asset path: {relative}")
if source is not None:
file = (source / relative).resolve()
if not file.is_relative_to(source) or file.stat().st_size > MAX_TOTAL:
raise ValueError(f"Invalid asset: {relative}")
data = file.read_bytes()
else:
url = f"https://raw.githubusercontent.com/SIGRobotics-UIUC/LeKiwi/{SOURCE['revision']}/{relative}"
with urllib.request.urlopen(url, timeout=60) as response:
data = response.read(MAX_TOTAL + 1)
total += len(data)
if total > MAX_TOTAL:
raise ValueError("Assets exceed 128 MiB")
files[relative] = data
return data
urdf = read("URDF/LeKiwi.urdf")
if digest(urdf) != SOURCE["urdfSha256"]:
raise ValueError("URDF revision/hash is not supported by lekiwi-v1")
tree = ET.fromstring(urdf)
for mesh in sorted({m.get("filename", "") for m in tree.iter("mesh")}):
if not mesh.startswith("meshes/") or not mesh.endswith(".stl"):
raise ValueError(f"Unexpected mesh reference: {mesh}")
read(f"URDF/{mesh}")
read("LICENSE.txt")
read("CITATION.cff")
manifest = {
"profileId": PROFILE["id"],
"profileVersion": PROFILE["version"],
"source": SOURCE,
"files": {p: digest(data) for p, data in files.items()},
}
files["source-manifest.json"] = json.dumps(manifest, indent=2).encode()
files["robot-profile.json"] = json.dumps(
{"id": PROFILE["id"], "version": PROFILE["version"]}
).encode()
files["SIMULATION-NOTICE.md"] = (
"# LeKiwi simulation input\n\nSource: "
+ SOURCE["repository"]
+ "\n\nRevision: "
+ SOURCE["revision"]
+ "\n\nOriginal URDF/STL files are unmodified, Apache-2.0. "
"The platform applies an explicit simulation-only profile after conversion: estimated "
"mass/inertia/limits/servos and simplified passive-roller contacts. Not a calibrated "
"hardware model; no cameras, training or grasping guarantee.\n"
).encode()
output.mkdir(parents=True, exist_ok=True)
for relative, data in files.items():
target = output / relative
if target.is_symlink() or not target.resolve().is_relative_to(output.resolve()):
raise ValueError("Unsafe output path")
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(data)
archive = output / "lekiwi-v1.zip"
if archive.is_symlink() or not archive.resolve().is_relative_to(output.resolve()):
raise ValueError("Unsafe archive output path")
with zipfile.ZipFile(archive, "w", zipfile.ZIP_DEFLATED) as bundle:
for relative, data in files.items():
info = zipfile.ZipInfo(relative, date_time=(1980, 1, 1, 0, 0, 0))
info.compress_type = zipfile.ZIP_DEFLATED
info.external_attr = 0o100644 << 16
bundle.writestr(info, data, compresslevel=9)
return archive
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument("--source", type=Path)
group.add_argument("--download", action="store_true")
parser.add_argument("--output", type=Path, default=ROOT / "build/lekiwi")
args = parser.parse_args()
print(prepare(args.source, args.output, args.download))
if __name__ == "__main__":
main()
@@ -0,0 +1,5 @@
# Offline cooking only. Do not install into training/LeRobot environments.
coacd==1.0.14
trimesh==4.12.2
numpy==2.1.3
scipy==1.17.0
+68
View File
@@ -0,0 +1,68 @@
#!/usr/bin/env python3
"""安装经验证的 CPU LeRobot 环境;不修改当前解释器、上游源码或训练环境。"""
import argparse
import json
import os
import subprocess
import sys
import venv
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--venv", type=Path, default=ROOT / "build/venvs/lerobot")
args = parser.parse_args()
target = args.venv.resolve()
if sys.version_info[:2] != (3, 12):
parser.error("此兼容环境仅验证 Python 3.12,请使用 Python 3.12 执行安装脚本")
if not target.is_relative_to(ROOT / "build") or target == Path(sys.prefix).resolve():
parser.error("目标必须位于仓库 build/ 下,并且不能是当前环境")
env = {**os.environ, "PYTHONNOUSERSITE": "1", "PYTHONPATH": ""}
if not (target / "bin/python").exists():
venv.EnvBuilder(with_pip=True).create(target)
python = str(target / "bin/python")
prefix = subprocess.check_output(
[python, "-I", "-c", "import sys; print(sys.prefix)"], env=env, text=True
).strip()
config = (target / "pyvenv.cfg").read_text().lower()
if Path(prefix).resolve() != target or "include-system-site-packages = false" not in config:
parser.error("目标不是独立 venv;拒绝对共享/其他 Python 环境执行 pip")
def run(*arguments):
subprocess.run([python, "-m", "pip", *arguments], env=env, cwd=ROOT, check=True)
run(
"install",
"--index-url",
"https://download.pytorch.org/whl/cpu",
"torch==2.11.0+cpu",
"torchvision==0.26.0+cpu",
)
constraints = "integrations/lerobot/requirements-cpu.txt"
run("install", "-r", constraints)
run(
"install",
"--no-build-isolation",
"-c",
constraints,
"-e",
"control_bridge",
"-e",
"integrations/lerobot",
)
run("check")
# Record versions, not pip freeze's private VCS origins or credentials.
result = subprocess.check_output(
[python, "-m", "pip", "list", "--format=json"], env=env, text=True
)
report = ROOT / "build/lerobot-environment.json"
report.write_text(json.dumps({"python": sys.version, "packages": json.loads(result)}, indent=2))
print(f"完成:{python}\n版本记录:{report}\n请在此环境执行插件测试和示例。")
if __name__ == "__main__":
main()
+132
View File
@@ -0,0 +1,132 @@
"""Linux terminal base + joint-space arm teleop; no hardware or pynput required."""
import argparse
import math
import os
import select
import sys
import termios
import time
import tty
CONTROL_HZ = 30
KEY_PULSE_SECONDS = 0.18
# Positive / negative in LeRobot units: degrees, or 0–100 gripper opening.
ARM_KEYS = {
"arm_shoulder_pan.pos": ("u", "j"),
"arm_shoulder_lift.pos": ("i", "k"),
"arm_elbow_flex.pos": ("o", "l"),
"arm_wrist_flex.pos": ("t", "g"),
"arm_wrist_roll.pos": ("y", "h"),
"arm_gripper.pos": ("v", "b"),
}
class KeyboardController:
"""Compose base velocities and incremental arm targets in one writer/lease."""
def __init__(self, robot, arm_speed=20.0, gripper_speed=50.0):
self.robot = robot
self.arm_speed = arm_speed
self.gripper_speed = gripper_speed
self.targets = robot.get_observation() # Never jump to a hard-coded zero pose.
self.active = {}
self.speed_keys = {robot.teleop_keys[k] for k in ("speed_up", "speed_down")}
self.motion_keys = {
robot.teleop_keys[k]
for k in ("forward", "backward", "left", "right", "rotate_left", "rotate_right")
} | {key for pair in ARM_KEYS.values() for key in pair}
def step(self, keys, now):
keys = set(keys)
if " " in keys:
self.active.clear()
pressed = set() # Space wins over every movement/speed key in the batch.
else:
self.active = {key: expiry for key, expiry in self.active.items() if expiry > now}
self.active.update(dict.fromkeys(keys & self.motion_keys, now + KEY_PULSE_SECONDS))
# Speed changes are input events, not a 180 ms pulse repeated every frame.
pressed = set(self.active) | (keys & self.speed_keys)
self.robot.get_observation() # Require fresh feedback even while holding still.
action = self.robot._from_keyboard_to_base_action(pressed)
for channel, (positive, negative) in ARM_KEYS.items():
direction = int(positive in pressed) - int(negative in pressed)
if direction:
speed = self.gripper_speed if channel == "arm_gripper.pos" else self.arm_speed
# At most one control tick per update: no large catch-up jump after a stall.
action[channel] = self.targets[channel] + direction * speed / CONTROL_HZ
# The plugin holds omitted arm channels and clamps via the browser descriptor.
# Accumulate from confirmed targets only, including any joint/gripper clipping.
self.targets = self.robot.send_action(action)
return self.targets
def read_keys(fd):
keys = []
# Bound input draining so queued terminal data cannot starve the control watchdog.
for _ in range(64):
if not select.select([fd], [], [], 0)[0]:
break
value = os.read(fd, 1)
if not value:
raise EOFError("终端输入已断开")
keys.append(value.decode(errors="ignore"))
return keys
def main():
parser = argparse.ArgumentParser(description="LeKiwi 底盘 + 机械臂终端遥操作(仿真专用)")
parser.add_argument(
"--endpoint", default=os.environ.get("MUJOCO_CONTROL_ENDPOINT", "http://127.0.0.1:8766")
)
parser.add_argument(
"--arm-speed", type=float, default=20.0, help="臂关节点动速度,0–90 度/秒,默认20"
)
parser.add_argument(
"--gripper-speed", type=float, default=50.0, help="夹爪点动速度,0–100 百分点/秒,默认50"
)
args = parser.parse_args()
for name, speed, limit in (
("--arm-speed", args.arm_speed, 90),
("--gripper-speed", args.gripper_speed, 100),
):
if not math.isfinite(speed) or not 0 < speed <= limit:
parser.error(f"{name} 必须为大于0且不超过{limit}的有限数值")
if not sys.stdin.isatty():
raise ValueError("请在交互式终端运行;自动测试请用 demo_control.py")
# Keep --help and argument validation usable without importing the LeRobot stack.
from demo_control import make_robot
fd = sys.stdin.fileno()
previous = termios.tcgetattr(fd)
robot = make_robot(args.endpoint)
try:
robot.connect()
controller = KeyboardController(robot, args.arm_speed, args.gripper_speed)
tty.setcbreak(fd)
print(
"底盘:w/s 前后,a/d 左右,z/x 旋转,r/f 底盘调速;空格停止点动,q 退出。\n"
"机械臂(前键增大/后键减小):u/j 肩转,i/k 肩俯仰,o/l 肘,t/g 腕俯仰,"
"y/h 腕旋转;v/b 夹爪开/合。\n"
f"臂 {args.arm_speed:g} 度/秒,夹爪 {args.gripper_speed:g} 百分点/秒。"
"终端按键为180ms脉冲;脉冲结束后底盘停止、机械臂保持目标。",
flush=True,
)
while True:
start = time.monotonic()
keys = read_keys(fd)
if robot.teleop_keys["quit"] in keys:
return
controller.step(keys, start)
time.sleep(max(0, 1 / CONTROL_HZ - (time.monotonic() - start)))
except KeyboardInterrupt:
pass
finally:
try:
robot.disconnect()
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, previous)
if __name__ == "__main__":
main()
@@ -0,0 +1,175 @@
"""Offline cooker contracts: no real CAD download or CoACD run required here."""
import copy
import hashlib
import importlib.util
import tempfile
import unittest
from pathlib import Path
import numpy as np
SCRIPT = Path(__file__).resolve().parents[1] / "generate_full_collisions.py"
spec = importlib.util.spec_from_file_location("collision_cooker", SCRIPT)
cooker = importlib.util.module_from_spec(spec)
spec.loader.exec_module(cooker)
def cube(center, size=0.004):
return (
np.array([[x, y, z] for x in (-size, size) for y in (-size, size) for z in (-size, size)])
+ center
)
class CookingTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.source = Path(self.temp.name)
(self.source / "meshes").mkdir()
(self.source / "meshes/m.stl").write_bytes(b"source")
links = "".join(
f'<link name="{name}"><visual name="{name}_visual"><geometry>'
'<mesh filename="meshes/m.stl" scale=".001 .001 .001"/>'
"</geometry></visual></link>"
for name in ("base", "motor", "arm")
)
self.xml = (
'<robot name="fixture">'
+ links
+ """
<joint name="fixed" type="fixed"><parent link="base"/><child link="motor"/></joint>
<joint name="hinge" type="continuous">
<parent link="motor"/><child link="arm"/><axis xyz="0 0 1"/>
</joint>
</robot>"""
)
self.config = {
"rootLink": "base",
"meshes": {"meshes/m.stl": hashlib.sha256(b"source").hexdigest()},
"assemblyCores": {
"hinge": {"radiusM": 0.015, "axialHalfExtentM": 0.02, "reason": "fixture bearing"}
},
}
self.save_source()
def save_source(self):
data = self.xml.encode()
(self.source / "LeKiwi.urdf").write_bytes(data)
self.config["urdfSha256"] = hashlib.sha256(data).hexdigest()
def test_covers_fixed_and_moving_visuals_and_preserves_scale(self):
items = cooker.visuals(self.source, self.config)
self.assertEqual(
[x["visual"] for x in items], ["base_visual", "motor_visual", "arm_visual"]
)
self.assertTrue(all(x["scale"] == [0.001] * 3 for x in items))
def test_rejects_changed_source_urdf_and_mesh(self):
(self.source / "LeKiwi.urdf").write_text("modified")
with self.assertRaisesRegex(ValueError, "URDF"):
cooker.visuals(self.source, self.config)
self.save_source()
(self.source / "meshes/m.stl").write_bytes(b"modified")
with self.assertRaisesRegex(ValueError, "SHA-256"):
cooker.visuals(self.source, self.config)
def test_rejects_unsupported_scale_and_path_escape(self):
for scale in ("nan .001 .001", "-.001 .001 .001", ".001 .001", "inf 1 1"):
original = self.xml
self.xml = self.xml.replace(".001 .001 .001", scale)
self.save_source()
with self.assertRaisesRegex(ValueError, "缩放"):
cooker.visuals(self.source, self.config)
self.xml = original
self.xml = self.xml.replace("meshes/m.stl", "../escape.stl")
self.save_source()
with self.assertRaisesRegex(ValueError, "未批准"):
cooker.visuals(self.source, self.config)
def test_rejects_missing_or_incomplete_coverage(self):
self.config["meshes"]["meshes/extra.stl"] = "unknown"
with self.assertRaisesRegex(ValueError, "覆盖"):
cooker.visuals(self.source, self.config)
self.config["rootLink"] = "not_present"
self.config["meshes"] = {}
with self.assertRaisesRegex(ValueError, "覆盖"):
cooker.visuals(self.source, self.config)
def test_radial_minimum_uses_polygon_edges_not_only_vertices(self):
bounds = cooker.swept_bounds(cube([2.0, 0, 0], 0.5), np.array([0, 0, 1.0]))
self.assertAlmostEqual(bounds[2], 1.5)
self.assertGreater(np.linalg.norm(cube([2.0, 0, 0], 0.5)[:, :2], axis=1).min(), bounds[2])
self.assertTrue(cooker.intervals_overlap(bounds, (-0.1, 0.1, 1.51, 1.52)))
self.assertFalse(cooker.intervals_overlap(bounds, (1.0, 2.0, 1.51, 1.52)))
for angle in np.linspace(-np.pi, np.pi, 37):
rotation = np.array(
[[np.cos(angle), -np.sin(angle), 0], [np.sin(angle), np.cos(angle), 0], [0, 0, 1]]
)
np.testing.assert_allclose(
cooker.swept_bounds(cube([2.0, 0, 0], 0.5) @ rotation.T, np.array([0, 0, 1.0])),
bounds,
atol=1e-12,
)
def test_core_partition_preserves_volume_and_separates_mating_from_structure(self):
from scipy.spatial import ConvexHull
coverage = cooker.visuals(self.source, self.config)
parts = [
{
"visual": item["visual"],
"name": item["link"],
"vertices": " ".join(map(str, cube([0.012, 0, 0], 0.01).ravel())),
}
for item in coverage
]
for item in coverage:
item["parts"] = 1
divided = cooker.partition_assembly_cores(self.source, self.config, parts, coverage)
policy = cooker.joint_policy(self.source, self.config, divided, coverage)[0]
for visual in (item["visual"] for item in coverage):
group = [piece for piece in divided if piece["visual"] == visual]
volume = 0
for piece in group:
hull = ConvexHull(np.fromstring(piece["vertices"], sep=" ").reshape(-1, 3))
volume += hull.volume
# The old unsplit hull locked the bearing although this point is inside its core.
if np.max(hull.equations[:, :3] @ [0.012, 0, 0] + hull.equations[:, 3]) < 1e-8:
self.assertIn(piece["name"], policy["coreParts"])
if np.max(hull.equations[:, :3] @ [0.021, 0, 0] + hull.equations[:, 3]) < 1e-8:
self.assertNotIn(piece["name"], policy["coreParts"])
self.assertAlmostEqual(volume / (0.02**3), 1, delta=0.001)
self.assertGreater(len(group), 1)
self.assertTrue(policy["pairs"])
def test_exceptions_are_bounded_core_parts_not_whole_adjacent_bodies(self):
coverage = cooker.visuals(self.source, self.config)
parts = [
{"name": name, "visual": visual, "vertices": " ".join(map(str, cube(center).ravel()))}
for name, visual, center in [
("base_core", "base_visual", [0, 0, 0]),
("base_structural", "base_visual", [0.05, 0, 0]),
("far_axial", "base_visual", [0.05, 0, 0.2]),
("arm_core", "arm_visual", [0, 0, 0]),
("arm_structural", "arm_visual", [0.05, 0, 0]),
]
]
policy = cooker.joint_policy(self.source, self.config, parts, coverage)[0]
self.assertEqual(policy["coreParts"], ["base_core", "arm_core"])
self.assertEqual(policy["pairs"], [["base_structural", "arm_structural"]])
self.assertGreater(policy["sweptPrunedPairs"], 0)
self.assertEqual([x["weldJoint"] for x in coverage], ["base", "base", "hinge"])
bad = copy.deepcopy(self.config)
bad["assemblyCores"]["hinge"]["radiusM"] = 1
with self.assertRaisesRegex(ValueError, "尺寸预算"):
cooker.joint_policy(self.source, bad, parts, coverage)
with self.assertRaisesRegex(ValueError, "整对"):
cooker.joint_policy(
self.source, self.config, [p for p in parts if "core" in p["name"]], coverage
)
if __name__ == "__main__":
unittest.main()