ROS2 入门应用 发布和订阅（Python）

  • 1. 创建功能包

    2. 创建源文件

    2.1. 话题发布

    2.2. 话题订阅

    3. 添加依赖关系

    4. 添加入口点

    5. 编译和运行

1. 创建功能包

在《ROS2 入门应用 工作空间》中已创建和加载了ros2_ws工作空间
在《ROS2 入门应用 元功能包（Python）》中已创建和加载了my_package功能包
那么就创建一个独立的py_pubsub功能包来做话题发布和订阅的功能

cd ~/ros2_ws/src
ros2 pkg create --build-type ament_python py_pubsub

2. 创建源文件

进入py_pubsub功能包的py_pubsub文件夹

cd ~/ros2_ws/src/py_pubsub/py_pubsub

2.1. 话题发布

新建publisher_member_function.py话题发布源文件

nano publisher_member_function.py

复制 ROS2 官方例程 的内容到文件中：

import rclpy
from rclpy.node import Node

from std_msgs.msg import String


class MinimalPublisher(Node):
    '''
    发布器节点类
    '''
    def __init__(self):
        # 初始化节点名、发布器、每0.5s回调的定时器和计数器
        super().__init__('minimal_publisher')
        self.publisher_ = self.create_publisher(String, 'topic', 10)
        timer_period = 0.5
        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):
    # 初始化ROS2
    rclpy.init(args=args)

    # 创建节点
    minimal_publisher = MinimalPublisher()

    # 运行节点
    rclpy.spin(minimal_publisher)

    # 销毁节点，退出ROS2
    minimal_publisher.destroy_node()
    rclpy.shutdown()


if __name__ == '__main__':
    main()

2.2. 话题订阅

新建subscriber_member_function.py话题订阅源文件

nano subscriber_member_function.py

复制 ROS2 官方例程 的内容到文件中：

import rclpy
from rclpy.node import Node

from std_msgs.msg import String


class MinimalSubscriber(Node):
    '''
    订阅器节点类
    '''
    def __init__(self):
        # 初始化节点
        super().__init__('minimal_subscriber')

        # 初始化订阅器，话题类型String，话题topic，回调函数listener_callback
        self.subscription = self.create_subscription(String, 'topic', self.listener_callback, 10)
        self.subscription  # 防止未使用变量警告

    def listener_callback(self, msg):
        '''
        订阅器回调函数
        '''
        # 打印订阅话题的消息数据
        self.get_logger().info('I heard: "%s"' % msg.data)


def main(args=None):
    # 初始化ROS2
    rclpy.init(args=args)

    # 创建节点
    minimal_subscriber = MinimalSubscriber()

    # 运行节点
    rclpy.spin(minimal_subscriber)

    # 销毁节点，退出ROS2
    minimal_subscriber.destroy_node()
    rclpy.shutdown()


if __name__ == '__main__':
    main()

3. 添加依赖关系

在package.xml清单文件中
找到<buildtool_depend>ament_cmake</buildtool_depend>依赖项，在其下面添加源文件所需的依赖（depend）：

<depend>rclpy</depend>
<depend>std_msgs</depend>

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

4. 添加入口点

打开 setup.py 文件
在entry_points入口字段的console_scripts控制台脚本的括号中添加以下两行：

entry_points={
    'console_scripts': [
        'talker = py_pubsub.publisher_member_function:main',
        'listener = py_pubsub.subscriber_member_function:main',
    ],
},

添加一个入口点py_pubsub功能包的publisher_member_function源文件的main函数，并命名为talker
添加一个入口点py_pubsub功能包的subscriber_member_function源文件的main函数，并命名为listener

5. 编译和运行

进入工作空间根目录

cd ~/ros2_ws

在编译之前检查缺失的依赖项（可跳过）：

rosdep install -i --from-path src --rosdistro humble -y

编译：

colcon build --packages-select py_pubsub

打开一个新终端，运行话题发布节点：

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 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"

谢谢