Python 软件包迁移示例

本指南展示了如何将一个 Python 示例软件包从 ROS 1 迁移到 ROS 2。

前提条件

你需要一个正常工作的 ROS 2 安装，例如 ROS DISTRO。

ROS 1 代码

本指南中不会使用 catkin，因此你不需要正常工作的 ROS 1 安装。你将使用 ROS 2 的构建工具 Colcon 来代替。

本节为你提供了 ROS 1 Python 软件包的代码。该软件包名为 talker_py，它有一个名为 talker_py_node 的节点。为了方便后续运行 Colcon，这些说明会让你在 Colcon 工作空间中创建该软件包。

首先，在 ~/ros2_talker_py 创建一个文件夹作为 Colcon 工作空间的根目录。

Linux

mkdir -p ~/ros2_talker_py/src

macOS

mkdir -p ~/ros2_talker_py/src

Windows

md \ros2_talker_py\src

接下来，创建 ROS 1 软件包的文件。

Linux

cd ~/ros2_talker_py
mkdir -p src/talker_py/src/talker_py
mkdir -p src/talker_py/scripts
touch src/talker_py/package.xml
touch src/talker_py/CMakeLists.txt
touch src/talker_py/src/talker_py/__init__.py
touch src/talker_py/scripts/talker_py_node
touch src/talker_py/setup.py

macOS

cd ~/ros2_talker_py
mkdir -p src/talker_py/src/talker_py
mkdir -p src/talker_py/scripts
touch src/talker_py/package.xml
touch src/talker_py/CMakeLists.txt
touch src/talker_py/src/talker_py/__init__.py
touch src/talker_py/scripts/talker_py_node
touch src/talker_py/setup.py

Windows

cd \ros2_talker_py
md src\talker_py\src\talker_py
md src\talker_py\scripts
type nul > src\talker_py\package.xml
type nul > src\talker_py\CMakeLists.txt
type nul > src\talker_py\src\talker_py\__init__.py
type nul > src\talker_py\scripts/talker_py_node
type nul > src\talker_py\setup.py

将以下内容放入每个文件中。

src/talker_py/package.xml：

<?xml version="1.0"?>
<?xml-model href="http://download.ros.org/schema/package_format2.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
<package format="2">
    <name>talker_py</name>
    <version>1.0.0</version>
    <description>The talker_py package</description>
    <maintainer email="gerkey@example.com">Brian Gerkey</maintainer>
    <license>BSD</license>

    <buildtool_depend>catkin</buildtool_depend>

    <depend>rospy</depend>
    <depend>std_msgs</depend>
</package>

src/talker_py/CMakeLists.txt：

cmake_minimum_required(VERSION 3.20)
project(talker_py)

find_package(catkin REQUIRED)

catkin_python_setup()

catkin_package()

catkin_install_python(PROGRAMS
    scripts/talker_py_node
    DESTINATION ${CATKIN_PACKAGE_BIN_DESTINATION}
)

src/talker/src/talker_py/__init__.py：

import rospy
from std_msgs.msg import String

def main():
    rospy.init_node('talker')
    pub = rospy.Publisher('chatter', String, queue_size=10)
    rate = rospy.Rate(10)  # 10hz
    while not rospy.is_shutdown():
        hello_str = "hello world %s" % rospy.get_time()
        rospy.loginfo(hello_str)
        pub.publish(hello_str)
        rate.sleep()

src/talker_py/scripts/talker_py_node：

#!/usr/bin/env python

import talker_py

if __name__ == '__main__':
    talker_py.main()

src/talker_py/setup.py：

from setuptools import setup
from catkin_pkg.python_setup import generate_distutils_setup

setup_args = generate_distutils_setup(
    packages=['talker_py'],
    package_dir={'': 'src'}
)

setup(**setup_args)

这就是完整的 ROS 1 Python 软件包。

迁移 `package.xml`

将软件包迁移到 ROS 2 时，先迁移构建系统文件，这样你就可以在过程中通过构建和运行代码来检查你的工作。始终从迁移 package.xml 开始。

