下面是一个完整的 ROS 2 Python Launch 文件编写教程,包含从创建包到编写复杂 Launch 文件的全过程。

1. 准备工作

1.1 创建 ROS 2 工作空间

mkdir -p ~/ros2_launch_ws/src
cd ~/ros2_launch_ws
colcon build
source install/setup.bash

1.2 创建新包

cd src
ros2 pkg create my_launch_pkg --build-type ament_python
cd my_launch_pkg

2. 创建基本节点

2.1 创建简单节点

在 my_launch_pkg/my_launch_pkg 目录下创建 simple_node.py:

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node

class SimpleNode(Node):
    def __init__(self):
        super().__init__('simple_node')
        self.get_logger().info('Simple Node 已启动!')
        self.timer = self.create_timer(1.0, self.timer_callback)
        self.counter = 0
    
    def timer_callback(self):
        self.counter += 1
        self.get_logger().info(f'计数器: {self.counter}')

def main(args=None):
    rclpy.init(args=args)
    node = SimpleNode()
    rclpy.spin(node)
    rclpy.shutdown()

if __name__ == '__main__':
    main()

2.2 设置可执行权限

chmod +x my_launch_pkg/simple_node.py

3. 编写第一个 Launch 文件

3.1 创建 Launch 目录

mkdir -p launch
touch launch/first_launch.py
chmod +x launch/first_launch.py

3.2 基本 Launch 文件内容

编辑 launch/first_launch.py:

from launch import LaunchDescription
from launch_ros.actions import Node

def generate_launch_description():
    return LaunchDescription([
        Node(
            package='my_launch_pkg',
            executable='simple_node',
            name='my_simple_node',
            output='screen',
            parameters=[{'counter_start': 5}]
        )
    ])

4. 修改 setup.py 安装文件

编辑 setup.py:

from setuptools import setup
import os
from glob import glob

package_name = 'my_launch_pkg'

