🔄 第14课：ros2bag与回放

📋 课程目标

• 深入理解ros2bag回放机制与时钟控制
• 掌握回放时的QoS和话题重映射
• 学会数据过滤与切片回放
• 掌握bag文件的转换与合并
• 学会回放驱动的离线算法调试

🧠 回放机制详解

⏰ 时钟与use_sim_time

回放时最关键的概念是仿真时钟。当 --clock 选项启用时，ros2bag会发布 /clock 话题，所有使用 use_sim_time:=true 的节点将使用这个时钟而非系统时钟。

# 回放时发布时钟
ros2 bag play my_recording --clock

# 节点使用仿真时钟
ros2 run my_pkg my_node --ros-args -p use_sim_time:=true

# 在Launch中设置
Node(
    package='my_pkg',
    executable='my_node',
    parameters=[{'use_sim_time': True}]
)

🔧 高级回放技巧

1. 话题重映射

# 回放时重映射话题
ros2 bag play my_recording \
  --remap /scan:=/robot1/scan \
  --remap /odom:=/robot1/odom

# 用于多机器人回放对比

2. 选择性回放

# 只回放特定话题
ros2 bag play my_recording --topics /scan /odom

# 排除话题
ros2 bag play my_recording --exclude-topics /camera/image_raw

3. 时间切片

# 从第10秒开始回放
ros2 bag play my_recording --start-offset 10

# 只回放30秒
ros2 bag play my_recording --duration 30

# 从第10秒开始回放30秒
ros2 bag play my_recording --start-offset 10 --duration 30

4. 速率控制

# 加速回放
ros2 bag play my_recording --rate 2.0

# 慢放（调试用）
ros2 bag play my_recording --rate 0.25

# 暂停/恢复（需要信号处理）
# kill -SIGUSR1 <pid>  # 暂停
# kill -SIGUSR2 <pid>  # 恢复

🐍 Python：自定义回放控制器

#!/usr/bin/env python3
"""自定义bag回放控制器 - 支持暂停/步进"""

import rclpy
from rclpy.node import Node
from rosbag2_py import SequentialReader, StorageOptions, ConverterOptions
from std_msgs.msg import String
import threading


class BagReplayController(Node):
    """带控制功能的bag回放器"""

    def __init__(self):
        super().__init__('bag_replay_controller')

        self.declare_parameter('bag_path', '')
        self.declare_parameter('topics', [''])
        self.declare_parameter('rate', 1.0)

        self.is_paused = False
        self.step_mode = False

        # 订阅控制命令
        self.cmd_sub = self.create_subscription(
            String, '/replay_cmd', self.cmd_callback, 10)

        self.get_logger().info('回放控制器就绪')
        self.get_logger().info('  命令: pause / resume / step / stop')

    def cmd_callback(self, msg):
        """处理回放控制命令"""
        cmd = msg.data.lower()
        if cmd == 'pause':
            self.is_paused = True
            self.get_logger().info('⏸️ 回放已暂停')
        elif cmd == 'resume':
            self.is_paused = False
            self.step_mode = False
            self.get_logger().info('▶️ 回放已恢复')
        elif cmd == 'step':
            self.step_mode = True
            self.is_paused = False
            self.get_logger().info('⏭️ 单步模式')
        elif cmd == 'stop':
            self.get_logger().info('⏹️ 回放停止')


def main(args=None):
    rclpy.init(args=args)
    node = BagReplayController()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

📊 Bag文件转换

# 转换存储格式
ros2 bag convert -i my_recording -o converted \
  --storage-id sqlite3

# 合并多个bag
ros2 bag convert -i bag1 bag2 -o merged_bag

🎯 练习题

🏆 成就解锁