话题

节点实现了机器人各种各样的功能,但这些功能并不是独立的,之间会有千丝万缕的联系,其中最重要的一种联系方式就是话题,它是节点间传递数据的桥梁

通信模型

以两个机器人节点为例。A节点的功能是驱动相机这个硬件设备,获取得到相机拍摄的图像信息,B节点的功能是视频监控,将相机拍摄到的图像实时显示给用户查看。

image-20220524141342311

大家可以想一下,这两个节点是不是必然存在某种关系?没错,节点A要将获取的图像数据传输给节点B,有了数据,节点B才能做这样可视化的渲染。

此时从节点A到节点B传递图像数据的方式,在ROS中,我们就称之为话题,它作为一个桥梁,实现了节点之间某一个方向上的数据传输。

发布/订阅模型

从话题本身的实现角度来看,使用了基于DDS的发布/订阅模型,什么叫发布和订阅呢?

image8

话题数据传输的特性是从一个节点到另外一个节点,发送数据的对象称之为发布者,接收数据的对象称之为订阅者,每一个话题都需要有一个名字,传输的数据也需要有固定的数据类型。

image-20220527224626092

打一个比方,大家平时应该也会看微信公众号,比如有一个公众号,它的名字叫做“古月居”,这个古月居就是话题名称,公众号的发布者是古月居的小编,他会把组织好的机器人知识排版成要求格式的公众号文章,发布出去,这个文章格式,就是话题的数据类型。如果大家对这个话题感兴趣,就可以订阅“古月居”,成为订阅者之后自然就可以收到古月居的公众号文章,没有订阅的话,也就无法收到。

类似这样的发布/订阅模型在生活中随处可见,比如订阅报纸、订阅杂志等等。

多对多通信

image9

大家再仔细想下这些可以订阅的东西,是不是并不是唯一的,我们每个人可以订阅很多公众号、报纸、杂志,这些公众号、报纸、杂志也可以被很多人订阅,没错,ROS里的话题也是一样,发布者和订阅者的数量并不是唯一的,可以称之为是多对多的通信模型。

因为话题是多对多的模型,发布控制指令的摇杆可以有一个,也可以有2个、3个,订阅控制指令的机器人可以有1个,也可以有2个、3个,大家可以想象一下这个画面,似乎还是挺魔性的,如果存在多个发送指令的节点,建议大家要注意区分优先级,不然机器人可能不知道该听谁的了。

异步通信

话题通信还有一个特性,那就是异步,这个词可能有同学是第一次听说?所谓异步,只要是指发布者发出数据后,并不知道订阅者什么时候可以收到,类似古月居公众号发布一篇文章,你什么时候阅读的,古月居根本不知道,报社发出一份报纸,你什么时候收到,报社也是不知道的。这就叫做异步。

异步的特性也让话题更适合用于一些周期发布的数据,比如传感器的数据,运动控制的指令等等,如果某些逻辑性较强的指令,比如修改某一个参数,用话题传输就不太合适了。

消息接口

最后,既然是数据传输,发布者和订阅者就得统一数据的描述格式,不能一个说英文,一个理解成了中文。在ROS中,话题通信数据的描述格式称之为消息,对应编程语言中数据结构的概念。比如这里的一个图像数据,就会包含图像的长宽像素值、每个像素的RGB等等,在ROS中都有标准定义。

消息是ROS中的一种接口定义方式,与编程语言无关,我们也可以通过.msg后缀的文件自行定义,有了这样的接口,各种节点就像积木块一样,通过各种各样的接口进行拼接,组成复杂的机器人系统。


案例一:Hello World话题通信

了解了话题的基本原理,接下来我们就要开始编写代码啦。

image-20220524141514506

还是从Hello World例程开始,我们来创建一个发布者,发布话题“chatter”,周期发送“Hello World”这个字符串,消息类型是ROS中标准定义的String,再创建一个订阅者,订阅“chatter”这个话题,从而接收到“Hello World”这个字符串。

运行效果

启动第一个终端,运行话题的发布者节点:

$ ros2 run learning_topic topic_helloworld_pub

image-20220524141606868

启动第二个终端,运行话题的订阅者节点:


$ ros2 run learning_topic topic_helloworld_sub

image-20220524141614633

可以看到发布者循环发布“Hello World”字符串消息,订阅者也以几乎同样的频率收到该话题的消息数据。

发布者代码解析

我们来看下发布者的实现方法。

程序实现

