go2_ros2_sdk开发者手册自定义ROS2消息与服务的完整教程【免费下载链接】go2_ros2_sdkUnofficial ROS2 SDK support for Unitree GO2 AIR/PRO/EDU项目地址: https://gitcode.com/gh_mirrors/go/go2_ros2_sdk如何为Unitree GO2机器人定制ROS2消息与服务面向初学者的终极指南go2_ros2_sdk项目为Unitree GO2 AIR/PRO/EDU机器人提供了完整的ROS2 SDK支持让开发者能够轻松实现机器人控制、传感器数据处理和自主导航功能。本文将深入讲解如何自定义ROS2消息与服务帮助你充分发挥GO2机器人的潜力打造个性化的机器人应用。 为什么需要自定义ROS2消息与服务在机器人开发中标准消息类型往往无法满足特定需求。go2_ros2_sdk项目已经定义了大量专用消息类型但你可能需要扩展功能添加新的传感器数据处理优化性能精简数据传输格式集成第三方与自定义硬件或软件对接特殊应用实现特定业务逻辑 项目消息定义结构go2_ros2_sdk的消息定义位于go2_interfaces/msg目录中。让我们先了解现有的消息结构现有消息类型概览项目已经定义了丰富的消息类型包括机器人状态消息Go2State.msg - 包含机器人模式、位置、速度等状态信息运动控制消息Go2Move.msg - 运动控制命令传感器数据消息IMU.msg - 惯性测量单元数据WebRTC请求消息WebRtcReq.msg - WebRTC通信请求电池管理消息BmsState.msg - 电池状态信息电机控制消息MotorCmds.msg - 电机控制命令消息定义语法示例让我们查看一个典型的消息定义文件# Go2State.msg - 机器人状态消息 uint8 mode # 机器人模式 int32 progress # 任务进度 uint8 gait_type # 步态类型 float32 foot_raise_height # 脚部抬起高度 float32[3] position # 位置坐标 (x, y, z) float32 body_height # 身体高度 float32[3] velocity # 速度向量 float32[4] range_obstacle # 障碍物距离 int16[4] foot_force # 脚部受力 float32[12] foot_position_body # 脚部位置 float32[12] foot_speed_body # 脚部速度 自定义ROS2消息的完整步骤步骤1创建新的消息文件在go2_interfaces/msg目录中创建你的自定义消息文件导航到消息目录cd go2_interfaces/msg创建新消息文件例如CustomSensorData.msgnano CustomSensorData.msg定义消息结构# CustomSensorData.msg - 自定义传感器数据消息 std_msgs/Header header # 标准消息头 string sensor_type # 传感器类型 float32[3] position # 传感器位置 float32[9] orientation # 传感器方向 float32[] data # 传感器数据数组 bool is_calibrated # 是否已校准 uint32 timestamp # 时间戳毫秒步骤2更新CMakeLists.txt编辑go2_interfaces/CMakeLists.txt文件添加新的消息定义# 在find_package部分确保包含依赖 find_package(rosidl_default_generators REQUIRED) # 在rosidl_generate_interfaces部分添加新消息 rosidl_generate_interfaces(${PROJECT_NAME} msg/AudioData.msg msg/BmsCmd.msg msg/BmsState.msg # ... 其他现有消息 msg/CustomSensorData.msg # 添加这一行 DEPENDENCIES std_msgs )步骤3更新package.xml编辑go2_interfaces/package.xml文件添加必要的依赖dependstd_msgs/depend buildtool_dependrosidl_default_generators/buildtool_depend exec_dependrosidl_default_runtime/exec_depend member_of_grouprosidl_interface_packages/member_of_group步骤4重新编译项目cd ~/ros2_ws colcon build --packages-select go2_interfaces source install/setup.bash步骤5验证消息生成使用ROS2命令验证消息是否正确生成# 查看消息定义 ros2 interface show go2_interfaces/msg/CustomSensorData # 发布测试消息 ros2 topic pub /custom_sensor go2_interfaces/msg/CustomSensorData \ header: stamp: sec: 0 nanosec: 0 frame_id: base_link sensor_type: temperature position: [0.0, 0.0, 0.5] orientation: [1.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 1.0] data: [25.5, 26.0, 24.8] is_calibrated: true timestamp: 1234567890 --once️ 创建自定义ROS2服务步骤1创建服务定义文件在go2_interfaces目录中创建srv目录如果不存在mkdir -p go2_interfaces/srv cd go2_interfaces/srv创建服务定义文件例如CustomControl.srv# CustomControl.srv - 自定义控制服务 # 请求部分 string command # 控制命令 float32[3] target_pose # 目标位置 float32 speed # 运动速度 bool use_obstacle_avoidance # 是否启用避障 --- # 响应部分 bool success # 执行是否成功 string message # 返回消息 float32 execution_time # 执行时间秒步骤2更新CMakeLists.txt添加服务在go2_interfaces/CMakeLists.txt中添加服务定义rosidl_generate_interfaces(${PROJECT_NAME} # ... 现有的消息定义 srv/CustomControl.srv # 添加服务定义 DEPENDENCIES std_msgs )步骤3创建服务服务器在go2_robot_sdk/go2_robot_sdk/application/services目录中创建服务实现创建custom_control_service.py# custom_control_service.py - 自定义控制服务实现 import logging from typing import Tuple from go2_interfaces.srv import CustomControl from ...domain.interfaces import IRobotController logger logging.getLogger(__name__) class CustomControlService: 自定义控制服务 def __init__(self, controller: IRobotController): self.controller controller async def handle_custom_control( self, request: CustomControl.Request, robot_id: str default ) - CustomControl.Response: 处理自定义控制请求 try: logger.info(f收到自定义控制命令: {request.command}) # 解析命令 if request.command move_to_position: # 移动到指定位置 success await self._move_to_position( request.target_pose, request.speed, request.use_obstacle_avoidance, robot_id ) message 位置移动完成 if success else 位置移动失败 elif request.command custom_gait: # 执行自定义步态 success await self._execute_custom_gait(robot_id) message 自定义步态执行完成 if success else 步态执行失败 else: success False message f未知命令: {request.command} return CustomControl.Response( successsuccess, messagemessage, execution_time0.5 # 示例执行时间 ) except Exception as e: logger.error(f自定义控制服务错误: {e}) return CustomControl.Response( successFalse, messagef服务错误: {str(e)}, execution_time0.0 ) async def _move_to_position( self, target_pose: Tuple[float, float, float], speed: float, use_obstacle_avoidance: bool, robot_id: str ) - bool: 移动到指定位置 # 实现位置移动逻辑 # 这里可以调用现有的机器人控制接口 try: # 示例通过现有接口发送移动命令 self.controller.send_movement_command( robot_id, target_pose[0], # x target_pose[1], # y target_pose[2] # z ) return True except Exception as e: logger.error(f移动失败: {e}) return False async def _execute_custom_gait(self, robot_id: str) - bool: 执行自定义步态 # 实现自定义步态逻辑 # 这里可以通过WebRTC发送自定义命令 try: # 示例通过WebRTC发送步态控制命令 self.controller.send_webrtc_request( robot_id, api_id1001, # 示例API ID parameter{gait_type: custom}, topicrt/api/sport/request ) return True except Exception as e: logger.error(f步态执行失败: {e}) return False步骤4集成服务到主节点在go2_robot_sdk/go2_robot_sdk/presentation/go2_driver_node.py中添加服务# 在适当的导入部分添加 from go2_interfaces.srv import CustomControl from ..application.services.custom_control_service import CustomControlService # 在节点初始化部分添加服务 class Go2DriverNode(Node): def __init__(self): super().__init__(go2_driver_node) # ... 现有初始化代码 # 创建自定义控制服务 self.custom_control_service CustomControlService(self.controller) # 创建ROS2服务 self.custom_control_srv self.create_service( CustomControl, custom_control, self.handle_custom_control ) self.get_logger().info(自定义控制服务已启动) async def handle_custom_control(self, request, response): 处理自定义控制服务请求 # 从请求中获取机器人ID默认为第一个机器人 robot_id self.robot_ids[0] if self.robot_ids else default # 调用服务处理逻辑 service_response await self.custom_control_service.handle_custom_control( request, robot_id ) response.success service_response.success response.message service_response.message response.execution_time service_response.execution_time return response步骤5重新编译并测试服务# 重新编译 cd ~/ros2_ws colcon build --packages-select go2_interfaces go2_robot_sdk source install/setup.bash # 启动机器人节点 export ROBOT_IPyour_robot_ip export CONN_TYPEwebrtc ros2 launch go2_robot_sdk robot.launch.py # 在另一个终端测试服务 ros2 service call /custom_control go2_interfaces/srv/CustomControl \ {command: move_to_position, target_pose: [1.0, 0.0, 0.0], speed: 0.5, use_obstacle_avoidance: true} 高级消息与服务设计模式模式1复合消息类型对于复杂的数据结构可以创建复合消息# ComplexCommand.msg - 复合命令消息 CustomSensorData sensor_data # 传感器数据 Go2State current_state # 当前状态 Go2Move movement_command # 运动命令 uint32 sequence_id # 序列号 bool emergency_stop # 紧急停止标志模式2动作服务器接口创建动作服务器来处理长时间运行的任务# 在服务中实现动作服务器 from rclpy.action import ActionServer from go2_interfaces.action import CustomNavigation class CustomNavigationAction: 自定义导航动作服务器 def __init__(self, node, controller): self._action_server ActionServer( node, CustomNavigation, custom_navigation, self.execute_callback ) self.controller controller async def execute_callback(self, goal_handle): 执行导航动作 feedback_msg CustomNavigation.Feedback() result CustomNavigation.Result() try: # 执行导航逻辑 target goal_handle.request.target_position feedback_msg.current_position [0, 0, 0] # 发布反馈 for i in range(10): feedback_msg.progress i * 10 goal_handle.publish_feedback(feedback_msg) await asyncio.sleep(1) # 完成动作 result.success True result.final_position target goal_handle.succeed() except Exception as e: result.success False result.error_message str(e) goal_handle.abort() return result模式3消息适配器模式创建适配器来处理不同格式的消息class MessageAdapter: 消息格式适配器 staticmethod def go2_state_to_custom(state_msg): 将Go2State转换为自定义格式 return { mode: state_msg.mode, position: list(state_msg.position), velocity: list(state_msg.velocity), timestamp: time.time() } staticmethod def custom_to_go2_move(custom_cmd): 将自定义命令转换为Go2Move move_msg Go2Move() move_msg.velocity_x custom_cmd.get(vx, 0.0) move_msg.velocity_y custom_cmd.get(vy, 0.0) move_msg.yaw_rate custom_cmd.get(yaw, 0.0) return move_msg 调试与测试技巧调试自定义消息使用ROS2命令行工具# 查看消息定义 ros2 interface show go2_interfaces/msg/CustomSensorData # 监听消息 ros2 topic echo /custom_sensor_data # 发布测试消息 ros2 topic pub /custom_sensor_data go2_interfaces/msg/CustomSensorData message.json创建测试节点# test_custom_messages.py import rclpy from rclpy.node import Node from go2_interfaces.msg import CustomSensorData class TestNode(Node): def __init__(self): super().__init__(test_node) self.publisher self.create_publisher( CustomSensorData, test_topic, 10 ) self.timer self.create_timer(1.0, self.publish_message) def publish_message(self): msg CustomSensorData() msg.sensor_type test_sensor msg.data [1.0, 2.0, 3.0] self.publisher.publish(msg) self.get_logger().info(消息已发布)性能优化建议使用数组而非多个字段# 优化前 float32 sensor1 float32 sensor2 float32 sensor3 # 优化后 float32[3] sensors合理使用数据类型# 根据需求选择合适的数据类型 uint8 status_flags # 状态标志0-255 int16 sensor_value # 传感器值-32768到32767 float32 position # 位置坐标 bool is_active # 布尔状态消息频率控制# 控制发布频率 class OptimizedPublisher(Node): def __init__(self): super().__init__(optimized_publisher) self.publisher self.create_publisher(CustomMsg, topic, 10) # 根据需求调整频率 self.timer self.create_timer(0.1, self.callback) # 10Hz 实战示例创建激光雷达数据处理服务让我们创建一个实际的示例激光雷达数据处理服务步骤1创建激光雷达服务定义在go2_interfaces/srv目录创建LidarProcessing.srv# LidarProcessing.srv - 激光雷达数据处理服务 # 请求 sensor_msgs/PointCloud2 pointcloud # 原始点云数据 float32 min_range # 最小范围 float32 max_range # 最大范围 bool filter_noise # 是否过滤噪声 --- # 响应 sensor_msgs/PointCloud2 filtered_pointcloud # 过滤后的点云 uint32 original_points # 原始点数 uint32 filtered_points # 过滤后点数 float32 processing_time # 处理时间秒步骤2实现服务处理逻辑# lidar_processing_service.py import numpy as np import time from sensor_msgs.msg import PointCloud2 from sensor_msgs_py import point_cloud2 from go2_interfaces.srv import LidarProcessing class LidarProcessingService: 激光雷达数据处理服务 def __init__(self): self.get_logger().info(激光雷达处理服务已初始化) async def process_lidar_data(self, request): 处理激光雷达数据 start_time time.time() try: # 读取点云数据 points point_cloud2.read_points( request.pointcloud, field_names(x, y, z, intensity), skip_nansTrue ) points_list list(points) original_count len(points_list) # 过滤点云 filtered_points [] for point in points_list: x, y, z, intensity point distance np.sqrt(x**2 y**2 z**2) # 范围过滤 if distance request.min_range or distance request.max_range: continue # 噪声过滤如果启用 if request.filter_noise and intensity 0.1: continue filtered_points.append(point) filtered_count len(filtered_points) # 创建过滤后的点云消息 header request.pointcloud.header fields request.pointcloud.fields filtered_pc2 point_cloud2.create_cloud_xyz32( header, [(p[0], p[1], p[2]) for p in filtered_points] ) processing_time time.time() - start_time return LidarProcessing.Response( filtered_pointcloudfiltered_pc2, original_pointsoriginal_count, filtered_pointsfiltered_count, processing_timeprocessing_time ) except Exception as e: self.get_logger().error(f激光雷达处理错误: {e}) return LidarProcessing.Response( filtered_pointcloudPointCloud2(), original_points0, filtered_points0, processing_time0.0 )步骤3集成到现有系统# 在go2_driver_node.py中添加 from go2_interfaces.srv import LidarProcessing from .lidar_processing_service import LidarProcessingService class Go2DriverNode(Node): def __init__(self): # ... 现有初始化 # 添加激光雷达处理服务 self.lidar_processor LidarProcessingService() self.lidar_service self.create_service( LidarProcessing, lidar_processing, self.handle_lidar_processing ) async def handle_lidar_processing(self, request, response): 处理激光雷达服务请求 return await self.lidar_processor.process_lidar_data(request) 最佳实践总结1.保持消息简洁只包含必要的数据字段使用合适的数据类型避免过度嵌套2.服务设计原则服务应具有明确的单一职责响应时间应尽可能短提供有意义的错误信息3.版本兼容性向后兼容现有接口使用可选字段进行扩展提供迁移指南4.文档与示例为每个消息和服务编写文档提供使用示例说明数据单位和范围5.性能考虑控制消息频率优化数据结构使用高效的序列化格式 开始你的自定义开发之旅通过本教程你已经掌握了在go2_ros2_sdk项目中自定义ROS2消息与服务的完整流程。从简单的消息定义到复杂的服务实现你现在可以✅创建自定义消息类型扩展机器人数据接口✅实现ROS2服务提供新的控制功能✅集成到现有系统保持兼容性✅优化性能确保实时性要求✅调试和测试保证功能稳定性go2_ros2_sdk项目为Unitree GO2机器人提供了强大的ROS2支持框架而自定义消息与服务能力让你能够充分发挥机器人的潜力。无论是学术研究、工业应用还是创意项目这些技能都将帮助你构建更智能、更灵活的机器人系统。记住良好的消息和服务设计是机器人系统成功的关键。从简单的需求开始逐步构建复杂的功能你将成为ROS2机器人开发的专家 ✨下一步行动从简单的消息扩展开始实践创建一个测试服务验证概念集成到你的具体应用场景分享你的成果给社区祝你开发顺利期待看到你基于go2_ros2_sdk创造的精彩应用【免费下载链接】go2_ros2_sdkUnofficial ROS2 SDK support for Unitree GO2 AIR/PRO/EDU项目地址: https://gitcode.com/gh_mirrors/go/go2_ros2_sdk创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考