setup(
    name=package_name,
    version='0.0.0',
    packages=[package_name],
    data_files=[
        ('share/ament_index/resource_index/packages',
            ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
        # 添加 launch 文件
        (os.path.join('share', package_name, 'launch'), glob('launch/*.py')),
    ],
    install_requires=['setuptools'],
    zip_safe=True,
    maintainer='your_name',
    maintainer_email='your_email@example.com',
    description='ROS 2 Launch 文件示例包',
    license='Apache License 2.0',
    tests_require=['pytest'],
    entry_points={
        'console_scripts': [
            'simple_node = my_launch_pkg.simple_node:main',
        ],
    },
)

5. 构建并运行 Launch 文件

5.1 构建包

cd ~/ros2_launch_ws
colcon build --packages-select my_launch_pkg
source install/setup.bash

5.2 运行 Launch 文件

ros2 launch my_launch_pkg first_launch.py

6. 进阶 Launch 文件功能

6.1 启动多个节点

创建 launch/multi_node_launch.py:

from launch import LaunchDescription
from launch_ros.actions import Node

def generate_launch_description():
    # 创建第一个节点
    node1 = Node(
        package='my_launch_pkg',
        executable='simple_node',
        name='node_1',
        output='screen',
        parameters=[{'counter_start': 10}]
    )
    
    # 创建第二个节点
    node2 = Node(
        package='my_launch_pkg',
        executable='simple_node',
        name='node_2',
        output='screen',
        parameters=[{'counter_start': 20}],
        # 设置命名空间
        namespace='group1'
    )
    
    # 创建第三个节点(带重映射)
    node3 = Node(
        package='my_launch_pkg',
        executable='simple_node',
        name='node_3',
        output='screen',
        remappings=[
            ('/topic_name', '/new_topic_name')
        ]
    )
    
    return LaunchDescription([
        node1,
        node2,
        node3
    ])

6.2 使用参数和条件

创建 launch/advanced_launch.py:

from launch import LaunchDescription
from launch.actions import DeclareLaunchArgument, LogInfo
from launch.conditions import IfCondition, UnlessCondition
from launch.substitutions import LaunchConfiguration, TextSubstitution
from launch_ros.actions import Node

def generate_launch_description():
    # 声明启动参数
    use_sim_time_arg = DeclareLaunchArgument(
        'use_sim_time',
        default_value='false',
        description='是否使用仿真时间'
    )
    
    log_level_arg = DeclareLaunchArgument(
        'log_level',
        default_value='info',
        description='日志级别 (debug, info, warn, error, fatal)'
    )
    
    # 创建带条件的节点
    sim_node = Node(
        package='my_launch_pkg',
        executable='simple_node',
        name='simulation_node',
        output='screen',
        # 使用仿真时间
        parameters=[{'use_sim_time': LaunchConfiguration('use_sim_time')}],
        # 仅当 use_sim_time 为 true 时启动
        condition=IfCondition(LaunchConfiguration('use_sim_time'))
    )
    
    # 创建另一个节点
    real_node = Node(
        package='my_launch_pkg',
        executable='simple_node',
        name='real_node',
        output='screen',
        # 设置日志级别
        arguments=['--ros-args', '--log-level', LaunchConfiguration('log_level')],
        # 当不使用仿真时间时启动
        condition=UnlessCondition(LaunchConfiguration('use_sim_time'))
    )
    
    # 信息日志
    info_log = LogInfo(
        msg=['使用仿真时间: ', LaunchConfiguration('use_sim_time')],
        condition=IfCondition(LaunchConfiguration('use_sim_time'))
    )
    
    return LaunchDescription([
        use_sim_time_arg,
        log_level_arg,
        sim_node,
        real_node,
        info_log
    ])

6.3 包含其他 Launch 文件

创建 launch/include_launch.py:

from launch import LaunchDescription
from launch.actions import IncludeLaunchDescription
from launch.launch_description_sources import PythonLaunchDescriptionSource
from launch.substitutions import PathJoinSubstitution
from launch_ros.substitutions import FindPackageShare

def generate_launch_description():
    # 包含其他 Launch 文件
    included_launch = IncludeLaunchDescription(
        PythonLaunchDescriptionSource([
            PathJoinSubstitution([
                FindPackageShare('my_launch_pkg'),
                'launch',
                'first_launch.py'
            ])
        ]),
        # 可以覆盖参数
        launch_arguments={
            'use_sim_time': 'true'
        }.items()
    )
    
    # 添加本地节点
    local_node = Node(
        package='my_launch_pkg',
        executable='simple_node',
        name='local_node',
        output='screen'
    )
    
    return LaunchDescription([
        included_launch,
        local_node
    ])

7. 使用事件处理程序

创建 launch/event_launch.py:

from launch import LaunchDescription
from launch.actions import RegisterEventHandler, EmitEvent
from launch.event_handlers import OnProcessStart, OnProcessExit
from launch.events import Shutdown
from launch_ros.actions import Node

def generate_launch_description():
    node = Node(
        package='my_launch_pkg',
        executable='simple_node',
        name='event_node',
        output='screen'
    )
    
    # 节点启动后执行的操作
    on_start = RegisterEventHandler(
        event_handler=OnProcessStart(
            target_action=node,
            on_start=[
                # 可以在这里执行其他操作
                # 例如:LogInfo(msg='节点已启动!')
            ]
        )
    )
    
    # 节点退出时执行的操作
    on_exit = RegisterEventHandler(
        event_handler=OnProcessExit(
            target_action=node,
            on_exit=[
                # 节点退出时关闭整个系统
                EmitEvent(event=Shutdown(reason='节点已退出'))
            ]
        )
    )
    
    return LaunchDescription([
        node,
        on_start,
        on_exit
    ])

8. 完整示例:带生命周期的节点启动

8.1 创建生命周期节点

在 my_launch_pkg 目录下创建 lifecycle_node.py:

#!/usr/bin/env python3
import rclpy
from rclpy.lifecycle import LifecycleNode, LifecycleState, TransitionCallbackReturn
from rclpy.lifecycle import State
from rclpy.timer import Timer

class MyLifecycleNode(LifecycleNode):
    def __init__(self):
        super().__init__('lifecycle_node')
        self.timer = None
        self.counter = 0
    
    def on_configure(self, previous_state: State) -> TransitionCallbackReturn:
        self.get_logger().info('正在配置节点...')
        # 初始化资源
        return TransitionCallbackReturn.SUCCESS
    
    def on_activate(self, previous_state: State) -> TransitionCallbackReturn:
        self.get_logger().info('正在激活节点...')
        # 启动定时器
        self.timer = self.create_timer(1.0, self.timer_callback)
        return super().on_activate(previous_state)
    
    def on_deactivate(self, previous_state: State) -> TransitionCallbackReturn:
        self.get_logger().info('正在停用节点...')
        # 停止定时器
        self.timer.cancel()
        self.timer = None
        return super().on_deactivate(previous_state)
    
    def on_cleanup(self, previous_state: State) -> TransitionCallbackReturn:
        self.get_logger().info('正在清理节点...')
        # 清理资源
        return TransitionCallbackReturn.SUCCESS
    
    def on_shutdown(self, previous_state: State) -> TransitionCallbackReturn:
        self.get_logger().info('正在关闭节点...')
        return TransitionCallbackReturn.SUCCESS
    
    def timer_callback(self):
        self.counter += 1
        self.get_logger().info(f'生命周期节点计数: {self.counter}')

def main(args=None):
    rclpy.init(args=args)
    node = MyLifecycleNode()
    rclpy.spin(node)
    rclpy.shutdown()

if __name__ == '__main__':
    main()

8.2 创建生命周期节点 Launch 文件

创建 launch/lifecycle_launch.py:

from launch import LaunchDescription
from launch.actions import DeclareLaunchArgument, ExecuteProcess, TimerAction
from launch_ros.actions import LifecycleNode
from launch.substitutions import LaunchConfiguration

def generate_launch_description():
    # 声明参数
    autostart_arg = DeclareLaunchArgument(
        'autostart', default_value='true',
        description='是否自动启动节点管理'
    )
    
    # 创建生命周期节点
    lifecycle_node = LifecycleNode(
        package='my_launch_pkg',
        executable='lifecycle_node',
        name='lifecycle_node',
        output='screen',
        namespace='',
        parameters=[],
        # 启用自动配置
        autostart=LaunchConfiguration('autostart')
    )
    
    # 配置节点
    configure_node = ExecuteProcess(
        cmd=['ros2 lifecycle set /lifecycle_node configure'],
        shell=True,
        # 延迟执行,确保节点已启动
        output='screen'
    )
    
    # 激活节点(延迟执行)
    activate_node = TimerAction(
        period=2.0,  # 延迟2秒
        actions=[
            ExecuteProcess(
                cmd=['ros2 lifecycle set /lifecycle_node activate'],
                shell=True,
                output='screen'
            )
        ]
    )
    
    return LaunchDescription([
        autostart_arg,
        lifecycle_node,
        configure_node,
        activate_node
    ])

9. 构建并测试

cd ~/ros2_launch_ws
colcon build --packages-select my_launch_pkg
source install/setup.bash

测试不同 Launch 文件:

# 测试基本 Launch
ros2 launch my_launch_pkg first_launch.py

# 测试多节点 Launch
ros2 launch my_launch_pkg multi_node_launch.py

# 测试带参数的 Launch
ros2 launch my_launch_pkg advanced_launch.py use_sim_time:=true log_level:=debug

# 测试生命周期节点 Launch
ros2 launch my_launch_pkg lifecycle_launch.py

10. 调试技巧

10.1 查看 Launch 文件描述

ros2 launch --print-description my_launch_pkg advanced_launch.py

10.2 查看启动参数

ros2 launch --show-args my_launch_pkg advanced_launch.py

10.3 调试模式

ros2 launch --debug my_launch_pkg first_launch.py

Python Launch 文件提供了比 XML 更强大的灵活性,允许使用完整的 Python 功能来创建复杂的启动场景。在实际项目中,可以根据需要组合这些技术来构建适合你系统的启动配置。

Logo

立足具身智能前沿赛道,致力于搭建全球化、开源化、全栈式技术交流与实践共创平台。

更多推荐