learning_topic/topic_helloworld_pub.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
@作者: 古月居(www.guyuehome.com)
@说明: ROS2话题示例-发布“Hello World”话题
"""

import rclpy                                     # ROS2 Python接口库
from rclpy.node import Node                      # ROS2 节点类
from std_msgs.msg import String                  # 字符串消息类型

"""
创建一个发布者节点
"""
class PublisherNode(Node):

    def __init__(self, name):
        super().__init__(name)                                    # ROS2节点父类初始化
        self.pub = self.create_publisher(String, "chatter", 10)   # 创建发布者对象(消息类型、话题名、队列长度)
        self.timer = self.create_timer(0.5, self.timer_callback)  # 创建一个定时器(单位为秒的周期,定时执行的回调函数)

    def timer_callback(self):                                     # 创建定时器周期执行的回调函数
        msg = String()                                            # 创建一个String类型的消息对象
        msg.data = 'Hello World'                                  # 填充消息对象中的消息数据
        self.pub.publish(msg)                                     # 发布话题消息
        self.get_logger().info('Publishing: "%s"' % msg.data)     # 输出日志信息,提示已经完成话题发布

def main(args=None):                                 # ROS2节点主入口main函数
    rclpy.init(args=args)                            # ROS2 Python接口初始化
    node = PublisherNode("topic_helloworld_pub")     # 创建ROS2节点对象并进行初始化
    rclpy.spin(node)                                 # 循环等待ROS2退出
    node.destroy_node()                              # 销毁节点对象
    rclpy.shutdown()                                 # 关闭ROS2 Python接口

完成代码的编写后需要设置功能包的编译选项,让系统知道Python程序的入口,打开功能包的setup.py文件,加入如下入口点的配置:

    entry_points={
        'console_scripts': [
         'topic_helloworld_pub  = learning_topic.topic_helloworld_pub:main',
        ],
    },

流程总结

对以上程序进行分析,如果我们想要实现一个发布者,流程如下:

  • 编程接口初始化
  • 创建节点并初始化
  • 创建发布者对象
  • 创建并填充话题消息
  • 发布话题消息
  • 销毁节点并关闭接口

订阅者代码解析

我们再来看下订阅者的实现方法。

程序实现

learning_topic/topic_helloworld_sub.py


#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
@作者: 古月居(www.guyuehome.com)
@说明: ROS2话题示例-订阅“Hello World”话题消息
"""

import rclpy                      # ROS2 Python接口库
from rclpy.node   import Node     # ROS2 节点类
from std_msgs.msg import String   # ROS2标准定义的String消息

"""
创建一个订阅者节点
"""
class SubscriberNode(Node):

    def __init__(self, name):
        super().__init__(name)                             # ROS2节点父类初始化
        self.sub = self.create_subscription(\
            String, "chatter", self.listener_callback, 10) # 创建订阅者对象(消息类型、话题名、订阅者回调函数、队列长度)

    def listener_callback(self, msg):                      # 创建回调函数,执行收到话题消息后对数据的处理
        self.get_logger().info('I heard: "%s"' % msg.data) # 输出日志信息,提示订阅收到的话题消息

def main(args=None):                               # ROS2节点主入口main函数
    rclpy.init(args=args)                          # ROS2 Python接口初始化
    node = SubscriberNode("topic_helloworld_sub")  # 创建ROS2节点对象并进行初始化
    rclpy.spin(node)                               # 循环等待ROS2退出
    node.destroy_node()                            # 销毁节点对象
    rclpy.shutdown()                               # 关闭ROS2 Python接口

完成代码的编写后需要设置功能包的编译选项,让系统知道Python程序的入口,打开功能包的setup.py文件,加入如下入口点的配置:

    entry_points={
        'console_scripts': [
         'topic_helloworld_pub  = learning_topic.topic_helloworld_pub:main',
         'topic_helloworld_sub  = learning_topic.topic_helloworld_sub:main',
        ],
    },

流程总结

对以上程序进行分析,如果我们想要实现一个订阅者,流程如下:

  • 编程接口初始化
  • 创建节点并初始化
  • 创建订阅者对象
  • 回调函数处理话题数据
  • 销毁节点并关闭接口

好啦,Hello World例程大家一定还不过瘾,接下来我们基于话题通信,继续优化下之前的机器视觉例程。


案例二:机器视觉识别

