【多喝热水系列】从零开始的ROS2之旅——Day13 服务与参数人脸检测项目:基本命令和Python项目实现

大家好~ 欢迎来到「多喝热水系列」从零开始的ROS2之旅,今天是我们ROS2学习的第13天,也是实战性拉满的一天!前面我们已经初步接触了ROS2的通信机制,今天就将理论落地,结合服务与参数通信,动手实现一个人脸检测项目,既能巩固服务通信的使用
一、核心目标
今天我们的核心任务是掌握:

  • 服务与参数通信介绍

  • 用Python服务通信实现人脸检测

二、服务与参数通信介绍

在ROS2中,服务通信是一种“请求-响应”式的同步通信方式,适合需要明确获取反馈的场景(比如我们今天的人脸检测:客户端发送图像请求,服务端返回检测结果)。参数通信则用于节点间共享配置信息,可动态修改,十分灵活。我们先从最基础的命令入手,快速熟悉服务通信的使用。

打开turtlesim_node之后,
ros2 service list -t 	#查看一般ros2服务
ros2 interface show turtlesim/srv/Spawn	#查看turtlesim/srv/Spawn的详细定义
ros2 service  call /spawn turtlesim/srv/Spawn "{x: 1,  y: 1}"
#通过调用服务生成新的海龟

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

通过rqt来实现服务

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

基于服务的参数通信

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

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

三、用Python服务通信实现人脸检测
本次人脸检测项目采用「服务端+客户端」架构:服务端负责加载图像、执行人脸检测逻辑,返回检测结果(人脸数量、耗时、人脸位置);客户端负责发送图像请求(或使用默认图像),接收服务端响应,并绘制人脸位置、显示结果。整个项目分为4个步骤,逐步实现。
3.1 自定义服务接口
ROS2默认的服务接口无法满足我们的人脸检测需求(需要传递图像、返回人脸相关信息),因此需要自定义服务接口。接口定义如下,分为请求(request)和响应(response)两部分:

#定义图像消息接口
sensor_msgs/Image image #原始图像,request的内容
---
int16 number            #人脸数,以下为response的内容 
float32 use_time          #识别耗时
int32[] top             #人脸在图像中的位置
int32[] right
int32[] bottom
int32[] left

说明:接口中使用sensor_msgs/Image类型传递图像,这是ROS2中标准的图像消息类型;响应部分的坐标列表用于存储多个人脸的位置信息,方便后续绘制矩形框。
3.2 人脸检测
在实现服务之前,我们先单独编写人脸检测的基础代码,验证检测逻辑是否可行。这里使用face_recognition库(人脸检测核心)和cv2库(图像读取、绘制),代码如下:

import face_recognition
import cv2
from ament_index_python.packages import get_package_share_directory

def main():
    # 获取资源文件路径
    defaut_image_path=get_package_share_directory('demo_python_service')+"/resource/ceshi.jpg"
    # 加载默认图像并提取人脸特征
    image=cv2.imread(defaut_image_path)
    #查找图像中的人脸位置
    face_locations=face_recognition.face_locations(image,number_of_times_to_upsample=1,model="hog")
    #绘制人脸位置
    for (top,right,bottom,left) in face_locations:
        cv2.rectangle(image,(left,top),(right,bottom),(0,255,0),4)
    #显示图像
    cv2.imshow("Face Detection",image)
    cv2.waitKey(0)

运行上述代码,会加载功能包中resource目录下的ceshi.jpg图像,检测到人脸后用绿色矩形框标注,运行效果如下:

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

3.3 人脸检测服务实现
服务端的核心功能:创建服务节点,接收客户端的图像请求,执行人脸检测,返回检测结果。代码中整合了基础检测逻辑,同时添加了ROS2服务的相关操作(创建服务、回调函数、日志输出),具体代码如下:

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图像消息之间的转换
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.upsample_times=1#设置人脸检测的上采样次数,默认为1
        self.model='hog'#设置人脸检测模型,默认为hog


    
    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客户端库

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

3.4 人脸检测客户端实现
客户端的核心功能:创建客户端节点,加载测试图像,向服务端发送图像请求,接收响应后,绘制人脸位置并显示结果。代码如下:

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
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')#日志记录服务调用的结果

        self.show_face_locations(response)


    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.send_request()
    rclpy.shutdown()![请添加图片描述]

提示:运行时需先启动服务端,再启动客户端,否则客户端会一直等待服务上线。
运行结果如下图所示:
请添加图片描述

四、今日总结
今天的学习的核心是“服务与参数通信”的实战应用,从基础命令到Python项目实现。

  1. 掌握了ROS2服务通信的基础命令(ros2 service list、ros2 service call等),以及rqt可视化调用服务的方法;

  2. 理解了参数通信的作用,知道如何结合服务实现节点配置的灵活调整;

  3. 学会了自定义ROS2服务接口,满足实际项目的需求;

  4. 用Python实现了人脸检测的服务端和客户端,掌握了ROS2服务通信的完整开发流程,以及cv_bridge的图像格式转换用法。

我们一步步完成了人脸检测服务的开发,明天我们将继续深入ROS2的学习。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