<think>我们正在讨论的是ROS2中替代rospy的库。根据用户提供的引用，我们知道在ROS1中，rospy是用于Python的客户端库，提供了与roscpp相似的功能，但API设计有所不同。在ROS2中，原来的ROS1核心概念被重新设计，并使用了新的中间件实现。因此，ROS2中Python客户端库是`rclpy`。 ### ROS2中的Python客户端库：rclpy 在ROS2中，`rclpy`（ROS Client Library for Python）是用于编写ROS2节点的Python库，它替代了ROS1中的`rospy`。`rclpy`提供了创建节点、发布者、订阅者、服务、客户端等的基本接口。 #### 主要模块和类 1. **节点（Node）**： - 使用`rclpy.create_node()`或通过继承`rclpy.node.Node`类来创建节点。 - 示例： ```python import rclpy from rclpy.node import Node class MyNode(Node): def __init__(self): super().__init__('my_node_name') # 创建发布者、订阅者等 def main(): rclpy.init() node = MyNode() rclpy.spin(node) node.destroy_node() rclpy.shutdown() ``` 2. **发布者（Publisher）**： - 使用节点的`create_publisher()`方法创建发布者。 - 示例： ```python from std_msgs.msg import String self.publisher = self.create_publisher(String, 'topic_name', 10) ``` 3. **订阅者（Subscriber）**： - 使用节点的`create_subscription()`方法创建订阅者。 - 示例： ```python self.subscription = self.create_subscription( String, 'topic_name', self.listener_callback, 10) ``` 4. **服务（Service）**： - 服务端：使用`create_service()`创建服务端。 - 客户端：使用`create_client()`创建客户端。 - 示例（服务端）： ```python from example_interfaces.srv import AddTwoInts self.srv = self.create_service(AddTwoInts, 'add_two_ints', self.add_two_ints_callback) ``` 5. **参数（Parameters）**： - 声明参数：`declare_parameter('param_name', default_value)` - 获取参数：`get_parameter('param_name').value` 6. **定时器（Timer）**： - 使用`create_timer(period_sec, callback)`创建定时器。 - 示例： ```python self.timer = self.create_timer(0.5, self.timer_callback) ``` 7. **日志（Logging）**： - 使用节点的`get_logger()`方法获取日志记录器。 - 示例： ```python self.get_logger().info('Hello, ROS2!') ``` #### 与rospy的主要区别 1. **节点创建**：在`rclpy`中，节点必须显式创建和销毁，且通常通过继承`Node`类来实现。 2. **命名空间**：节点名、话题名、服务名等可以通过节点构造函数的参数（如`namespace`）来设置。 3. **执行模型**：ROS2使用执行器（`Executor`）来管理节点的回调，支持单线程、多线程等模式。 4. **生命周期**：节点有明确的生命周期管理（创建、销毁），且需要显式初始化（`rclpy.init()`）和关闭（`rclpy.shutdown()`）。 #### 示例：一个简单的ROS2节点 ```python import rclpy 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) self.timer = self.create_timer(0.5, 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): rclpy.init(args=args) minimal_publisher = MinimalPublisher() rclpy.spin(minimal_publisher) minimal_publisher.destroy_node() rclpy.shutdown() if __name__ == '__main__': main() ``` ### 注意事项 - **异常处理**：ROS2中的异常处理类似于ROS1，但异常类在`rclpy.exceptions`模块中定义，例如`RCLError`。 - **时间**：使用`self.get_clock().now()`获取当前时间。 - **参数类型**：参数可以是整数、浮点数、字符串、列表等基本类型。 ### 结论 在ROS2中，使用`rclpy`库作为Python编程的客户端库，它提供了与ROS1中`rospy`类似的功能，但API设计更面向对象，并且与ROS2的架构（如DDS）紧密结合。 §§相关问题§§ 1. ROS2中如何创建和使用自定义消息？ 2. 如何在ROS2中实现节点间的同步通信？ 3. ROS2中的执行器（Executor）有几种类型？如何选择？ 4. ROS2与ROS1在参数服务器上有哪些主要区别？ </think> 在 ROS2 中，替代 ROS1 `rospy` 的 Python 客户端库是 **`rclpy`**（ROS Client Library for Python）。这是 ROS2 的官方 Python 客户端库，提供了与 `rospy` 类似的功能但采用全新设计的 API。以下是关键类和功能的对比： ### ROS1 (rospy) → ROS2 (rclpy) 核心映射表 | ROS1 (rospy) 功能 | ROS2 (rclpy) 替代方案 | 使用示例 | |---------------------------|-----------------------------------------------|--------------------------------------------------------------------------| | `rospy.init_node()` | `rclpy.init()` + `Node` 类 | `node = rclpy.create_node('node_name')` | | `rospy.Publisher` | `Node.create_publisher()` | `pub = node.create_publisher(msg_type, 'topic', qos_profile)` | | `rospy.Subscriber` | `Node.create_subscription()` | `sub = node.create_subscription(msg_type, 'topic', callback, qos)` | | `rospy.Service` | `Node.create_service()` | `srv = node.create_service(srv_type, 'service_name', callback)` | | `rospy.ServiceProxy` | `Node.create_client()` | `client = node.create_client(srv_type, 'service_name')` | | `rospy.Timer` | `Node.create_timer()` | `timer = node.create_timer(0.1, timer_callback)` | | `rospy.Rate` | `Node.create_rate()` 或 `rclpy.time.Duration` | `rate = node.create_rate(10)` | | `rospy.get_param()` | `Node.declare_parameter()` | `node.declare_parameter('param_name', default_value)` | | `rospy.loginfo()` | `Node.get_logger().info()` | `node.get_logger().info("Message")` | | `rospy.spin()` | `rclpy.spin(node)` | `rclpy.spin(node)` | | `rospy.is_shutdown()` | `rclpy.ok()` | `while rclpy.ok(): ...` | | `rospy.on_shutdown()` | `context.on_shutdown()` | `context = rclpy.Context()`<br>`context.on_shutdown(callback)` | ### 关键差异说明 1. **节点生命周期管理**： ```python # ROS2 显式生命周期管理 import rclpy from rclpy.node import Node rclpy.init() # 替代 rospy.init_node() node = Node('my_node') # 创建节点对象 rclpy.spin(node) # 启动节点循环 node.destroy_node() # 显式销毁节点 rclpy.shutdown() # 关闭ROS2上下文 ``` 2. **QoS 策略**（重要新特性）： ```python from rclpy.qos import QoSProfile, ReliabilityPolicy qos = QoSProfile( depth=10, # 队列深度 reliability=ReliabilityPolicy.RELIABLE # 或 BEST_EFFORT ) pub = node.create_publisher(String, 'topic', qos) ``` 3. **参数声明式处理**： ```python # 声明参数并获取值 node.declare_parameter('velocity_limit', 5.0) max_vel = node.get_parameter('velocity_limit').value ``` 4. **多节点执行模型**： ```python # 使用 Executor 管理多节点 from rclpy.executors import MultiThreadedExecutor executor = MultiThreadedExecutor() executor.add_node(node1) executor.add_node(node2) executor.spin() # 同时运行多个节点 ``` ### 完整 ROS2 节点示例 ```python import rclpy 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) self.timer = self.create_timer(0.5, self.timer_callback) self.counter = 0 # 声明参数 self.declare_parameter('prefix', 'Message') def timer_callback(self): msg = String() prefix = self.get_parameter('prefix').value msg.data = f'{prefix}: {self.counter}' self.publisher.publish(msg) self.get_logger().info(f'Publishing: "{msg.data}"') self.counter += 1 def main(args=None): rclpy.init(args=args) publisher = MinimalPublisher() rclpy.spin(publisher) publisher.destroy_node() rclpy.shutdown() if __name__ == '__main__': main() ``` ### 迁移注意事项 1. **命名空间处理**： ```python # ROS2 通过节点构造参数设置命名空间 node = Node('node_name', namespace='my_ns') ``` 2. **时间处理**： ```python # 获取当前时间 now = self.get_clock().now() duration = rclpy.time.Duration(seconds=2.5) ``` 3. **异常处理**： ```python from rclpy.exceptions import ROSException try: response = client.call(request) except ROSException as e: self.get_logger().error(f'Service call failed: {e}') ```