python核心基础，这关于基于Moveltg加 Ros2实战Python编程基础实课

ROS2 服务（Service）客户端 + 服务端完整代码梳理文档（第一版简易加法示例）目的是让了速了解整 个流程，

核心看点：

ROS2 中服务端代码中可以包含订阅功能，里常规标准写法。单个节点既能创建订阅者接收图像数据，同时又创建服务端对外提供查询接口，完全合在一起解决事情。

一、ROS2 服务通信整体概述

1. 通信模型分类

ROS2 两大常用同步 / 异步通信模式：

• 话题（Topic）：异步单向流式通信，发布者不停发、订阅者被动接收，一对多；无应答机制。
• 服务（Service）：同步问答式点对点通信，严格一问一答，固定客户端发请求、服务端处理后返回应答，一对一交互。

2. 服务通信核心角色

1. 服务端（Server）：常驻后台运行，注册服务名称，监听客户端请求，执行业务逻辑，计算结果后返回应答；你截图里service_adder_server.py就是服务端。
2. 客户端（Client）：主动发起请求，携带请求数据，等待服务端运算完成后接收返回结果；service_adder_client.py为客户端。
3. 自定义服务接口（.srv 文件）约定请求数据格式和应答数据格式，是两端统一的数据协议，相当于双方通信的 “数据契约”。接口独立存放于单独功能包learning_interface，不和客户端、服务端业务代码混在一起，实现解耦复用。

3. 完整通信流程

1. 先启动服务端：节点初始化 → 创建服务，绑定服务名 + 回调函数 → 持续自旋等待客户端接入。
2. 后启动客户端：节点初始化 → 创建客户端，指定相同服务名，循环等待服务端上线。
3. 客户端封装请求参数，异步发送请求。
4. 服务端捕获请求，触发回调函数执行业务（加法计算），填充应答数据。
5. 应答原路返回客户端，客户端解析结果，一次完整问答通信结束。

二、本次示例任务说明

任务目标

搭建 ROS2 标准服务通信框架，实现远程两整数加法运算：

1. 服务端：永久运行，提供加法计算能力，接收客户端传入的两个整数a、b，求和后把sum返回。
2. 客户端：从命令行读取两个数字，打包成请求发给服务端，等待并打印求和结果。

配套文件分工（意思是本次涉及到三个文件）

表格

三、逐段代码完整解析

（一）服务端代码 service_adder_server.py

python

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ @作者: 古月居(www.guyuehome.com) @说明: ROS2服务示例-提供加法器的服务器处理功能 """ import rclpy # ROS2 Python核心库 from rclpy.node import Node # ROS2节点父类 from learning_interface.srv import AddTwoInts # 导入自定义服务接口 # 自定义服务端节点类，继承Node父类 class adderServer(Node): def __init__(self, name): super().__init__(name) # 调用父类构造，初始化节点 # ========== 核心：创建服务对象 ========== # 参数1：接口类型AddTwoInts；参数2：全局服务名add_two_ints；参数3：收到请求后的回调函数 self.srv = self.create_service(AddTwoInts, 'add_two_ints', self.adder_callback) # 服务回调函数：客户端每发一次请求，自动触发一次 def adder_callback(self, request, response): # 取出请求里的a、b，相加，赋值给应答的sum response.sum = request.a + request.b # 日志打印收到的请求参数 self.get_logger().info(f'Incoming request\na: {request.a} b: {request.b}') # 必须return应答对象，数据传回客户端 return response # 程序入口函数 def main(args=None): rclpy.init(args=args) # ROS2环境初始化 node = adderServer("service_adder_server") # 实例化服务端节点 rclpy.spin(node) # 节点持续自旋，阻塞等待客户端请求 node.destroy_node() # 销毁节点 rclpy.shutdown() # 关闭ROS2环境 if __name__ == '__main__': main()

关键代码拆解

1. create_service(接口类型, 服务名, 回调函数)
   • 接口类型：AddTwoInts，来自独立接口包，规定收发数据结构；
   • 服务名add_two_ints：全网唯一标识，客户端必须填写完全一致的名称才能连上；
   • 回调函数：专门处理业务逻辑，入参固定request（请求）、response（应答）。
2. rclpy.spin(node)：让节点持续运行、监听服务请求，不会运行完直接退出。

（二）客户端代码 service_adder_client.py

python

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ @作者: 古月居(www.guyuehome.com) @说明: ROS2服务示例-发送两个加数，请求加法器计算 """ import sys import rclpy from rclpy.node import Node from learning_interface.srv import AddTwoInts # 和服务端导入完全相同的接口 # 自定义客户端节点类 class adderClient(Node): def __init__(self, name): super().__init__(name) # 创建客户端对象：同样填写接口类型+完全一致的服务名 self.client = self.create_client(AddTwoInts, 'add_two_ints') # 循环轮询，等待服务端启动上线，1秒超时重试 while not self.client.wait_for_service(timeout_sec=1.0): self.get_logger().info('service not available, waiting again...') # 创建请求数据包实例 self.request = AddTwoInts.Request() # 封装发送请求逻辑 def send_request(self): # 从终端命令行参数读取两个数字，赋值给请求里的a、b self.request.a = int(sys.argv[1]) self.request.b = int(sys.argv[2]) # 异步发送请求，不会阻塞程序卡死 self.future = self.client.call_async(self.request) def main(args=None): rclpy.init(args=args) node = adderClient("service_adder_client") node.send_request() # 发起请求 # 循环自旋，等待服务端返回结果 while rclpy.ok(): rclpy.spin_once(node) # 判断应答是否返回完成 if node.future.done(): try: # 取出应答数据 response = node.future.result() except Exception as e: node.get_logger().info(f'Service call failed %r' % (e,)) else: # 打印最终求和结果 node.get_logger().info(f'Result of add_two_ints: {response.sum}') break node.destroy_node() rclpy.shutdown() if __name__ == '__main__': main()

