如何在ROS2中开发一种计算机视觉模块呢？

一、视觉模块架构设计

考虑的几个方面：

传感器主要使用话题（topic）通信机制持续向外部发布图像信息；

图像接受与处理、以及发送处理结果节点有两种形式，一种是使用服务（service）通信机制，一种是使用话题机制，两者均可。本人在网上搜集了一些信息，并且参考了chatgpt的意见，得到了一个不错的结果：话题速度更快，并且实现更简单，开发中一般默认使用话题，如果随着开发的进行，话题不再满足我们的需求，可以转至service机制。由于工业机械臂中的传感器（camera）和AI模型（一般一个机械臂只会用到一个）并不复杂，所以我选择topic通信机制开发图像数据的接受处理、结果发送模块。

使用订阅者来接受CV model的处理结果。

[sensor publisher (camera_pub.py)] --> [subscriber and publisher node (cam_sub_and_detection_pub.py)] --> [subscriber (detection_results_sub.py)]

二、代码编写

一、新建工作空间

1. 创建src文件夹以存放源码；

2. 在src目录下新建cv_devel_pkg与interfaces_pkg，分别存放视觉开发模块的源码与topic数据接口（interface）文件；

2.1 interfaces_pkg编写

需要注意的是，在新建interface pkg时，build type暂时只能选择c++（信息来源：ros官方文档），并且我们需要修改cmakelists.txt与package.xml：

cmakelists.txt:

新增：

# find_package( REQUIRED)


find_package(geometry_msgs REQUIRED)
find_package(rosidl_default_generators REQUIRED)


rosidl_generate_interfaces(${PROJECT_NAME}
"msg/DetectionResults.msg"
s  DEPENDENCIES geometry_msgs
)

package.xml：

新增：

 
geometry_msgs
rosidl_default_generators
rosidl_default_runtime
rosidl_interface_packages
 

其中，DetectionResults.msg中存放的信息是结果中心坐标的msg：

int32 position_x

int32 position_y

至此，interface pkg代码编写结束。

2.2 cv_devel_pkg编写

(1) camera_pub.py

import rclpy
from rclpy.node import Node
from cv_bridge import CvBridge
from sensor_msgs.msg import Image


import cv2




class CameraPubNode(Node):
def __init__(self, name):
super().__init__(name)
self.pub = self.create_publisher(Image, 'image_raw', 10)
self.timer = self.create_timer(0.5, self.timer_callback)
self.cap = cv2.VideoCapture(0)
self.cv_bridge = CvBridge()




def timer_callback(self):
        ret = self.cap.grab()
if ret:
            flag, frame = self.cap.retrieve()
if flag:
self.pub.publish(self.cv_bridge.cv2_to_imgmsg(frame, 'bgr8'))
self.get_logger().info('Publish image successfully!')
else:
self.get_logger().info('Did not get image info!')




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

编写完成后，在pkg目录下的setup.py中注册节点，并分别执行colcon build、source install/local_setup.sh、ros2 run cv_devel_pkg camera_pub。

如图，正常运行：

(2) cam_sub_and_detection_pub.py

import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Image
from interfaces_pkg.msg import DetectionResults
from cv_bridge import CvBridge


import cv2
import numpy as np




class CamSubAndDetectionPubNode(Node):
def __init__(self, name):
        super().__init__(name)
        self.sub = self.create_subscription(Image, 'image_raw', self.listen_callback, 10)
        self.pub = self.create_publisher(DetectionResults, 'detection_results', 10)
        self.cv_bridge = CvBridge()
        self.position_x = 0
        self.position_y = 0


def listen_callback(self, data):
        self.get_logger().info('Get image! I will process it!')
        image = self.cv_bridge.imgmsg_to_cv2(data, 'bgr8')
        self.detect(image)
        position = DetectionResults()
        position.position_x = self.position_x
        position.position_y = self.position_y
        self.get_logger().info('Position is: ({}, {})'.format(self.position_x, self.position_y))
        self.pub.publish(position)


def detect(self, image):
pass#这里可以嵌入自己的机器视觉或者AI视觉代码


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

（3）detection_results_sub.py

import rclpy
from rclpy.node import Node
from interfaces_pkg.msg import DetectionResults




class DetectionResultsSubNode(Node):
def __init__(self, name):
super().__init__(name)
self.sub = self.create_subscription(DetectionResults, 'detection_results', self.listen_callback, 10)


def listen_callback(self, data):
self.get_logger().info('I get the position: ({},{})'.format(data.position_x, data.position_y))




def main(args=None):
    rclpy.init(args=args)
    node = DetectionResultsSubNode('detection_results_sub_node')
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

三、完工

cv_devel_pkg中的节点代码全部编写完成后，在setup.py中注册，然后build & run。

检测结果展示：

三个节点可正常运行：

审核编辑：刘清