How to develop a computer vision module in ROS2?

1. Visual module architecture design

Several aspects to consider:

Mainly use the (toc) mechanism to continuously publish image information to the outside world;

There are two forms of image reception and processing, as well as sending processing result nodes. One is to use the service communication mechanism, and the other is to use the topic mechanism. Both are acceptable. I collected some information online and referred to the opinions of chatgpt, and got a good result: topics are faster and simpler to implement. Topics are generally used by default during development. If topics no longer meet our needs as development progresses, we can switch to the service mechanism. Since the sensors (camera) and models in the arm (usually only one is used in a robotic arm) are not complicated, I chose the topic communication mechanism to develop the image data reception and processing and result sending modules.

Use subscribers to receive the processing results of the CV model.

[sens publisher (camera_pub.py)] --> [scriber and publisher node (cam_sub_and_detecon_pub.py)] --> [subscriber (detection_results_sub.py)]

2. Code Writing

1. Create a new workspace

1. Create a src folder to store the source code;

2. Create cv_devel_pkg and interfaces_pkg in the src directory to store the source code of the vision development module and topic data (interface) files respectively;

2.1 interfaces_pkg writing

It should be noted that when creating a new interface pkg, build type can only be selected temporarily (information source: official document), and we need to modify cmakelists.txt and package.xml:

cmakelists.txt:

Added:

# find_package(REQUIRED)


find_package(geometry_msgs REQUIRED)
find_package(rosidl_default_generators REQUIRED)


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

package.xml:

Added:

 
geometry_msgs
rosidl_default_generators
rosidl_default_runtime
rosidl_interface_packages
 

Among them, the information stored in DetectionResults.msg is the msg of the result coordinates:

int32 position_x

int32 position_y

At this point, the interface pkg code writing is completed.

2.2 cv_devel_pkg writing

(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_callbk)
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()

After writing, register the node in setup.py under the pkg directory, and execute colcon build, source install/local_setup.sh, and run cv_devel_pkg camera_pub respectively.

As shown in the figure, normal operation:

(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#You can embed your own or AI vision code here


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 DetectionResulubNode(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()

3. Completion

After all the node codes in cv_devel_pkg are written, register them in setup.py, then build & run.

Results:

The three nodes are running normally:

Review editor: Liu Qing