关键代码拆解

1. create_client：接口类型、服务名必须和服务端一字不差，否则无法配对通信；
2. wait_for_service：容错机制，防止客户端先启动、服务端没开导致连接失败；
3. call_async异步调用：非阻塞发送请求，不会卡死终端，配合spin_once轮询等待结果；
4. sys.argv[1]/[2]：运行客户端时，在命令行追加两个数字，动态传入加法参数。

（三）配套自定义接口 AddTwoInts.srv 内容

接口文件分为上下两段，---分隔请求、应答：

plaintext

# 请求段（客户端发给服务端） int64 a int64 b --- # 应答段（服务端返回客户端） int64 sum

编译后 ROS2 自动生成 Python 可导入的类，两端都能调用Request()、Response()封装解析数据。

四、代码运行操作步骤

1. 编译工作空间

bash

colcon build --packages-select learning_interface learning_service source install/setup.bash

1. 终端 1：启动服务端

bash

ros2 run learning_service service_adder_server

终端持续阻塞，等待客户端接入。

1. 终端 2：启动客户端，传入两个加数（示例 10 20）

bash

ros2 run learning_service service_adder_client 10 20

• 客户端日志输出最终sum=30；
• 服务端日志打印收到的a=10,b=20，一次问答通信完成。

五、以下是如果使用ai 开发本程的提示词参考，估记再过一段时间都 不需要以下提示词了，学习的目产最好自己会代码）

1. 接口开发心得

1. 接口单独新建独立功能包存放（示例learning_interface），多个业务包可以重复调用，不用重复定义；
2. .srv请求、应答变量名、数据类型两端严格统一，类型不匹配会直接通信报错；
3. CMakeLists.txt、package.xml 必须开启接口编译配置，否则编译不会自动生成 Python 接口类，代码 import 会爆红。

2. 服务端通用开发套路（固定 5 步模板）

1. 导入rclpy、Node+ 自定义服务接口；
2. 新建节点类继承 Node，构造函数内调用父类初始化；
3. create_service()绑定接口、服务名、回调函数；
4. 编写回调函数：读取 request 数据 → 业务逻辑处理 → 填充 response → return response；
5. main 函数初始化 rclpy、实例节点、spin()常驻运行，最后销毁节点关闭环境。

3. 客户端通用开发套路（固定 6 步模板）

