Python ROS2【机器人中间件框架】 简介

ROS2（Robot Operating System 2） 是一个开源的机器人中间件框架，用于构建复杂的机器人系统。相比 ROS1，ROS2 在实时性、安全性、跨平台支持和分布式架构上进行了全面升级。

核心优势

1. DDS 通信协议
   使用 Data Distribution Service (DDS) 实现去中心化通信，支持实时数据传输和 QoS 策略。
2. 跨平台支持
   支持 Linux、Windows、macOS 和实时操作系统（RTOS）。
3. 多语言兼容
   支持 Python、C++、Java 等，Python 是最常用的开发语言。
4. 微服务架构
   节点可独立编译/部署，适合模块化开发。

Python ROS2 核心概念

1. 节点 (Node)

import rclpy
from rclpy.node import Node

class MyNode(Node):
    def __init__(self):
        super().__init__('my_node_name')
        self.get_logger().info("Node started!")

def main():
    rclpy.init()
    node = MyNode()
    rclpy.spin(node)  # 保持节点运行
    rclpy.shutdown()

2. 话题通信 (Topic)

# 发布者
from std_msgs.msg import String
publisher = self.create_publisher(String, 'chatter', 10)
publisher.publish(String(data="Hello ROS2"))

# 订阅者
def callback(msg):
    self.get_logger().info(f"Received: {msg.data}")
self.create_subscription(String, 'chatter', callback, 10)

3. 服务 (Service)

# 服务端
from example_interfaces.srv import AddTwoInts

def add(request, response):
    response.sum = request.a + request.b
    return response

self.srv = self.create_service(AddTwoInts, 'add_two_ints', add)

# 客户端
client = self.create_client(AddTwoInts, 'add_two_ints')
request = AddTwoInts.Request(a=2, b=3)
future = client.call_async(request)

4. 参数 (Parameters)

# 声明参数
self.declare_parameter('speed', 0.5)

# 读取参数
speed = self.get_parameter('speed').value

开发流程示例

1. 创建 ROS2 包

ros2 pkg create my_py_pkg --build-type ament_python

2. 编写节点代码

保存到 my_py_pkg/my_py_pkg/node_script.py

3. 配置入口点

在 setup.py 中添加：

entry_points={
    'console_scripts': ['my_node = my_py_pkg.node_script:main'],
}

4. 编译运行

colcon build --packages-select my_py_pkg
source install/setup.bash
ros2 run my_py_pkg my_node

关键工具

学习资源

1. 官方文档
2. ROS2 Python 示例
3. 命令行工具：ros2 --help
4. 教程：ros2 run demo_nodes_py talker/listener

提示：使用 colcon build 编译包，通过 source install/setup.bash 激活环境。

ROS2 应用领域

ROS2 广泛应用于现代机器人系统的开发，涵盖以下领域：

1. 工业自动化

• 机械臂控制
• 物流搬运机器人
• 自动化质检系统

# 机械臂关节控制示例
from control_msgs.action import FollowJointTrajectory
from trajectory_msgs.msg import JointTrajectoryPoint

class RobotArmController(Node):
    def __init__(self):
        super().__init__('arm_controller')
        self._action_client = ActionClient(self, FollowJointTrajectory, '/arm_controller/follow_joint_trajectory')
    
    def move_to_position(self, positions):
        goal_msg = FollowJointTrajectory.Goal()
        point = JointTrajectoryPoint()
        point.positions = positions  # [rad]
        point.time_from_start = Duration(seconds=2).to_msg()
        goal_msg.trajectory.points.append(point)
        self._action_client.send_goal_async(goal_msg)

2. 自动驾驶

• 环境感知（激光雷达/摄像头）
• 路径规划
• 车辆控制

# 激光雷达数据处理
from sensor_msgs.msg import LaserScan

