• 本次就使用PaddleInference将PaddleHub上的两个模型串联起来，并部署在CPU桌面平台上，搭建一个简单的人脸识别系统

先展示一下效果

• 因为不想露脸所以使用摄像头拍摄手机播放的视频来大概演示一下效果
• 请自行忽略这个极简GUI，准确说是没有GUI
• 测试平台为Windows10，CPU为I5-6500
• 可以看到效果还是不错的

部署方案

• PaddleHub本身其实包含一个可直接调用的预测端口，但是使用起来不够灵活
• 所以本次使用的大概方案是：
  1. 将两个模型导出为推理模型
  2. 使用PaddleInference重写预测端口，优化预测效率
  3. 串联两个模型，实现人脸识别
  4. 使用预测结果实现一些小功能

导出推理模型

• 对于人脸检测模型这类PaddleHub的预置模型来讲，导出推理模型是非常简单的，通过直接调用模型的save_inference_model函数即可一键导出推理模型
• 对于人脸验证模型这类经过Finetune的模型来讲，导出推理模型需要先搭建配置模型，然后加载训练结束的模型参数，然后再调用save_inference_model函数即可导出推理模型

In [ ]

# 安装新版PaddleHub
!pip install paddlehub==1.8.1

In [ ]

# 人脸检测模型导出
import paddlehub as hub

# 加载模型
face_detector = hub.Module(name="pyramidbox_lite_mobile")

# 导出推理模型
face_detector.save_inference_model(
    dirname='inference/face_detection',
    model_filename='__model__',
    params_filename='__params__')

In [ ]

# 人脸验证模型导出
import paddlehub as hub
from paddlehub.dataset.base_cv_dataset import BaseCVDataset

# 新建临时train.txt
with open('data/train.txt', 'w', encoding='UTF-8') as f:
    for x in range(100):
        f.write('test 0\\n')

import paddlehub as hub
from paddlehub.dataset.base_cv_dataset import BaseCVDataset

# 自定义数据集
class FaceDataset(BaseCVDataset):	
   def __init__(self):	
       # 数据集存放位置
       self.dataset_dir = "/home/aistudio/data/"
       super(FaceDataset, self).__init__(
           base_path=self.dataset_dir,
           train_list_file="train.txt",
           label_list=['0','1']
           )

dataset = FaceDataset()

# 使用mobilenet_v3_large_imagenet_ssld预训练模型进行finetune
module = hub.Module(name="mobilenet_v3_large_imagenet_ssld")

# 数据读取器
data_reader = hub.reader.ImageClassificationReader(
    image_width=module.get_expected_image_width(),
    image_height=module.get_expected_image_height(),
    images_mean=module.get_pretrained_images_mean(),
    images_std=module.get_pretrained_images_std(),
    dataset=dataset)

# 优化器配置
strategy = hub.AdamWeightDecayStrategy(
    learning_rate=1e-3,
    lr_scheduler="linear_decay",
    warmup_proportion=0.1,
    weight_decay=0.0001,
    optimizer_name="adam")

# 总体配置
config = hub.RunConfig(
    use_cuda=False,
    num_epoch=10,
    checkpoint_dir="mobilenet_v3",
    batch_size=100,
    eval_interval=100,
    strategy=strategy)

# 任务构建
input_dict, output_dict, program = module.context(trainable=True)

img = input_dict["image"]

feature_map = output_dict["feature_map"]

feed_list = [img.name]

task = hub.ImageClassifierTask(
    data_reader=data_reader,
    feed_list=feed_list,
    feature=feature_map,
    num_classes=dataset.num_labels,
    config=config)

# 加载best_model
task.init_if_load_best_model()

# 导出推理模型
task.save_inference_model(
    dirname='inference/face_verification',
    model_filename='__model__',
    params_filename='__params__')

使用推理模型进行预测

