机器人系统——ROS2文档—编写简单的发布者和订阅者(Python)

目标：使用 Python 创建并运行一个发布者和一个订阅者节点。教程级别：入门级时间： 20 分钟

目录

• 背景
• 前置条件
• 任务
• 1 创建包
• 2 编写发布者节点
• 3 编写订阅者节点
• 4 构建并运行
• 总结
• 后续步骤
• 相关资料

背景

在本教程中，您将创建通过话题以字符串消息形式相互传递信息的节点。这里使用的示例是一个简单的“talker”和“listener”系统；一个节点发布数据，另一个节点订阅该话题以便接收该数据。

这些示例中使用的代码可以在这里找到。

前置条件

在之前的教程中，您学习了如何创建工作空间和创建包。建议对 Python 有基本的了解，但不是完全必要的。

任务

1创建包

打开一个新终端并加载您的 ROS 2 安装，以便ros2命令能够工作。

导航到在上一个教程中创建的ros2_ws目录。

回想一下，包应该创建在src目录中，而不是工作空间的根目录。所以，导航到ros2_ws/src，并运行包创建命令：

$ ros2 pkg create --build-type ament_python --license Apache-2.0 py_pubsub

您的终端将返回一条消息，确认您的包py_pubsub及其所有必要的文件和文件夹已创建。

2编写发布者节点

导航到ros2_ws/src/py_pubsub/py_pubsub。回想一下，这个目录是一个Python 包，名称与其嵌套的 ROS 2 包相同。

通过输入以下命令下载示例 talker 代码：

Linux：

$ wget https://raw.githubusercontent.com/ros2/examples/lyrical/rclpy/topics/minimal_publisher/examples_rclpy_minimal_publisher/publisher_member_function.py

现在将有一个名为publisher_member_function.py的新文件，与__init__.py相邻。

使用您喜欢的文本编辑器打开该文件。

import rclpy

from rclpy.executors import ExternalShutdownException

from rclpy.node import Node

from std_msgs.msg import String

class MinimalPublisher(Node):

def __init__(self):

super().__init__('minimal_publisher')

self.publisher_ = self.create_publisher(String, 'topic', 10)

timer_period = 0.5# seconds

self.timer = self.create_timer(timer_period, self.timer_callback)

self.i = 0

def timer_callback(self):

msg = String()

msg.data = 'Hello World: %d' % self.i

self.publisher_.publish(msg)

self.get_logger().info('Publishing: "%s"' % msg.data)

self.i += 1

def main(args=None):

try:

with rclpy.init(args=args):

minimal_publisher = MinimalPublisher()

rclpy.spin(minimal_publisher)

except (KeyboardInterrupt, ExternalShutdownException):

pass

if __name__ == '__main__':

main()

2.1 检查代码

注释后的前几行代码导入了rclpy，以便可以使用其Node类。

import rclpy

from rclpy.executors import ExternalShutdownException

from rclpy.node import Node

下一个语句导入了内置的std_msgs/msg/String消息类型，节点使用该消息类型来构建其在话题上传递的数据结构。

from std_msgs.msg import String

这些行代表了节点的依赖项。回想一下，依赖项必须添加到package.xml中，您将在下一节中执行此操作。

接下来，创建MinimalPublisher类，它继承自Node。

class MinimalPublisher(Node):

接下来是类构造函数的定义。super().__init__调用Node类的构造函数并给它您的节点名称，在本例中为minimal_publisher。

create_publisher声明节点在名为topic的话题上发布类型为std_msgs/msg/String的消息，并且“队列大小”为 10。队列大小是必需的服务质量（QoS）设置，限制如果订阅者接收不够快时排队消息的数量。

接下来，使用create_timer创建一个每 0.5 秒执行一次的回调。self.i是回调中使用的计数器。

def __init__(self):

super().__init__('minimal_publisher')

self.publisher_ = self.create_publisher(String, 'topic', 10)

timer_period = 0.5# seconds

self.timer = self.create_timer(timer_period, self.timer_callback)

self.i = 0

timer_callback创建一个带有计数器值的消息，发布它，并使用get_logger()的info()函数将其打印到控制台。

def timer_callback(self):

msg = String()

msg.data = 'Hello World: %d' % self.i

self.publisher_.publish(msg)

self.get_logger().info('Publishing: "%s"' % msg.data)

self.i += 1

最后，定义了 main 函数。

def main(args=None):

try:

with rclpy.init(args=args):

minimal_publisher = MinimalPublisher()

rclpy.spin(minimal_publisher)

except (KeyboardInterrupt, ExternalShutdownException):

pass

首先初始化rclpy库，然后创建节点，然后“旋转”节点（使用spin()）以便调用其回调。

2.2 添加依赖项

返回上一级目录到ros2_ws/src/py_pubsub，在那里已经为您创建了setup.py、setup.cfg和package.xml文件。

使用文本编辑器打开package.xml。

正如上一个教程中提到的，确保填写<description>、<maintainer>和<license>标签：

<description>Examples of minimal publisher/subscriber using rclpy</description>

<maintainer email="you@email.com">Your Name</maintainer>

<license>Apache-2.0</license>

在上述行之后，添加与您节点的 import 语句相对应的以下依赖项：

<exec_depend>rclpy</exec_depend>

<exec_depend>std_msgs</exec_depend>

这声明了包在执行其代码时需要rclpy和std_msgs。

确保保存文件。

2.3 添加入口点

打开setup.py文件。再次将maintainer、maintainer_email、description和license字段与您的package.xml匹配：

maintainer='YourName',

maintainer_email='you@email.com',

