TSINGSEE青犀视频开发人脸识别AI接口的实现

Posted 2021-12-31 TSINGSEE

大家知道我们的人脸识别已经在进行内测了，并会在不久的将来于EasyCVR及EasyGBS中进行测试。目前人脸识别AI是基于Python实现，在输入RTSP流的时候会直接开始识别人脸，并进行对比人脸的相似度，来判断是不是同一个人。大致实现如下：

face = my_face_recognition.my_face()
root_path = root + '/image/test_images'
known_people_list = os.listdir(root_path)

index = 1
for i in known_people_list:
    image_path = os.path.join(root_path, i)
    image = face_recognition.load_image_file(image_path)
    face.add_user(image, index, i.replace('.jpg', ''))
    index = index + 1


# path = root + '/image/test.mp4'
path = 'rtsp://admin:a1234567@192.168.99.114:554/cam/realmonitor?channel=1&subtype=0'
face.face_search_from_video(path)

def face_search_from_video(self, video_path, model='hog'):
    '''
    从一段视频中逐帧进行人脸识别,并且保存,
    :param video_path: 视频的路径
    :param model:人脸检测的模型，默认为hog，可选为cnn
    :return:
    '''

    fourcc = cv2.VideoWriter_fourcc(*'XVID')

    input_video = cv2.VideoCapture(video_path)

    ret, frame = input_video.read()
    print("frame")
    print(ret)
    # 帧数为每秒20帧
    out_video = cv2.VideoWriter(('RTSP' if video_path.find('rtsp') >= 0 else video_path.replace('.mp4', '')) + '_result.avi', fourcc, 5,
                                (frame.shape[1], frame.shape[0]), True)

    while ret:
        timestamp = int(round(time.time() * 1000))
        print("timestamp:%d", timestamp)
        frame = self.face_serch_from_picture(frame, model=model, show_result=False)
        cv2.imshow('frame', frame)
        cv2.waitKey(1)
        # out_video.write(frame)
        ret, frame = input_video.read()

以上方法是直接使用RTSP流来进行人脸识别，如果想要进行所有的语言都要识别人脸，最快的方法就是将人脸识别做成http接口用来调用，所以就要分离各个识别的方法。

具体思路先安装Python的http库：flask。安装方法：pip install flask。

下面是实现的http post接口及代码的实现：

1、先实现http接口

from flask import Flask, request, make_response, redirect, render_template
app = Flask(__name__)
if __name__ == "__main__":
    app.run('0.0.0.0', port=PORT, threaded=False, debug=False)

2、http实现人脸的录入，接口是以json的格式传入

@app.route('/add_user', methods=['POST'])  # application/json
def add_user():
    global idx
    data = request.get_data()
    body = "success": False, "message": "no data or no json data"
    if not data:
        return json.dumps(body, ensure_ascii=False)
    data_json = json.loads(data)
    if "image" not in data_json:
        body["message"] = "empty image"
        return json.dumps(body, ensure_ascii=False)
    if "name" not in data_json:
        body["message"] = "empty name"
        return json.dumps(body, ensure_ascii=False)

    im = face.base64_cv2(str(data_json["image"]))
    if im is None:
        body["message"] = "image format error"
        return json.dumps(body, ensure_ascii=False)
    isFace = face.add_user(im, idx, data_json["name"], model='hog')
    if not isFace:
        body["message"] = "entry failed"
        return json.dumps(body, ensure_ascii=False)
    idx += 1
    body["success"] = True
    body["message"] = ""
    return json.dumps(body, ensure_ascii=False)

3、http实现人脸对比，json的格式

@app.route('/search_user', methods=['POST'])
def search_user():
    body = "success": False, "message": "no search user", "data": []
    data = request.get_data()
    if idx <= 1:
        return json.dumps(body, ensure_ascii=False)
    if not data:
        body["message"] = "empty data"
        return json.dumps(body, ensure_ascii=False)
    data_json = json.loads(data)
    if "image" not in data_json:
        body["message"] = "empty image"
        return json.dumps(body, ensure_ascii=False)
    im = face.base64_cv2(str(data_json["image"]))
    if im is None:
        body["message"] = "image format error"
        return json.dumps(body, ensure_ascii=False)
    show = False
    if "show" in data_json:
        show = data_json["show"]
    result_json, images = face.face_search_from_image(im, show, model='hog')
    body["success"] = len(result_json) > 0
    body["data"] = result_json
    if images is not None:
        body["image"] = images
    body["message"] = "" if len(result_json) > 0 else "empty person"
    return json.dumps(body, ensure_ascii=False)

4、最后就是验证http是否可以，采用的是直接写html+js实现接口测试，代码如下：

// 注册人脸
AddUser(params) 
    this.isLoading = true
    let URL = `http://$this.HOST:$this.PORT`
    return axios.post(`$URL/add_user`, params)
,
// 查找录入的人脸
SearchUser(params) 
    this.isLoading = true
    let URL = `http://$this.HOST:$this.PORT`
    return axios.post(`$URL/search_user`, params)
,