如何在ROS 2 Python节点中批量发布形状为[12,3,224,224]的图像？

我来帮你搞定这个批量图像发布的问题！首先得明确：ROS 2自带的sensor_msgs/Image消息是专门用来传输单张图像的，直接传列表肯定行不通。咱们得用自定义消息类型来实现批量图像的传输，下面是具体步骤：

1. 创建自定义批量图像消息 #

首先在你的ROS 2功能包下创建msg目录，新建BatchImage.msg文件，内容如下：

sensor_msgs/Image[] images

这个消息类型定义了一个包含多个sensor_msgs/Image的数组，刚好满足咱们批量传输的需求。

配置编译依赖 #

接下来需要修改功能包的package.xml和setup.py，让ROS 2能编译这个自定义消息：

修改package.xml

添加以下依赖项：

<build_depend>sensor_msgs</build_depend>
<exec_depend>sensor_msgs</exec_depend>
<build_depend>rosidl_default_generators</build_depend>
<exec_depend>rosidl_default_runtime</exec_depend>
<member_of_group>rosidl_interface_packages</member_of_group>

修改setup.py

先在文件顶部导入所需模块：

import os
from glob import glob

然后在setup()函数中更新配置：

setup(
    # ... 保留你原来的其他配置 ...
    packages=['your_package_name'],  # 替换成你的功能包名称
    data_files=[
        ('share/ament_index/resource_index/packages',
            ['resource/your_package_name']),
        ('share/your_package_name', ['package.xml']),
        # 添加自定义消息的路径
        (os.path.join('share', 'your_package_name', 'msg'), glob('msg/*.msg')),
    ],
    # ... 保留其他配置 ...
    entry_points={
        'console_scripts': [
            # 添加节点的入口脚本
            'batch_image_publisher = your_package_name.batch_publisher:main',
        ],
    },
)

编译并生效 #

在工作空间根目录执行编译：

colcon build --packages-select your_package_name

编译完成后，别忘了source环境：

source install/setup.bash

2. 修改发布节点代码 #

现在可以编写支持批量图像发布的节点了，下面是完整代码：

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node
from cv_bridge import CvBridge
from sensor_msgs.msg import Image
# 导入自定义的批量图像消息（替换成你的功能包名）
from your_package_name.msg import BatchImage
import cv2
import numpy as np

class BatchImagePublisher(Node):
    def __init__(self):
        super().__init__('batch_image_publisher')
        # 创建批量图像发布者，话题名可按需修改
        self.publisher_ = self.create_publisher(BatchImage, 'batch_images', 10)
        timer_period = 0.5  # 发布周期（秒）
        self.timer = self.create_timer(timer_period, self.timer_callback)
        self.bridge = CvBridge()
        # 加载你的批量图像（形状[12,3,224,224]）
        self.batch_images = self.load_batch_data()

    def load_batch_data(self):
        batch_msg_list = []
        # 假设你的原始批量数据是shape为[12,3,224,224]的numpy数组
        # 这里可以替换成你实际的加载/生成逻辑
        your_batch_array = np.random.randint(0, 255, (12,3,224,224), dtype=np.uint8)
        
        for img_array in your_batch_array:
            # 将CHW格式（通道在前）转换为OpenCV需要的HWC格式（通道在后）
            cv_img = np.transpose(img_array, (1, 2, 0))
            # 如果你的数组是RGB格式，需要转成OpenCV的BGR格式
            cv_img = cv2.cvtColor(cv_img, cv2.COLOR_RGB2BGR)
            # 转换为ROS Image消息
            img_msg = self.bridge.cv2_to_imgmsg(cv_img, "bgr8")
            batch_msg_list.append(img_msg)
        return batch_msg_list

    def timer_callback(self):
        # 构造批量消息并发布
        batch_msg = BatchImage()
        batch_msg.images = self.batch_images
        self.publisher_.publish(batch_msg)
        self.get_logger().info(f'Publishing batch of {len(self.batch_images)} images')

def main(args=None):
    rclpy.init(args=args)
    batch_publisher = BatchImagePublisher()
    rclpy.spin(batch_publisher)
    batch_publisher.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

关键说明 #

• 格式转换：你的原始数据是[批量数,通道数,宽度,高度]的CHW格式，而OpenCV和cv_bridge需要的是[宽度,高度,通道数]的HWC格式，所以必须用np.transpose做维度转换。
• 颜色空间：如果你的原始图像是RGB格式，记得转成BGR，因为OpenCV默认使用BGR格式存储图像，否则发布的图像颜色会失真。
• 自定义消息优势：这种方式比把多张图像拼接成一张大图再拆分更规范，也更灵活，后续订阅端可以直接遍历数组处理每张图像。

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