ROS2参数与自定义接口:运行时配置与消息定义实战
参数系统让节点在运行时可配置,自定义接口让通信数据结构不再受限于内置类型。基于Jazzy Jalisco LTS,从参数声明、约束、回调到YAML加载,从自定义msg/srv/action到构建验证,所有代码可直接运行。
一、ROS2参数系统
参数是节点的运行时配置,以键值对形式存储在节点内部。跟ROS1的参数服务器完全不同——ROS2的参数是节点本地的,不存在全局参数服务器这个单点。
ROS1 vs ROS2参数对比
| 特性 | ROS1 | ROS2 |
|---|---|---|
| 存储位置 | 全局参数服务器(集中式) | 节点本地(分布式) |
| 类型系统 | 弱类型,任意XML-RPC值 | 强类型,编译期检查 |
| 修改方式 | rosparam set/get | ros2 param set/get + 回调 |
| 持久化 | YAML文件加载 | YAML文件 + launch参数 |
| 动态更新 | 需要手动刷新 | add_on_set_parameters_callback |
| 命名空间 | 全局扁平 | 节点命名空间隔离 |
支持的数据类型
| 类型 | Python映射 | 说明 |
|---|---|---|
bool |
bool |
布尔开关 |
int64 |
int |
整数配置 |
float64 |
float |
浮点配置 |
string |
str |
字符串配置 |
bool_array |
list[bool] |
布尔数组 |
int64_array |
list[int] |
整数数组 |
float64_array |
list[float] |
浮点数组 |
string_array |
list[str] |
字符串数组 |
注意:没有
int32、float32类型。ROS2参数只支持64位整数和64位浮点数。
二、参数声明与约束
ROS2的参数必须先声明再使用,这是跟ROS1的重要区别。未声明的参数直接访问会抛异常。
基础声明
import rclpy
from rclpy.node import Node
class SensorNode(Node):
def __init__(self):
super().__init__('sensor_node')
# 基础参数声明:参数名 + 默认值
self.declare_parameter('sensor_id', 'front_camera')
self.declare_parameter('publish_rate', 30.0)
self.declare_parameter('enable_denoise', True)
self.declare_parameter('topics', ['/camera/image', '/camera/depth'])
# 读取参数
sensor_id = self.get_parameter('sensor_id').value
rate = self.get_parameter('publish_rate').value
self.get_logger().info(f'传感器: {sensor_id}, 频率: {rate}Hz')
def main():
rclpy.init()
node = SensorNode()
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
带描述和范围约束的参数
生产环境中,参数需要描述文档和取值范围约束。ParameterDescriptor提供元数据,IntegerRange和FloatingPointRange限制取值。
import rclpy
from rclpy.node import Node
from rcl_interfaces.msg import ParameterDescriptor, IntegerRange, FloatingPointRange
class MotorController(Node):
def __init__(self):
super().__init__('motor_controller')
# 带描述的参数
motor_id_desc = ParameterDescriptor(
description='电机ID编号,范围1-8',
additional_constraints='必须是已注册的电机ID'
)
self.declare_parameter('motor_id', 1, motor_id_desc)
# 整数范围约束:1-8
motor_id_range = IntegerRange(from_value=1, to_value=8, step=1)
motor_id_desc.integer_range = [motor_id_range]
# 重新声明带范围约束的参数(需要先undeclare再declare)
self.undeclare_parameter('motor_id')
self.declare_parameter('motor_id', 1, motor_id_desc)
# 浮点范围约束:0.0-100.0,步长0.1
speed_desc = ParameterDescriptor(
description='电机目标速度百分比',
floating_point_range=[FloatingPointRange(from_value=0.0, to_value=100.0, step=0.1)]
)
self.declare_parameter('target_speed', 50.0, speed_desc)
# 只读参数
firmware_desc = ParameterDescriptor(
description='固件版本号(只读)',
read_only=True
)
self.declare_parameter('firmware_version', 'v2.1.0', firmware_desc)
self.get_logger().info(
f'电机{self.get_parameter("motor_id").value}, '
f'速度{self.get_parameter("target_speed").value}%'
)
def main():
rclpy.init()
node = MotorController()
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
尝试设置超出范围的值会被拒绝:
# 正常设置
ros2 param set /motor_controller target_speed 75.5
# 超出范围,被拒绝
ros2 param set /motor_controller target_speed 150.0
# 报错:Parameter 'target_speed' doesn't satisfy the constraints
三、参数变更回调
参数变更时需要触发业务逻辑更新——比如速度参数变了,要重新配置电机驱动。add_on_set_parameters_callback就是干这个的。
参数变更回调机制
完整代码示例
import rclpy
from rclpy.node import Node
from rcl_interfaces.msg import ParameterDescriptor, FloatingPointRange
from rcl_interfaces.msg import SetParametersResult
class AdaptiveController(Node):
def __init__(self):
super().__init__('adaptive_controller')
# 声明参数
speed_desc = ParameterDescriptor(
description='控制速度百分比',
floating_point_range=[FloatingPointRange(from_value=0.0, to_value=100.0, step=0.1)]
)
self.declare_parameter('target_speed', 50.0, speed_desc)
self.declare_parameter('mode', 'normal') # normal / sport / eco
# 当前运行状态
self.current_speed = 50.0
self.current_mode = 'normal'
# 注册参数变更回调
self.add_on_set_parameters_callback(self.on_parameter_change)
self.get_logger().info(f'控制器启动: 速度={self.current_speed}%, 模式={self.current_mode}')
def on_parameter_change(self, params):
"""参数变更回调,返回SetParametersResult"""
for param in params:
if param.name == 'target_speed':
new_speed = param.value
# 业务校验:eco模式速度上限60%
if self.current_mode == 'eco' and new_speed > 60.0:
self.get_logger().warn(f'eco模式速度不能超过60%,拒绝设置{new_speed}%')
return SetParametersResult(successful=False)
self.current_speed = new_speed
self.get_logger().info(f'速度更新: {new_speed}%')
elif param.name == 'mode':
new_mode = param.value
if new_mode not in ('normal', 'sport', 'eco'):
self.get_logger().warn(f'未知模式: {new_mode}')
return SetParametersResult(successful=False)
# 切换eco模式时自动降速
if new_mode == 'eco' and self.current_speed > 60.0:
self.current_speed = 60.0
self.get_logger().info(f'eco模式自动降速至60%')
self.current_mode = new_mode
self.get_logger().info(f'模式切换: {new_mode}')
return SetParametersResult(successful=True)
def main():
rclpy.init()
node = AdaptiveController()
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
验证回调效果:
# 终端1:启动节点
ros2 run my_package adaptive_controller
# 终端2:正常修改速度
ros2 param set /adaptive_controller target_speed 80.0
# 输出:速度更新: 80.0%
# 切换eco模式(速度超过60%会自动降速)
ros2 param set /adaptive_controller mode eco
# 输出:eco模式自动降速至60%
# eco模式下尝试设置高速,被拒绝
ros2 param set /adaptive_controller target_speed 90.0
# 输出:eco模式速度不能超过60%,拒绝设置90.0%
四、YAML参数文件
参数写死在代码里不灵活,YAML文件让配置与代码分离。不同环境(仿真/实车/测试)用不同YAML文件即可。
YAML文件格式
创建config/params.yaml:
# 节点全名:命名空间/节点名
motor_controller:
ros__parameters:
motor_id: 2
target_speed: 75.0
firmware_version: "v2.1.0"
adaptive_controller:
ros__parameters:
target_speed: 60.0
mode: "eco"
# 带命名空间的节点
robot:
sensor_node:
ros__parameters:
sensor_id: "rear_camera"
publish_rate: 15.0
enable_denoise: false
topics: ["/camera/rear/image", "/camera/rear/depth"]
启动时加载YAML
# 命令行加载参数文件
ros2 run my_package motor_controller --ros-args --params-file config/params.yaml
# 带命名空间启动
ros2 run my_package sensor_node --ros-args \
--params-file config/params.yaml \
--remap __ns:=/robot
# launch文件中加载
# Python launch文件写法
在launch文件中加载:
from launch import LaunchDescription
from launch_ros.actions import Node
import os
def generate_launch_description():
params_file = os.path.join(
os.path.dirname(__file__), '..', 'config', 'params.yaml'
)
return LaunchDescription([
Node(
package='my_package',
executable='motor_controller',
name='motor_controller',
parameters=[params_file],
),
Node(
package='my_package',
executable='sensor_node',
name='sensor_node',
namespace='robot',
parameters=[params_file],
),
])
运行时参数操作
# 查看节点所有参数
ros2 param list /motor_controller
# 获取参数值
ros2 param get /motor_controller target_speed
# 输出:Double value is: 75.0
# 设置参数值
ros2 param set /motor_controller target_speed 80.0
# 导出参数为YAML(方便保存当前配置)
ros2 param dump /motor_controller
# 输出保存到 ./motor_controller.yaml
# 导出指定路径
ros2 param dump /motor_controller --output-dir config/
命令行工具速查
| 命令 | 功能 | 示例 |
|---|---|---|
ros2 param list |
列出节点参数 | ros2 param list /node_name |
ros2 param get |
获取参数值 | ros2 param get /node_name param_name |
ros2 param set |
设置参数值 | ros2 param set /node_name param_name value |
ros2 param dump |
导出参数为YAML | ros2 param dump /node_name |
ros2 param describe |
查看参数描述 | ros2 param describe /node_name param_name |
--params-file |
启动时加载YAML | --ros-args --params-file config.yaml |
五、自定义接口概述
内置的std_msgs/String、geometry_msgs/Twist不够用时,就得自定义接口。ROS2支持三种接口类型:
| 接口类型 | 文件后缀 | 通信方式 | 特点 |
|---|---|---|---|
| 消息 | .msg |
话题(Topic) | 单向、持续发送 |
| 服务 | .srv |
服务(Service) | 请求-响应、一次性 |
| 动作 | .action |
动作(Action) | 请求-响应+反馈、长时间任务 |
支持的数据类型
| 类别 | 类型 | 说明 |
|---|---|---|
| 基本类型 | bool int8 int16 int32 int64 uint8 uint16 uint32 uint64 float32 float64 string |
标量值 |
| 数组类型 | 基本类型[] |
变长数组 |
| 定长数组 | 基本类型[N] |
固定长度 |
| 特殊类型 | Header |
时间戳+坐标系 |
| 嵌套类型 | 包名/消息名 |
引用其他消息 |
接口包目录结构
my_interfaces/
├── CMakeLists.txt
├── package.xml
├── msg/
│ └── RobotStatus.msg
├── srv/
│ └── CalibrateSensor.srv
└── action/
└── NavigateToPose.action
一个包可以同时包含msg、srv、action,也可以只包含其中一种。推荐统一放在
my_interfaces包中管理。
六、自定义消息(.msg)
RobotStatus.msg
创建msg/RobotStatus.msg:
std_msgs/Header header
string robot_name
float64 battery_level
float64 temperature
bool is_charging
int32 error_code
字段说明:
| 字段 | 类型 | 含义 |
|:-----|:-----|:-----|
| header | std_msgs/Header | 时间戳+坐标系,消息标配 |
| robot_name | string | 机器人标识 |
| battery_level | float64 | 电池电量百分比 |
| temperature | float64 | CPU温度(℃) |
| is_charging | bool | 是否充电中 |
| error_code | int32 | 错误码,0=正常 |
构建配置
CMakeLists.txt中添加消息生成配置:
cmake_minimum_required(VERSION 3.8)
project(my_interfaces)
# 必须是C++17
if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang")
add_compile_options(-Wall -Wextra -Wpedantic)
endif()
find_package(ament_cmake REQUIRED)
find_package(rosidl_default_generators REQUIRED)
find_package(std_msgs REQUIRED)
find_package(builtin_interfaces REQUIRED)
# 生成消息代码
rosidl_generate_interfaces(${PROJECT_NAME}
"msg/RobotStatus.msg"
DEPENDENCIES std_msgs builtin_interfaces
)
ament_package()
package.xml中添加依赖:
<?xml version="1.0"?>
<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
<package format="3">
<name>my_interfaces</name>
<version>0.1.0</version>
<description>自定义ROS2接口</description>
<maintainer email="dev@linuxros.cn">linuxros</maintainer>
<license>MIT</license>
<buildtool_depend>ament_cmake</buildtool_depend>
<buildtool_depend>rosidl_default_generators</buildtool_depend>
<depend>std_msgs</depend>
<depend>builtin_interfaces</depend>
<exec_depend>rosidl_default_runtime</exec_depend>
<!-- 关键:声明此包是接口包 -->
<member_of_group>rosidl_interface_packages</member_of_group>
<export>
<build_type>ament_cmake</build_type>
</export>
</package>
构建和验证
# 构建
cd ~/ros2_ws
colcon build --packages-select my_interfaces
# source环境
source install/setup.bash
# 验证消息定义
ros2 interface show my_interfaces/msg/RobotStatus
# 输出:
# std_msgs/Header header
# string robot_name
# float64 battery_level
# float64 temperature
# bool is_charging
# int32 error_code
七、自定义服务(.srv)
服务接口分两部分:请求(---上方)和响应(---下方)。
CalibrateSensor.srv
创建srv/CalibrateSensor.srv:
# 请求
string sensor_id
---
# 响应
bool success
string message
字段说明:
| 部分 | 字段 | 类型 | 含义 |
|:-----|:-----|:-----|:-----|
| 请求 | sensor_id | string | 要校准的传感器ID |
| 响应 | success | bool | 校准是否成功 |
| 响应 | message | string | 结果描述或错误信息 |
在CMakeLists.txt中追加:
rosidl_generate_interfaces(${PROJECT_NAME}
"msg/RobotStatus.msg"
"srv/CalibrateSensor.srv"
DEPENDENCIES std_msgs builtin_interfaces
)
验证:
colcon build --packages-select my_interfaces
source install/setup.bash
ros2 interface show my_interfaces/srv/CalibrateSensor
# 输出:
# string sensor_id
# ---
# bool success
# string message
八、自定义动作(.action)
动作接口分三部分:目标(---分隔)、结果(---分隔)、反馈。
NavigateToPose.action
创建action/NavigateToPose.action:
# 目标
geometry_msgs/PoseStamped target_pose
float64 timeout_sec
---
# 结果
geometry_msgs/Pose current_pose
bool reached
---
# 反馈
float64 distance_remaining
float32 percent_complete
字段说明:
| 部分 | 字段 | 类型 | 含义 |
|:-----|:-----|:-----|:-----|
| 目标 | target_pose | geometry_msgs/PoseStamped | 目标位姿 |
| 目标 | timeout_sec | float64 | 超时时间(秒) |
| 结果 | current_pose | geometry_msgs/Pose | 最终位姿 |
| 结果 | reached | bool | 是否到达目标 |
| 反馈 | distance_remaining | float64 | 剩余距离(米) |
| 反馈 | percent_complete | float32 | 完成百分比 |
在CMakeLists.txt中追加:
find_package(geometry_msgs REQUIRED)
rosidl_generate_interfaces(${PROJECT_NAME}
"msg/RobotStatus.msg"
"srv/CalibrateSensor.srv"
"action/NavigateToPose.action"
DEPENDENCIES std_msgs builtin_interfaces geometry_msgs
)
在package.xml中追加:
<depend>geometry_msgs</depend>
验证:
colcon build --packages-select my_interfaces
source install/setup.bash
ros2 interface show my_interfaces/action/NavigateToPose
# 输出:
# geometry_msgs/PoseStamped target_pose
# float64 timeout_sec
# ---
# geometry_msgs/Pose current_pose
# bool reached
# ---
# float64 distance_remaining
# float32 percent_complete
九、接口构建与使用
完整CMakeLists.txt
cmake_minimum_required(VERSION 3.8)
project(my_interfaces)
if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang")
add_compile_options(-Wall -Wextra -Wpedantic)
endif()
find_package(ament_cmake REQUIRED)
find_package(rosidl_default_generators REQUIRED)
find_package(std_msgs REQUIRED)
find_package(builtin_interfaces REQUIRED)
find_package(geometry_msgs REQUIRED)
rosidl_generate_interfaces(${PROJECT_NAME}
"msg/RobotStatus.msg"
"srv/CalibrateSensor.srv"
"action/NavigateToPose.action"
DEPENDENCIES std_msgs builtin_interfaces geometry_msgs
)
ament_package()
完整package.xml
<?xml version="1.0"?>
<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
<package format="3">
<name>my_interfaces</name>
<version>0.1.0</version>
<description>自定义ROS2接口</description>
<maintainer email="dev@linuxros.cn">linuxros</maintainer>
<license>MIT</license>
<buildtool_depend>ament_cmake</buildtool_depend>
<buildtool_depend>rosidl_default_generators</buildtool_depend>
<depend>std_msgs</depend>
<depend>builtin_interfaces</depend>
<depend>geometry_msgs</depend>
<exec_depend>rosidl_default_runtime</exec_depend>
<member_of_group>rosidl_interface_packages</member_of_group>
<export>
<build_type>ament_cmake</build_type>
</export>
</package>
构建流程
cd ~/ros2_ws
colcon build --packages-select my_interfaces
source install/setup.bash
# 逐个验证
ros2 interface show my_interfaces/msg/RobotStatus
ros2 interface show my_interfaces/srv/CalibrateSensor
ros2 interface show my_interfaces/action/NavigateToPose
# 列出包中所有接口
ros2 interface list | grep my_interfaces
Python中使用自定义接口
import rclpy
from rclpy.node import Node
from rclpy.action import ActionServer
from my_interfaces.msg import RobotStatus
from my_interfaces.srv import CalibrateSensor
from my_interfaces.action import NavigateToPose
class RobotNode(Node):
def __init__(self):
super().__init__('robot_node')
# 使用自定义消息发布
self.status_pub = self.create_publisher(RobotStatus, 'robot_status', 10)
self.timer = self.create_timer(1.0, self.publish_status)
# 使用自定义服务
self.calibrate_srv = self.create_service(
CalibrateSensor, 'calibrate_sensor', self.calibrate_callback
)
# 使用自定义动作
self.navigate_action = ActionServer(
self, NavigateToPose, 'navigate_to_pose', self.navigate_callback
)
self.get_logger().info('机器人节点启动完成')
def publish_status(self):
"""发布机器人状态"""
msg = RobotStatus()
msg.header.stamp = self.get_clock().now().to_msg()
msg.header.frame_id = 'base_link'
msg.robot_name = 'robot_01'
msg.battery_level = 85.5
msg.temperature = 42.3
msg.is_charging = False
msg.error_code = 0
self.status_pub.publish(msg)
def calibrate_callback(self, request, response):
"""传感器校准服务回调"""
self.get_logger().info(f'校准传感器: {request.sensor_id}')
response.success = True
response.message = f'传感器{request.sensor_id} 校准完成'
return response
def navigate_callback(self, goal_handle):
"""导航动作回调"""
self.get_logger().info(
f'导航目标: {goal_handle.request.target_pose}, '
f'超时: {goal_handle.request.timeout_sec}s'
)
# 模拟导航过程,发布反馈
for i in range(10):
feedback = NavigateToPose.Feedback()
feedback.distance_remaining = 5.0 * (1.0 - (i + 1) / 10.0)
feedback.percent_complete = (i + 1) * 10.0
goal_handle.publish_feedback(feedback)
# 返回结果
result = NavigateToPose.Result()
result.reached = True
result.current_pose = goal_handle.request.target_pose.pose
return result
def main():
rclpy.init()
node = RobotNode()
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
ros2 interface命令速查
| 命令 | 功能 | 示例 |
|---|---|---|
ros2 interface list |
列出所有接口 | ros2 interface list |
ros2 interface show |
显示接口定义 | ros2 interface show my_interfaces/msg/RobotStatus |
ros2 interface packages |
列出接口包 | ros2 interface packages |
ros2 interface proto |
显示接口原型 | ros2 interface proto my_interfaces/msg/RobotStatus |
十、常见问题
Q1:自定义接口编译后找不到?
现象:Python中from my_interfaces.msg import RobotStatus报ModuleNotFoundError。
原因:package.xml中缺少member_of_group声明。
解决:确认package.xml包含以下行:
<member_of_group>rosidl_interface_packages</member_of_group>
同时确认构建后执行了source install/setup.bash。自定义接口包必须source后才能被其他包发现。
Q2:YAML参数加载失败?
现象:启动节点时YAML参数没有生效,节点使用默认值。
原因:YAML中的节点名与实际运行的节点名不匹配,包括命名空间。
解决:用ros2 node list查看实际节点名,确保YAML中的键名完全一致:
# 查看实际节点名
ros2 node list
# 输出:/robot/sensor_node
# YAML中必须写全名(含命名空间)
# robot:
# sensor_node:
# ros__parameters:
# ...
Q3:参数范围约束不生效?
现象:设置了IntegerRange/FloatingPointRange,但ros2 param set仍然能设置超出范围的值。
原因:范围约束必须通过ParameterDescriptor传入declare_parameter,仅声明不传描述符无效。
解决:
# 错误:只声明了描述,没关联范围
self.declare_parameter('speed', 50.0)
speed_desc = ParameterDescriptor(
floating_point_range=[FloatingPointRange(from_value=0.0, to_value=100.0, step=0.1)]
)
# 范围约束不会生效!
# 正确:声明时传入描述符
speed_desc = ParameterDescriptor(
description='速度百分比',
floating_point_range=[FloatingPointRange(from_value=0.0, to_value=100.0, step=0.1)]
)
self.declare_parameter('speed', 50.0, speed_desc)
十一、总结
核心要点
- 参数必须先声明后使用,
declare_parameter是唯一入口 - ParameterDescriptor提供描述、范围约束和只读控制,生产环境必备
- add_on_set_parameters_callback实现参数变更时的业务联动,返回
SetParametersResult控制是否生效 - YAML文件让配置与代码分离,
--params-file加载,ros2 param dump导出 - 自定义接口统一放在独立包中,
member_of_group是关键配置 - msg/srv/action三种接口覆盖单向推送、请求响应、长任务反馈三种通信模式
速查表
| 操作 | 命令/代码 |
|---|---|
| 声明参数 | self.declare_parameter('name', default_value, descriptor) |
| 读取参数 | self.get_parameter('name').value |
| 参数回调 | self.add_on_set_parameters_callback(self.callback) |
| 加载YAML | --ros-args --params-file config.yaml |
| 查看参数 | ros2 param list /node_name |
| 设置参数 | ros2 param set /node_name param_name value |
| 导出参数 | ros2 param dump /node_name |
| 查看接口 | ros2 interface show pkg_name/msg/MsgName |
| 构建接口包 | colcon build --packages-select my_interfaces |
本文首发于linuxros.cn,转载请注明出处。