from compreface_utils import FaceUtils, DataUtils from compreface import CompreFace from compreface.service import RecognitionService from compreface.collections import FaceCollection from compreface.collections.face_collections import Subjects import yaml import tempfile import json from flask_cors import CORS import os from flask import Flask, request, jsonify app = Flask(__name__) CORS(app) clock_in_dict = {} # 初始化 CompreFace 客户端 with open('config.yaml', 'r') as file: configData = yaml.safe_load(file) DOMAIN: str = configData['compreface_service']['DOMAIN'] PORT: str = configData['compreface_service']['PORT'] API_KEY: str = configData['compreface_service']['API_KEY'] compre_face: CompreFace = CompreFace(DOMAIN, PORT) # recognition 是服务实例 recognition: RecognitionService = compre_face.init_face_recognition(API_KEY) # 获取人员列表 @app.route('/getSubjects') def getSubjects(): subjects: Subjects = recognition.get_subjects() subjects_list = subjects.list() return jsonify({"data": subjects_list}) # 添加人员 @app.route('/addSubject', methods=['POST']) def addSubject(): subjects: Subjects = recognition.get_subjects() data = request.json subject_name = data.get('subject_name') result = subjects.add(subject_name) return jsonify({"data": result}) # 删除人员 @app.route('/deleteSubject',methods=['POST']) def deleteSubject(): subjects: Subjects = recognition.get_subjects() # face_collection是人脸集合 face_collection: FaceCollection = recognition.get_face_collection() data = request.json subject_name = data.get('subject_name') face_collection.delete_all(subject_name) result = subjects.delete(subject_name) return jsonify({"data": result}) # 更改人员名称 @app.route('/renameSubject', methods=['POST']) def renameSubject(): subjects: Subjects = recognition.get_subjects() data = request.json old_subject_name = data.get('old_subject_name') new_subject_name = data.get('new_subject_name') result = subjects.update(old_subject_name, new_subject_name) return jsonify({"data": result}) # 按名字查询 @app.route('/selectSubject', methods=['POST']) def selectSubject(): subjects: Subjects = recognition.get_subjects() data = request.json Query = data.get('query', []) subject_list = subjects.list() matching_subjects = [ name for name in subject_list['subjects'] if all(query in name for query in Query) ] return jsonify({"data": matching_subjects}),200 # 获取图片id集合 @app.route('/getFaceImgs', methods=['POST']) def getFaceImgs(): # face_collection是人脸集合 face_collection: FaceCollection = recognition.get_face_collection() data = request.json subject_name = data.get('subject_name') face_list = face_collection.list() print(face_list) filtered_ids = DataUtils.filter_images_by_subject(face_list, subject_name) return jsonify({"data": filtered_ids}) # 根据人名添加图片 @app.route('/addFaceImg', methods=['POST']) def addFaceImg(): # face_collection是人脸集合 face_collection: FaceCollection = recognition.get_face_collection() # 获取图片文件和主题名称 imgFile = request.files['img'] subject_name = request.form.get('subject_name') # 检查是否成功获取了图片文件和主题名称 if not imgFile or not subject_name: return jsonify({'data': 'error'}), 400 try: # 创建一个临时文件 temp_dir = tempfile.mkdtemp() temp_path = os.path.join(temp_dir, imgFile.filename) imgFile.save(temp_path) # 这里可以继续执行你的人脸识别集合添加逻辑 result = face_collection.add(image_path=temp_path, subject=subject_name) print(result) return jsonify({'data': result}) finally: # 删除临时文件和目录 if os.path.exists(temp_path): os.remove(temp_path) os.rmdir(temp_dir) # 根据image_id删除图片 @app.route('/deleteFaceImg', methods=['POST']) def deleteFaceImg(): # face_collection是人脸集合 face_collection: FaceCollection = recognition.get_face_collection() data = request.json image_id = data.get('image_id') print(image_id) result = face_collection.delete(image_id) return jsonify({"data": result}) # 修改识别配置 @app.route('/config', methods=['POST']) def config(): data = request.json start_time_config = data.get('start_time') end_time_config = data.get('end_time') frequency_face_config = data.get('frequency_face') # @app.route('/login', methods=['POST']) # def login(): # data = request.json # username = data.get('username') # password = data.get('password') # if username == 'admin' and password == '12345678Aa': # return 'success', 200 # TODO # 1. 先接收半个小时内的告警图片保存到本地 # 2. 定时任务方法，将半个小时内的图片依次调用compreface接口，返回result 整理数据，删除本地图片 # 2.1 整理数据 三种情况，工人，陌生人 ，脏数据 # 置信值 > 0.8 ==>工人 0.5 < 置信值 < 0.8 ==>陌生人 置信值 < 0.5 ==>抛弃 @app.route('/compreface', methods=['POST']) def receiveFrameToComprefaceOnly(): if 'file' not in request.files: return jsonify({"error": "No file part"}), 400 file = request.files['file'] if file.filename == '': return jsonify({"error": "No selected file"}), 400 if file: # 保存文件到临时位置 file_path = os.path.join("/tmp", file.filename) file.save(file_path) # 使用 CompreFace SDK 进行面部识别 result = recognition.recognize(file_path) # 删除临时文件 os.remove(file_path) return jsonify(result) if __name__ == '__main__': app.run(port=12500 ,debug=False, host='0.0.0.0') # { # "result" : [ { # "age" : { # "probability": 0.9308982491493225, # "high": 32, # "low": 25 # }, # "gender" : { # "probability": 0.9898611307144165, # "value": "female" # }, # "mask" : { # "probability": 0.9999470710754395, # "value": "without_mask" # }, # "embedding" : [ 9.424854069948196E-4, "...", -0.011415496468544006 ], # "box" : { # "probability" : 1.0,表示检测到的面部位置的置信度 TODO 可以通过这部分检测是否有人脸 # "x_max" : 1420,表示检测到的面部在图像中的边界框坐标 TODO 画框 # "y_max" : 1368, # "x_min" : 548, # "y_min" : 295 # }, # "landmarks" : [ [ 814, 713 ], [ 1104, 829 ], [ 832, 937 ], [ 704, 1030 ], [ 1017, 1133 ] ], # "subjects" : [ { # "similarity" : 0.97858, 百分之0.9的可能性是subject1 # "subject" : "subject1" # } ], # "execution_time" : { # "age" : 28.0, # "gender" : 26.0, # "detector" : 117.0, # "calculator" : 45.0, # "mask": 36.0 # } # } ], # "plugins_versions" : { # "age" : "agegender.AgeDetector", # "gender" : "agegender.GenderDetector", # "detector" : "facenet.FaceDetector", # "calculator" : "facenet.Calculator", # "mask": "facemask.MaskDetector" # } # }