Python创建ROS串口写入服务self参数报错的正确实现方法
问题根源
报错核心是服务回调函数绑定方式错误:你把类实例方法send_serial定义在类外,注册rospy.Service时传入的是未绑定实例的普通函数,ROS触发服务回调时只会传入请求参数,不会自动传入类实例self,因此触发缺少位置参数的错误。原代码还存在串口写入参数错误、返回值不符合srv规范、节点初始化顺序错误、客户端缺少服务类型导入等问题。
跨节点调用ROS服务不需要持有服务端实例对象,ROS Master会自动完成节点间的请求路由,只需要保证服务名、服务类型匹配,服务端节点正常运行即可。
正确实现方案
第一步:确认自定义srv定义
首先保证你的my_package/srv/SerialData.srv定义清晰,参考示例如下:
# 请求段:要写入串口的数据 string data --- # 响应段:写入是否成功 bool success
如果你自定义的srv字段名不同,后续代码替换为对应字段名即可。修改完srv后记得重新catkin编译工作空间、source环境变量。
第二步:修正服务端代码
修正点:
- 节点初始化放在类构造函数最前端
- 服务回调方法缩进放入类内部,注册服务时绑定当前实例的方法
self.send_serial - 串口初始化加异常判断,避免端口占用/不存在时节点直接崩溃
- 串口写入时取请求对象里的实际数据字段,转成字节类型后写入,写入后刷新缓冲区
- 服务返回值严格匹配srv定义的响应结构
import rospy import serial from my_package.srv import SerialData, SerialDataResponse class SerialServer: def __init__(self): rospy.init_node('serial_server', anonymous=True) # 初始化串口连接 try: self.ser = serial.Serial('/dev/ttyUSB0', baudrate=115200, timeout=1) rospy.loginfo("Serial port /dev/ttyUSB0 connected") except Exception as e: rospy.logfatal(f"Serial port open failed: {str(e)}") return # 注册服务,绑定实例方法 rospy.Service('send_serial', SerialData, self.send_serial) rospy.loginfo("Serial write service [send_serial] is ready") def send_serial(self, req): try: # 字符串转字节类型写入串口 write_content = req.data.encode('utf-8') if isinstance(req.data, str) else req.data self.ser.write(write_content) self.ser.flush() return SerialDataResponse(success=True) except Exception as e: rospy.logerr(f"Serial write error: {str(e)}") return SerialDataResponse(success=False) def main(): try: server = SerialServer() rospy.spin() except rospy.ROSInterruptException: rospy.loginfo("Serial server node stopped") if __name__ == '__main__': main()
第三步:修正客户端代码
修正点:
- 节点初始化放在所有ROS接口(订阅、服务代理)创建之前
- 导入自定义SerialData服务类型
- 提前初始化服务代理、等待服务上线,不用每次回调重复创建
- 调用服务时按srv字段传参,判断响应的success字段得到执行结果
import rospy from ros_igtl_bridge.msg import igtlstring from my_package.srv import SerialData class SerialClient: def __init__(self): # 节点初始化必须放在最前 rospy.init_node('serial_client', anonymous=True) # 提前建立服务代理 rospy.wait_for_service('send_serial') self.serial_proxy = rospy.ServiceProxy('send_serial', SerialData) rospy.loginfo("Connected to serial write service") # 注册话题订阅 rospy.Subscriber('IGTL_STRING_IN', igtlstring, self.callbackString) def callbackString(self, msg): if msg.name == 'INIT': init_condition = msg.data[4] + msg.data[5] try: resp = self.serial_proxy(data=init_condition) if resp.success: rospy.loginfo("Initialization successful") else: rospy.logwarn("Initialization command send failed") except rospy.ServiceException as e: rospy.logerr(f"Service call failed: {str(e)}") def main(): try: client = SerialClient() rospy.spin() except rospy.ROSInterruptException: rospy.loginfo("Serial client node stopped") if __name__ == '__main__': main()
部署注意事项
- 串口是独占资源,整个机器人系统中只需要
serial_server一个节点持有串口连接,其他所有需要操作串口的节点都通过调用该服务实现,禁止多个节点同时访问同一个串口端口 - 给两个python脚本添加可执行权限:
chmod +x 脚本路径.py - 运行顺序:先启动
serial_server节点确认串口连接成功、服务注册完成,再启动客户端节点
内容的提问来源于stack exchange,提问作者Mariana Bernardes
相关产品推荐
相关产品推荐

