【多喝热水系列】从零开始的ROS2之旅——Day15 在Python节点中使用参数：Python

大家好，今天是ROS2学习的第15天，咱们继续围绕人脸检测项目展开，重点学习如何在Python节点中使用参数。参数就像是节点的“配置开关”，不用修改代码，就能动态调整节点的运行逻辑，比如切换人脸检测模型、调整检测精度。
本项目在原先人脸项目的源码进行实验——具体可参考多喝热水系列的Day13。
一、核心目标
今天我们的核心任务是掌握：

  • 参数的声明与设置
  • 订阅参数更新
  • 修改其他节点参数

二、具体实验
本次实验所有修改都围绕之前的服务端和客户端节点展开，核心是给人脸检测功能添加参数配置，实现动态调整检测参数的效果。

2.1 参数的声明与设置（服务端）
参数的声明就是告诉ROS2节点“我有这些可配置的参数”，同时可以设置默认值，避免参数未配置时节点报错。我们在face_detect_node的初始化函数中添加参数声明和获取的代码，具体如下：

    def __init__(self):
        super().__init__('face_detect_node')#创建一个名为face_detect_node的节点,继承自Node类
        self.bridge=CvBridge()#实例化CvBridge对象
        self.service=self.create_service(FaceDetector,'/face_detect',self.face_detect_callback)
        #创建一个名为/face_detect的服务,使用FaceDetector服务类型,并指定回调函数为face_detect_callback

        self.get_logger().info('人脸检测服务已启动')#日志输出,表示服务已启动

        self.default_image_path=os.path.join(get_package_share_directory('demo_python_service'),'resource','ceshi2.jpg')#获取默认图像路径

        #声明和获取参数
        self.declare_parameter('face_locations_upsample_times', 1)
        #声明参数face_locations_upsample_times,默认值为1
        self.declare_parameter('face_locations_model', 'hog')
        #声明参数face_locations_model,默认值为'hog'

        self.upsample_times=self.get_parameter('face_locations_upsample_times').value
        #获取参数face_locations_upsample_times的值,并赋值给实例变量
        self.model=self.get_parameter('face_locations_model').value
        #获取参数face_locations_model的值,并赋值给实例变量

        self.add_on_set_parameters_callback(self.parameter_callback)

注意：

  • declare_parameter()：用于声明参数，第一个参数是参数名（唯一），第二个是默认值，ROS2会自动管理这些参数。
  • get_parameter().value：用于获取参数的当前值，赋值给实例变量后，后续的人脸检测逻辑就能直接使用这些参数。

此为查看参数和设置参数指令

ros2 param set /face_detect_service face_locations_model cnn

ros2 run demo_python_service face_detect_node --ros-agrs -p face_detect_node face_locations_model :=cnn

2.2 订阅参数修改部分（服务端）
仅仅声明参数还不够，我们需要让节点“感知”到参数被修改，并及时更新自身的配置（比如切换检测模型后，后续检测要立即使用新模型）。这就需要添加参数更新回调函数：

    def __init__(self):
		******
        self.add_on_set_parameters_callback(self.parameter_callback)
    
    def parameter_callback(self,parameters):
        #动态参数设置回调函数,当参数被修改时调用
        for parameter in parameters:
            self.get_logger().info(f'参数{parameter.name}被设置为{parameter.value}')
            #日志输出,表示哪个参数被修改以及修改后的值
            if parameter.name=='face_locations_upsample_times':
                self.upsample_times=parameter.value
                #如果修改的参数是face_locations_upsample_times,则更新实例变量的值
            elif parameter.name=='face_locations_model':
                self.model=parameter.value
                #如果修改的参数是face_locations_model,则更新实例变量的值
        return SetParametersResult(successful=True)
        #返回参数设置结果,表示参数设置成功

注意：
add_on_set_parameters_callback() 会注册一个回调函数，每当节点的参数被修改时，ROS2就会调用这个回调函数，传入被修改的参数列表。我们在回调函数中判断参数名，更新对应的实例变量，这样后续的人脸检测逻辑就能使用新的参数值了。

2.3 服务端源码
将上面的参数声明、回调函数整合到face_detect_node中，完整源码如下（包含人脸检测核心逻辑）：

#此为face_detect_node在实验的完整代码
import rclpy
from rclpy.node import Node
from chapt4_interfaces.srv import FaceDetector
from ament_index_python.packages import get_package_share_directory#获取资源文件路径
from cv_bridge import CvBridge      #opencv和ROS图像消息之间的转换
from rcl_interfaces.msg import SetParametersResult #用于动态参数设置的结果消息类型
import cv2
import face_recognition
import time
import os