在节点概念的讲解过程中,我们通过一个节点驱动了相机,并且实现了对红色物体的识别。功能虽然没问题,但是对于机器人开发来讲,并没有做到程序的模块化,更好的方式是将相机驱动和视觉识别做成两个节点,节点间的联系就是这个图像数据,通过话题周期传输即可。

image-20220524141708449

运行效果

这个图像消息在ROS中是标准定义好的,如果未来要更换另一个相机,只需要修改驱动节点,视觉识别节点完全是软件功能,就可以保持不变了,这种模块化的设计思想,可以更好的保证软件的可移植性。

好啦,说干就干,我们先来看下效果如何?

启动两个终端,分别运行以下两个节点,第一个节点驱动相机并发布图像话题,第二个节点订阅图像话题并实现视觉识别。

$ ros2 run learning_topic topic_webcam_pub
$ ros2 run learning_topic topic_webcam_sub

将红色物体放入相机范围内,即可看到识别效果。

image-20220527231048652

发布者代码解析

learning_topic/topic_webcam_pub.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
@作者: 古月居(www.guyuehome.com)
@说明: ROS2话题示例-发布图像话题
"""

import rclpy                        # ROS2 Python接口库
from rclpy.node import Node         # ROS2 节点类
from sensor_msgs.msg import Image   # 图像消息类型
from cv_bridge import CvBridge      # ROS与OpenCV图像转换类
import cv2                          # Opencv图像处理库

"""
创建一个发布者节点
"""
class ImagePublisher(Node):

    def __init__(self, name):
        super().__init__(name)                                           # ROS2节点父类初始化
        self.publisher_ = self.create_publisher(Image, 'image_raw', 10)  # 创建发布者对象(消息类型、话题名、队列长度)
        self.timer = self.create_timer(0.1, self.timer_callback)         # 创建一个定时器(单位为秒的周期,定时执行的回调函数)
        self.cap = cv2.VideoCapture(0)                                   # 创建一个视频采集对象,驱动相机采集图像(相机设备号)
        self.cv_bridge = CvBridge()                                      # 创建一个图像转换对象,用于稍后将OpenCV的图像转换成ROS的图像消息

    def timer_callback(self):
        ret, frame = self.cap.read()                         # 一帧一帧读取图像

        if ret == True:                                      # 如果图像读取成功
            self.publisher_.publish(
                self.cv_bridge.cv2_to_imgmsg(frame, 'bgr8')) # 发布图像消息

        self.get_logger().info('Publishing video frame')     # 输出日志信息,提示已经完成图像话题发布

def main(args=None):                                 # ROS2节点主入口main函数
    rclpy.init(args=args)                            # ROS2 Python接口初始化
    node = ImagePublisher("topic_webcam_pub")        # 创建ROS2节点对象并进行初始化
    rclpy.spin(node)                                 # 循环等待ROS2退出
    node.destroy_node()                              # 销毁节点对象
    rclpy.shutdown()                                 # 关闭ROS2 Python接口

完成代码的编写后需要设置功能包的编译选项,让系统知道Python程序的入口,打开功能包的setup.py文件,加入如下入口点的配置:

    entry_points={
        'console_scripts': [
         'topic_helloworld_pub  = learning_topic.topic_helloworld_pub:main',
         'topic_helloworld_sub  = learning_topic.topic_helloworld_sub:main',
         'topic_webcam_pub      = learning_topic.topic_webcam_pub:main',
        ],
    },

订阅者代码解析

learning_topic/topic_webcam_sub.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
@作者: 古月居(www.guyuehome.com)
@说明: ROS2话题示例-订阅图像话题
"""

import rclpy                            # ROS2 Python接口库
from rclpy.node import Node             # ROS2 节点类
from sensor_msgs.msg import Image       # 图像消息类型
from cv_bridge import CvBridge          # ROS与OpenCV图像转换类
import cv2                              # Opencv图像处理库
import numpy as np                      # Python数值计算库

lower_red = np.array([0, 90, 128])      # 红色的HSV阈值下限
upper_red = np.array([180, 255, 255])   # 红色的HSV阈值上限