1. 导入依赖，导入和服务端一模一样的接口；
2. 构造函数创建客户端，同名服务号；循环等待服务上线；
3. 实例化Request()对象，填充请求参数；
4. call_async()异步发送请求；
5. 循环spin_once轮询 future 状态，等待应答返回；
6. 解析 result () 拿到应答数据，做后续业务处理。

4. 开发特别要注意思的地方

1. 服务名必须两端完全一致，大小写、字符不能有差别；
2. 必须先编译接口包，再编译业务包，否则找不到接口；
3. 每次新开终端必须source install/setup.bash，否则 ros2 run 找不到可执行文件；
4. 服务是一问一答，服务端 spin 持续运行，客户端单次请求执行完毕自动退出；
5. 回调函数必须 return response，否则客户端收不到任何返回值。

接下我们还做一个更复杂的示例，这个示例是从摄像头中读出识别物本的坐标，

ROS2 进阶版：图像目标识别服务端完整解析文档

承接上一篇简易加法服务示例，本次是订阅图像话题 + OpenCV 颜色识别 + 服务应答坐标复合型服务端，融合「话题订阅」+「服务通信」两大 ROS2 核心机制，下面分模块完整梳理。

一、整体功能总览

1. 节点双重能力

这个节点不再是单纯的服务端，同时兼具两个角色：

1. 话题订阅者：持续订阅相机发布的image_raw图像话题，实时接收画面帧；
2. 服务端 Server：对外提供get_target_position服务，客户端发起查询请求时，把识别到的红色物体像素坐标 X、Y 原路返回。

2. 完整业务流水线

相机发布图像话题 → 当前节点订阅拿到 ROS 图像消息 → CvBridge 转 OpenCV 格式 → HSV 红色阈值筛选 + 轮廓检测 → 计算物体中心像素坐标并全局保存 → 客户端调用服务 → 服务回调把坐标打包成应答发回客户端。

3. 用到的依赖拆解

表格

二、自定义接口约定（GetObjectPosition.srv）

先明确接口结构，服务端、客户端必须统一：

srv

# 请求区（客户端发给服务端） bool get --- # 应答区（服务端返回客户端） int32 x int32 y

• get: True：代表合法查询指令，返回最新识别到的目标坐标；
• get: False：非法指令，应答 x、y 强制置 0 并打印提示。

三、逐段代码详细拆解

1. 全局常量定义

python

lower_red = np.array([0, 90, 128]) # HSV红色下限阈值 upper_red = np.array([180, 255, 255])# HSV红色上限阈值

固定红色 HSV 筛选区间，用来在画面里抠出红色物体。

2. 节点类初始化init

python

class ImageSubscriber(Node): def __init__(self, name): super().__init__(name) # 1. 创建图像订阅者：订阅/image_raw话题 self.sub = self.create_subscription( Image, 'image_raw', self.listener_callback, 10) self.cv_bridge = CvBridge() # 图像转换实例 # 2. 创建服务端：绑定接口、服务名、服务回调函数 self.srv = self.create_service( GetObjectPosition, 'get_target_position', self.object_position_callback) # 全局变量：缓存最新目标物体中心坐标 self.objectX = 0 self.objectY = 0

关键要点：

1. 同一个节点内可以同时创建订阅者 + 服务端，ROS2 允许单节点多通信实体；
2. self.objectX/Y是成员变量，图像回调实时更新，服务回调直接读取，实现数据跨函数共享。

3. 图像核心处理函数 object_detect (image)

python

def object_detect(self, image): hsv_img = cv2.cvtColor(image, cv2.COLOR_BGR2HSV) # BGR转HSV mask_red = cv2.inRange(hsv_img, lower_red, upper_red) # 二值化抠红色 # 轮廓检测 contours, hierarchy = cv2.findContours( mask_red, cv2.RETR_LIST, cv2.CHAIN_APPROX_NONE) for cnt in contours: if cnt.shape[0] < 150: # 过滤过小轮廓，去除噪声小点 continue # 获取轮廓外接矩形左上角(x,y)、宽w、高h (x, y, w, h) = cv2.boundingRect(cnt) cv2.drawContours(image, [cnt], -1, (0, 255, 0), 2) # 绘制轮廓 cv2.circle(image, (int(x+w/2), int(y+h/2)), 5,(0,255,0),-1) # 中心点标记 # 更新全局中心坐标 self.objectX = int(x + w/2) self.objectY = int(y + h/2) cv2.imshow("object", image) # 弹窗显示处理后图像 cv2.waitKey(50)

