Isaac Lab Mimic 数据生成模块(isaaclab_mimic.datagen)深度解析:从源演示到可执行轨迹
【免费下载链接】IsaacLabUnified framework for robot learning built on NVIDIA Isaac Sim项目地址: https://gitcode.com/GitHub_Trending/is/IsaacLab
isaaclab_mimic.datagen是 Isaac Lab 中负责"基于少量人工采集源演示生成新数据集"的核心子包,其实现受 MimicGen 启发:将演示解析为以物体为参考系的子任务片段,再根据新场景的物体位姿对片段进行位姿变换、拼接与插值,最终形成机械臂可执行的航点轨迹。本文将以该子包的官方 API 文档骨架为主线,结合仓库源码逐类剖析DataGenerator、DatagenInfo、DataGenInfoPool、三种SelectionStrategy以及航点类族的设计与调用关系,并给出基于 generate_dataset.py 的实战使用方法,帮助你完整掌握 Isaac Lab Mimic 数据生成的底层原理与二次开发入口。
一、模块定位与文件构成
datagen子包位于 source/isaaclab_mimic/isaaclab_mimic/datagen/,包级入口init.py 一次性导出全部核心符号。从源码结构看,该子包按职责分为六个文件:
| 文件 | 核心类/函数 | 职责 |
|---|---|---|
data_generator.py | DataGenerator及 3 个位姿变换函数 | 数据生成主流水线 |
datagen_info.py | DatagenInfo | 单个时间步的环境信息载体 |
datagen_info_pool.py | DataGenInfoPool | 多 episode 信息池与子任务边界解析 |
selection_strategy.py | SelectionStrategy及 3 个策略实现 | 源演示子任务片段选择 |
waypoint.py | Waypoint、WaypointSequence、WaypointTrajectory、MultiWaypoint | 航点/轨迹表示与执行 |
generation.py | run_data_generator、env_loop、setup_env_config、setup_async_generation | asyncio 驱动的批量生成框架 |
整体数据流为:源 HDF5 数据集 →DataGenInfoPool(解析为DatagenInfo并计算子任务边界)→DataGenerator.generate()(选源片段 → 随机化边界 → 位姿变换 → 构建航点轨迹)→ 多环境并行执行并导出新演示。下面按数据流顺序逐层展开。
二、DatagenInfo:数据生成的统一信息载体
DatagenInfo定义在 datagen_info.py,它的作用是把数据生成过程中反复需要的环境信息集中到一处,避免每次生成都反复查询环境。构造器签名如下:
DatagenInfo( eef_pose=None, # 机器人末端执行器位姿,形状 [..., 4, 4] object_poses=None, # dict:物体名 -> 物体位姿 [..., 4, 4] subtask_start_signals=None, # dict:子任务名 -> 0/1 起始信号(skillgen 必需) subtask_term_signals=None, # dict:子任务名 -> 0/1 完成信号 target_eef_pose=None, # 各时间步的目标末端位姿 [..., 4, 4] gripper_action=None, # 夹爪动作 [..., D],D 为夹爪执行维度 )关键设计点:
- 非强制全量:类文档明确指出 "not all information must be present",六个字段均可为
None,这为不同类型任务(如无夹爪、无参考物体)留出灵活性。 - 子任务信号的数据类型宽容:
subtask_start_signals与subtask_term_signals的每个值既可以是 Python 的float/int,也可以是形状[..., 1]的torch.Tensor,构造器对两种情况分别处理。这些信号是 skillgen 识别演示中子任务精确起止的关键依据(见datagen_info.py第 64-83 行)。 to_dict():将非None字段导出为字典,其中object_poses、子任务信号使用deepcopy防止调用方误改内部状态,方便与录制器/导出管线对接。
三、DataGenInfoPool:演示池与子任务边界解析
DataGenInfoPool定义在 datagen_info_pool.py,是一个存放DatagenInfo的容器,支持通过asyncio.Lock实现"边消费边添加"的线程安全,因而可以被多个 mimic 数据生成器共享。
3.1 构造与配置提取
构造器接收env、env_cfg、device和可选的asyncio_lock。初始化时从env_cfg.subtask_configs(按末端执行器名组织的SubTaskConfig列表)提取三类信息(第 45-54 行):
subtask_term_signal_names[eef_name]:每个子任务的终止信号名;subtask_start_offset_ranges[eef_name]/subtask_term_offset_ranges[eef_name]:子任务起止边界的随机偏移范围。
3.2 添加 episode 与边界计算
add_episode(episode)是异步方法(有锁时加锁),内部调用_add_episode。该方法的处理流程(第 89-204 行):
- 从 episode 观测的
obs["datagen_info"]中提取eef_pose、object_pose、target_eef_pose、subtask_term_signals及可选的subtask_start_signals;缺失时抛出ValueError; - 通过
env.actions_to_gripper_actions(ep_grp["actions"])把原始动作转换成夹爪动作; - 组装
DatagenInfo并追加到self._datagen_infos; - 为每个末端执行器逐个解析子任务边界
(start_index, end_index):- skillgen 模式:子任务起点由演示数据中的
subtask_start_signals显式给出,通过计算信号序列相邻差分(diffs = indicators[1:] - indicators[:-1])找到第一个 0→1 跳变位置作为起点;终点同样由subtask_term_signals的第一个 0→1 跳变决定;最后一个子任务没有终止信号,以动作序列长度作为终点; - 非 skillgen 模式:子任务按顺序衔接,前一个子任务的终点即后一个子任务的起点;
- 若
end_index <= start_index则抛出ValueError(终止信号未递增)。
- skillgen 模式:子任务起点由演示数据中的
此外_add_episode还对偏移范围做了两层一致性断言:最坏情况下子任务不为空(skillgen),以及最大偏移下子任务之间不重叠(两种模式各有独立的断言分支)。
3.3 从数据集文件加载
load_from_dataset_file(file_path, select_demo_keys=None)使用isaaclab.utils.datasets.HDF5DatasetFileHandler打开 HDF5 文件,遍历全部 episode(可选按select_demo_keys过滤),逐个调用_add_episode完成装载。
四、SelectionStrategy:源子任务片段选择策略
选择策略解决"当前场景下该选哪条源演示的子任务片段"的问题,实现在 selection_strategy.py。
4.1 注册机制
模块维护全局字典REGISTERED_SELECTION_STRATEGIES,采用元类SelectionStrategyMeta自动注册:任何继承SelectionStrategy且定义了NAME属性的类,定义时即被登记。make_selection_strategy(name)按名称查表实例化,assert_selection_strategy_exists(name)用于校验。抽象基类SelectionStrategy定义了抽象方法:
def select_source_demo(self, eef_pose, object_pose, src_subtask_datagen_infos): # eef_pose: 当前 4x4 末端位姿 # object_pose: 当前子任务参考物体的 4x4 位姿 # src_subtask_datagen_infos: 各源演示该子任务片段的 DatagenInfo 列表 # 返回选中的源演示索引 source_demo_ind4.2 三种内置策略
| 注册名 | 类 | 选择逻辑 |
|---|---|---|
random | RandomStrategy | 均匀随机返回一个源演示索引 |
nearest_neighbor_object | NearestNeighborObjectStrategy | 选择"当前场景物体位姿与源演示片段起始物体位姿最近"的源演示 |
nearest_neighbor_robot_distance | NearestNeighborRobotDistanceStrategy | 选择"变换后片段首点位姿距当前末端位姿最近"的源演示 |
两个最近邻策略共享参数pos_weight(位置权重,默认 1.0)、rot_weight(旋转权重,默认 1.0)、nn_k(从最近的前 k 个候选中均匀随机选一,默认 3),并采用相同的距离度量实现:
- 位置距离取 L2 欧氏距离;
- 旋转距离通过旋转矩阵差的对角线计算轴角表示下的夹角:
arc_cos_in = (diag(delta_R).sum(-1) - 1) / 2并 clamp 到[-1, 1]保证数值稳定; - 加权距离
pos_weight * pos_dists + rot_weight * rot_dists排序后取前nn_k个,再从中均匀随机抽取,兼顾"贴近最佳"与"多样性"。
NearestNeighborRobotDistanceStrategy的实现细节值得注意:它先把源末端位姿变换到源物体坐标系,再应用当前物体位姿得到"若使用该源片段,变换后子任务的起始末端位姿",最后度量其与当前末端位姿的距离——这正是DataGenerator中位姿变换逻辑的镜像,选择与变换严格自洽。
这些策略的选择正确性由 test_selection_strategy.py 验证:测试构造两组距离悬殊的位姿簇,反复调用select_source_demo(最多 1000 次)并断言选中索引始终落在正确簇内,覆盖了位置噪声扰动下的鲁棒性。
五、Waypoint 类族:轨迹的表示、拼接与执行
Waypoint、WaypointSequence、WaypointTrajectory、MultiWaypoint四个类定义在 waypoint.py,构成了轨迹的层次化数据结构。
Waypoint:单个 6-DoF 目标点,含pose(4x4 位姿)、gripper_action(夹爪动作)与可选的noise(该时间步的动作噪声幅值,只作用于手臂动作)。WaypointSequence:Waypoint的有序列表。from_poses(poses, gripper_actions, action_noise)类方法支持把形状(T, 4, 4)的位姿序列批量转为航点序列,action_noise既可以是标量(广播为常值)也可以是逐时间步张量。支持__add__拼接、split(ind)切分、last_waypoint取尾。WaypointTrajectory:由若干WaypointSequence组成的完整轨迹。merge()实现子任务间衔接:按num_steps_interp线性插值(基于isaaclab.utils.math.interpolate_poses)、按num_steps_fixed重复目标位姿,并把other的首元素弹出以与插值段衔接;pop_first()用于丢弃为插值临时加入的起点;add_waypoint_sequence_for_target_pose()支持"向目标位姿插值"或"恒定保持"两种模式。MultiWaypoint:同一时间步多个末端执行器航点的集合,execute()将{eef_name: waypoint}组装为target_eef_pose_dict与gripper_action_dict,调用环境的target_eef_pose_to_action()(优先传action_noise_dict,旧签名走noise参数兼容路径),然后env.step()执行并用success_term判定任务成功,返回布尔结果。
六、DataGenerator:数据生成主流水线
DataGenerator是 data_generator.py 的核心类,也是官方 API 文档列出的头号类。其文档明确描述了设计意图:受 MimicGen 启发,用少量人工采集的源演示生成新数据集——把演示解析为以物体为参考系的子任务片段(存入DataGenInfoPool),再按新场景上下文变换各片段并拼接成连贯的末端执行器轨迹。
6.1 构造与入口
DataGenerator( env: ManagerBasedRLMimicEnv, src_demo_datagen_info_pool: DataGenInfoPool | None = None, dataset_path: str | None = None, demo_keys: list[str] | None = None, )构造时校验env.cfg为MimicEnvCfg,并对任务规范做健全性检查:每个末端执行器的最后一个子任务的subtask_term_offset_range必须为 (0, 0)(终段不允许边界随机化)。源演示池与数据集路径二选一,否则抛ValueError。
6.2 子任务边界随机化
randomize_subtask_boundaries()根据任务规范为每个源演示、每个子任务采样随机的起止偏移:
- 第一个子任务的起点偏移来自
first_subtask_start_offset_range; - skillgen 模式下每个子任务的起点可独立随机(
subtask_start_offset_range),非 skillgen 模式则起点固定为前一子任务终点; - 每个子任务的终点偏移采样自
subtask_term_offset_range; - 随后做三重断言:子任务非空、起止索引严格递增、展平后索引不降序。
6.3 源片段选择与位姿变换
select_source_demo()把当前末端位姿、物体位姿与各源演示子任务片段信息交给所选策略。其约束是:若子任务没有object_ref(参考物体为 None),只支持random策略。
generate_eef_subtask_trajectory()是单子任务轨迹生成的核心,关键步骤:
- 查询当前场景物体位姿(
env.get_object_poses),判断是否需要重新选源:默认仅首个子任务选源,generation_select_src_per_subtask=True时每个子任务独立选源,generation_select_src_per_arm=False时同一源演示在各臂间复用; - 处理协调约束:若当前子任务与并发任务有协调约束且并发任务已选定源演示,则复用其源索引与 delta 变换;
- 按随机化边界切片源演示的末端位姿、目标位姿、夹爪动作;
- 若为首个子任务或
generation_transform_first_robot_pose=True,在目标位姿序列前拼接源片段的首个末端位姿(保证从真实机器人位姿开始插值而非直接跳到第一个目标点),并同步补齐一条夹爪动作; - 位姿变换分三种情形:
- 使用并发任务的delta 变换:
transform_source_data_segment_using_delta_object_pose; - 按协调方案计算 delta:
get_delta_pose_with_scheme支持SubTaskConstraintCoordinationScheme的TRANSFORM(完整位姿差)、TRANSLATE(仅平移差)、REPLAY(恒等),并可叠加coordination_scheme_pos_noise_scale/coordination_scheme_rot_noise_scale噪声; - 常规情形使用物体坐标系变换
transform_source_data_segment_using_object_pose:先把源末端位姿变换到源物体坐标系,再应用到当前物体坐标系,从而保持"目标末端位姿与物体位姿之间的相对关系"不变;无参考物体时原样保留。
- 使用并发任务的delta 变换:
- 用
WaypointSequence.from_poses构造航点序列并封装为WaypointTrajectory。
6.4 子任务轨迹合并
merge_eef_subtask_trajectory()负责把上一步生成的子任务轨迹并入整体可执行轨迹:若generation_interpolate_from_last_target_pose=True且非首个子任务,则从上一段轨迹最后一个航点开始插值(该航点未必已到达);否则从当前机器人末端位姿开始。随后调用WaypointTrajectory.merge()执行num_interpolation_steps步插值与num_fixed_steps步保持,噪声为apply_noise_during_interpolation * action_noise;最后pop_first()丢弃为插值临时加入的起点。
6.5 generate():整段演示的异步主循环
generate()(第 628 行起,标注noqa: C901)是数据生成的最外层入口,返回字典含initial_state、states、observations、datagen_infos、actions、success、src_demo_inds等字段。其主流程:
- skillgen 前置校验:
use_skillgen=True时motion_planner必须提供,否则抛ValueError; - 重置环境:通过
recorder_manager.reset()与env_reset_queue异步配合,记录新初始状态; - 展开
task_constraint_configs为运行时约束字典; - 进入逐时间步 while 循环:在
asyncio.Lock保护下检查源演示池是否有新增(有则重新随机化边界),为处于子任务起点的末端执行器生成/合并轨迹; - skillgen 路径:调用
motion_planner.update_world_and_plan_motion()规划子任务间的无碰撞过渡轨迹(目标为下一子任务首航点位姿与夹爪动作),规划成功则先执行规划轨迹、暂存原子任务轨迹,失败则返回{"success": False}中止本次尝试;规划轨迹通过_convert_planned_trajectory_to_waypoints()转成Waypoint列表(噪声取自motion_planner.config.motion_noise_scale); - 非 skillgen 路径:直接用插值合并过渡;
- 按约束处理每臂每步的航点:
_SEQUENTIAL_LATTER约束在满足min_time_diff前停在原地等待前置子任务完成;COORDINATION约束计算synchronous_steps(两条并发子任务长度的最小值)并支持coordination_synchronize_start同步启动,保证并发任务同时结束; - 组装
MultiWaypoint并execute(),用success_term判定成功; - 支持
use_navigation_controller的移动操作模式(要求存在名为"body"的末端执行器配置):导航未达目标时重复最后一步动作,提前到达则跳过剩余导航轨迹; - 全部末端执行器完成所有子任务后,通过
recorder_manager.set_success_to_episodes()写成功标志并按export_demo导出。
七、配置体系:DataGenConfig、SubTaskConfig 与约束
配置类集中在 mimic_env_cfg.py(MimicEnvCfg通过datagen_config、subtask_configs、task_constraint_configs、mimic_recorder_config四个字段把生成逻辑接入任务环境)。
DataGenConfig关键字段(默认值以仓库源码为准):
| 字段 | 默认值 | 说明 |
|---|---|---|
name | "demo" | 生成过程名称 |
generation_guarantee | True | 是否重试直到产出generation_num_trials条成功演示 |
generation_keep_failed | False | 是否保留失败试次(调试低成功率时有用) |
max_num_failures | 50 | 停止生成前允许的最大失败次数 |
seed | 1 | 随机种子,保证可复现 |
source_dataset_path/generation_path | None | 源数据集路径与生成结果保存路径 |
generation_num_trials | 10 | 生成试次数量 |
generation_select_src_per_subtask | False | 是否每个子任务独立选源(要求子任务间有合理时间约束) |
generation_select_src_per_arm | False | 是否每只机械臂独立选源 |
generation_transform_first_robot_pose | False | 是否变换首段机器人位姿 |
generation_interpolate_from_last_target_pose | True | 是否从上个目标位姿开始插值 |
use_skillgen | False | 是否使用 skillgen(运动规划器生成过渡轨迹) |
use_navigation_controller | False | 是否使用导航控制器生成移动操作轨迹 |
SubTaskConfig关键字段:object_ref(参考物体,可 None)、subtask_term_signal(终止信号名)、selection_strategy(random/nearest_neighbor_object/nearest_neighbor_robot_distance,后两者要求object_ref非 None,通常成功率更高)、selection_strategy_kwargs、first_subtask_start_offset_range、subtask_start_offset_range(仅 skillgen 使用,且会覆盖前者的语义)、subtask_term_offset_range、action_noise(默认 0.03)、num_interpolation_steps(默认 5)、num_fixed_steps(默认 0)、apply_noise_during_interpolation(默认 False)。
SubTaskConstraintConfig支持两种约束:SEQUENTIAL(顺序约束,把约束展开为_SEQUENTIAL_FORMER/_SEQUENTIAL_LATTER一对运行时条目,通过fulfilled标志让后者等待前者完成,sequential_min_time_diff=-1表示后者等前者完全结束才启动)与COORDINATION(并发协调,通过共享selected_src_demo_ind与transform保证两条并发子任务使用同一源片段与同一变换,synchronous_steps取两段长度最小值实现同时收尾)。协调方案枚举SubTaskConstraintCoordinationScheme含REPLAY/TRANSFORM/TRANSLATE,与 6.3 节的变换逻辑一一对应。
实战配置示例可参考 franka_stack_ik_rel_mimic_env_cfg.py:FrankaCubeStackIKRelMimicEnvCfg继承任务环境与MimicEnvCfg,在__post_init__中覆盖datagen_config(如generation_num_trials=10、generation_select_src_per_subtask=True、max_num_failures=25),并为抓取红方块、堆叠、抓取绿方块、堆叠四个子任务各配置一个SubTaskConfig——每个子任务给出object_ref(cube_1/2/3)、subtask_term_signal(grasp_1/stack_1/grasp_2)、subtask_term_offset_range=(10, 20)、selection_strategy="nearest_neighbor_object"、selection_strategy_kwargs={"nn_k": 3}等参数;最后一个子任务subtask_term_signal=None、偏移 (0,0),与DataGenerator构造器的健全性检查要求一致。
八、实战运行:generate_dataset.py 与异步生成框架
命令行入口 generate_dataset.py 定义了完整参数集:
python scripts/imitation_learning/isaaclab_mimic/generate_dataset.py \ --task <TaskName> \ --num_envs 4 \ --input_file /path/to/source_dataset.hdf5 \ --output_file ./datasets/output_dataset.hdf5 \ --generation_num_trials 10 \ [--use_skillgen] [--pause_subtask] [--disable_dataset_compression] [--external_callback path.to.fn]参数说明:--task指定任务名(缺省时从输入数据集元数据自动读取环境名,见get_env_name_from_dataset());--input_file为源数据集路径(必需);--output_file默认./datasets/output_dataset.hdf5;--use_skillgen启用 CuRobo 运动规划器(每个环境一个CuroboPlanner,仅第一个环境开启可视化以控制开销);--external_callback可注入自定义环境注册回调。脚本在启动时依据datagen_config.seed统一设置random/np.random/torch三套随机种子。
生成过程复用 generation.py 的异步框架:
setup_env_config():解析任务配置、提取terminations.success作为success_term(缺失则抛NotImplementedError),配置录制器输出路径/文件名/压缩,并按generation_keep_failed决定导出模式(EXPORT_SUCCEEDED_FAILED_IN_SEPARATE_FILES或EXPORT_SUCCEEDED_ONLY);setup_async_generation():创建共享DataGenInfoPool并载入源数据集,为每个环境创建独立的run_data_generator异步任务;env_loop():主环境步进循环,在torch.inference_mode()下消费动作队列、处理重置队列,并实时打印生成统计(num_success/num_attempts与成功率百分比),按generation_guarantee决定以成功数或尝试数作为终止条件;同时监听数据生成任务是否异常退出(data_gen_tasks.done()时重抛异常)。
九、测试与验证
该模块的测试集中在 source/isaaclab_mimic/test/:
test_selection_strategy.py:如第四节所述,验证两种最近邻策略在两组位姿簇下的选择正确性与随机性;test_generate_dataset_franka_state.py、test_generate_dataset_franka_visuomotor.py、test_generate_dataset_gr1t2_nutpour.py、test_generate_dataset_gr1t2_pickplace.py:端到端验证generate_dataset.py在不同任务上的数据集生成流程;test_generate_dataset_skillgen.py与test_curobo_planner_*.py:验证 skillgen 路径下运动规划器与数据生成器的集成。
这些测试表明datagen模块与ManagerBasedRLMimicEnv、录制器、CuRobo 规划器之间存在完整的可运行闭环,可作为二次开发与回归验证的参考入口。
十、总结
isaaclab_mimic.datagen以"物体坐标系下的子任务片段复用"为设计主线,把数据生成拆解为信息采集(DatagenInfo/DataGenInfoPool)、源片段选择(SelectionStrategy)、位姿变换与轨迹装配(DataGenerator+Waypoint类族)、异步批量执行(generation.py)四个层次,并通过DataGenConfig/SubTaskConfig/SubTaskConstraintConfig实现任务的参数化配置。无论你是想基于现有任务直接运行generate_dataset.py扩充数据集,还是希望自定义选择策略、加入新任务约束或接入运动规划器,上述源码路径与配置入口都能作为可靠的开发起点。
【免费下载链接】IsaacLabUnified framework for robot learning built on NVIDIA Isaac Sim项目地址: https://gitcode.com/GitHub_Trending/is/IsaacLab
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考