ESC
输入关键词搜索文章标题和内容

ROS2参数与自定义接口:运行时配置与消息定义实战

本文由 linuxROS 整理发布,首发于 linuxros.cn,转载请注明出处。

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就是干这个的。

参数变更回调机制

flowchart TB A(["参数变更请求"]) --> B["SetParametersCallback"] B --> C{"参数值合法?"} C -->|"是"| D["更新业务逻辑"] C -->|"否"| E["拒绝变更"] D --> F["返回successful=True"] E --> G["返回successful=False"] F --> H(["参数生效"]) G --> I(["参数不变"]) style A fill:#E3F2FD,stroke:#1976D2 style C fill:#FFF8E1,stroke:#F57C00 style D fill:#E8F5E9,stroke:#388E3C style E fill:#FFEBEE,stroke:#D32F2F style H fill:#E8F5E9,stroke:#388E3C style I fill:#FFEBEE,stroke:#D32F2F

完整代码示例

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中添加消息生成配置:

来自 linuxros.cn · linuxROS
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)

动作接口分三部分:目标(---分隔)、结果(---分隔)、反馈。

创建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)

十一、总结

核心要点

  1. 参数必须先声明后使用,declare_parameter是唯一入口
  2. ParameterDescriptor提供描述、范围约束和只读控制,生产环境必备
  3. add_on_set_parameters_callback实现参数变更时的业务联动,返回SetParametersResult控制是否生效
  4. YAML文件让配置与代码分离,--params-file加载,ros2 param dump导出
  5. 自定义接口统一放在独立包中,member_of_group是关键配置
  6. 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,转载请注明出处。

版权声明

作者linuxROS
协议本作品采用 CC BY-NC-SA 4.0 许可协议:署名-非商业性使用-相同方式共享
关注欢迎关注微信公众号 linuxROS,获取更多机器人 / 嵌入式 / Linux 干货
返回首页