如何在ros2中从包文件导出图像和视频数据

在前面的回答的基础上,这里有一个 不是 依赖于ros的代码可以将ros2包反序列化为图像(帧),从那里你可以对它们做任何你想做的事情。

from rosbags.highlevel import AnyReader
from pathlib import Path
import cv2
from rosbags.image import message_to_cvimage
count = 0
with AnyReader([Path('path_to_bag_dir_which_contains_metadata.yamal_file')]) as reader:
    # topic and msgtype information is available on .connections list
    for connection in reader.connections:
        print(connection.topic, connection.msgtype)
    for connection, timestamp, rawdata in reader.messages():
        if connection.topic == '/basler/image': # topic Name of images
            msg = reader.deserialize(rawdata, connection.msgtype)
            img = message_to_cvimage(msg, 'bgr8') # change encoding type if needed
            cv2.imwrite("output/folder/frame%06i.png" % count, img)
            count += 1

您需要安装rosbags python lib

pip install rosbags
pip install rosbags-image

以下是订阅的python脚本 /image_raw/compressed 并将视频另存为 output_video.mp4 .

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


class ImageToVideoConverter(Node):
    def __init__(self):
        super().__init__('image_to_video_converter')
        self.bridge = cv_bridge.CvBridge()
        self.subscription = self.create_subscription(
            CompressedImage,
            '/image_raw/compressed',
            self.image_callback,
            10
        )
        self.video_writer = None

    def image_callback(self, msg):
        try:
            np_arr = np.frombuffer(msg.data, np.uint8)
            cv_image = cv2.imdecode(np_arr, cv2.IMREAD_COLOR)
            if self.video_writer is None:
                self.init_video_writer(cv_image)
            self.video_writer.write(cv_image)
        except Exception as e:
            self.get_logger().error('Error processing image: %s' % str(e))

    def init_video_writer(self, image):
        try:
            height, width, _ = image.shape
            video_format = 'mp4'  # or any other video format supported by OpenCV
            video_filename = 'output_video.' + video_format
            fourcc = cv2.VideoWriter_fourcc(*'mp4v')
            fps = 30  # Frames per second
            self.video_writer = cv2.VideoWriter(video_filename, fourcc, fps, (width, height))
        except Exception as e:
            self.get_logger().error('Error initializing video writer: %s' % str(e))

    def destroy_node(self):
        if self.video_writer is not None:
            self.video_writer.release()
        super().destroy_node()


def main(args=None):
    rclpy.init(args=args)
    image_to_video_converter = ImageToVideoConverter()
    rclpy.spin(image_to_video_converter)
    image_to_video_converter.destroy_node()
    rclpy.shutdown()


if __name__ == '__main__':
    main()