首先，ROS 2 不使用 catkin。删除对它的 <buildtool_depend>。

<!-- delete this -->
<buildtool_depend>catkin</buildtool_depend>

接下来，ROS 2 使用 rclpy 而不是 rospy。删除对 rospy 的依赖。

<!-- Delete this -->
<depend>rospy</depend>

将其替换为对 rclpy 的新依赖。

<depend>rclpy</depend>

添加 <export> 部分，告诉 ROS 2 的构建工具 Colcon 这是一个 ament_python 软件包而非 catkin 软件包。

 <export>
   <build_type>ament_python</build_type>
 </export>

你的 package.xml 已完全迁移。它现在应该如下所示：

<?xml version="1.0"?>
<?xml-model href="http://download.ros.org/schema/package_format2.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
<package format="2">
    <name>talker_py</name>
    <version>1.0.0</version>
    <description>The talker_py package</description>
    <maintainer email="gerkey@example.com">Brian Gerkey</maintainer>
    <license>BSD</license>

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

    <export>
        <build_type>ament_python</build_type>
    </export>
</package>

删除 `CMakeLists.txt`

ROS 2 中的 Python 软件包不使用 CMake，因此删除 CMakeLists.txt。

迁移 `setup.py`

setup.py 中 setup() 的参数不能再通过 catkin_pkg 自动生成。你必须手动传递这些参数，这意味着会与你的 package.xml 有一些重复。

首先，删除从 catkin_pkg 的导入。

# Delete this
from catkin_pkg.python_setup import generate_distutils_setup

将给 generate_distutils_setup() 的所有参数移到 setup() 的调用中，然后添加 install_requires 和 zip_safe 参数。你对 setup() 的调用应该如下所示：

setup(
    packages=['talker_py'],
    package_dir={'': 'src'},
    install_requires=['setuptools'],
    zip_safe=True,
)

删除对 generate_distutils_setup() 的调用。

# Delete this
setup_args = generate_distutils_setup(
    packages=['talker_py'],
    package_dir={'': 'src'}
)

setup() 的调用需要一些从 package.xml 复制的额外元数据：

通过 name 参数设置软件包名称
通过 version 参数设置软件包版本
通过 maintainer 和 maintainer_email 参数设置维护者
通过 description 参数设置描述
通过 license 参数设置许可证

软件包名称将被多次使用。在 setup() 调用上方创建一个名为 package_name 的变量。

package_name = 'talker_py'

将所有剩余信息复制到 setup.py 中 setup() 的参数中。你对 setup() 的调用应该如下所示：

setup(
    name=package_name,
    version='1.0.0',
    install_requires=['setuptools'],
    zip_safe=True,
    packages=['talker_py'],
    package_dir={'': 'src'},
    maintainer='Brian Gerkey',
    maintainer_email='gerkey@example.com',
    description='The talker_py package',
    license='BSD',
)

ROS 2 软件包必须安装两个数据文件：

一个 package.xml
一个软件包标记文件

你的软件包已经有了一个 package.xml。它描述了软件包的依赖项。软件包标记文件告诉 ros2 run 等工具在哪里可以找到你的软件包。

在 package.xml 旁边创建一个名为 resource 的目录。在 resource 目录中创建一个与软件包同名的空文件。

Linux

mkdir resource
touch resource/talker_py

macOS

mkdir resource
touch resource/talker_py

Windows

md resource
type nul > resource\talker_py

setup.py 中的 setup() 调用必须告诉 setuptools 如何安装这些文件。将以下 data_files 参数添加到 setup() 调用中。

data_files=[
    ('share/ament_index/resource_index/packages',
        ['resource/' + package_name]),
    ('share/' + package_name, ['package.xml']),
],

你的 setup.py 几乎完成了。

迁移 Python 脚本并创建 `setup.cfg`

