ROS2系列 (17) : Python服务通信实例——Server端（人脸检测服务端）

在上一篇教程中，我们创建了人脸检测服务的自定义接口chapt4_interfaces/srv/FaceDetector。本文将基于该接口实现服务端（Server），完成图像接收、人脸检测、结果返回的核心逻辑，专注于服务通信的核心流程。

一、功能与技术要点

服务端功能：接收客户端发送的图像，使用face_recognition库检测人脸位置，返回人脸数量、耗时及边界框信息。

核心技术点：

• 服务端创建：通过rclpy的create_service绑定服务回调函数。
• 图像格式转换：利用cv_bridge实现ROS图像消息与OpenCV格式的转换。
• 人脸检测：使用face_recognition库完成人脸位置检测。

二、环境准备与功能包创建

2.1 安装依赖库

# 安装人脸识别库
pip3 install face_recognition
# 安装OpenCV
sudo apt install python3-opencv
# 安装cv_bridge（ROS2图像格式转换工具）
sudo apt install ros-humble-cv-bridge

2.2 创建服务端功能包

在工作空间chapt4_ws/src下创建功能包：

cd ~/chapt4/chapt4_ws/src
ros2 pkg create demo_python_service --build-type ament_python \
  --dependencies rclpy chapt4_interfaces cv_bridge sensor_msgs \
  --license Apache-2.0

2.3 准备测试图像

在功能包中创建资源目录并放入测试图像：

mkdir -p demo_python_service/demo_python_service/resource
# 下载或复制一张测试图像到该目录，命名为default.jpg
wget -O demo_python_service/demo_python_service/resource/default.jpg https://img.freepik.com/free-photo/group-people-having-fun-outdoors_23-2149077463.jpg

三、编写服务端代码

在demo_python_service/demo_python_service/目录下创建face_detection_server.py：

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  # 用于ROS图像与OpenCV格式转换
import cv2
import face_recognition  # 人脸识别库
import time

class FaceDetectorionNode(Node):
    def __init__(self):
        super().__init__('face_detection_node')
        # 初始化图像格式转换器
        self.bridge = CvBridge()
        # 创建服务端：服务名/face_detect，接口类型FaceDetector，绑定回调函数
        self.service = self.create_service(FaceDetector, '/face_detect', self.detect_face_callback)
        # 配置默认图像路径（当客户端未发送图像时使用）
        self.defaut_image_path = get_package_share_directory('demo_python_service') + '/resource/default.jpg'
        # 初始化检测参数（默认值）
        self.upsample_times = 1  # 上采样次数（影响检测精度）
        self.model = "hog"       # 检测模型（hog快速但精度低，cnn精确但慢）
        self.get_logger().info('人脸检测服务端已启动，服务名：/face_detect')

    def detect_face_callback(self, request, response):
        """服务回调函数：处理人脸检测请求"""
        # 1. 接收图像（客户端发送或使用默认图像）
        if request.image.data:  # 若客户端发送了图像
            cv_image = self.bridge.imgmsg_to_cv2(request.image)  # ROS图像→OpenCV格式
        else:  # 客户端未发送图像，使用默认图像
            cv_image = cv2.imread(self.defaut_image_path)
            self.get_logger().info('未接收到图像，使用默认图像进行检测')

        # 2. 执行人脸检测
        start_time = time.time()
        self.get_logger().info('加载完图像，开始检测')
        # 调用face_recognition库检测人脸位置
        face_locations = face_recognition.face_locations(
            cv_image, 
            number_of_times_to_upsample=self.upsample_times, 
            model=self.model  
        )
        end_time = time.time()
        self.get_logger().info(f'检测完成，耗时{end_time - start_time:.2f}秒')

        # 3. 组装响应结果
        response.number = len(face_locations)  # 人脸数量
        response.use_time = end_time - start_time  # 检测耗时
        # 提取每个人脸的边界框（上、右、下、左）
        for top, right, bottom, left in face_locations:
            response.top.append(top)
            response.right.append(right)
            response.bottom.append(bottom)
            response.left.append(left)
        
        return response


def main(args=None):
    rclpy.init(args=args)
    node = FaceDetectorionNode()
    rclpy.spin(node)  # 保持节点运行
    rclpy.shutdown()

四、配置功能包（setup.py）

修改demo_python_service/setup.py，添加节点入口和资源文件安装规则：

from setuptools import setup
import os
from glob import glob

package_name = 'demo_python_service'

setup(
    name=package_name,
    version='0.0.0',
    packages=[package_name],
    data_files=[
        ('share/ament_index/resource_index/packages',
            ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
        (os.path.join('share', package_name, 'resource'), glob('demo_python_service/resource/*')),
    ],
    install_requires=['setuptools'],
    zip_safe=True,
    maintainer='your_name',
    maintainer_email='your@email.com',
    description='人脸检测服务端（Python）',
    license='Apache-2.0',
    tests_require=['pytest'],
    entry_points={
        'console_scripts': [
            'face_detection_server = demo_python_service.face_detection_server:main',
        ],
    },
)

五、构建与运行服务端

5.1 构建功能包

cd ~/chapt4/chapt4_ws
colcon build --packages-select demo_python_service

5.2 运行服务端节点

source install/setup.bash
ros2 run demo_python_service face_detection_server

终端输出服务启动信息：

[INFO] [1693521000.123456] [face_detection_node]: 人脸检测服务端已启动，服务名：/face_detect

六、核心模块解析

6.1 服务端创建

self.service = self.create_service(FaceDetector, '/face_detect', self.detect_face_callback)

• FaceDetector：自定义服务接口类型；
• /face_detect：服务名称（客户端需匹配该名称）；
• detect_face_callback：处理请求的回调函数。

6.2 图像格式转换

cv_bridge.CvBridge()实现ROS图像消息（sensor_msgs/Image）与OpenCV图像（cv::Mat）的转换：

• 若客户端发送图像，通过bridge.imgmsg_to_cv2转换；
• 若客户端未发送图像，读取默认图像default.jpg。

6.3 人脸检测逻辑

使用face_recognition.face_locations检测人脸：

• 输入：OpenCV图像、上采样次数（upsample_times）、模型类型（model）；
• 输出：人脸边界框列表（top, right, bottom, left）；
• 响应结果包含人脸数量、耗时及边界框坐标，返回给客户端。

七、总结

本文实现了人脸检测服务的服务端，核心流程包括：

• 服务端创建：通过create_service绑定回调函数，处理客户端请求；
• 图像处理：使用cv_bridge转换格式，支持默认图像 fallback；
• 人脸检测：基于face_recognition库完成检测并组装响应结果。

下一篇将实现客户端（Client），发送图像请求并解析服务端返回的人脸检测结果，完成整个服务通信闭环。