（八）用于人脸识别的Web API

目录

介绍

客户端应用

服务器端应用程序

在容器中运行系统

下一步

在这里，我们将开发用于将检测到的人脸图像发送到识别服务器的Python代码，使用简单的Web API包装AI人脸识别以从边缘设备接收人脸图像，并展示它如何协同工作。

• 下载 net.zip - 80.9 MB

人脸识别是人工智能(AI)的一个领域，现代深度学习(DL)方法在过去十年中取得了巨大成功。最好的人脸识别系统可以以与人类相同的精度识别图像和视频中的人物，甚至更好。

我们关于这个主题的系列文章分为两个部分：

• 人脸检测，客户端应用程序检测图像或视频源中的人脸，对齐检测到的人脸图片，并将它们提交给服务器。
• 人脸识别（这部分），服务器端应用程序进行人脸识别。

我们假设您熟悉DNN、Python、Keras和TensorFlow。

在之前的文章中，我们学习了如何在Raspberry Pi设备上使用MTCNN库检测人脸，以及如何使用FaceNet模型识别人脸。在本文中，我们将看到如何在一个简单的Web客户端-服务器系统中一起使用这些组件。

让我们从在边缘设备上运行的客户端部分开始。首先，我们为一个将人脸图像发送到服务器的简单类编写代码：

class ImgSend:
    def __init__(self, host, port, debug_mode=False):
        self.host = host
        self.port = port
        self.url = host+":"+str(port)+"/api/faceimg"
        self.dbg_mode = debug_mode
    
    def send(self, img):
        (_, encoded) = cv2.imencode(".png", img)
        
        data = encoded.tostring()
        headers = { "content-type": "image/png" }
        if self.dbg_mode:
            print("Sending request... ")
            #print(data)
        t1 = time.time()
        response = requests.post(self.url, data=data, headers=headers)
        t2 = time.time()
        dt = t2-t1
        if self.dbg_mode:
            print("Request processed: "+str(dt)+" sec")
        
        result = json.loads(response.text)
        
        return result

构造函数接收host和port参数，形成带有特殊路径/api/faceimg的最终URL，将请求路由到人脸识别方法。在send方法中，我们将图像编码为png格式，将其转换为字符串，并通过requests.post函数将字符串发送到服务器。

我们还必须修改本文（参考“Raspberry Pi 上的人脸检测”部分）文章中描述的人脸检测器。

class VideoWFR:    
    def __init__(self, detector, sender):
        self.detector = detector
        self.sender = sender
    
    def process(self, video, align=False, save_path=None):
        detection_num = 0;
        rec_num = 0
        capture = cv2.VideoCapture(video)
        img = None

        dname = 'AI face recognition'
        cv2.namedWindow(dname, cv2.WINDOW_NORMAL)
        cv2.resizeWindow(dname, 960, 720)
        
        frame_count = 0
        dt = 0
        if align:
            fa = Face_Align_Mouth(160)
            
        # Capture all frames
        while(True):    
            (ret, frame) = capture.read()
            if frame is None:
                break
            frame_count = frame_count+1
            
            t1 = time.time()
            faces = self.detector.detect(frame)
            f_count = len(faces)
            detection_num += f_count
            
            names = None
            if (f_count>0) and (not (self.sender is None)):
                names = [None]*f_count
                for (i, face) in enumerate(faces):
                    if align:
                        (f_cropped, f_img) = fa.align(frame, face)
                    else:
                        (f_cropped, f_img) = self.detector.extract(frame, face)
                    if (not (f_img is None)) and (not f_img.size==0):
                        response = self.sender.send(f_img)
                        is_recognized = response["message"]=="RECOGNIZED"
                        print(response["message"])
                        if is_recognized:
                            print(response["name"]+": "+response["percent"])
                        
                        if is_recognized:
                            rec_num += 1
                            name = response["name"]
                            percent = int(response["percent"])
                            conf = percent*0.01
                            names[i] = (name, conf)
                            if not (save_path is None):
                                ps = ("%03d" % rec_num)+"_"+name+"_"+("%03d" % percent)+".png"
                                ps = os.path.join(save_path, ps)
                                cv2.imwrite(ps, f_img)
                        
            t2 = time.time()
            dt = dt + (t2-t1)
                    
            if len(faces)>0:
                Utils.draw_faces(faces, (0, 0, 255), frame, True, True, names)
            
            # Display the resulting frame
            cv2.imshow(dname,frame)
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
        
        capture.release()
        cv2.destroyAllWindows()
        
        if dt>0:
            fps = detection_num/dt
        else:
            fps = 0
        
        return (detection_num, rec_num, fps)

现在我们有了人脸识别器，而不仅仅是人脸检测器。它包括一个内部MTCNN检测器和一个图像发送器。当检测到人脸时，将其发送到服务器。当收到来自服务器的响应时，它会被解析并保存到指定的文件夹中。

