一、视觉模块架构设计
考虑的几个方面：
传感器主要使用话题（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 rclpyfrom rclpy.node import Nodefrom cv_bridge import CvBridgefrom sensor_msgs.msg import Imageimport cv2class 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 rclpyfrom rclpy.node import Nodefrom sensor_msgs.msg import Imagefrom interfaces_pkg.msg import DetectionResultsfrom cv_bridge import CvBridgeimport cv2import numpy as npclass 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 = 0def 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 rclpyfrom rclpy.node import Nodefrom interfaces_pkg.msg import DetectionResultsclass 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。
检测结果展示：
三个节点可正常运行：
审核编辑：刘清