ROS 2 Python 软件包使用 console_scripts 入口点将 Python 脚本安装为可执行文件。配置文件 setup.cfg 告诉 setuptools 将这些可执行文件安装到软件包特定的目录中，以便 ros2 run 等工具能够找到它们。在 package.xml 旁边创建一个 setup.cfg 文件。

Linux

touch setup.cfg

macOS

touch setup.cfg

Windows

type nul > touch setup.cfg

将以下内容放入其中：

[develop]
script_dir=$base/lib/talker_py
[install]
install_scripts=$base/lib/talker_py

你需要使用 console_scripts 入口点来定义要安装的可执行文件。每个条目的格式为 executable_name = some.module:function。第一部分指定要创建的可执行文件的名称。第二部分指定可执行文件启动时应运行的函数。此软件包需要创建一个名为 talker_py_node 的可执行文件，该可执行文件需要调用 talker_py 模块中的 main 函数。将以下入口点规范作为另一个参数添加到 setup.py 中的 setup() 调用中。

entry_points={
    'console_scripts': [
        'talker_py_node = talker_py:main',
    ],
},

talker_py_node 文件不再需要。删除文件 talker_py_node 并删除 scripts/ 目录。

Linux

rm scripts/talker_py_node
rmdir scripts

macOS

rm scripts/talker_py_node
rmdir scripts

Windows

del scripts/talker_py_node
rd scripts

添加 console_scripts 是对 setup.py 的最后一项更改。你最终的 setup.py 应该如下所示：

from setuptools import setup

package_name = 'talker_py'

