第一步：
首先在工作空间/src下创建数据结构目录service_interfaces
ros2 pkg create service_interfaces --build-type ament_cmake然后再创建一个srv目录
在里面创建FaceDetect.srv（注意，首字母要大写）
sensor_msgs/Image photo
---
uint16 number
float32 use_time
int32[] top
int32[] right
int32[] bottom
int32[] leftCmakeList.txt
cmake_minimum_required(VERSION 3.8)
project(service_interfaces)
if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang")
add_compile_options(-Wall -Wextra -Wpedantic)
endif()
# find dependencies
find_package(ament_cmake REQUIRED)
# uncomment the following section in order to fill in
# further dependencies manually.
# find_package(<dependency> REQUIRED)
# 添加依赖
find_package(sensor_msgs REQUIRED) # 确保这个包在这里被找到
# 声明srv文件所属的工程名字, 文件位置, 依赖DEPENDENCIES
find_package(rosidl_default_generators REQUIRED)
rosidl_generate_interfaces(${PROJECT_NAME}
"srv/FaceDetect.srv"
DEPENDENCIES
sensor_msgs
)
if(BUILD_TESTING)
find_package(ament_lint_auto REQUIRED)
# the following line skips the linter which checks for copyrights
# comment the line when a copyright and license is added to all source files
set(ament_cmake_copyright_FOUND TRUE)
# the following line skips cpplint (only works in a git repo)
# comment the line when this package is in a git repo and when
# a copyright and license is added to all source files
set(ament_cmake_cpplint_FOUND TRUE)
ament_lint_auto_find_test_dependencies()
endif()
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>service_interfaces</name>
<version>0.0.0</version>
<description>TODO: Package description</description>
<maintainer email="root@todo.todo">root</maintainer>
<license>TODO: License declaration</license>
<buildtool_depend>ament_cmake</buildtool_depend>
<depend>sensor_msgs</depend>
<build_depend>rosidl_default_generators</build_depend>
<exec_depend>rosidl_default_runtime</exec_depend>
<member_of_group>rosidl_interface_packages</member_of_group>
<test_depend>ament_lint_auto</test_depend>
<test_depend>ament_lint_common</test_depend>
<export>
<build_type>ament_cmake</build_type>
</export>
</package>
第二步：
要准备好人脸识别的库，不懂的可以看我的另一篇文章《Ubuntu如何使用pip》
要安装好face_recognition的库
pip install face_recognition安装成功后即可
第三步
在工作空间/src下创建功能包
ros2 pkg create hj_service --build-type ament_cmake --dependencies rclcpp service_interfaces接着创建face_detect_client.py
在这个代码里面，你需要准备好一张里面有人脸的图片
import rclpy
from rclpy.node import Node
from service_interfaces.srv import FaceDetect
from sensor_msgs.msg import Image
from cv_bridge import CvBridge, CvBridgeError
import cv2
import sys
class FaceDetectClientNode(Node):
def __init__(self):
super().__init__('face_detect_client_node')
self.client = self.create_client(FaceDetect, 'face_detect_service')
while not self.client.wait_for_service(timeout_sec=1.0):
self.get_logger().info('Service not available, waiting again...')
self.bridge = CvBridge()
self.cv_image = None # 初始化cv_image为类属性
def send_request(self, image_path):
try:
# 读取图像文件
self.cv_image = cv2.imread(image_path) # 保存cv_image为类属性
if self.cv_image is None:
self.get_logger().info('Failed to read image')
return
# 将OpenCV图像转换为ROS图像消息
msg = self.bridge.cv2_to_imgmsg(self.cv_image, encoding="bgr8")
# 构建服务请求
request = FaceDetect.Request()
request.photo = msg
# 调用服务
self.future = self.client.call_async(request)
except CvBridgeError as e:
self.get_logger().info('CvBridge Error: {0}'.format(e))
def get_response(self):
while rclpy.ok():
rclpy.spin_once(self)
if self.future.done():
try:
response = self.future.result()
except Exception as e:
self.get_logger().info('Service call failed %r' % (e,))
else:
self.get_logger().info('Service response: use_time: %f, number: %d' % (response.use_time, response.number))
for i in range(response.number):
self.get_logger().info('Face %d: top: %d, right: %d, bottom: %d, left: %d' %
(i, response.top[i], response.right[i], response.bottom[i], response.left[i]))
# 绘制人脸边框
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.cv_image, (left, top), (right, bottom), (0, 255, 0), 2)
# 显示图像
cv2.imshow('Detected Faces', self.cv_image)
cv2.waitKey(0) # 等待按键
cv2.destroyAllWindows() # 关闭所有窗口
break
def main(args=None):
rclpy.init(args=args)
node = FaceDetectClientNode()
# 替换'image.jpg'为你的图像文件路径
node.send_request('/home/phj/Downloads/test.jpeg')
node.get_response()
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()face_detect_server.py
import rclpy
from rclpy.node import Node
from service_interfaces.srv import FaceDetect
from sensor_msgs.msg import Image
from cv_bridge import CvBridge, CvBridgeError
import cv2
import face_recognition
import time
class FaceDetectServerNode(Node):
def __init__(self, node_name):
super().__init__(node_name)
self.service = self.create_service(FaceDetect, 'face_detect_service', self.face_detect_callback)
self.bridge = CvBridge()
self.get_logger().info('Service created.')
def face_detect_callback(self, request, response):
if request.photo.data:
try:
cv_image = self.bridge.imgmsg_to_cv2(request.photo, "bgr8")
except CvBridgeError as e:
self.get_logger().info('CvBridge Error: %s' % e)
return
else:
self.get_logger().info("No image data received")
return
start = time.time()
face_locations = face_recognition.face_locations(cv_image, model="hog")
response.use_time = time.time() - start
response.number = len(face_locations)
# Initialize lists in the response
response.top = []
response.right = []
response.bottom = []
response.left = []
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 = FaceDetectServerNode('face_detect_server_node')
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()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>hj_service</name>
<version>0.0.0</version>
<description>TODO: Package description</description>
<maintainer email="root@todo.todo">root</maintainer>
<license>TODO: License declaration</license>
<depend>rclpy</depend>
<depend>service_interfaces</depend>
<depend>sensor_msgs</depend>
<test_depend>ament_copyright</test_depend>
<test_depend>ament_flake8</test_depend>
<test_depend>ament_pep257</test_depend>
<test_depend>python3-pytest</test_depend>
<export>
<build_type>ament_python</build_type>
</export>
</package>
setup.py
from setuptools import find_packages, setup
package_name = 'hj_service'
setup(
name=package_name,
version='0.0.0',
packages=find_packages(exclude=['test']),
data_files=[
('share/ament_index/resource_index/packages',
['resource/' + package_name]),
('share/' + package_name, ['package.xml']),
],
install_requires=['setuptools'],
zip_safe=True,
maintainer='root',
maintainer_email='root@todo.todo',
description='TODO: Package description',
license='TODO: License declaration',
tests_require=['pytest'],
entry_points={
'console_scripts': [
'service = hj_service.service_member_function:main',
'client = hj_service.client_member_function:main',
'param_sample = hj_service.param:main',
"face_detect_server_node = hj_service.face_detect_server:main",
"face_detect_client_node = hj_service.face_detect_client:main"
],
},
)
第四步：
打开一个终端
ros2 run hj_service face_detect_server_node
打开另一个终端
ros2 run hj_service face_detect_client_node
成功！
