Server-cpp/project-responder-hs/api/Task.py

from hub import methods, Global

import cv2
import base64


class Task(object):

    @classmethod
    def task002(cls, **params):
        """获取特征值接口"""
        run_at = methods.now_ts()
        try:

            # --- get face_features ---
            call_id = params.get('call_id')
            save_is = params.get('save_is')
            rdb = Global.get_redis_client()
            image_bytes = rdb.get_one(key=call_id)

            # --- 人脸裁剪逻辑 ---
            # methods.debug_log('Task.task002', f"m-24: {type(image_bytes)}")
            image_array = Global.scrfd_agent.image_bytes_to_image_array(image_bytes)
            # methods.debug_log('Task.task002', f"m-24: {type(image_array)}")
            # inference_result = Global.scrfd_agent.inference(image_array)
            inference_result = Global.scrfd_agent.inference_with_image_array(image_array)

            # --- save file ---
            if save_is:
                save_at = methods.now_string('%Y-%m%d-%H%M%S-%f')
                frame = Global.arcface_agent.image_bytes_to_image_array(image_bytes)
                face_file_path = f"/home/server/resources/vms-files/{save_at}-raw.jpg"
                cv2.imwrite(face_file_path, frame)
            else:
                face_file_path = ''

            # --- check ---
            if not inference_result:
                return dict(code=1, details=f"something is wrong.")

            # --- 歪脸矫正 ---
            face_image = inference_result[0].get('align_face')
            probability = 0.0

            # --- 提取人脸特征 ---
            face_features = Global.arcface_agent.get_face_features_normalization_by_image_array(face_image)

            # --- check ---
            if face_features is None:
                return dict(code=2, details=f"something is wrong.")

            # --- debug to close --- todo 存在歪脸矫正后，图像图像变模糊的问题
            # save_at = methods.now_string('%Y-%m%d-%H%M%S-%f')
            # face_file_path_x = f"/home/server/resources/vms-files/{save_at}-align.jpg"
            # cv2.imwrite(face_file_path_x, face_image)

            # --- 增加戴口罩人脸底库特征与图片 ---
            masked_face_image = Global.face_to_mask_face.add_mask_one(face_image)
            if masked_face_image is not None:
                face_features_2 = Global.arcface_agent.get_face_features_normalization_by_image_array(masked_face_image)
            else:
                face_features_2 = face_features
                masked_face_image = face_image

            # --- save file ---
            if save_is:
                save_at = methods.now_string('%Y-%m%d-%H%M%S-%f')
                face_file_path_2 = f"/home/server/resources/vms-files/{save_at}-masked.jpg"
                cv2.imwrite(face_file_path_2, masked_face_image)
            else:
                face_file_path_2 = ''

            # --- update call_id value ---
            rdb.set_one(key=call_id, data=(face_features, probability, face_file_path,
                                           face_features_2, probability, face_file_path_2), expire_time=600)
            return dict(code=0, details=f"congratulations. | use time {methods.now_ts() - run_at}s")

        except Exception as exception:

            methods.debug_log('Task.task002', f"m-76: exception | {exception}")
            methods.debug_log('Task.task002', f"m-76: traceback | {methods.trace_log()}")
            return dict(code=-1, details=f"{methods.trace_log()}")

    @classmethod
    def task003(cls, **params):
        """
        支持8301接口
        """
        run_at = methods.now_ts()
        try:
            # --- get data ---
            call_id = params.get('call_id')
            rdb = Global.get_redis_client()
            input_data = rdb.get_one(key=call_id)

            # --- 人脸裁剪逻辑 ---
            image_array = Global.scrfd_agent.image_bytes_to_image_array(input_data.get('image_bytes'))
            inference_result = Global.scrfd_agent.inference_with_image_array(image_array)

            # --- 歪脸矫正 ---
            face_image = inference_result[0].get('align_face')

            # --- 提取人脸特征 ---
            face_features = Global.arcface_agent.get_face_features_normalization_by_image_array(face_image)
            # methods.debug_log('Task.task003', f"m-105: face_features | {face_features}")

            # --- fill face_type_name_dict ---
            """
            face_type_name_dict = {<type_uuid>: <name>}
            """
            face_type_name_dict = dict()
            for item in Global.mdb.get_all('FaceType'):
                uuid = str(item.get('_id'))
                face_type_name_dict[uuid] = item.get('name')

            # --- fill d5 ---
            d5 = dict()  # {<face_uuid>: {info}}
            for item in Global.mdb.get_all('Face'):
                face_uuid = str(item.get('_id'))
                d5[face_uuid] = item.copy()

            # --- fill d2 ---
            """
            FaceLog: 陌生人访问日志表
            FaceLog.face_image_path: 抓拍人脸图像路径
            FaceLog.face_probability: 人脸概率
            FaceLog.base_face_uuid: 底库人脸id
            FaceLog.base_face_image_path: 底库人脸图片路径
            FaceLog.face_similarity: 对比相似度
            FaceLog.count: 条目计数
            """
            d2 = list()
            start_ts = methods.string_to_ts(f"{input_data.get('start_date_at')}", pattern='%Y-%m-%d %H:%M:%S')
            end_ts = methods.string_to_ts(f"{input_data.get('end_date_at')}", pattern='%Y-%m-%d %H:%M:%S') + 1
            items = list(Global.xdb.filter_by_time_range('FaceLog', start_ts, end_ts))
            for item in items:

                # --- check ---
                base_face_uuid = item.get('base_face_uuid')
                if base_face_uuid not in d5:
                    continue

                # --- check ---
                # face = Global.mdb.get_one_by_id('Face', base_face_uuid)
                face = d5.get(base_face_uuid)
                face_features_1 = methods.pickle_loads(face.get('face_feature_info_list')[0].get('face_features'))
                if face_features_1 is None:
                    continue

                # --- check ---
                dist = Global.arcface_agent.compare_faces_by_normalization(face_features, face_features_1)
                if dist < input_data.get('face_similarity'):
                    continue

                # --- fill ---
                d4 = item, face, dist
                d2.append(d4)

            # --- fill d3 ---
            d3 = list()
            page = input_data.get('page')
            size = input_data.get('size')
            for item, face, dist in d2[(page - 1) * size: page * size]:

                # --- fill d1 --- todo 返回数据 -> 抓拍照片、检索照片、相似度
                d1 = {
                    'face_uuid': item.get('base_face_uuid'),
                    'face_name': face.get('face_name'),
                    'record_at': methods.string_to_ts(item.get('time'), pattern='%Y-%m-%dT%H:%M:%S.%fZ'),
                    'snap_face_image_b64': str(),  # 抓拍照片
                    'base_face_image_b64': str(),  # 对比照片
                    'face_type_name_list': list(),  # 类别名称
                    'face_similarity': dist,  # 相似度
                }
                # methods.debug_log('actions.action_8201', f"m-86: d1 is {d1}")

                # --- fill snap_face_image_b64 ---
                file_path = item.get('face_image_path')
                if file_path and methods.is_file(file_path):
                    frame = cv2.imread(file_path)
                    if frame is not None:
                        _, image = cv2.imencode('.jpg', frame)
                        base64_data = base64.b64encode(image)  # byte to b64 byte
                        s = base64_data.decode()  # byte to str
                        d1['snap_face_image_b64'] = f'data:image/jpeg;base64,{s}'

                # --- fill base_face_image_b64 ---
                # file_path = item.get('base_face_image_path')
                # if file_path and methods.is_file(file_path):
                #     frame = cv2.imread(file_path)
                #     if frame is not None:
                #         _, image = cv2.imencode('.jpg', frame)
                #         base64_data = base64.b64encode(image)  # byte to b64 byte
                #         s = base64_data.decode()  # byte to str
                #         d1['base_face_image_b64'] = f'data:image/jpeg;base64,{s}'

                # --- fill base_face_image_b64 ---
                frame = Global.arcface_agent.image_bytes_to_image_array(input_data.get('image_bytes'))
                if frame is not None:
                    _, image = cv2.imencode('.jpg', frame)
                    base64_data = base64.b64encode(image)  # byte to b64 byte
                    s = base64_data.decode()  # byte to str
                    d1['base_face_image_b64'] = f'data:image/jpeg;base64,{s}'

                # --- fill face_type_name_list ---
                face_type_uuid_list = face.get('face_type_uuid_list')
                if face_type_uuid_list:
                    d1['face_type_name_list'] = [face_type_name_dict.get(i)
                                                 for i in face_type_uuid_list if face_type_name_dict.get(i)]

                # --- append d1 ---
                d3.append(d1)

            # --- update call_id value ---
            rdb.set_one(key=call_id, data=(d3, len(d2)), expire_time=600)
            return dict(code=0, details=f"congratulations. | use time {methods.now_ts() - run_at}s")

        except Exception as exception:

            methods.debug_log('Task.task003', f"m-76: exception | {exception}")
            methods.debug_log('Task.task003', f"m-76: traceback | {methods.trace_log()}")
            return dict(code=-1, details=f"{methods.trace_log()}")