Flask 服务器端应用程序：视频姿态估计和人脸检测 API

该代码为一个基于 Flask 框架的服务器端应用程序，主要提供了三个接口：

/upload：接收客户端上传的视频文件，保存到服务器端指定的文件夹中。
/pred3D：对上传的视频文件进行姿态估计，返回每一帧的 3D 关键点坐标。
/face2D 和 /face2DAll：对上传的视频文件进行人脸检测，返回人脸区域的关键点坐标、眼睛、嘴巴的比例以及头部姿态角度。

其中，allowed_file 函数用于判断上传的文件是否为 mp4 格式，progress 函数用于在控制台中打印进度条，Pose3D 和 Face2D 分别为两个模型的类对象，用于进行姿态估计和人脸检测。app.run() 函数用于在本地运行服务器，serve 函数用于在生产环境中运行服务器。

import os
from flask import Flask, flash, request, send_from_directory, jsonify
from waitress import serve

import cv2
import numpy as np
from model3d import Pose3D

from face import Face2D

app = Flask(__name__)

currpath = os.path.dirname(os.path.realpath(__file__))
UPLOAD_FOLDER = os.path.join(currpath, 'data')
ALLOWED_EXTENSIONS = {'mp4'}
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER

def allowed_file(filename):
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

@app.route('/upload', methods=['POST'])
def upload():
    if request.method == 'POST':
        data = {'succeed': False, 'msg': 'No file received'}
        # check if the post request has the file part
        if 'video' not in request.files:
            return jsonify(data)
        video = request.files['video']
        filename = request.form['name']
        if (not allowed_file(filename)):
            data['msg'] = 'not mp4 data'
            return jsonify(data)
        else:
            save_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
            if (os.path.exists(save_path)):
                os.remove(save_path)
            video.save(save_path)
            data['succeed'] = True
            data['msg'] = filename
            return jsonify(data)


def progress(percent, width=50):
    '''进度打印功能'''
    if percent >= 100:
        percent = 100

    show_str = ('[%%-%ds]' % width) % (int(width * percent / 100) * "#")  # 字符串拼接的嵌套使用
    print('
%s %d%%' % (show_str, percent), end='')


pose3d_model = Pose3D()
@app.route('/pred3D', methods=['POST'])
def pred3D():
    if request.method == 'POST':
        print("************ enter pose video process ************")
        data = {'succeed': False, 'msg': 'no video find', 'kps': [], 'framenum': 0}
        if 'name' not in request.form:
            return jsonify(data)
        else:
            filename = request.form['name']
            video_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
            if (not os.path.exists(video_path)):
                data['msg'] = 'Video not uploaded in server'
                return jsonify(data)
            cap = cv2.VideoCapture(video_path)
        allkps3D = []
        frame_idx = 0
        print("total frame:", cap.get(7))
        while (cap.isOpened()):
            progress(frame_idx / cap.get(7) * 100, width=30)
            ret, img = cap.read()
            if ret is not False:
                # start 2D predict
                kps2d = pose3d_model.infer2d(img)
                # start 3D predict
                kps3d = pose3d_model.infer3d(img, kps2d)
                allkps3D.append(kps3d.tolist())  # kps3d.reshape((-1,3)).tolist()
                frame_idx = frame_idx + 1
            else:
                data['succeed'] = True
                data['msg'] = 'Finish'
                allkps3D = np.squeeze(np.array(allkps3D, dtype=np.float32).reshape((-1, 1))).tolist()
                data['kps'] = allkps3D
                data['framenum'] = frame_idx
                return jsonify(data)

facemodel = Face2D()
@app.route('/face2D', methods=['POST'])
def face2D():
    data = {'succeed': False, 'leye': None, 'reye': None, 'mouth': None, 'euler': None}  #
    if request.method == 'POST':
        if 'image' not in request.files:
            return jsonify()
        else:
            imgfile = request.files['image']
            img = cv2.imdecode(np.asarray(bytearray(imgfile.read()), dtype='uint8'), cv2.IMREAD_COLOR)
            _, _, leye_ratio, reye_ratio, mouth_ratio, euler_angles = facemodel.detect(img)
            data['succeed'] = True
            data['leye'] = leye_ratio.item()
            data['reye'] = reye_ratio.item()
            data['mouth'] = mouth_ratio.item()
            data['euler'] = euler_angles
            return jsonify(data)

@app.route('/face2DAll', methods=['POST'])
def face2DAll():
    if request.method == 'POST':
        print("************ enter face video process ************")
        data = {'succeed': False, 'msg': 'no video find', 'all_leyes': [], 'all_reyes': [], 'all_mouthes': [], 'all_eulers': [], 'framenum': 0}
        if 'name' not in request.form:
            return jsonify(data)
        else:
            filename = request.form['name']
            video_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
            if not os.path.exists(video_path):
                data['msg'] = "video not uploaded in server"
                return jsonify(data)
            cap = cv2.VideoCapture(video_path)
            alldata = []
            frame_idx = 0
            print("total frame:", cap.get(7))
            leyes = []
            reyes = []
            mouthes = []
            eulers = []
            while (cap.isOpened()):
                progress(frame_idx / cap.get(7) * 100, width=30)
                ret, img = cap.read()
                if ret is not False:
                    _, _, leye_ratio, reye_ratio, mouth_ratio, euler_angles = facemodel.detect(img)
                    leyes.append(leye_ratio.item())
                    reyes.append(reye_ratio.item())
                    mouthes.append(mouth_ratio.item())
                    eulers.append(euler_angles)
                    frame_idx = frame_idx + 1
                else:
                    data['succeed'] = True
                    data['msg'] = 'Finish'
                    data['all_leyes'] = leyes
                    data['all_reyes'] = reyes
                    data['all_mouthes'] = mouthes
                    data['all_eulers'] = np.squeeze(np.array(eulers, dtype=np.float32).reshape((-1, 1))).tolist()
                    data['framenum'] = frame_idx
                    return jsonify(data)

if __name__ == "__main__":
    app.run(debug=True, host='0.0.0.0', port=6666)  # deployment
    # serve(app, host="0.0.0.0", port=5555)  # product