As a simple example, we can subscribe to the camera topic in a ROS2 Python node using OpenCV (via cv_bridge )
import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Image
from cv_bridge import CvBridge
import cv2
class ImageProcessor(Node):
def __init__(self):
super().__init__('image_processor')
self.bridge = CvBridge()
# Subscribe to the color image topic from the vision node
self.sub = self.create_subscription(Image, '/camera/color/image_raw', self.image_callback, 10)
def image_callback(self, msg):
# Convert ROS Image message to OpenCV image
cv_img = self.bridge.imgmsg_to_cv2(msg, 'bgr8')
# Perform edge detection (Canny)
edges = cv2.Canny(cv_img, 100, 200)
# Display the result
cv2.imshow('Edges', edges)
cv2.waitKey(1)
def main(args=None):
rclpy.init(args=args)
node = ImageProcessor()
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()In this script, we subscribe to /camera/color/image_raw , which is published by the vision node . (Depth images are on /camera/depth/image_raw .) The callback converts the image to a CV image and runs Canny edge detection on it.
We can simplify running the file by creating a launch file.
from launch import LaunchDescription
from launch_ros.actions import Node
def generate_launch_description():
return LaunchDescription([
Node(package='kinova_vision', executable='kinova_vision_node', name='kinova_vision'),
Node(package='my_pkg', executable='image_processor', name='image_processor')
])Launch with:
ros2 launch my\_pkg process\_and\_vision.launch.py
This ensures the camera stream is running when your processing node starts.