图像处理完整步骤：

1. 色彩空间转换：OpenCV 默认 BGR，HSV 做颜色分割鲁棒性更强；
2. 二值掩码：只保留红色像素，其余全部变黑；
3. 轮廓提取 + 噪声过滤：太小的色块直接丢弃，避免噪点干扰；
4. 外接矩形计算中心像素点，赋值给节点成员变量持久保存。

4. 话题订阅回调 listener_callback

相机每发布一帧图像，该函数自动触发：

python

def listener_callback(self, data): self.get_logger().info('Receiving video frame') # ROS图像消息 → OpenCV图像Mat image = self.cv_bridge.imgmsg_to_cv2(data, 'bgr8') self.object_detect(image) # 送入识别函数更新坐标

核心：CvBridge做格式转换，ROS 的Image消息不能直接给 cv2 处理，必须转换。

5. 服务回调函数 object_position_callback（应答客户端请求）

python

def object_position_callback(self, request, response): if request.get == True: # 合法请求：把缓存好的坐标填入应答 response.x = self.objectX response.y = self.objectY self.get_logger().info('Object position\nx: %d y: %d' % (response.x, response.y)) else: # 非法指令，坐标清零 response.x = 0 response.y = 0 self.get_logger().info('Invalid command') return response

服务端标准固定格式：入参固定request、response，修改 response 字段后必须 return 返回。

6. main 入口函数

python

def main(args=None): rclpy.init(args=args) node = ImageSubscriber("service_object_server") rclpy.spin(node) # 持续自旋：同时监听话题+等待服务调用 node.destroy_node() rclpy.shutdown()

rclpy.spin(node)会同时驱动订阅回调、服务回调两个事件，不用额外多写自旋逻辑。

四、完整运行逻辑时序梳理

1. 终端 1：启动相机节点，持续发布/image_raw图像话题；
2. 终端 2：运行本图像识别服务端节点；
   • 自动订阅图像话题，每一帧实时做红色物体检测，不断刷新objectX/Y；
   • 同时服务端静默等待客户端连接；
3. 终端 3：运行客户端，发送get: True请求；
4. 服务端触发服务回调，把最新识别好的像素坐标打包返回客户端；
5. 客户端接收并打印 X、Y 坐标，单次服务问答完成。

ROS2 视觉目标坐标查询客户端完整代码解析

承接上一节图像识别复合服务端代码，这份是配套客户端service_object_client.py，作用是主动发起请求，向服务端索要识别到的目标物体像素坐标，下面完整拆解、注释、梳理逻辑。

一、客户端整体功能说明

1. 客户端节点启动后，持续尝试连接名为get_target_position的服务；
2. 连接成功后，自动封装请求指令get=True发送给服务端；
3. 阻塞等待服务端返回目标物体x、y像素坐标；
4. 接收应答后打印坐标，程序自动结束。

二、完整带注释客户端代码