• 使用推理模型部署一般分为如下5步：
  1. 读取数据
  2. 数据预处理
  3. 模型预测
  4. 结果后处理
  5. 结果展示
• 上述模型预测阶段也分为5步：
  1. 配置推理选项
  2. 创建Predictor
  3. 准备模型输入
  4. 模型推理
  5. 获取模型输出
• 下面就来演示一下如何使用刚刚导出的推理模型完成人脸的检测和验证
• 代码上有详细的注释，更多更详细的使用方法请参考[PaddleInference]

In [1]

# 检测模型预测
%matplotlib inline

import cv2
import numpy as np
from PIL import Image
from  matplotlib import pyplot as plt

from paddle.fluid.core import AnalysisConfig, PaddleTensor
from paddle.fluid.core import create_paddle_predictor

# 数据预处理
def pre_det(org_im, shrink):
    image = org_im.copy()
    image_height, image_width, image_channel = image.shape

    # 图像缩放
    if shrink != 1:
        image_height, image_width = int(image_height * shrink), int(
            image_width * shrink)
        image = cv2.resize(image, (image_width, image_height),
                           cv2.INTER_NEAREST)
    # HWC to CHW
    if len(image.shape) == 3:
        image = np.swapaxes(image, 1, 2)
        image = np.swapaxes(image, 1, 0)

    # 归一化
    mean = [104., 117., 123.]
    scale = 0.007843
    image = image.astype('float32')
    image -= np.array(mean)[:, np.newaxis, np.newaxis].astype('float32')
    image = image * scale
    image = np.expand_dims(image, axis=0).astype('float32')
    return image