class ObstacleDetector(Node):
    def __init__(self):
        super().__init__('obstacle_detector')
        self.subscription = self.create_subscription(
            LaserScan,
            '/scan',
            self.scan_callback,
            10)
    
    def scan_callback(self, msg):
        # 检测前方障碍物（简化示例）
        front_scan = msg.ranges[len(msg.ranges)//4:3*len(msg.ranges)//4]
        if min(front_scan) < 1.0:  # 1米内有障碍物
            self.get_logger().warn("Obstacle detected!")

3. 服务机器人

• 室内导航
• 人机交互
• 任务调度

# 自主导航到目标点
from nav2_msgs.action import NavigateToPose
from geometry_msgs.msg import PoseStamped

class DeliveryRobot(Node):
    def send_to_location(self, x, y):
        goal_pose = PoseStamped()
        goal_pose.header.frame_id = 'map'
        goal_pose.pose.position.x = x
        goal_pose.pose.position.y = y
        
        goal_msg = NavigateToPose.Goal()
        goal_msg.pose = goal_pose
        self.nav_client.wait_for_server()
        self.nav_client.send_goal_async(goal_msg)

4. 无人机系统

• 飞行控制
• 集群协同
• 航拍测绘

# 无人机航点飞行
from geographic_msgs.msg import GeoPoseStamped
from mavros_msgs.srv import CommandBool

class DroneController(Node):
    async def fly_waypoints(self, coordinates):
        # 解锁无人机
        arm = self.create_client(CommandBool, '/mavros/cmd/arming')
        await arm.call_async(CommandBool.Request(value=True))
        
        # 发送航点
        for lat, lon, alt in coordinates:
            goal = GeoPoseStamped()
            goal.position.latitude = lat
            goal.position.longitude = lon
            goal.position.altitude = alt
            self.waypoint_publisher.publish(goal)
            await asyncio.sleep(5)  # 等待到达

5. 医疗机器人

• 手术辅助
• 医疗物流
• 康复训练

# 手术器械精准控制
from sensor_msgs.msg import Joy

class SurgicalController(Node):
    def __init__(self):
        self.sub = self.create_subscription(Joy, '/surgical_joystick', self.joy_callback, 10)
        self.tool_pub = self.create_publisher(Twist, '/surgical_tool/control', 10)
    
    def joy_callback(self, msg):
        # 将游戏手柄输入转换为器械控制
        control = Twist()
        control.linear.x = msg.axes[1] * 0.1  # 精度控制
        control.linear.y = msg.axes[0] * 0.1
        self.tool_pub.publish(control)

综合案例：智能仓库机器人

# warehouse_robot.py
import rclpy
from rclpy.action import ActionClient
from rclpy.node import Node
from nav2_msgs.action import NavigateToPose
from std_msgs.msg import String
from warehouse_interfaces.srv import ItemRequest

class WarehouseRobot(Node):
    def __init__(self):
        super().__init__('warehouse_robot')
        
        # 导航客户端
        self.nav_client = ActionClient(self, NavigateToPose, 'navigate_to_pose')
        
        # 库存服务
        self.inventory_client = self.create_client(ItemRequest, 'item_location')
        
        # 任务订阅
        self.create_subscription(String, 'delivery_tasks', self.task_callback, 10)
        
        # 当前状态
        self.current_task = None
    
    async def task_callback(self, msg):
        item_id = msg.data
        self.get_logger().info(f"New task: Fetch {item_id}")
        
        # 查询物品位置
        item_loc = await self.get_item_location(item_id)
        if not item_loc:
            self.get_logger().error(f"Item {item_id} not found!")
            return
            
        # 导航到物品位置
        await self.navigate_to(item_loc.x, item_loc.y)
        
        # 模拟拾取操作
        await asyncio.sleep(2)
        
        # 导航到发货区
        await self.navigate_to(5.0, 3.0)
        
        self.get_logger().info(f"Item {item_id} delivered!")
    
    async def get_item_location(self, item_id):
        req = ItemRequest.Request()
        req.item_id = item_id
        future = self.inventory_client.call_async(req)
        await future
        return future.result().location
    
    async def navigate_to(self, x, y):
        goal_pose = PoseStamped()
        goal_pose.header.frame_id = 'map'
        goal_pose.pose.position.x = x
        goal_pose.pose.position.y = y
        
        goal_msg = NavigateToPose.Goal()
        goal_msg.pose = goal_pose
        
        self.nav_client.wait_for_server()
        send_goal_future = self.nav_client.send_goal_async(goal_msg)
        await send_goal_future
        goal_handle = send_goal_future.result()
        
        # 等待导航完成
        result_future = goal_handle.get_result_async()
        await result_future
        return result_future.result().result

系统架构图

+-------------------+     +----------------+     +-----------------+
| 任务管理系统      |     | 导航系统       |     | 库存数据库      |
| (发布任务)        |---->| (路径规划/避障)|<----| (物品位置查询)  |
+-------------------+     +-------+--------+     +-----------------+
                                  |
                          +-------v--------+
                          | 机器人执行器   |
                          | - 移动底盘     |
                          | - 机械臂       |
                          | - 传感器       |
                          +----------------+

开发进阶技巧

1. 组件化设计

   # 独立导航组件
   class NavigationComponent(Node):
       def __init__(self):
           super().__init__('navigation_component')
           # ... 导航实现 ...
   
   # 独立任务处理组件
   class TaskHandlerComponent(Node):
       def __init__(self):
           super().__init__('task_handler')
           # ... 任务处理 ...
   

2. QoS策略配置

   from rclpy.qos import QoSProfile, QoSReliabilityPolicy
   
   # 确保关键指令可靠传输
   reliable_qos = QoSProfile(
       depth=10,
       reliability=QoSReliabilityPolicy.RELIABLE
   )
   
   self.cmd_pub = self.create_publisher(
       Twist, 
       '/cmd_vel', 
       reliable_qos
   )
   

3. 生命周期管理

   from rclpy.lifecycle import LifecycleNode
   
   class SafetyMonitor(LifecycleNode):
       def on_activate(self):
           self.get_logger().info("Safety system activated")
           # 启动监控线程
   
       def on_deactivate(self):
           self.get_logger().info("Safety system deactivated")
           # 停止监控
   

行业应用案例

1. 亚马逊物流机器人：使用ROS2实现仓库自主导航
2. 达芬奇手术系统：基于ROS2的手术器械控制
3. NASA火星车：ROS2用于太空探测系统
4. 宝马生产线：ROS2协调工业机械臂
5. 农业无人机：ROS2实现自动喷洒和监测

实际开发建议：

1. 使用ros2 launch管理复杂系统启动
2. 利用ros2 bag记录和回放传感器数据
3. 使用ros2 doctor定期检查系统健康状态
4. 通过ros2 security启用通信加密
5. 使用ros2 tracing进行性能分析

完整项目示例参考：ROS2 Warehouse Robot Demo
官方进阶教程：ROS2 Navigation Guide

Python 图书推荐

JAVA 图书推荐