让我们转到服务器应用程序。我们使用Flask微框架将我们的面部识别代码与Web API包装起来：

import flask
from flask import Flask, request, Response

print(flask.__version__)

# Initialize the Flask application
app = Flask(__name__)

rec = None
f_db = None
rec_data = None
save_path = None

@app.route("/api/faceimg", methods=['POST'])
def test():
    response = {}
    r_status = 200
    r = request
    
    print("Processing recognition request... ")
    t1 = time.time()
    nparr = np.fromstring(r.data, np.uint8)
    img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
    
    embds = rec.embeddings(img)
    data = rec.recognize(embds, f_db)
    t2 = time.time()
    dt = t2-t1
    print("Recognition request processed: "+str(dt)+" sec")
    
    rec_data.count()
    ps = ""
    info = ""
    if not (data is None):
        (name, dist, p_photo) = data
        conf = 1.0 - dist
        percent = int(conf*100)
        info = "Recognized: "+name+" "+str(conf)
        ps = ("%03d" % rec_data.get_count())+"_"+name+"_"+("%03d" % percent)+".png"
        response = { "message": "RECOGNIZED",
                     "name": name,
                     "percent": str(percent) }
    else:
        info = "UNRECOGNIZED"
        ps = ("%03d" % rec_data.get_count())+"_unrecognized"+".png"
        response = { "message": "UNRECOGNIZED" }
    
    print(info)
    
    if not (save_path is None):
       ps = os.path.join(save_path, ps)
       cv2.imwrite(ps, img)
        
    # encode response using jsonpickle
    response_pickled = jsonpickle.encode(response)
    return Response(response=response_pickled, status=r_status, mimetype="application/json")

初始化Flask应用程序后，我们应用route装饰器来触发test具有指定URL（客户端应用程序使用的URL）的方法。在这种方法中，我们解码接收到的人脸PNG图像，获取嵌入，识别人脸，并将响应发送回客户端。

最后，这是运行我们的Web应用程序的代码：

if __name__ == "__main__":
    host = str(sys.argv[1])
    port = int(sys.argv[2])

    # FaceNet recognizer
    m_file = r"/home/pi_fr/net/facenet_keras.h5"
    rec = FaceNetRec(m_file, 0.5)
    rec_data = RecData()
    print("Recognizer loaded.")
    print(rec.get_model().inputs)
    print(rec.get_model().outputs)
    
    # Face DB 
    save_path = r"/home/pi_fr/rec"
    db_path = r"/home/pi_fr/db"
    f_db = FaceDB()
    f_db.load(db_path, rec)
    db_f_count = len(f_db.get_data())
    print("Face DB loaded: "+str(db_f_count))
    
    print("Face recognition running")
          
    #host = "0.0.0.0"
    #port = 50
    app.run(host=host, port=port, threaded=False)

由于我们要在创建的Docker容器中运行Web应用程序（参考上一部分），我们需要使用适当的网络设置启动这个容器。使用以下命令从镜像创建一个新容器：

c:\>docker network create my-net
c:\>docker create --name FR_2 --network my-net --publish 5050:50 sergeylgladkiy/fr:v1

现在我们可以在容器中运行应用程序（注意，因为它在容器内部，我们指定内部端口 50）：

# python /home/pi_fr/pi_fr_facenet.run_align_dock_flask.lnx.py 0.0.0.0 50

if __name__ == "__main__":
    #v_file = str(sys.argv[1])
    #host = str(sys.argv[2])
    #port = int(sys.argv[3])
    
    v_file = r"/home/pi/Desktop/PI_FR/video/5_2.mp4"
    host = "http://192.168.2.135"
    port = 5050
    
    # Video Web recognition 
    save_path = r"/home/pi/Desktop/PI_FR/rec"
    d = MTCNN_Detector(50, 0.95)
    sender = ImgSend(host, port, True)
    vr = VideoWFR(d, sender)

    (f_count, rec_count, fps) = vr.process(v_file, True, save_path)

    print("Face detections: "+str(f_count))
    print("Face recognitions: "+str(rec_count))
    print("FPS: "+str(fps))

注意客户端使用的是宿主机的IP和端口（5050），而不是容器的IP和内部端口号（50）。

如您所见，识别请求的处理速度非常快；只用了大约0.07秒。这是正确系统架构的证明。客户端仅将裁剪和对齐的检测到的人脸图像发送到服务器，从而减少网络负载，而识别算法在功能强大的服务器计算机上运行。

在本系列的下一篇文章中，我们将展示如何在Kubernetes上运行人脸识别服务器。

https://www.codeproject.com/Articles/5305701/Web-API-for-Face-Recognition