class FaceDetectorinNode(Node):
    def __init__(self):
        super().__init__('face_detect_node')#创建一个名为face_detect_node的节点,继承自Node类
        self.bridge=CvBridge()#实例化CvBridge对象
        self.service=self.create_service(FaceDetector,'/face_detect',self.face_detect_callback)
        #创建一个名为/face_detect的服务,使用FaceDetector服务类型,并指定回调函数为face_detect_callback

        self.get_logger().info('人脸检测服务已启动')#日志输出,表示服务已启动

        self.default_image_path=os.path.join(get_package_share_directory('demo_python_service'),'resource','ceshi2.jpg')#获取默认图像路径

        #声明和获取参数
        self.declare_parameter('face_locations_upsample_times', 1)
        #声明参数face_locations_upsample_times,默认值为1
        self.declare_parameter('face_locations_model', 'hog')
        #声明参数face_locations_model,默认值为'hog'

        self.upsample_times=self.get_parameter('face_locations_upsample_times').value
        #获取参数face_locations_upsample_times的值,并赋值给实例变量
        self.model=self.get_parameter('face_locations_model').value
        #获取参数face_locations_model的值,并赋值给实例变量

        self.add_on_set_parameters_callback(self.parameter_callback)


    def parameter_callback(self,parameters):
        #动态参数设置回调函数,当参数被修改时调用
        for parameter in parameters:
            self.get_logger().info(f'参数{parameter.name}被设置为{parameter.value}')
            #日志输出,表示哪个参数被修改以及修改后的值
            if parameter.name=='face_locations_upsample_times':
                self.upsample_times=parameter.value
                #如果修改的参数是face_locations_upsample_times,则更新实例变量的值
            elif parameter.name=='face_locations_model':
                self.model=parameter.value
                #如果修改的参数是face_locations_model,则更新实例变量的值
        return SetParametersResult(successful=True)
        #返回参数设置结果,表示参数设置成功




    def face_detect_callback(self,request,response):
        #TODO:实现人脸检测逻辑
        
        if request.image.data:
            #如果请求中包含图像数据,则进行人脸检测
            cv_image=self.bridge.imgmsg_to_cv2(request.image)#将ROS图像消息转换为OpenCV图像格式

        else:
            #如果请求中不包含图像数据,则使用默认图像进行人脸检测
            cv_image=cv2.imread(self.default_image_path)

        start_time=time.time()#记录开始时间
        self.get_logger().info('正在进行人脸检测...')#日志输出,表示正在进行人脸检测

        face_locations=face_recognition.face_locations(cv_image,number_of_times_to_upsample=self.upsample_times,model=self.model)
        #使用face_recognition库进行人脸检测,返回人脸位置列表
        #cv_image为输入图像,number_of_times_to_upsample为上采样次数,model为检测模型

        end_time=time.time()#记录结束时间
        self.get_logger().info(f'人脸检测完成,共检测到{len(face_locations)}张人脸,耗时{end_time-start_time:.2f}秒')
        #日志输出,表示人脸检测完成,并显示检测到的人脸数量和耗时

        response.number=len(face_locations)
        #将检测到的人脸数量赋值给响应对象的number字段


        response.use_time=end_time-start_time
        #将识别耗时赋值给响应对象的use_time字段

        for (top,right,bottom,left) in face_locations:
            response.top.append(top)#append方法将人脸位置的坐标添加到响应对象的相应字段列表中
            response.right.append(right)#append方法将人脸位置的坐标添加到响应对象的相应字段列表中
            response.bottom.append(bottom)#append方法将人脸位置的坐标添加到响应对象
            response.left.append(left)#append方法将人脸位置的坐标添加到响应对象的相应字段列表中 
        #将人脸在图像中的位置赋值给响应对象的相应字段



        return response
    
def main(args=None):
    rclpy.init(args=args)#初始化ROS2客户端库
    node=FaceDetectorinNode()#创建FaceDetectorinNode节点实例
    rclpy.spin(node)#进入ROS2事件循环,等待服务请求
    rclpy.shutdown()#关闭ROS2客户端库

