用 Rerun Chunk Processing API 构建机器人数据预处理管线：多源数据合并、记录修复与 URDF 正向运动学

用 Rerun Chunk Processing API 构建机器人数据预处理管线多源数据合并、记录修复与 URDF 正向运动学【免费下载链接】rerunVisualize, query, and stream to train on multimodal robotics data.项目地址: https://gitcode.com/GitHub_Trending/re/rerun本指南以examples/python/robot_data_preprocessing示例为蓝本系统讲解如何用 Rerun 的 Chunk Processing API 将散落在多个文件、多种格式中的机器人数据MCAP 主录制、JSON 静态偏移、URDF 模型组装为一份连贯的可视化录制并在组装过程中完成错误修复、数据增强与正向运动学计算。读完本文你将掌握LazyChunkStream管线的构建方法、MutateLens/DeriveLens两类透镜的实战用法、Selector的列式查询语法以及如何利用 recording ID 把多个 RRD 文件组织成单一逻辑录制。场景概述一个真实的机器人数据整理难题双机械臂工作台的数据分散在三类文件中每一类都有各自的毛病episode.mcapoffsets.jsonURDF 文件基础录制视频、传感器等每台机器人的世界系静态偏移机器人及场景模型URDF 格式部分相机内参错误未录制动态 3D 变换只有自定义 Protobuf 模式编码的关节状态保存在基础录制之外robot.urdf、scene.urdf及网格数据示例刻意使用模拟数据但涵盖了机器人学家工作中常见的四类现实挑战需要预处理的不完整数据、自定义数据类型、录制数据中的 bug、分布在多个不同格式文件中的数据。目标是把这些数据源统一处理并合并成一份连贯的录制读取、转换并修复 MCAP 数据使用 MCAP 中的关节状态和 URDF 计算 3D 变换正向运动学处理 URDF加入scene.urdf和 2 份robot.urdf并按机器人分别修改视觉网格的颜色与透明度从 JSON 添加静态变换。所有输入文件都位于 examples/python/robot_data_preprocessing/input_data 目录下完整实现见 robot_data_preprocessing.py。管线骨架Chunk Stream 作为胶水核心概念Chunk 是 Rerun 的核心数据结构而 Chunk Processing API概念文档提供以 chunk 为中心的数据摄取、转换与转换管线。示例用chunk streams作为整个管线的胶水LazyChunkStream允许我们声明式地定义 Chunk 如何经过过滤、变换和输出步骤被路由顾名思义这些流是惰性求值的——构建 DAG 只是元数据操作与输入规模无关我们使用富有表达力的 Python API 定义管线但最终执行发生在一个多线程、无 GIL 的 Rust 执行引擎中以获得最高效率内存开销由单次流过管线的 chunk 决定而不是由录制总大小决定因此可以处理超大数据集。Python 侧的LazyChunkStream定义在 rerun_py/rerun_sdk/rerun/chunk/_lazy_chunk_stream.py它支持结构化过滤filter/drop、分支split、汇聚merge、重塑lenses以及任意逐 chunk 操作map/flat_map。一个重要语义是move 语义stream.filter(...)、stream.lenses(...)、stream.split(...)、LazyChunkStream.merge(...)都会消耗输入流重复使用已被消耗的流会抛出ValueError而to_chunks()、collect()、write_rrd()等终端方法不会消耗流可对同一流执行多次代价是重复执行整条管线。三类数据源本例使用了三个可以产生LazyChunkStream的源McapReader.stream()——读取 MCAP 录制。它复用 Rerun 的 MCAP importer与查看器或rerun mcap convertCLI 相同因此我们拿到的是可以直接在流中处理的 Rerun 组件UrdfTree.stream()——产生 URDF 模型的 chunk 流手工构造的LazyChunkStream——用Chunk.from_columns(…)读取自定义 JSON 文件中的静态变换。第一步读取 MCAP 并修复相机内参示例 MCAP 中外部相机的标定存在一个典型 bugPinhole:resolution分辨率的宽高被写反了。我们用MutateLens修复mcap_stream.lenses( MutateLens( Pinhole:resolution, Selector(.).pipe( lambda resolution: pa.array( [(height, width) for width, height in resolution.to_pylist()], typeresolution.type ) ), ), content[/external/cam_low, /external/cam_high], output_modeforward_unmatched, )关键参数content过滤器确保该透镜只作用于外部相机实体output_modeforward_unmatched保证不匹配的其他 pinhole 实体这里指无需修复的机器人相机原样转发。MutateLens在 rerun_py/rerun_sdk/rerun/chunk/_lens.py 中定义它就地修改输入组件input_component用selector的变换结果替换 chunk 中的该组件默认生成新的 row ID可传keep_row_idsTrue保留原始 row ID。lenses()方法的output_mode共有三种取值见_lazy_chunk_stream.py源码注释forward_all同时转发变换后数据与原始数据forward_unmatched匹配则转发变换结果不匹配则转发原始数据默认drop_unmatched只转发变换后的数据。此外content是可选的实体路径过滤器一旦设置透镜只作用于实体路径匹配的 chunk不匹配的 chunk 无论output_mode如何都原样通过。第二步用 URDF 计算正向运动学这是本例最复杂的透镜组合目标是从关节值角度、距离计算 3D 变换。这需要两方面的输入MCAP 中录制的关节状态以及描述运动学结构的 URDF。自定义 Protobuf 模式MCAP 中的关节状态使用自定义 Protobuf 模式编码message JointState { google.protobuf.Timestamp timestamp 1; repeated string joint_names 2; repeated double joint_positions 3; repeated double joint_velocities 4; repeated double joint_efforts 5; }该自定义模式并不属于 MCAP importer 直接支持的消息类型如视频流。但借助模式反射schema reflection我们依然能拿到带有可查询 Rerun 组件的 chunk从而在流中处理。导入后其组件名形如schemas.proto.JointState:messagejoint_names/joint_positions等字段可通过Selector访问。两段式 DeriveLens每一条关节状态输入行包含N个关节值对应N个 3D 变换而每个变换需要一条独立的输出行每条带一个Transform3D。由于输入与输出的行数不匹配示例使用两个顺序执行的透镜透镜 1批量计算变换joints_batch_lens对每条关节状态消息选出关节名与关节值调用UrdfTree.compute_joint_transform_batches定义于 rerun_py/rerun_sdk/rerun/urdf.py要求names为ListArrayUtf8、values为可转成Float64的ListArray输出是含translation、quaternion、parent_frame、child_frame字段的ListArray外层行数与输入一致输出单行、含N个 3D 变换的列表。def joints_batch_lens(robot_urdf: UrdfTree, to_entity: str /tmp) - DeriveLens: return DeriveLens(schemas.proto.JointState:message, output_entityto_entity).to_component( rerun.urdf.JointTransformBatch, Selector(.).pipe( lambda joint_state_messages: robot_urdf.compute_joint_transform_batches( namesSelector(.joint_names).execute(joint_state_messages), valuesSelector(.joint_positions).execute(joint_state_messages), ) ), )透镜 2展开为逐关节 Transform3Doutput_transforms_lens把每个计算出的行展开为N行每行携带Transform3D组件列def output_transforms_lens() - DeriveLens: return ( DeriveLens(rerun.urdf.JointTransformBatch, output_entity/tf, scatterTrue) .to_component(rr.Transform3D.descriptor_translation(), Selector(.[].translation)) .to_component(rr.Transform3D.descriptor_quaternion(), Selector(.[].quaternion)) .to_component(rr.Transform3D.descriptor_parent_frame(), Selector(.[].parent_frame)) .to_component(rr.Transform3D.descriptor_child_frame(), Selector(.[].child_frame)) )其中scatterTrue启用1:N 行映射将列表炸开成多行Selector(.[].translation)使用[]运算符遍历列表的每个元素。DeriveLens的完整能力见 rerun_py/rerun_sdk/rerun/chunk/_lens.py从输入组件派生全新组件to_component或时间列to_timeline通过output_entity把输出写到不同实体cast_toauto可自动把列转型为组件的规范 Arrow 类型例如 parquet 常见的f64列转成Transform3D:translation期望的f32内置便捷方法to_translation、to_quaternion、to_scale、to_scalars、to_rotation_axis_angle等。在主函数中两套透镜按左右机器人分别作用并使用不同output_modemcap_stream ( mcap_stream .lenses(joints_batch_lens(robot_urdf_left), content/robot_left/joint_states, output_modeforward_all) .lenses(output_transforms_lens(), content/tmp, output_modedrop_unmatched) .lenses(joints_batch_lens(robot_urdf_right), content/robot_right/joint_states, output_modeforward_all) .lenses(output_transforms_lens(), content/tmp, output_modedrop_unmatched) )第一阶段forward_all保留原始关节状态的同时追加中间批量值第二阶段drop_unmatched丢弃临时批量值/tmp下的rerun.urdf.JointTransformBatch只留下最终的Transform3D避免污染录制。content的作用在此体现得淋漓尽致两个机器人共用同一个组件名schemas.proto.JointState:message正是靠content/robot_left/joint_states与content/robot_right/joint_states的实体路径作用域来区分左右两侧的输入。第三步处理 URDF——加载、着色与裁剪加载同一 URDF 两次左右两台机器人使用同一个robot.urdf但需要不同的实体路径与帧名前缀因此UrdfTree.from_file_path被调用两次其签名见 rerun_py/rerun_sdk/rerun/urdf.pyrobot_urdf_left UrdfTree.from_file_path( DATA_DIR / robot.urdf, entity_path_prefixrobot_left, frame_prefixleft_, static_transform_entity_path/tf_static/left_robot, ) robot_urdf_right UrdfTree.from_file_path( DATA_DIR / robot.urdf, entity_path_prefixrobot_right, frame_prefixright_, static_transform_entity_path/tf_static/right_robot, ) scene_urdf UrdfTree.from_file_path(DATA_DIR / scene.urdf, static_transform_entity_path/tf_static/scene)参数说明entity_path_prefix给该 URDF 产生的所有实体路径加前缀保证左右两套模型不冲突frame_prefix给所有 frame ID 加前缀用于多机器人场景下生成唯一的帧名static_transform_entity_path静态变换如 URDF 中的固定关节写入的实体路径默认/tf_static不受entity_path_prefix影响。用 MutateLens 修改视觉网格颜色与透明度默认的robot_urdf.stream()输出的网格外观是统一的示例用MutateLens修改Asset3D:albedo_factor反照率因子可表达颜色与透明度让左右机器人拥有不同的外观def change_albedo_factor_lens(new_albedo: rr.components.AlbedoFactor) - MutateLens: return MutateLens( Asset3D:albedo_factor, Selector(.).pipe(lambda old_albedo: pa.array([new_albedo] * len(old_albedo), typeold_albedo.type)), ) robot_urdf_left_stream robot_urdf_left.stream().lenses( change_albedo_factor_lens(rr.components.AlbedoFactor([80, 120, 175, 125])), content/robot_left/wxai/visual_geometries/**, output_modeforward_unmatched, ) robot_urdf_right_stream robot_urdf_right.stream().lenses( change_albedo_factor_lens(rr.components.AlbedoFactor([200, 120, 90, 125])), content/robot_right/wxai/visual_geometries/**, output_modeforward_unmatched, )AlbedoFactor([r, g, b, a])的第四个通道即透明度这里125让网格呈现半透明效果content使用/**通配符匹配该机器人全部视觉几何实体forward_unmatched让场景中其他不匹配实体原样通过。丢弃碰撞网格UrdfTree.stream()同时会产生视觉visual_geometries与碰撞collision_geometries网格。碰撞网格在查看器中本可关闭这里演示如何在管线中直接丢弃robot_urdf_left_stream robot_urdf_left_stream.drop(content/robot_left/wxai/collision_geometries/**) robot_urdf_right_stream robot_urdf_right_stream.drop(content/robot_right/wxai/collision_geometries/**)drop是filter的补集filter保留匹配部分、丢弃其余drop丢弃匹配部分、保留其余。两者都支持按content、has_timeline、is_static、components组合条件过滤。第四步从 JSON 添加静态变换机器人相对世界的偏移保存在独立的offsets.json中见 input_data/offsets.json内容是transforms数组每个元素包含parent、child、translation与quaternion_xyzw{ transforms: [ { parent: world, child: left_base_link, translation: [-0.4575, -0.019, 0.02], quaternion_xyzw: [0.0, 0.0, 0.0, 1.0] }, { parent: world, child: right_base_link, translation: [0.4575, -0.019, 0.02], quaternion_xyzw: [0.0, 0.0, -1.0, 0.0] } ] }通过Chunk.from_columns手工构造一个静态 chunk再用LazyChunkStream.from_iter提升为单元素流def json_transforms_stream(json_path: Path) - LazyChunkStream: with json_path.open() as f: transforms json.load(f)[transforms] chunk Chunk.from_columns( /tf_static/robot_offsets, indexes[], columnsrr.Transform3D.columns( translation[transform[translation] for transform in transforms], quaternion[transform[quaternion_xyzw] for transform in transforms], parent_frame[transform[parent] for transform in transforms], child_frame[transform[child] for transform in transforms], ), ) return LazyChunkStream.from_iter([chunk])要点indexes[]表示这是一个静态 chunk无时间列rr.Transform3D.columns(...)与rr.send_columns使用同一套.columns(...)辅助方法因此任何能通过rr.send_columns记录的数据都可以打包成Chunk注入处理管线更多细节见 ChunksChunk.from_record_batch与Chunk.from_dataframe则对应rr.send_record_batch与rr.send_dataframe。第五步合并流并写出两个 RRD按逻辑分组合并LazyChunkStream.merge是扇入操作多个输入流汇成一个流。所有输入并发执行chunk 一就绪即产出每个输入内部保持顺序跨输入之间顺序不确定。示例把流分成两组data_stream LazyChunkStream.merge(mcap_stream, robot_offsets_stream) urdf_stream LazyChunkStream.merge( robot_urdf_left_stream, robot_urdf_right_stream, scene_urdf.stream(), )这样基础录制 静态偏移与URDF 模型各自独立成层如果想合并成一个流也是可行的。物化、优化并写出collect(optimize...)是终端方法运行管线并把所有 chunk 物化到内存中的ChunkStore。OptimizationProfile见 rerun_py/rerun_sdk/rerun/chunk/_optimization_profile.py控制物化后的额外优化OptimizationProfile.LIVE面向实时 Viewer 工作流的小 chunk 调优OptimizationProfile.OBJECT_STORE面向对象存储查询与流式应用的大 chunk 调优。本例为查询与流式场景选用OBJECT_STORE配置然后写出两个 RRDdata_stream.collect(optimizeOptimizationProfile.OBJECT_STORE).write_rrd( OUTPUT_DIR / data.rrd, application_idrerun_example_robot_data_preprocessing, recording_idepisode, ) urdf_stream.collect(optimizeOptimizationProfile.OBJECT_STORE).write_rrd( OUTPUT_DIR / urdf.rrd, application_idrerun_example_robot_data_preprocessing, recording_idepisode, )逻辑录制 vs 物理录制示例刻意生成两个物理 RRD 文件但由于二者指定了相同的recording_idepisode与application_id在 Rerun Viewer 或 Catalog 中它们共同构成一个逻辑录制logical recording参见 Recordings 概念两个 RRD 作为同一逻辑录制的两个层layers出现。这种设计便于后续继续追加更多层例如元数据或额外传感器数据。运行示例安装与执行项目使用hatchling构建见 pyproject.toml依赖仅pyarrow与rerun-sdk并注册了robot_data_preprocessing控制台脚本pip install -e examples/python/robot_data_preprocessing python -m robot_data_preprocessing两条命令效果等价后者的入口在pyproject.toml的[project.scripts]中声明。运行后会在examples/python/robot_data_preprocessing/output/下生成data.rrd与urdf.rrd。查看结果rerun examples/python/robot_data_preprocessing/output/*.rrd由于两个文件使用一致的 recording ID两条 RRD 层在查看器中显示为单一录制。关键 API 参考API作用源码位置LazyChunkStream惰性、可组合的 chunk 管线filter/drop/split/merge/lenses/map/flat_maprerun_py/rerun_sdk/rerun/chunk/_lazy_chunk_stream.pyMcapReader从 MCAP 文件读取 chunk支持时间范围、topic 正则过滤、recover恢复损坏摘要等rerun_py/rerun_sdk/rerun/chunk/_mcap_reader.pyUrdfTree加载 URDF 模型stream()输出模型 chunkcompute_joint_transform_batches批量计算关节变换rerun_py/rerun_sdk/rerun/urdf.pyMutateLens就地修改 chunk 中的输入组件rerun_py/rerun_sdk/rerun/chunk/_lens.pyDeriveLens从输入组件派生新组件/时间列支持scatter1:N 展开rerun_py/rerun_sdk/rerun/chunk/_lens.pySelectorjq 风格的 Arrow 数组查询语言支持.field、[]、[N]、?、!、\|与.pipe()rerun_py/rerun_sdk/rerun/chunk/_selector.pyOptimizationProfile物化后的 chunk 优化配置LIVE/OBJECT_STORErerun_py/rerun_sdk/rerun/chunk/_optimization_profile.pySelector的语法要点jq 风格、面向 Arrow 数组按列操作.field访问 struct 的命名字段[]遍历列表每个元素[N]按位置索引?错误抑制!断言非空|管道.pipe(func)把选择器结果接入 Python 变换函数或另一个选择器。总结与延伸本例展示了如何通过结构化数据管线优雅地解决一个非平凡的机器人学问题用不到 200 行 Python 代码把 Chunk Processing API 的读者、透镜、选择器组合成紧凑的自定义管线同时让强大的多线程执行引擎在底层负责性能。它同时演示了 recording ID 如何把 RRD 结构化为逻辑录制为后续追加元数据或额外传感器数据层留出空间。在实际生产环境中这类处理通常只是数据整理data curation的第一步——在把多个原始录制最终送入中心存储前完成定型。在 Rerun 中这意味着把数据集注册到 catalog server企业级扩展可借助 Rerun Hub小规模本地开发可使用开源的rerun server从而支持跨录制查询分析或导出训练数据参见 dataframe queries。继续深入可阅读仓库内的相关文档Chunk Processing API、Lenses、Recordings、MCAP 使用指南、URDF 模型加载指南以及本示例的完整源码 robot_data_preprocessing.py。【免费下载链接】rerunVisualize, query, and stream to train on multimodal robotics data.项目地址: https://gitcode.com/GitHub_Trending/re/rerun创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考