description='Examples of minimal publisher/subscriber using rclpy',

license='Apache-2.0',

在entry_points字段的console_scripts括号内添加以下行：

entry_points={

'console_scripts': [

'talker = py_pubsub.publisher_member_function:main',

],

},

不要忘记保存。

2.4 检查 setup.cfg

setup.cfg文件的内容应该自动正确填充，如下所示：

[develop]

script_dir=$base/lib/py_pubsub

[install]

install_scripts=$base/lib/py_pubsub

这只是告诉setuptools将您的可执行文件放在lib中，因为ros2 run会在那里寻找它们。

您现在可以构建您的包，加载本地设置文件，并运行它，但让我们先创建订阅者节点，以便您可以看到完整的系统运行。

3编写订阅者节点

返回到ros2_ws/src/py_pubsub/py_pubsub来创建下一个节点。在您的终端中输入以下代码：

Linux：

$ wget https://raw.githubusercontent.com/ros2/examples/lyrical/rclpy/topics/minimal_subscriber/examples_rclpy_minimal_subscriber/subscriber_member_function.py

现在目录中应该有以下文件：

__init__.pypublisher_member_function.pysubscriber_member_function.py

3.1 检查代码

使用文本编辑器打开subscriber_member_function.py。

import rclpy

from rclpy.executors import ExternalShutdownException

from rclpy.node import Node

from std_msgs.msg import String

class MinimalSubscriber(Node):

def __init__(self):

super().__init__('minimal_subscriber')

self.subscription = self.create_subscription(

String,

'topic',

self.listener_callback,

10)

self.subscription# prevent unused variable warning

def listener_callback(self, msg):

self.get_logger().info('I heard: "%s"' % msg.data)

def main(args=None):

try:

with rclpy.init(args=args):

minimal_subscriber = MinimalSubscriber()

rclpy.spin(minimal_subscriber)

except (KeyboardInterrupt, ExternalShutdownException):

pass

if __name__ == '__main__':

main()

订阅者节点的代码与发布者的几乎相同。构造函数使用create_subscription创建了一个与发布者参数相同的订阅者。回想一下话题教程，发布者和订阅者使用的话题名称和消息类型必须匹配才能允许它们通信。

self.subscription = self.create_subscription(

String,

'topic',

self.listener_callback,

10)

订阅者的构造函数和回调不包含任何计时器定义，因为它不需要。它的回调在收到消息时立即被调用。

回调定义只是将一个信息消息连同收到的数据打印到控制台。回想一下，发布者定义了msg.data = 'Hello World: %d'% self.i。

def listener_callback(self, msg):

self.get_logger().info('I heard: "%s"' % msg.data)

main定义几乎完全相同，将发布者的创建和旋转替换为订阅者。

minimal_subscriber = MinimalSubscriber()

rclpy.spin(minimal_subscriber)

由于此节点与发布者具有相同的依赖项，因此无需向package.xml添加新内容。setup.cfg文件也可以保持不变。

3.2 添加入口点

重新打开setup.py并在发布者入口点下面添加订阅者节点的入口点。entry_points字段现在应该如下所示：

entry_points={

'console_scripts': [

'talker = py_pubsub.publisher_member_function:main',

'listener = py_pubsub.subscriber_member_function:main',

],

},

确保保存文件，然后您的发布/订阅系统应该已准备就绪。

4 构建并运行

作为 ROS 2 系统的一部分，您可能已经安装了rclpy和std_msgs包。最佳实践是在工作空间根目录（ros2_ws）运行rosdep来检查构建前是否缺少依赖项：

Linux：

$ rosdep install -i --from-path src --rosdistro lyrical -y

仍在工作空间根目录ros2_ws中，构建您的新包：

Linux：

$ colcon build --packages-select py_pubsub

打开一个新终端，导航到ros2_ws，并加载设置文件：

Linux：

$ source install/setup.bash

现在运行 talker 节点。终端应该每 0.5 秒开始发布信息消息，如下所示：

$ ros2 run py_pubsub talker

[INFO] [minimal_publisher]: Publishing: "Hello World: 0"

[INFO] [minimal_publisher]: Publishing: "Hello World: 1"

[INFO] [minimal_publisher]: Publishing: "Hello World: 2"

[INFO] [minimal_publisher]: Publishing: "Hello World: 3"

[INFO] [minimal_publisher]: Publishing: "Hello World: 4"

...

打开另一个终端，再次从ros2_ws内部加载设置文件，然后启动 listener 节点。listener 将开始向控制台打印消息，从发布者当时所处的任何消息计数开始，如下所示：

$ ros2 run py_pubsub listener

[INFO] [minimal_subscriber]: I heard: "Hello World: 10"

[INFO] [minimal_subscriber]: I heard: "Hello World: 11"

[INFO] [minimal_subscriber]: I heard: "Hello World: 12"

[INFO] [minimal_subscriber]: I heard: "Hello World: 13"

[INFO] [minimal_subscriber]: I heard: "Hello World: 14"

...

在每个终端中输入Ctrl+C来停止节点旋转。

总结

您创建了两个节点来通过话题发布和订阅数据。在运行它们之前，您将它们的依赖项和入口点添加到包配置文件中。

后续步骤

接下来，您将使用服务/客户端模型创建另一个简单的 ROS 2 包。同样，您可以选择使用C++或Python编写。

相关资料

用 Python 编写发布者和订阅者有几种方法；请查看ros2/examples仓库中的minimal_publisher和minimal_subscriber包。

本文档完整翻译自 ROS 2 Lyrical 官方文档，仅用于学习交流。