运行结果如下图所示：
在这里插入图片描述
2.4 修改其他节点参数（客户端）
除了在终端用指令修改参数，我们也可以通过客户端节点，在代码中主动修改服务端节点的参数（跨节点参数修改）。核心思路是：创建SetParameters服务客户端，向服务端发送参数修改请求。
关键代码片段（客户端中新增）：

    def __init__(self):
		******
        #获取测试图像的路径
        self.image=cv2.imread(self.test1_image_path)#使用OpenCV加载测试图像


    def send_request(self):
        #发送服务请求，并处理响应
        while self.client.wait_for_service(timeout_sec=1.0) is False:
            self.get_logger().info(f'等待服务上线...')#等待服务上线,每隔1秒检查一次
        
        #2.构造request对象,并填充请求数据
        request=FaceDetector.Request()
        #创建一个FaceDetector服务的请求对象
        request.image=self.bridge.cv2_to_imgmsg(self.image)
        #将OpenCV图像转换为ROS图像消息,并将其赋值给请求对象的image字段

        #3.发送spin等待服务处理完成
        future=self.client.call_async(request)
        #异步调用服务,将请求对象作为参数传递给call_async方法,该方法会返回一个Future对象,用于跟踪服务调用的状态和结果
        rclpy.spin_until_future_complete(self,future)
        #使用rclpy.spin_until_future_complete函数等待服务调用完成,该函数会阻塞当前线程,
        #直到Future对象的状态变为完成(无论是成功还是失败

        #4.根据服务调用结果进行处理
        response=future.result()
        #获取服务调用的结果,如果服务调用成功,则response将包含服务器返回的数据;如果服务调用失败,则response将为None
        self.get_logger().info(f'服务调用完成,结果: 图像多少张脸：{response.number}，耗时{response.use_time}ms')#日志记录服务调用的结果

        #注释show_face_locations函数,防止显示堵塞无法多次请求
        #self.show_face_locations(response)

    def call_set_parameters(self,parameters):
        #1.创建一个客户端，并等待服务上线
        client=self.create_client(SetParameters,'/face_detect_node/set_parameters')
        
        while not client.wait_for_service(timeout_sec=1.0):
            self.get_logger().info('等待参数服务上线...')
        #2.构造请求对象
        request=SetParameters.Request()
        request.parameters=parameters
        #3.发送请求并等待响应
        future=client.call_async(request)
        rclpy.spin_until_future_complete(self,future)
        #4.处理响应结果
        response=future.result()
        
        return response

    def update_detect_model(self,model):
        #1.创建一个Parameter对象

        param=Parameter()
        param.name='face_locations_model'
        #2.设置参数值,并赋值给Parameter对象
        new_model_value=ParameterValue()
        new_model_value.type=ParameterType.PARAMETER_STRING
        new_model_value.string_value=model
        param.value=new_model_value

        #3.请求更新参数，并处理响应结果
        response=self.call_set_parameters([param])
        for result in response.results:
            if result.successful:
                self.get_logger().info(f'参数更新成功: {result.reason}')
            else:
                self.get_logger().error(f'参数更新失败: {result.reason}')

注意：

  • SetParameters是ROS2内置的服务类型，用于跨节点修改参数，服务地址为“目标节点名/set_parameters”。

  • Parameter对象需要指定参数名和参数值，参数值要明确类型（如字符串、整数），避免类型不匹配导致修改失败。

2.5 客户端源码

import rclpy
from rclpy.node import Node
from chapt4_interfaces.srv import FaceDetector
from cv_bridge import CvBridge
from sensor_msgs.msg import Image  #sensor_msgs是ROS中用于处理图像数据的消息类型库,其中Image是其中的一种消息类型,用于表示图像数据
from ament_index_python.packages import get_package_share_directory#获取资源文件路径
from rcl_interfaces.srv import SetParameters
#用于动态参数设置的服务类型
from rcl_interfaces.msg import Parameter,ParameterValue,ParameterType 
#用于动态参数设置的消息类型
import os
import cv2