"""
创建一个订阅者节点
"""
class ImageSubscriber(Node):
    def __init__(self, name):
        super().__init__(name)                                # ROS2节点父类初始化
        self.sub = self.create_subscription(
            Image, 'image_raw', self.listener_callback, 10)   # 创建订阅者对象(消息类型、话题名、订阅者回调函数、队列长度)
        self.cv_bridge = CvBridge()                           # 创建一个图像转换对象,用于OpenCV图像与ROS的图像消息的互相转换

    def object_detect(self, image):
        hsv_img = cv2.cvtColor(image, cv2.COLOR_BGR2HSV)      # 图像从BGR颜色模型转换为HSV模型
        mask_red = cv2.inRange(hsv_img, lower_red, upper_red) # 图像二值化
        contours, hierarchy = cv2.findContours(
            mask_red, cv2.RETR_LIST, cv2.CHAIN_APPROX_NONE)   # 图像中轮廓检测

        for cnt in contours:                                  # 去除一些轮廓面积太小的噪声
            if cnt.shape[0] < 150:
                continue

            (x, y, w, h) = cv2.boundingRect(cnt)              # 得到苹果所在轮廓的左上角xy像素坐标及轮廓范围的宽和高
            cv2.drawContours(image, [cnt], -1, (0, 255, 0), 2)# 将苹果的轮廓勾勒出来
            cv2.circle(image, (int(x+w/2), int(y+h/2)), 5,
                       (0, 255, 0), -1)                       # 将苹果的图像中心点画出来

        cv2.imshow("object", image)                           # 使用OpenCV显示处理后的图像效果
        cv2.waitKey(10)

    def listener_callback(self, data):
        self.get_logger().info('Receiving video frame')     # 输出日志信息,提示已进入回调函数
        image = self.cv_bridge.imgmsg_to_cv2(data, 'bgr8')  # 将ROS的图像消息转化成OpenCV图像
        self.object_detect(image)                           # 苹果检测


def main(args=None):                            # ROS2节点主入口main函数
    rclpy.init(args=args)                       # ROS2 Python接口初始化
    node = ImageSubscriber("topic_webcam_sub")  # 创建ROS2节点对象并进行初始化
    rclpy.spin(node)                            # 循环等待ROS2退出
    node.destroy_node()                         # 销毁节点对象
    rclpy.shutdown()                            # 关闭ROS2 Python接口

完成代码的编写后需要设置功能包的编译选项,让系统知道Python程序的入口,打开功能包的setup.py文件,加入如下入口点的配置:

    entry_points={
        'console_scripts': [
         'topic_helloworld_pub  = learning_topic.topic_helloworld_pub:main',
         'topic_helloworld_sub  = learning_topic.topic_helloworld_sub:main',
         'topic_webcam_pub      = learning_topic.topic_webcam_pub:main',
         'topic_webcam_sub      = learning_topic.topic_webcam_sub:main',
        ],
    },

案例三:机器视觉识别优化

通过话题对原本节点功能的解耦,似乎让视觉识别的例程焕然一新了,不过似乎还有哪里不太对劲,大家有感觉到么?

ROS的目标不是提高软件复用率么,现在视觉识别的节点可以复用了,相机驱动节点好像不行呀,每换一个相机,是不是都得换一个驱动节点?这当然是不可能的!

常用的usb相机驱动一般都是通用的,ROS中也集成了usb相机的标准驱动,我们只需要通过这样一行指令,就可以安装好,无论你用什么样的相机,只要符合usb接口协议,就可以直接使用ROS中的相机驱动节点,发布标准的图像话题了。


$ sudo apt install ros-humble-usb-cam

这样,我们的代码又得到了进一步精简,刚才自己写的图像发布节点换成了这样一句指令,视觉识别节点不需要做任何变化。

$ ros2 run usb_cam usb_cam_node_exe
$ ros2 run learning_topic topic_webcam_sub

image-20220527231048652


话题命令行操作

话题命令的常用操作如下:


$ ros2 topic list                # 查看话题列表
$ ros2 topic info <topic_name>   # 查看话题信息
$ ros2 topic hz <topic_name>     # 查看话题发布频率
$ ros2 topic bw <topic_name>     # 查看话题传输带宽
$ ros2 topic echo <topic_name>   # 查看话题数据
$ ros2 topic pub <topic_name> <msg_type> <msg_data>   # 发布话题消息

image-20220524141906673

思考题

image-20220524141933643

关于话题通信的原理和实现方法我们就讲到这里,给大家留一个思考题:话题通信的特性是单向传输,适合周期性的数据传递,对于一个复杂的机器人系统来讲,这种特性肯定无法满足所有数据传输的需求,大家是否能够举几个例子,是话题通信无法完成的呢?

参考链接

https://docs.ros.org/en/humble/Tutorials/Topics/Understanding-ROS2-Topics.html

本系列视频公开课:https://class.guyuehome.com/detail/p_628f4288e4b01c509ab5bc7a/6