如何在ROS2中将bag文件中的图像数据导出并转换为MP4

前提依赖 #

确保系统已安装以下工具：

• ROS2对应版本的cv_bridge（如Humble版本执行sudo apt install ros-humble-cv-bridge）
• ffmpeg（执行sudo apt install ffmpeg，macOS用brew install ffmpeg）

方法一：离线提取图像并合成MP4（推荐） #

通过Python脚本直接读取Bag中的压缩图像，解码后用FFmpeg合成视频，适合批量处理。

步骤1：创建转换脚本 #

保存以下代码为bag_to_mp4.py：

import rclpy
from rclpy.node import Node
from sensor_msgs.msg import CompressedImage
import cv2
import numpy as np
import subprocess
import sys

class BagToVideo(Node):
    def __init__(self, bag_path, target_topic, output_path):
        super().__init__('bag_to_video_converter')
        self.bag_path = bag_path
        self.target_topic = target_topic
        self.output_path = output_path
        self.frame_list = []
        self.fps = 30.0  # 默认帧率，无法检测时使用

        # 从Bag信息中提取话题帧率
        info_cmd = f"ros2 bag info {self.bag_path} --yaml | grep '{self.target_topic}' -A 5"
        info_result = subprocess.run(info_cmd, shell=True, capture_output=True, text=True)
        if info_result.returncode == 0:
            for line in info_result.stdout.split('\n'):
                if 'frequency' in line:
                    try:
                        self.fps = float(line.split(':')[1].strip())
                        self.get_logger().info(f"检测到话题帧率: {self.fps} FPS")
                    except ValueError:
                        pass

        # 读取Bag中的所有压缩图像帧
        bag_reader = rclpy.node.BagReader(self.bag_path)
        while bag_reader.has_next():
            topic, msg, _ = bag_reader.read_next()
            if topic == self.target_topic:
                # 解码压缩图像
                img_buffer = np.frombuffer(msg.data, np.uint8)
                frame = cv2.imdecode(img_buffer, cv2.IMREAD_COLOR)
                self.frame_list.append(frame)
                if len(self.frame_list) % 100 == 0:
                    self.get_logger().info(f"已加载 {len(self.frame_list)} 帧")

        if not self.frame_list:
            self.get_logger().error("Bag中未找到目标话题的图像帧")
            return

        # 获取图像尺寸
        height, width = self.frame_list[0].shape[:2]
        # 调用FFmpeg合成视频
        ffmpeg_cmd = [
            'ffmpeg', '-y',
            '-f', 'rawvideo', '-vcodec', 'rawvideo', '-pix_fmt', 'bgr24',
            '-s', f'{width}x{height}', '-r', str(self.fps), '-i', '-',
            '-c:v', 'libx264', '-pix_fmt', 'yuv420p', self.output_path
        ]
        ffmpeg_process = subprocess.Popen(ffmpeg_cmd, stdin=subprocess.PIPE)
        for frame in self.frame_list:
            ffmpeg_process.stdin.write(frame.tobytes())
        ffmpeg_process.stdin.close()
        ffmpeg_process.wait()
        self.get_logger().info(f"视频已保存至: {self.output_path}")

def main():
    rclpy.init()
    if len(sys.argv) != 4:
        print("使用方式: python3 bag_to_mp4.py <Bag文件路径> <目标图像话题> <输出MP4路径>")
        print("示例: python3 bag_to_mp4.py rosbag2_2023_06_21-22_18_10_0 /image_raw/compressed output.mp4")
        rclpy.shutdown()
        return
    bag_path = sys.argv[1]
    target_topic = sys.argv[2]
    output_path = sys.argv[3]
    converter_node = BagToVideo(bag_path, target_topic, output_path)
    converter_node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

步骤2：运行脚本 #

1. 赋予脚本执行权限：

   chmod +x bag_to_mp4.py
   

2. 执行转换命令（替换为你的实际路径和话题）：

   python3 bag_to_mp4.py rosbag2_2023_06_21-22_18_10_0 /image_raw/compressed output.mp4
   

方法二：实时播放Bag并录制 #

适合需要边看边录的场景，通过image_transport转码后用FFmpeg录制。

步骤1：转换压缩话题为原始图像话题 #

新开终端运行republish节点，将压缩图像转为原始图像：

ros2 run image_transport republish compressed raw --ros-args \
  --remap in/compressed:=/image_raw/compressed \
  --remap out:=/image_raw/uncompressed

步骤2：用FFmpeg录制视频 #

再开一个终端，执行录制命令（替换640x480为你的图像实际分辨率，30为实际帧率）：

ros2 topic echo -n -1 /image_raw/uncompressed --field data | \
ffmpeg -f rawvideo -pixel_format bgr8 -video_size 640x480 -framerate 30 -i - output.mp4

注意事项 #

• 若不知道图像分辨率，可通过ros2 topic echo /image_raw/compressed --field format查看格式信息，或用rqt_image_view查看时获取窗口分辨率。
• 若FFmpeg提示编码错误，确保已安装libx264库（大部分系统默认包含）。

内容的提问来源于stack exchange，提问作者Mubashir