class FaceDetectorClientNode(Node):
    def __init__(self):
        super().__init__('face_detector_client_node')#创建一个名为face_detector_client_node的ROS节点
        self.client=self.create_client(FaceDetector,'/face_detect')#创建一个名为/face_detect的服务客户端,该客户端将用于向服务器发送请求并接收响应
        self.bridge=CvBridge()#创建一个CvBridge对象,该对象用于在ROS图像消息和OpenCV图像之间进行转换
        self.get_logger().info('FaceDetectorClientNode has been started.')#日志记录器
        self.test1_image_path=get_package_share_directory('demo_python_service')+'/resource/ceshi2.jpg'
        #获取测试图像的路径
        self.image=cv2.imread(self.test1_image_path)#使用OpenCV加载测试图像


    def send_request(self):
        #发送服务请求，并处理响应
        while self.client.wait_for_service(timeout_sec=1.0) is False:
            self.get_logger().info(f'等待服务上线...')#等待服务上线,每隔1秒检查一次
        
        #2.构造request对象,并填充请求数据
        request=FaceDetector.Request()
        #创建一个FaceDetector服务的请求对象
        request.image=self.bridge.cv2_to_imgmsg(self.image)
        #将OpenCV图像转换为ROS图像消息,并将其赋值给请求对象的image字段

        #3.发送spin等待服务处理完成
        future=self.client.call_async(request)
        #异步调用服务,将请求对象作为参数传递给call_async方法,该方法会返回一个Future对象,用于跟踪服务调用的状态和结果
        rclpy.spin_until_future_complete(self,future)
        #使用rclpy.spin_until_future_complete函数等待服务调用完成,该函数会阻塞当前线程,
        #直到Future对象的状态变为完成(无论是成功还是失败

        #4.根据服务调用结果进行处理
        response=future.result()
        #获取服务调用的结果,如果服务调用成功,则response将包含服务器返回的数据;如果服务调用失败,则response将为None
        self.get_logger().info(f'服务调用完成,结果: 图像多少张脸：{response.number}，耗时{response.use_time}ms')#日志记录服务调用的结果

        #注释show_face_locations函数,防止显示堵塞无法多次请求
        #self.show_face_locations(response)

    def call_set_parameters(self,parameters):
        #1.创建一个客户端，并等待服务上线
        client=self.create_client(SetParameters,'/face_detect_node/set_parameters')
        
        while not client.wait_for_service(timeout_sec=1.0):
            self.get_logger().info('等待参数服务上线...')
        #2.构造请求对象
        request=SetParameters.Request()
        request.parameters=parameters
        #3.发送请求并等待响应
        future=client.call_async(request)
        rclpy.spin_until_future_complete(self,future)
        #4.处理响应结果
        response=future.result()
        
        return response

    def update_detect_model(self,model):
        #1.创建一个Parameter对象

        param=Parameter()
        param.name='face_locations_model'
        #2.设置参数值,并赋值给Parameter对象
        new_model_value=ParameterValue()
        new_model_value.type=ParameterType.PARAMETER_STRING
        new_model_value.string_value=model
        param.value=new_model_value

        #3.请求更新参数，并处理响应结果
        response=self.call_set_parameters([param])
        for result in response.results:
            if result.successful:
                self.get_logger().info(f'参数更新成功: {result.reason}')
            else:
                self.get_logger().error(f'参数更新失败: {result.reason}')



    def show_face_locations(self,response):
        #在图像上绘制人脸位置

        for i in range(response.number):
            top=response.top[i]
            right=response.right[i]
            bottom=response.bottom[i]
            left=response.left[i]
            cv2.rectangle(self.image,(left,top),(right,bottom),(0,255,0),4)
            #在图像上绘制一个矩形框,框住检测到的人脸,矩形框的颜色为绿色(0,255,0),线宽为4
        cv2.imshow("Face Detection Result",self.image)
        #显示图像窗口,窗口标题为"Face Detection Result",显示的图像为self.image
        #该图像已经在上一步中绘制了检测到的人脸位置
        cv2.waitKey(0)


def main(args=None):
    rclpy.init(args=args)
    node=FaceDetectorClientNode()
    node.update_detect_model('hog')#调用update_detect_model方法,将检测模型更新为'hog'
    node.send_request()#调用send_request方法,发送服务请求并处理响应
    node.update_detect_model('cnn')#调用update_detect_model方法,将检测模型更新为'cnn'
    node.send_request()#调用send_request方法,发送服务请求并处理响应


    rclpy.spin(node)#进入ROS2事件循环,等待服务请求
    
    rclpy.shutdown()

运行结果如下图所示：
请添加图片描述

运行结果如下图所示：
在这里插入图片描述

三、今日总结

今天通过人脸检测项目，实战掌握了ROS2 Python节点中参数的使用，核心知识点总结如下：

  1. 参数声明与获取：用declare_parameter()声明参数（指定默认值），用get_parameter().value获取参数值，是参数使用的基础。

  2. 参数更新订阅：通过add_on_set_parameters_callback()注册回调函数，实现参数动态更新，无需重启节点就能生效。

  3. 跨节点参数修改：利用SetParameters服务，在客户端创建服务客户端，向目标节点发送参数修改请求，实现节点间的参数交互。

小技巧：参数名要保持一致（比如服务端声明的是face_locations_model，客户端修改时也要用这个名字），否则会修改失败；另外，参数类型要匹配（比如字符串类型不能传入整数）。

之后我们继续深入ROS2，探索更多实用的节点操作，一起加油

# python # 开发语言 # 人工智能
Logo
龙虾开发者社区

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用

更多推荐

cover

我用 NVIDIA SkillSpector 纯静态扫了一遍 GitHub 最火的 agent skills：能挖出多少注入？0 个

avatar 龙虾开发者社区
cover

周榜换血九成，Agent Skills 星数翻倍

avatar 龙虾开发者社区

写给 TypeScript 工程师的 OpenClaw.NET 上手指南：用你熟悉的 TS 思维，跑起一个生产级 AI Agent

读代码层：TS 和 C# 本来就是「远房亲戚」——asyncawait??、接口、泛型、严格空检查，你的心智模型几乎无缝平移。真正要适应的只有两件事：类型是具名而非结构化的（多一句显式声明），以及 NativeAOT 下的「禁反射」约束。写扩展层：你甚至可以不怎么碰 C#——Node.js 插件桥意味着你的 TypeScript 插件就是一等公民，而且能在生产 aot 车道跑。内核给你 .NET

avatar 龙虾开发者社区