# 数据后处理
# 输入原始图像，根据预测结果绘制人脸预测框，并裁剪人脸图像
def post_det(img, output_datas):
    img_h, img_w = img.shape[:2]
    new_img = img.copy()
    crops = []
    for data in output_datas:
        label, score, x1, y1, x2, y2 = data
        if score>0.8:
            x1, y1, x2, y2 = [int(_) for _ in [x1*img_w, y1*img_h, x2*img_w, y2*img_h]]
            crop = img[max(0, y1-50):min(y2+50,img_h),max(0, x1-50):min(x2+50,img_w),:]    
            h, w = crop.shape[:2]
            crop = cv2.resize(crop, (200, int(h/w*200))) if w>h else cv2.resize(crop, (int(w/h*200), 200))
            row_nums = 200-crop.shape[0]
            line_nums = 200-crop.shape[1]
            if row_nums%2 ==0:
                crop= np.pad(crop,((row_nums//2,row_nums//2),(0,0),(0,0)),'constant')
            else:
                crop= np.pad(crop,((row_nums//2,row_nums//2+1),(0,0),(0,0)),'constant')
            if line_nums%2 ==0:
                crop= np.pad(crop,((0,0),(line_nums//2,line_nums//2),(0,0)),'constant')
            else:
                crop= np.pad(crop,((0,0),(line_nums//2,line_nums//2+1),(0,0)),'constant')
            crops.append(crop)            
            cv2.rectangle(new_img, (x1, y1), (x2, y2), (255, 0, 0), 2)
    return new_img, crops

# 创建预测器
def create_predictor(model_file, params_file):
    # 创建配置
    config = AnalysisConfig(model_file, params_file)
    # 关闭GPU
    config.disable_gpu()
    # 开启mkldnn加速intel平台的CPU推理速度
    config.enable_mkldnn()
    # 关闭log显示
    config.disable_glog_info()
    # 开启ir优化
    config.switch_ir_optim(True)
    # 使用feed和fetch的算子
    config.switch_use_feed_fetch_ops(True)
    # 根据配置创建预测器
    predictor = create_paddle_predictor(config)
    return predictor

# 模型预测
def predict_det(predictor, inputs):
    # 转换输入数据为PaddleTensor
    inputs = PaddleTensor(inputs.copy())
    # 执行前向计算
    result = predictor.run([inputs])
    # 转换输出数据为ndarray
    output_data = result[0].as_ndarray()
    return output_data

# 实例化检测模型预测器
predictor = create_predictor('inference/face_detection/__model__', 'inference/face_detection/__params__')

# 读取图片
img = cv2.imread('img/test.jpg')

# 原始图片展示
plt.imshow(img[:,:,::-1])
plt.show()

# 图像预处理
img1 = pre_det(img, 0.5)

# 模型预测
output_data = predict_det(predictor, img1)

# 结果后处理
img, crops = post_det(img, output_data)

# 结果图片展示
plt.imshow(img[:,:,::-1])
plt.show()

with 1 Axes>

with 1 Axes>

In [3]

# 验证模型预测
%matplotlib inline

import cv2
import numpy as np
from PIL import Image
from  matplotlib import pyplot as plt

from paddle.fluid.core import AnalysisConfig
from paddle.fluid.core import create_paddle_predictor

# 图片拼接
def concatenate(true_img, crop):
    new = np.concatenate([true_img,crop],1)
    return new

# 数据预处理
def pre_val(img):
    img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
    img = Image.fromarray(img)

    # 图像缩放
    image = img.resize((224, 224), Image.LANCZOS)

    # HWC to CHW
    mean = np.array([0.485,0.456,0.406]).reshape(3, 1, 1)
    std = np.array([0.229,0.224,0.225]).reshape(3, 1, 1)
    image = np.array(image).astype('float32')
    if len(image.shape) == 3:
        image = np.swapaxes(image, 1, 2)
        image = np.swapaxes(image, 1, 0)

    # 归一化
    image /= 255
    image -= mean
    image /= std
    image = image[[0, 1, 2], :, :]
    image = np.expand_dims(image, axis=0).astype('float32')
    return image

# 创建预测器
def create_predictor(model_file, params_file):
    # 创建配置
    config = AnalysisConfig(model_file, params_file)
    # 关闭GPU
    config.disable_gpu()
    # 开启mkldnn加速intel平台的CPU推理速度
    config.enable_mkldnn()
    # 关闭log显示
    config.disable_glog_info()
    # 开启ir优化
    config.switch_ir_optim(True)
    # 不使用feed和fetch的算子
    config.switch_use_feed_fetch_ops(False)
    # 根据配置创建预测器
    predictor = create_paddle_predictor(config)
    return predictor

# 模型预测
def predict_val(predictor, inputs):
    # 获取输入向量名
    input_names = predictor.get_input_names()
    # 根据输入向量名获取输入向量
    input_tensor = predictor.get_input_tensor(input_names[0])
    # 将输入数据拷贝进输入向量
    input_tensor.copy_from_cpu(inputs)
    # 执行前向计算
    predictor.zero_copy_run()
    # 获取输出向量名
    output_names = predictor.get_output_names()
    # 根据输出向量名获取输出向量
    output_tensor = predictor.get_output_tensor(output_names[0])
    # 从输出向量中拷贝输出数据到输出变量上
    output_data = output_tensor.copy_to_cpu()
    return output_data

# 实例化检测模型预测器
predictor = create_predictor('inference/face_verification/__model__', 'inference/face_verification/__params__')

# 读取图片
img1 = cv2.imread('img/crop_0.jpg')
img2 = cv2.imread('img/crop_1.jpg')

# 图像拼接
img_true = concatenate(img1, img1)
img_false = concatenate(img1, img2)

# 输入图片展示
plt.imshow(img_true[:,:,::-1])
plt.show()
plt.imshow(img_false[:,:,::-1])
plt.show()

# 数据预处理
img_true = pre_val(img_true)
img_false = pre_val(img_false)

# 数据拼接
imgs = np.concatenate([img_true, img_false], 0)

# 模型预测
output_data = predict_val(predictor, imgs)

# 结果后处理
results = np.argmax(output_data, 1)

for i, result in enumerate(results):
    if result:
        print('第%d个样本匹配' % (i+1))
    else:
        print('第%d个样本不匹配' % (i+1))

with 1 Axes>

with 1 Axes>

第1个样本匹配
第2个样本不匹配

完整程序

• 只需要将上面的两个代码稍微封装，串联起来，就能实现一个简单的实时人脸识别系统
• 完整程序存放在face_recognition目录下，目录结构如下：
  • inference -- 存放推理模型
  • preprocess.py -- 数据预处理
  • postprocess.py -- 数据后处理
  • inference.py -- 模型推理
  • main.py -- 主程序
• 仅作为测试使用，未封装GUI界面
• 使用下面的代码即可启动程序
• 将按钮窗口关闭，并使用Esc键退出视频窗口，即可退出程序

In [ ]

# 请下载代码并在有摄像头的系统环境中执行
!python face_recognition/main.py

程序流程

• 通过main.py来介绍一下大致的程序流程
• 具体细节请参考源码

# 导入所需的包
import cv2, threading  
import numpy as np
from inference import AnalysisModel
from preprocess import pre_det, pre_val
from postprocess import post_det

from tkinter import Tk, Button

# 按钮点击函数，用于切换人脸
def change_face():
    global change_flag
    change_flag = True

# 主线程
def main():
    global change_flag
    
    # 开启摄像头
    cap = cv2.VideoCapture(0)
    
    # 初始化两个模型
    model_det = AnalysisModel('inference/face_detection/__model__', 
                            'inference/face_detection/__params__',
                            True,
                            False)

    model_val = AnalysisModel('inference/face_verification/__model__', 
                            'inference/face_verification/__params__',
                            False,
                            True)
    tmp = None
    font = cv2.FONT_HERSHEY_SIMPLEX
    
    while True:
  			# 读取当前帧
        sucess, img = cap.read()
        # 检测数据预处理
        img_det = pre_det(img, 0.3)
        # 检测模型预测
        result_det = model_det.predict_det(img_det)
        # 检测结果后处理
        img, crops, bboxes = post_det(img, result_det)  
        
        # 如果当前人脸信息不为空，则启动人脸验证
        if type(tmp) is np.ndarray:
            for crop, bbox in zip(crops, bboxes):
            	  # 验证数据预处理
                img_val = pre_val(tmp, crop)
                x1, y1 = bbox[:2]
                # 验证模型预测
                result_val = model_val.predict_val(img_val)
                # 验证结果后处理
                if np.argmax(result_val[0]):
                    img = cv2.putText(img, 'Success', (x1, y1-4), font, 0.6, (0, 255, 0), 2)                
                else:
                    img = cv2.putText(img, 'Faild', (x1, y1-4), font, 0.6, (0, 0, 255), 2)   
                    
        # 若更换人脸的标识为真，则切换人脸信息
        if (len(crops)>0)  and change_flag:
            tmp = crops[0]
            crop = crops[0]
            cv2.imshow('Face', crop)
            change_flag=False 
            
        # 使用窗口显示结果图片
        cv2.imshow('Face recognition', img)
        k = cv2.waitKey(1)
        if k == 27:
            #通过esc键退出摄像
            cv2.destroyAllWindows()
            break
            
if __name__=='__main__':
    global change_flag
    change_flag = False
    
    # 初始化按钮界面
    root = Tk()
    root.title('Button')
    button = Button(root, text ="点击抓取人脸图片", command = change_face)
    button.pack()
    
    # 初始化主线程
    main_thread = threading.Thread(target=main)
    
    # 启动主线程
    main_thread.start()
    
    # 启动按钮界面线程
    root.mainloop()

总结

• 这个人脸识别系统实测可用，效果也还能够接受
• 如果项目有任何错误的地方，欢迎大家在评论区中评论指正

关于作者

• 肖培楷
• 东莞理工学院 智能制造工程专业 大一在读
• 感兴趣的方向为：机器视觉和强化学习等