ros2简单的案例，一个节点采集图片，一个节点推理

先说一下为什么要学ros2，：首先他的通信非常快,而且可以多语言编程。

比如说，如果要采集一张，然后多个模型推理，然后结果汇总，如果就单纯的用python的多线程，多进程，不仅速度慢，各种损耗，gil限制，而且多进程多了，管理起来比较复杂，并且ipc开销高，速度也不是很快。如果使用c++，如果模型中还要加几个大模型，各种算子没有，要自己写，如果采用部分模型c++推理，部分模型python推理，然后http通信，但是这样实时性又不好。采用ros2可以就可以非常好的解决这些问题，c++，python多语言编程，且通信速度非常快。

没有ros2环境需要根据系统版本安装对应的ros2.

# 安装 cv_bridge sudo apt install ros-jazzy-cv-bridge # 安装 OpenCV Python 包（推荐使用 pip 获取最新版本） pip install opencv-python

1，创建工作空间

mkdir -p ~/ros2_ws/src cd ~/ros2_ws

2, 创建python包

包名我们取为my_camera_pkg：

cd src ros2 pkg create --build-type ament_python my_camera_pkg --dependencies rclpy sensor_msgs cv_bridge

参数说明：

• --build-type ament_python：python包。

• --dependencies：自动在package.xml和setup.py中添加依赖。

3，编辑节点代码

进入包目录：

cd my_camera_pkg/my_camera_pkg

创建两个python文件：

（1）：camera_publisher.py (采集节点）

import rclpy from rclpy.node import Node from sensor_msgs.msg import Image import cv2 from cv_bridge import CvBridge class CameraPublisher(Node): def __init__(self): super().__init__('camera_publisher') self.publisher_ = self.create_publisher(Image, '/camera/image_raw', 10) self.timer = self.create_timer(1.0 / 30, self.timer_callback) # 30 Hz self.cap = cv2.VideoCapture(0) if not self.cap.isOpened(): self.get_logger().error('Cannot open camera') raise RuntimeError('Camera open failed') self.bridge = CvBridge() self.get_logger().info('Camera publisher started') def timer_callback(self): ret, frame = self.cap.read() if ret: msg = self.bridge.cv2_to_imgmsg(frame, encoding='bgr8') self.publisher_.publish(msg) else: self.get_logger().warn('Failed to capture frame') def destroy_node(self): self.cap.release() super().destroy_node() def main(args=None): rclpy.init(args=args) node = CameraPublisher() try: rclpy.spin(node) except KeyboardInterrupt: pass finally: node.destroy_node() rclpy.shutdown() if __name__ == '__main__': main()

（2）：canny_subscriber.py（推理节点）,为了减少搭建环境的工作量，我这里不使用yolo,用canny算子来演示。

import rclpy from rclpy.node import Node from sensor_msgs.msg import Image import cv2 from cv_bridge import CvBridge class CannySubscriber(Node): def __init__(self): super().__init__('canny_subscriber') self.subscription = self.create_subscription( Image, '/camera/image_raw', self.image_callback, 10) self.bridge = CvBridge() # 创建显示窗口 cv2.namedWindow('Canny Edge', cv2.WINDOW_NORMAL) self.get_logger().info('Canny subscriber started') def image_callback(self, msg): try: cv_image = self.bridge.imgmsg_to_cv2(msg, desired_encoding='bgr8') except Exception as e: self.get_logger().error(f'cv_bridge error: {e}') return # 转为灰度图 gray = cv2.cvtColor(cv_image, cv2.COLOR_BGR2GRAY) # Canny 边缘检测 edges = cv2.Canny(gray, 50, 150) # 显示结果 cv2.imshow('Canny Edge', edges) cv2.waitKey(1) def destroy_node(self): cv2.destroyAllWindows() super().destroy_node() def main(args=None): rclpy.init(args=args) node = CannySubscriber() try: rclpy.spin(node) except KeyboardInterrupt: pass finally: node.destroy_node() rclpy.shutdown() if __name__ == '__main__': main()

4 ，添加可执行权限

chmod +x camera_publisher.py canny_subscriber.py

4. 配置包信息

在~/ros2_ws/src/my_camera_pkg/setup.py中找到entry_points部分。修改成：

entry_points={ 'console_scripts': [ 'camera_publisher = my_camera_pkg.camera_publisher:main', 'canny_subscriber = my_camera_pkg.canny_subscriber:main', ], },

5. 编译工作空间

#回到工作空间根目录： cd ~/ros2_ws # 编译 colcon build --packages-select my_camera_pkg source install/setup.bash #运行采集节点代码： source ~/workspaces/ros2_ws/install/setup.bash ros2 run my_camera_pkg camera_publisher

6：运行推理节点代码：

source ~/workspaces/ros2_ws/install/setup.bash ros2 run my_camera_pkg canny_subscriber

7：结果