1. 项目概述为什么“把节点数据录进包”是ROS2开发绕不开的第一课在ROS2实际开发中我见过太多人卡在同一个地方写好了发布者节点也搭好了订阅者逻辑但一到调试阶段就抓瞎——消息到底发没发时间戳对不对丢包发生在哪一环频率是不是真稳定在50Hz靠ros2 topic echo看几眼根本没法回溯、没法定量分析更别说给同事复现问题或者做算法离线训练了。这时候“把节点数据录进一个包”就不是锦上添花的功能而是你从“能跑通”迈向“可验证、可复现、可交付”的分水岭。这个操作在ROS2里叫bag recording生成的文件后缀是.db3SQLite3数据库格式它本质上是一个带时间戳、带元数据、支持多话题并发、可精确回放的“数据黑匣子”。它不依赖你的节点是否还在运行也不关心你用的是C还是Python写的节点只要消息符合ROS2 IDL定义就能原样存下来。对新手来说这是你第一次真正触碰到ROS2底层数据流的“实体化”形态对老手而言它是构建CI/CD流水线、做性能压测、做传感器标定的基础数据载体。关键词——ROS2入门教程、ros2 bag record、节点数据记录、bag文件、离线回放、调试诊断——全围绕一个核心动作让流动的数据停下来变成你随时可以打开、分析、共享的“数字快照”。这事儿听起来简单执行起来却藏着不少坑。比如你用ros2 bag record /topic_name录下来的数据回放时发现时间戳全乱了又或者你明明启动了两个发布者但bag里只看到一个话题再比如录完一个2GB的bag想用Python脚本抽其中某段10秒的数据结果发现官方API文档写得像天书。这些都不是玄学而是ROS2设计哲学和底层机制共同作用的结果它默认按系统时钟ROS_TIME打戳但如果你的节点用了rclcpp::Clock::get_now()而没显式设置时钟类型时间戳就可能混用它默认只录活跃话题但有些话题是latched或on-demand的需要手动指定它的bag文件结构是分块存储索引优化的直接用SQLite工具打开只会看到一堆二进制blob。所以这篇教程不只教你怎么敲命令更要带你搞懂为什么必须加-a参数才能录全系统话题为什么--include-hidden-topics不是万能钥匙为什么--duration和--start-delay的组合能救命这些细节决定了你录出来的是一份可用的工程资产还是一堆无法解析的废文件。2. 核心原理与设计思路ROS2 Bag不是“录像机”而是一套精密的数据管道2.1 ROS2 Bag的本质一个基于SQLite3的实时流式写入系统很多人下意识把ros2 bag record当成Linux下的tcpdump——抓什么存什么。但ROS2 Bag的设计远比这复杂。它不是一个简单的“拷贝内存数据到磁盘”的过程而是一整套带缓冲、带序列化、带索引、带压缩选项的数据管道。当你执行ros2 bag record -a时背后发生的是话题发现与订阅建立ros2bag进程会向ROS2图ROS graph发起一次全局扫描获取当前所有活跃的、非隐藏的话题列表默认不包含以/parameter_events、/clock等开头的系统话题。然后它为每个目标话题创建一个独立的rclcpp::Subscription对象并绑定一个专用回调函数。消息接收与时间戳归一化每个回调收到消息后第一件事不是写磁盘而是检查该消息是否自带有效时间戳如std_msgs::msg::Header::stamp。如果没有ros2bag会自动注入当前rclcpp::Clock::now()的时间戳。这里的关键是它强制使用RCL_ROS_TIME时钟类型确保所有消息的时间基准统一避免不同节点因系统时钟漂移导致回放错乱。序列化与缓冲写入消息被序列化为cdrCommon Data Representation二进制格式而非JSON或YAML。这是因为cdr是ROS2 IDL的原生序列化协议零拷贝、无解析开销、跨语言兼容。序列化后的数据块先写入内存环形缓冲区默认大小为10MB当缓冲区满或达到--max-cache-size阈值时才批量刷入SQLite3数据库文件。这种设计极大降低了I/O压力实测在Jetson Orin上连续录制10个高频话题总带宽20MB/s时磁盘IO占用率稳定在35%以下而纯直写模式会瞬间飙到95%并导致丢包。索引构建与元数据写入每写入一个数据块ros2bag会同步更新数据库中的messages表存消息体、topics表存话题名、类型、QoS配置、schema表存IDL定义的MD5哈希。最关键的是chunk表——它记录每个数据块的起始时间、结束时间、消息数量、压缩方式默认无压缩可选zstd。正是这个chunk索引让你能用ros2 bag info秒级查出整个bag的时长、话题分布、最大消息延迟而不是像读普通日志一样逐行扫描。提示这就是为什么ros2 bag record命令本身不提供“实时预览”功能——它压根不维护消息内容的内存副本所有数据都在缓冲区和数据库之间流转。你想看实时数据老老实实用ros2 topic echo你想存可靠数据交给ros2 bag。2.2 为什么不能只录“看到的”话题——理解ROS2的“话题可见性”机制新手常犯的错误是ros2 topic list能看到/lidar_points就以为ros2 bag record /lidar_points一定能录到数据。结果回放时发现空空如也。根源在于ROS2的话题发现discovery机制是异步且有生命周期的。ROS2节点启动后会通过DDSData Distribution Service的“participant discovery”协议广播自己的存在并订阅其他节点的广播。这个过程需要几十到几百毫秒。而ros2 bag record进程启动时如果目标节点还没完成发现它就“看不见”那个话题。更麻烦的是有些话题是latched粘性的比如/tf_static它只在首次订阅时发送一次快照还有些是on-demand按需的比如某些诊断话题只有当有订阅者连接时才开始发布。ros2 bag record默认的“发现-订阅”流程很可能错过这些瞬态窗口。解决方案有两个层级战术层用--start-delay 5参数让ros2 bag进程启动后等待5秒再开始订阅给系统留出充分的发现时间。我实测在复杂网络环境下3秒成功率85%5秒升至99.2%。战略层用--include-hidden-topics强制包含所有以_开头的隐藏话题如/parameter_events再配合-aall topics确保不漏掉任何已注册话题。但注意--include-hidden-topics不会帮你“唤醒”那些尚未发布的on-demand话题它只是扩大发现范围。注意ros2 bag record -a并不等于“录下整个ROS2世界”。它只录当前DDS域内、已成功完成participant discovery、且QoS策略兼容主要是可靠性Reliability和历史深度History depth的话题。如果你的发布者设了BEST_EFFORT而订阅者要求RELIABLE它们根本连不上自然也录不到。2.3 Bag文件结构解析揭开.db3文件的神秘面纱一个典型的ROS2 bag文件如my_session_000.db3其实就是一个标准SQLite3数据库。你可以用任何SQLite浏览器如DB Browser for SQLite直接打开它看到6张核心表表名作用关键字段说明messages存储所有原始消息数据topic_id(外键),timestamp(纳秒级),data(BLOB二进制)topics存储话题元数据name(话题名),type(消息类型, 如sensor_msgs/msg/PointCloud2),serialization_format(固定为cdr)schema存储消息类型的IDL定义name(类型名),encoding(编码方式),data(IDL文本的SHA256哈希)chunk存储数据块索引start_timestamp,end_timestamp,message_count,compression(压缩算法)chunk_info存储每个chunk的详细信息chunk_id,message_count,uncompressed_size,compressed_sizeattachments存储附加文件如标定文件name,content_type,data(BLOB)重点看messages表它的data字段是纯二进制直接用SQLite工具打开就是乱码。这不是bug而是设计使然——ROS2 Bag必须保证序列化/反序列化的100%保真任何文本化转换都会破坏cdr的字节对齐。所以永远不要试图用文本编辑器或cat命令查看.db3文件内容。正确的做法是用ros2 bag play回放或用ros2 bag info查元数据或用Python API读取后面详述。我曾见过有同事为了“快速检查”把.db3拖进VS Code结果编辑器卡死强行退出后文件损坏2小时录制功亏一篑。记住.db3是数据库不是日志对待它要像对待生产数据库一样敬畏。3. 实操全流程从零开始录制、验证、切片、回放一个完整Bag3.1 环境准备与基础命令验证在动手前请确认你的ROS2环境已正确初始化。我以Humble HawksbillLTS版本为例这是目前工业界最稳定的发行版。打开终端执行source /opt/ros/humble/setup.bash # 验证ROS2核心服务是否正常 ros2 node list # 应返回 /_ros2cli_* 类似节点 ros2 topic list # 应返回 /parameter_events, /rosout 等基础话题如果报错command not found: ros2说明环境变量没加载。常见原因有两个一是你安装的是二进制包但没执行source二是你用apt install ros-humble-desktop安装后setup.bash路径写错了正确路径是/opt/ros/humble/setup.bash不是/opt/ros/humble/env.sh。别急着重装先检查ls -la /opt/ros/humble/setup.* # 看看有哪些setup文件 echo $ROS_DISTRO # 应输出 humble确认环境OK后我们先启动一个最简单的测试节点——talker它会以1Hz频率发布std_msgs/String消息# 终端1启动talker ros2 run demo_nodes_cpp talker # 终端2观察消息此时应看到 Hello World: 1, Hello World: 2 ... ros2 topic echo /chatter现在我们用最基础的命令录下这个话题# 终端3录制/chatter话题保存到当前目录的my_bag文件夹 ros2 bag record /chatter -o my_bag你会看到类似输出[INFO] [1715234567.890123456] [rosbag2_storage]: Opened database my_bag/my_bag_000.db3. [INFO] [1715234567.890123456] [rosbag2_transport]: Listening for topics... [INFO] [1715234567.890123456] [rosbag2_transport]: Subscribed to topic /chatter [INFO] [1715234567.890123456] [rosbag2_transport]: Recording 1 topics按CtrlC停止录制。此时my_bag/目录下会生成my_bag_000.db3主数据库文件metadata.yaml人类可读的元数据包含ROS2版本、录制时间、话题列表等实操心得永远用-o folder_name指定输出目录不要用默认的rosbag2_YYYY_MM_DD-HH_MM_SS命名。因为默认名含时间戳你在写自动化脚本时很难预测路径。我习惯用语义化命名如lidar_calib_20240510_run1方便后续归档和检索。3.2 录制策略详解何时用-a何时用-e何时必须加--start-delay真实场景远比/chatter复杂。假设你正在调试一个移动机器人它同时发布/scan激光雷达、/odom里程计、/tf坐标变换、/diagnostics诊断信息。你需要一份完整的系统快照。这时ros2 bag record的参数组合就至关重要。场景1录全所有活跃话题推荐新手起步用ros2 bag record -a -o full_robot_log --start-delay 5-a录制所有当前活跃话题不含隐藏话题--start-delay 5启动后等待5秒再开始订阅确保所有节点完成DDS发现优点简单粗暴覆盖90%调试场景缺点会录下大量无关的系统话题如/parameter_events增大bag体积场景2精准录制指定话题推荐CI/CD或资源受限时用ros2 bag record /scan /odom /tf /diagnostics -o precise_log --compression-mode file --compression-format zstd显式列出所有目标话题避免冗余--compression-mode file对整个.db3文件启用压缩而非单个chunk节省30%-50%空间--compression-format zstdZstandard算法压缩比和速度平衡最佳比bz2快3倍比lz4压缩率高15%优点bag体积小回放快目标明确缺点必须提前知道所有话题名漏一个就少一维数据场景3录制隐藏话题如/tf_static或自定义诊断话题ros2 bag record -a --include-hidden-topics -o with_hidden_log--include-hidden-topics强制包含所有以_开头的话题如/tf_static,/parameter_events注意它不会帮你“激活”那些on-demand话题只是扩大发现范围典型用途/tf_static是静态坐标系只在首次订阅时发一次必须录下来才能做完整TF树回放常见误区纠正很多人以为--include-hidden-topics能录到/clock话题。其实不能。/clock是ROS2内部时钟话题由rclcpp::Clock自动管理不走标准发布/订阅流程ros2 bag根本订阅不到。如果你需要记录仿真时间应该在Gazebo或Ignition仿真器中启用use_sim_time:true然后录/clock以外的业务话题它们的时间戳会自动对齐仿真时钟。3.3 录制后必做的三件事验证、切片、备份录完一个bag千万别直接关机必须立即执行以下三步验证否则可能白干步骤1用ros2 bag info快速体检ros2 bag info my_bag/输出示例Files: my_bag_000.db3 Bag size: 1.2 MiB Storage id: sqlite3 Duration: 10.0s Start: May 10 2024 14:23:45.123 (1715235825.123) End: May 10 2024 14:23:55.123 (1715235835.123) Messages: 10 Topic information: Topic: /chatter Type: std_msgs/msg/String Count: 10 Serialization Format: cdr关键看三项Duration是否符合预期时长如果录了10秒但显示只有2秒说明大部分时间没收到消息可能是话题名写错或节点没启动。Count每个话题的消息数是否合理/chatter是1Hz10秒应有10条如果只有5条说明丢包严重。Serialization Format必须是cdr如果是unknown说明消息类型未被正确识别回放会失败。步骤2用ros2 bag play做黄金验证# 新终端启动一个空的ROS2环境确保无干扰 source /opt/ros/humble/setup.bash # 回放bag同时监听/chatter ros2 bag play my_bag/ ros2 topic echo /chatter你应该看到和录制时完全一致的Hello World: X序列。如果ros2 topic echo没输出或者输出乱码说明bag损坏或话题名不匹配。实操心得回放时加--rate 0.5可以慢速播放0.5倍速加--loop可以循环播放这对反复调试某个异常帧特别有用。但注意--rate只改变消息发布时间间隔不改变bag内存储的时间戳所以算法处理时仍以原始时间戳为准。步骤3切片与备份——为长期项目管理打基础一个2小时的bag文件可能达20GB你不可能每次都回放全部。ROS2提供了强大的切片能力# 切出第30秒到第45秒的数据绝对时间戳 ros2 bag filter my_bag/ -o sliced_bag --start 1715235825.0 --end 1715235840.0 # 切出包含/chatter和/odom的子集话题过滤 ros2 bag filter my_bag/ -o subset_bag --topics /chatter /odom # 同时做时间话题双重过滤 ros2 bag filter my_bag/ -o precise_slice --start 1715235825.0 --end 1715235840.0 --topics /chatter /odom切片后用ros2 bag info sliced_bag/确认新bag的元数据是否正确。最后立即备份原始bag到NAS或云存储。我坚持“3-2-1备份原则”3份拷贝2种介质本地SSD NAS1份异地如公司私有云。因为.db3文件一旦损坏几乎无法修复——SQLite的WAL日志模式在异常断电时极易导致数据库崩溃。3.4 Python API进阶用代码控制录制实现自动化闭环命令行适合手动调试但工程化必须靠代码。ROS2官方提供了rosbag2_pyPython API让你能完全掌控录制流程。下面是一个生产级录制脚本的核心逻辑import rclpy from rclpy.node import Node from rosbag2_py import SequentialWriter, StorageOptions, ConverterOptions from rosbag2_py._writer import WriterError from std_msgs.msg import String import time class AutoRecorder(Node): def __init__(self): super().__init__(auto_recorder) # 配置存储选项 storage_options StorageOptions( uriauto_recorded_bag, # 输出目录 storage_idsqlite3, max_bagfile_size1024*1024*100, # 单文件最大100MB自动分卷 ) converter_options ConverterOptions( input_serialization_formatcdr, output_serialization_formatcdr ) self.writer SequentialWriter() try: self.writer.open(storage_options, converter_options) except WriterError as e: self.get_logger().error(fFailed to open writer: {e}) return # 注册/chatter话题必须提前声明否则无法写入 from rosbag2_py._storage import TopicMetadata topic_info TopicMetadata( name/chatter, typestd_msgs/msg/String, serialization_formatcdr, offered_qos_profiles[] # 空列表表示使用默认QoS ) self.writer.create_topic(topic_info) def record_message(self, msg_text: str): 模拟发布一条消息并写入bag msg String() msg.data msg_text # 关键必须手动设置时间戳否则writer会用系统时间可能和ROS2时钟不一致 msg.header.stamp self.get_clock().now().to_msg() # 写入bag注意这里是raw bytes不是msg对象 from rosidl_runtime_py import message_to_csv # 实际中你需要用rosidl_generator_py的序列化函数 # 这里简化为伪代码serialized_data serialize_msg_to_cdr(msg) # self.writer.write(/chatter, serialized_data, timestamp_ns) def main(argsNone): rclpy.init(argsargs) recorder AutoRecorder() # 模拟录制30秒 start_time time.time() while time.time() - start_time 30.0: recorder.record_message(fAutoRecord: {int(time.time())}) time.sleep(1.0) # 1Hz recorder.destroy_node() rclpy.shutdown() if __name__ __main__: main()这个脚本展示了三个关键点分卷录制max_bagfile_size参数让bag自动按大小切分避免单文件过大导致I/O瓶颈。手动时间戳msg.header.stamp self.get_clock().now().to_msg()确保时间戳与ROS2时钟严格对齐这是离线回放同步的基石。话题预注册create_topic()必须在写入前调用否则writer.write()会抛出WriterError。这和命令行的“自动发现”完全不同API要求你完全掌控。注意rosbag2_pyAPI的序列化部分比较底层官方文档确实简略。我建议直接参考ros2 bag record的源码在ros2bag包的record.py中那里有完整的serialize_message实现。或者用更成熟的方案先用ros2 topic echo -p将消息转成YAML再用PyYAML解析虽然慢一点但稳定可靠。4. 常见问题与硬核排查指南那些让你熬夜到凌晨三点的Bug4.1 问题速查表症状、原因、解决方案症状可能原因解决方案验证方法ros2 bag record启动后无输出ros2 topic list也看不到新话题DDS域不匹配ROS_DOMAIN_ID环境变量不同统一所有终端的export ROS_DOMAIN_ID0或在~/.bashrc中永久设置echo $ROS_DOMAIN_ID确保所有进程一致录制的bag中/chatter消息数为0但ros2 topic echo /chatter能看到数据话题QoS不兼容发布者BEST_EFFORTvs 订阅者RELIABLE在ros2 bag record命令后加--qos-profile-overrides-path qos_overrides.yaml强制设为BEST_EFFORT创建qos_overrides.yaml/chatter:subscription:reliability: best_effortros2 bag info显示Duration: 0.0sCount: 0bag文件被其他进程占用如SQLite浏览器开着lsof my_bag_000.db3查占用进程kill -9 PID释放fuser -v my_bag_000.db3Linux回放时/tf坐标系丢失rviz2报No transform from [base_link] to [map]未录制/tf_static话题录制时加--include-hidden-topics或显式指定/tf_staticros2 bag info my_bag/ | grep tf_staticbag文件体积异常大如1秒录了500MB消息类型含大数组如sensor_msgs/Image未压缩用--compression-format zstd --compression-mode file重新录制du -sh my_bag_000.db3对比压缩前后4.2 硬核排查案例时间戳错乱之谜现象用户反馈用ros2 bag play回放一个激光雷达bag在RViz2中看到点云“跳跃式”移动不像真实采集的平滑轨迹。ros2 topic hz /scan显示频率是10Hz但ros2 bag info显示Duration: 5.0s,Count: 200平均40Hz明显矛盾。排查过程先用ros2 bag play -r 0.10.1倍速回放用ros2 topic echo /scan/header/stamp打印时间戳sec: 1715235825 nanosec: 123456789 sec: 1715235825 nanosec: 123456789 ← 全部相同所有消息时间戳完全一致说明录制时没打上正确时间戳。检查发布者节点代码发现它用了rclcpp::Clock::now()但没设置时钟类型// 错误写法默认是SYSTEM_TIME和ROS_TIME不一致 auto now this-get_clock()-now(); msg.header.stamp now;正确写法应强制使用ROS_TIME// 正确确保和bag录制时钟一致 auto now this-get_clock()-now(); now rclcpp::Time(now.nanoseconds(), RCL_ROS_TIME); msg.header.stamp now;重新编译节点用ros2 bag record --start-delay 5录制再验证时间戳恢复正常。教训总结ROS2的时钟类型RCL_SYSTEM_TIME,RCL_STEADY_TIME,RCL_ROS_TIME不是可选项而是数据一致性的生命线。ros2 bag默认用RCL_ROS_TIME你的所有业务节点也必须对齐。在rclcpp::Node构造时可以显式传入rclcpp::NodeOptions().use_intra_process_comms(true).clock_type(RCL_ROS_TIME)来全局设定。4.3 性能瓶颈定位当录制开始丢包现象在Jetson AGX Orin上录制8个高清摄像头话题总带宽约80MB/sros2 bag info显示Count比理论值少15%。定位步骤监控系统资源htop看CPUiotop看磁盘IOnvidia-smi看GPU如果用GPU加速编码。发现iotop中ros2bag进程的WRITE列持续显示120.00 M/s而磁盘峰值只有100 MB/s说明I/O是瓶颈。查ros2 bag record --help发现--max-cache-size参数可调大内存缓冲ros2 bag record ... --max-cache-size 100000000 # 100MB缓存调整后iotop显示WRITE降为80.00 M/sCount误差降至0.3%。终极方案对于超高速数据流放弃单机录制改用分布式采集——让每个传感器节点自己录本地bag再用ros2 bag filter在后处理阶段按时间戳对齐合并。这需要你为每个节点添加NTP时间同步但这才是工业级的正确姿势。5. 工程化延伸从单次录制到数据资产管理体系5.1 自动化录制脚本让重复劳动消失手动敲命令终究不可持续。我为你写了一个生产就绪的录制脚本auto_bagger.sh它能检测目标节点是否存活ros2 node list \| grep target_node自动获取该节点发布的所有话题ros2 node info /target_node \| grep Publishing按预设模板生成bag名robot_model_20240510_142345_run01录制完成后自动上传到NASrsync -avz my_bag/ usernas:/data/bags/发送企业微信通知调用Webhook API脚本核心逻辑简化版#!/bin/bash # 参数机器人型号、测试场景、运行编号 ROBOT_MODEL$1 SCENE$2 RUN_ID$3 # 生成时间戳和bag名 TIMESTAMP$(date %Y%m%d_%H%M%S) BAG_NAME${ROBOT_MODEL}_${TIMESTAMP}_${SCENE}_run${RUN_ID} # 检查节点 if ! ros2 node list | grep -q target_node; then echo ERROR: target_node not running! exit 1 fi # 获取话题列表过滤掉/system/类话题 TOPICS$(ros2 node info /target_node | grep Publishing | awk {print $2} | grep -v ^/system) # 开始录制 echo Recording to ${BAG_NAME}... ros2 bag record $TOPICS -o ${BAG_NAME} --start-delay 5 --compression-format zstd # 验证 if [ $(ros2 bag info ${BAG_NAME}/ | grep Count: | wc -l) -eq 0 ]; then echo ERROR: Bag validation failed! exit 1 fi # 上传 echo Uploading to NAS... rsync -avz ${BAG_NAME}/ usernas:/data/bags/${BAG_NAME}/ # 通知 curl -X POST https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyxxx \ -H Content-Type: application/json \ -d {\msgtype\: \text\, \text\: {\content\: \Bag recorded: ${BAG_NAME}\}} echo Done: ${BAG_NAME}把它放进/usr/local/bin/auto_bagchmod x以后只需auto_bag orin_nav mapping 01全自动搞定。5.2 Bag元数据标准化让数据可搜索、可追溯一个没有元数据的bag就像一本没有目录的书。我在团队推行的metadata.yaml增强规范如下# 自动生成部分ros2 bag record 会写 ros_version: humble ros_distro: humble duration: 120.0 start_timestamp: 1715235825.123 end_timestamp: 1715235945.123 # 人工补充部分录制前用脚本注入 test_scenario: indoor_navigation environment: lab_office weather: indoor operator: zhang_san hardware_config: - robot_model: orin_nav_v2 - lidar: velodyne_vlp16 - camera: realsense_d435 software_version: - firmware: v2.3.1 - navigation_stack: nav2_humble_patch_202405 - custom_nodes: [obstacle_avoidance_v1.2] # 标签用于Elasticsearch全文检索 tags: [mapping, localization, tf_error_analysis]如何注入用ros2 bag info --output-yaml导出原始元数据用yqYAML处理器合并人工部分ros2 bag info my_bag/ --output-yaml metadata_orig.yaml yq eval-all . as $item ireduce ({}; . * $item) metadata_orig.yaml custom_metadata.yaml metadata_final.yaml # 然后用ros2 bag reindex重建索引需先stop录制这样当你在数据平台搜索tags: mapping AND hardware_config.lidar velodyne_vlp16就能秒级定位所有相关bag。5.3 离线回放的终极形态用Python做数据挖掘bag不只是用来回放的它是你的离线算法沙盒。下面这段代码从bag中提取/scan话题的强度intensity均值随时间变化曲线import rosbag2_py from rclpy.serialization import deserialize_message from sensor_msgs.msg import LaserScan import numpy as np import matplotlib.pyplot as plt def analyze_scan_intensity(bag_path: str): reader rosbag2_py.SequentialReader() storage_options rosbag2_py.StorageOptions(uribag_path, storage_idsqlite3) reader.open(storage_options,