python

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ @作者: 古月居(www.guyuehome.com) @说明: ROS2服务示例-请求目标识别，等待目标位置应答 """ # 1. 导入依赖库 import rclpy from rclpy.node import Node # 导入和服务端完全一致的自定义服务接口 from learning_interface.srv import GetObjectPosition # 自定义客户端节点类，继承ROS2基类Node class objectClient(Node): def __init__(self, name): super().__init__(name) # 创建客户端对象：接口类型 + 和服务端一模一样的服务名 self.client = self.create_client(GetObjectPosition, 'get_target_position') # 循环轮询：每隔1秒检测一次服务端是否上线 while not self.client.wait_for_service(timeout_sec=1.0): self.get_logger().info('service not available, waiting again...') # 实例化请求数据包对象 self.request = GetObjectPosition.Request() # 封装发送请求函数 def send_request(self): # 填充请求指令：True代表请求获取目标坐标 self.request.get = True # 异步发送请求，非阻塞 self.future = self.client.call_async(self.request) # 程序入口主函数 def main(args=None): # ROS2环境初始化 rclpy.init(args=args) # 实例化客户端节点，命名service_object_client node = objectClient("service_object_client") # 调用方法发起服务请求 node.send_request() # 循环自旋，持续等待服务端返回结果 while rclpy.ok(): # 单次自旋，不卡死程序 rclpy.spin_once(node) # 判断异步请求是否收到应答 if node.future.done(): try: # 取出应答数据包 response = node.future.result() except Exception as e: # 请求异常捕获（服务掉线、通信失败） node.get_logger().info('Service call failed %r' % (e,)) else: # 正常接收坐标，打印x、y像素值 node.get_logger().info( 'Result of object position:\n x: %d y: %d' % (response.x, response.y)) # 拿到结果后跳出循环 break # 销毁节点、关闭ROS2环境 node.destroy_node() rclpy.shutdown() if __name__ == '__main__': main()

三、逐模块逻辑拆解

1. 导入部分

python

from learning_interface.srv import GetObjectPosition

必须和服务端导入同一个自定义接口包、同一个服务类型，.srv中定义的bool get请求字段、int32 x/y应答字段两端完全对齐。

2. 客户端构造函数__init__

1. create_client(接口类型, 服务名)服务名字符串get_target_position必须和服务端create_service里填写的名称一字不差，大小写、符号不能改动，否则无法建立通信；
2. wait_for_service(timeout_sec=1.0)容错机制：如果客户端先启动、视觉服务端还没运行，不会直接报错退出，每隔 1 秒重试连接，控制台持续打印等待日志。

3.send_request()请求发送函数

python

self.request.get = True self.future = self.client.call_async(self.request)

• 给请求包的get赋值为True，对应服务端判断合法查询；
• call_async()异步调用：不会阻塞终端，通过future对象异步接收返回结果，是 ROS2 推荐写法。

4. 结果等待与解析主循环

python

while rclpy.ok(): rclpy.spin_once(node) if node.future.done():

• spin_once(node)：单次驱动节点回调，不永久阻塞；
• future.done()：标记异步请求是否完成、是否收到服务端回包；
• future.result()：安全取出应答数据，里面包含response.x、response.y目标像素坐标。

5. 收尾销毁流程

客户端单次问答完成后主动执行：destroy_node()销毁节点 →shutdown()关闭 ROS2 运行环境，客户端程序直接退出，符合服务一问一答的通信特性。

四、整套视觉服务完整运行流程（服务端 + 客户端 + 相机）

1. 终端 A：启动相机节点，持续发布/image_raw图像话题；
2. 终端 B：运行视觉识别服务端

   bash

   ros2 run learning_service service_object_server

   自动订阅图像、实时颜色轮廓检测、持续更新目标物体 XY 坐标，静默等待客户端调用；
3. 终端 C：运行本次客户端代码

   bash

   ros2 run learning_service service_object_client

   自动连接服务、发送查询指令、立刻打印识别出来的像素坐标，客户端运行结束。

五、和之前加法客户端的共性与区别

共性（ROS2 客户端标准通用模板）

1. 固定四步结构：创建客户端 → 等待服务上线 → 填充请求异步发送 → 循环自旋等待应答；
2. 都用future异步回调机制、异常捕获、spin_once轮询结果；
3. 最后统一销毁节点、关闭 rclpy。

本次视觉客户端独有差异

1. 请求不再是两个数字 a、b，而是布尔控制位get；
2. 应答不再是单一求和 sum，而是两个坐标 x、y；
3. 服务端不是当场计算数值，而是提前订阅图像持续预处理，客户端只是按需读取缓存好的结果。

六、开发踩坑注意点

1. 服务名严格匹配客户端create_client第二个字符串必须和服务端create_service第二个字符串完全一致，空格、大小写不能出错；
2. 接口包必须提前编译一定要先编译learning_interface接口包，再编译业务包，否则import GetObjectPosition会爆红；
3. 新开终端务必 source 环境

   bash

   source install/setup.bash

   不然ros2 run找不到客户端可执行文件；
4. 必须先开服务端，再运行客户端客户端虽然有等待重试逻辑，但服务端未启动图像识别时，返回的 x/y 默认是 0，拿不到真实坐标。