setup(
    name=package_name,
    version='1.0.0',
    packages=['talker_py'],
    package_dir={'': 'src'},
    install_requires=['setuptools'],
    zip_safe=True,
    data_files=[
        ('share/ament_index/resource_index/packages',
            ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
    ],
    maintainer='Brian Gerkey',
    maintainer_email='gerkey@example.com',
    description='The talker_py package',
    license='BSD',
    entry_points={
        'console_scripts': [
            'talker_py_node = talker_py:main',
        ],
    },
)

迁移 `src/talker_py/init.py` 中的 Python 代码

ROS 2 改变了很多 Python 代码的最佳实践。首先按原样迁移代码。在有了可以工作的东西之后，后续重构代码会更容易。

使用 `rclpy` 代替 `rospy`

ROS 2 软件包使用 rclpy 而不是 rospy。要使用 rclpy，你必须做两件事：

导入 rclpy
初始化 rclpy

删除导入 rospy 的语句。

# Remove this
import rospy

将其替换为导入 rclpy 的语句。

import rclpy

在 main() 函数的最开头添加 rclpy.init() 调用。

def main():
    # Add this line
    rclpy.init()

在后台执行回调

ROS 1 和 ROS 2 都使用回调。在 ROS 1 中，回调始终在后台线程中执行，用户可以自由地使用 rate.sleep() 等调用阻塞主线程。在 ROS 2 中，rclpy 使用 Executors 让用户更好地控制回调在哪里被调用。在移植使用 rate.sleep() 等阻塞调用的代码时，你必须确保这些调用不会干扰 executor。一种方法是为 executor 创建一个专用线程。

首先，添加这两个导入语句。

import threading

from rclpy.executors import ExternalShutdownException

接下来，添加一个名为 spin_in_background() 的顶层函数。此函数请求默认 executor 执行回调，直到有东西关闭它。

def spin_in_background():
    executor = rclpy.get_global_executor()
    try:
        executor.spin()
    except ExternalShutdownException:
        pass

在 main() 函数中，紧接 rclpy.init() 调用之后添加以下代码，以启动一个调用 spin_in_background() 的线程。

    # In rospy callbacks are always called in background threads.
    # Spin the executor in another thread for similar behavior in ROS 2.
    t = threading.Thread(target=spin_in_background)
    t.start()

最后，在程序结束时通过将以下语句放在 main() 函数底部来 join 线程。

    t.join()

创建节点

在 ROS 1 中，Python 脚本每个进程只能创建一个节点，API init_node() 创建它。在 ROS 2 中，单个 Python 脚本可以创建多个节点，创建节点的 API 名为 create_node。

删除对 rospy.init_node() 的调用：

rospy.init_node('talker')

添加对 rclpy.create_node() 的新调用，并将结果存储在名为 node 的变量中：

node = rclpy.create_node('talker')

我们必须告诉 executor 这个节点。在创建节点之后紧接着添加以下行：

rclpy.get_global_executor().add_node(node)

创建 publisher

在 ROS 1 中，用户通过实例化 Publisher 类来创建 publisher。在 ROS 2 中，用户通过节点的 create_publisher() API 来创建 publisher。 create_publisher() API 与 ROS 1 有一个不太方便的差异：topic 名称和 topic 类型参数的位置互换了。

删除 rospy.Publisher 实例的创建。

pub = rospy.Publisher('chatter', String, queue_size=10)

将其替换为 node.create_publisher() 调用。

pub = node.create_publisher(String, 'chatter', 10)

创建 rate

在 ROS 1 中，用户直接创建 Rate 实例，而在 ROS 2 中，用户通过节点的 create_rate() API 来创建。

删除 rospy.Rate 实例的创建。

rate = rospy.Rate(10)  # 10hz

将其替换为 node.create_rate() 调用。

rate = node.create_rate(10)  # 10hz

在 `rclpy.ok()` 上循环

在 ROS 1 中，rospy.is_shutdown() API 指示进程是否被要求关闭。在 ROS 2 中，rclpy.ok() API 执行此操作。

删除语句 not rospy.is_shutdown()

while not rospy.is_shutdown():

将其替换为 rclpy.ok() 调用。

while rclpy.ok():

创建带当前时间的 `String` 消息

你必须对以下行做一些更改：

hello_str = "hello world %s" % rospy.get_time()

在 ROS 2 中，你需要：

必须从 Clock 实例获取时间
应该使用 f-strings 来格式化 str 数据，因为 % 在活跃的 Python 版本中不推荐使用
必须实例化一个 std_msgs.msg.String 实例

首先获取时间。 ROS 2 节点有一个 Clock 实例。将 rospy.get_time() 调用替换为 node.get_clock().now() 以从节点的时钟获取当前时间。

接下来，将 % 的使用替换为 f-string：f'hello world {node.get_clock().now()}'。

最后，实例化一个 std_msgs.msg.String() 实例并将上述内容分配给该实例的 data 属性。你的最终代码应该如下所示：

hello_str = String()
hello_str.data = f'hello world {node.get_clock().now()}'

记录信息消息

在 ROS 2 中，你必须通过 Logger 实例发送日志消息，而节点就有一个。

删除对 rospy.loginfo() 的调用。

rospy.loginfo(hello_str)

将其替换为对节点 Logger 实例的 info() 调用。

node.get_logger().info(hello_str.data)

这是对 src/talker_py/__init__.py 的最后一项更改。你的文件应该如下所示：

import threading

import rclpy
from rclpy.executors import ExternalShutdownException
from std_msgs.msg import String


def spin_in_background():
    executor = rclpy.get_global_executor()
    try:
        executor.spin()
    except ExternalShutdownException:
        pass


def main():
    rclpy.init()
    # In rospy callbacks are always called in background threads.
    # Spin the executor in another thread for similar behavior in ROS 2.
    t = threading.Thread(target=spin_in_background)
    t.start()

    node = rclpy.create_node('talker')
    rclpy.get_global_executor().add_node(node)
    pub = node.create_publisher(String, 'chatter', 10)
    rate = node.create_rate(10)  # 10hz

    while rclpy.ok():
        hello_str = String()
        hello_str.data = f'hello world {node.get_clock().now()}'
        node.get_logger().info(hello_str.data)
        pub.publish(hello_str)
        rate.sleep()

    t.join()

构建并运行 `talker_py_node`

创建三个终端：

一个用于构建 talker_py
一个用于运行 talker_py_node
一个用于 echo talker_py_node 发布的消息

在第一个终端中构建工作空间。

Linux

cd ~/ros2_talker_py
. /opt/ros/{DISTRO}/setup.bash
colcon build

macOS

cd ~/ros2_talker_py
. /opt/ros/{DISTRO}/setup.bash
colcon build

Windows

cd \ros2_talker_py
call C:\dev\ros2\local_setup.bat
colcon build

在第二个终端中 source 你的工作空间，并运行 talker_py_node。

Linux

cd ~/ros2_talker_py
. install/setup.bash
ros2 run talker_py talker_py_node

macOS

cd ~/ros2_talker_py
. install/setup.bash
ros2 run talker_py talker_py_node

Windows

cd \ros2_talker_py
call install\setup.bat
ros2 run talker_py talker_py_node

在第三个终端中 echo 节点发布的消息：

Linux

. /opt/ros/{DISTRO}/setup.bash
ros2 topic echo /chatter

macOS

. /opt/ros/{DISTRO}/setup.bash
ros2 topic echo /chatter

Windows

call C:\dev\ros2\local_setup.bat
ros2 topic echo /chatter

你应该在第二个终端中看到带有当前时间的消息被发布，并在第三个终端中接收到相同的消息。

重构代码以使用 ROS 2 约定

你已经成功地将 ROS 1 Python 软件包迁移到了 ROS 2！现在你已经有了可以工作的东西，考虑重构它以更好地与 ROS 2 的 Python API 对齐。遵循这两个原则。

创建一个继承自 Node 的类。
所有工作都在回调中完成，并且永远不要阻塞这些回调。

例如，创建一个继承自 Node 的 Talker 类。至于在回调中完成工作，使用带有回调的 Timer（定时器）来代替 rate.sleep()。让定时器回调发布消息并返回。让 main() 创建一个 Talker 实例，而不是使用 rclpy.create_node()，并将主线程交给 executor 执行。

你重构后的代码可能如下所示：

import rclpy
from rclpy.node import Node
from rclpy.executors import ExternalShutdownException
from std_msgs.msg import String


class Talker(Node):

    def __init__(self, **kwargs):
        super().__init__('talker', **kwargs)

        self._pub = self.create_publisher(String, 'chatter', 10)
        self._timer = self.create_timer(1 / 10, self.do_publish)

    def do_publish(self):
        hello_str = String()
        hello_str.data = f'hello world {self.get_clock().now()}'
        self.get_logger().info(hello_str.data)
        self._pub.publish(hello_str)


def main():
    try:
        with rclpy.init():
            rclpy.spin(Talker())
    except (ExternalShutdownException, KeyboardInterrupt):
        pass

结论

你已经学习了如何将一个 Python ROS 1 示例软件包迁移到 ROS 2。从现在开始，在迁移你自己的 Python 软件包时，请参考 Python 软件包迁移参考页面。

Python 软件包迁移示例

前提条件

ROS 1 代码

Linux

macOS

Windows

Linux

macOS

Windows

迁移 package.xml

删除 CMakeLists.txt

迁移 setup.py

Linux

macOS

Windows

迁移 Python 脚本并创建 setup.cfg

Linux

macOS

Windows

Linux

macOS

Windows

迁移 src/talker_py/__init__.py 中的 Python 代码

使用 rclpy 代替 rospy

在后台执行回调

创建节点

创建 publisher

创建 rate

在 rclpy.ok() 上循环

创建带当前时间的 String 消息

记录信息消息

构建并运行 talker_py_node

Linux

macOS

Windows

Linux

macOS

Windows

Linux

macOS

Windows

重构代码以使用 ROS 2 约定

结论

迁移 `package.xml`

删除 `CMakeLists.txt`

迁移 `setup.py`

迁移 Python 脚本并创建 `setup.cfg`

迁移 `src/talker_py/init.py` 中的 Python 代码

使用 `rclpy` 代替 `rospy`

在 `rclpy.ok()` 上循环

创建带当前时间的 `String` 消息

构建并运行 `talker_py_node`