ASJPush/Object/SageMakerAiObject.py

# -*- encoding: utf-8 -*-
"""
@File    : SageMakerAiObject.py
@Time    : 2023/11/10 17:52
@Author  : stephen
@Email   : zhangdongming@asj6.wecom.work
@Software: PyCharm
"""
import base64
import json
import logging
import threading
import time
from io import BytesIO

import cv2
import numpy as np
from PIL import Image, UnidentifiedImageError

from AnsjerPush.Config.aiConfig import AI_IDENTIFICATION_TAGS_DICT
from AnsjerPush.config import CONFIG_EUR
from AnsjerPush.config import CONFIG_INFO
from AnsjerPush.config import PUSH_BUCKET
from Controller import AiController
from Object.AiEngineObject import AiEngine
from Object.OCIObjectStorage import OCIObjectStorage
from Object.enums.MessageTypeEnum import MessageTypeEnum
from Service.EquipmentInfoService import EquipmentInfoService
from Service.HuaweiPushService.HuaweiPushService import HuaweiPushObject
from Service.PushService import PushObject

LOGGER = logging.getLogger('time')


class SageMakerAiObject:

    @staticmethod
    def sage_maker_ai_server(uid, base64_list):

        try:
            start = time.time()
            url = 'ec2-34-192-147-108.compute-1.amazonaws.com:8001'
            model_name = 'pdchv'

            ai = AiEngine(url, model_name)

            if ai.health and ai.set_model(model_name):
                LOGGER.info(f'uid={uid} Health check passed and Model set successfully')
            else:
                LOGGER.info('SageMake fail')
                return False

            for i in range(len(base64_list)):
                base64_list[i] = base64_list[i].replace(' ', '+')

            img_list = list(map(base64.b64decode, base64_list))
            img_list = map(BytesIO, img_list)
            img_list = map(Image.open, img_list)
            img_list = map(np.array, img_list)
            img_list = map(lambda x: cv2.cvtColor(x, cv2.COLOR_RGB2BGR), img_list)
            img_list = np.array(list(img_list))

            nms_threshold = 0.45
            confidence = 0.5
            client_timeout = 3

            results = ai.yolo_infer(img_list, nms_threshold, confidence, client_timeout)

            end = time.time()
            LOGGER.info(f'uid={uid},{(end - start)}s,sageMaker Result={results}')

            ai.close()

            if results == 'e_timeout':
                return False
            if results == 'e_no_model':
                return False
            if results == 'e_connect_fail':
                return False

            return results
        except UnidentifiedImageError as e:
            LOGGER.info('***sagemakerUnidentifiedImageError***uid={},errLine={errLine}, errMsg={errMsg}'
                        .format(uid, errLine=e.__traceback__.tb_lineno, errMsg=repr(e)))
            return 'imageError'
        except Exception as e:
            LOGGER.info('***sagemakerException***uid={},errLine={errLine}, errMsg={errMsg}'
                        .format(uid, errLine=e.__traceback__.tb_lineno, errMsg=repr(e)))
            return 'imageError'

    @staticmethod
    def get_table_name(uid, ai_result, detect_group):
        """
            根据sageMaker识别结果得到标签组和坐标框
            :param uid: str，请求ID
            :param ai_result: dict，sageMaker识别结果
            :param detect_group: str，指定标签组
            :return: dict or None or False
        """
        try:
            LOGGER.info(f'***get_table_name***uid={uid}')
            # 定义标签组的映射关系
            ai_table_groups = {'person': 1, 'cat': 2, 'dog': 2, 'vehicle': 3, 'package': 4}
            event_group = []

            # 获取识别到的标签以及坐标信息
            for pic_key, pic_value in ai_result.items():
                for item in pic_value:
                    if 'class' in item:
                        event_group.append(ai_table_groups[item['class']])

            # 如果没有识别到标签则返回False
            if not event_group:
                LOGGER.info(f'***get_table_name***uid={uid},没有识别到标签={event_group}')
                return False

            # 去重标签组并转换为整数类型，同时将指定标签组转换为整数列表
            event_group = set(event_group)
            ai_group = list(map(int, detect_group.split(',')))
            is_success = False

            # 判断是否识别到了指定的标签
            for item in event_group:
                if item in ai_group:
                    is_success = True
                    break

            # 如果未识别到指定标签则返回False
            if not is_success:
                LOGGER.info(f'***get_table_name***uid={uid},没有识别到指定标签sageMakerLabel={event_group}')
                return False

            # 将标签组转换为整数表示的类型组
            event_type = int(''.join(map(str, sorted(event_group))))
            label_list = []

            # 根据标签组获取对应的标签名称
            for item in event_group:
                label_list.append(AI_IDENTIFICATION_TAGS_DICT[str(item)])
            LOGGER.info(f'***get_table_name***uid={uid},类型组={event_type},名称组={label_list}')

            # 返回包含事件类型、标签名称列表和坐标框的字典对象
            return {'event_type': event_type, 'label_list': label_list,
                    'coords': ai_result}
        except Exception as e:
            LOGGER.info('***get_table_name***uid={},errLine={errLine}, errMsg={errMsg}'
                        .format(uid, errLine=e.__traceback__.tb_lineno, errMsg=repr(e)))
            return False

    @staticmethod
    def save_push_message(uid, d_push_time, uid_push_qs, channel, res, file_list, notify):
        """
        保存推送消息
        """
        try:
            LOGGER.info(f'***save_push_message***uid={uid},res={res}')
            if not res:
                return False
            d_push_time = int(d_push_time)
            # 存储消息以及推送
            uid_push_list = [uid_push for uid_push in uid_push_qs]
            user_id_list = []
            nickname = uid_push_list[0]['uid_set__nickname']
            event_type = res['event_type']
            coords = json.dumps(res['coords']) if res['coords'] else ''  # 坐标框字典转json字符串
            now_time = int(time.time())
            region = 4 if CONFIG_INFO == CONFIG_EUR else 3
            # 推送表存储数据
            equipment_info_kwargs = {
                'device_uid': uid,
                'device_nick_name': nickname,
                'channel': channel,
                'is_st': 3,
                'storage_location': region,
                'event_type': event_type,
                'event_time': d_push_time,
                'add_time': now_time,
                'border_coords': coords
            }
            push_msg_list = []
            equipment_info_list = []
            equipment_info_model = EquipmentInfoService.randoms_choice_equipment_info()  # 随机获取其中一张推送表
            for up in uid_push_list:
                # 保存推送数据
                lang = up['lang']
                label_str = SageMakerAiObject().get_push_tag_message(lang, event_type)
                equipment_info_kwargs['alarm'] = label_str

                tz = up['tz']
                if tz is None or tz == '':
                    tz = 0
                user_id = up['userID_id']
                if user_id not in user_id_list:
                    equipment_info_kwargs['device_user_id'] = user_id
                    equipment_info_list.append(equipment_info_model(**equipment_info_kwargs))
                    user_id_list.append(user_id)

                # 推送
                push_type = up['push_type']
                app_bundle_id = up['appBundleId']
                token_val = up['token_val']

                # 推送标题和推送内容
                msg_title = PushObject.get_msg_title(nickname=nickname)
                msg_text = PushObject.get_ai_msg_text(channel=channel, n_time=d_push_time, lang=lang, tz=tz,
                                                      label=label_str)
                kwargs = {
                    'nickname': nickname,
                    'app_bundle_id': app_bundle_id,
                    'token_val': token_val,
                    'n_time': d_push_time,
                    'event_type': event_type,
                    'msg_title': msg_title,
                    'msg_text': msg_text,
                    'uid': uid,
                    'channel': channel,
                    'push_type': push_type
                }
                push_msg_list.append(kwargs)

            if equipment_info_list:  # 消息存表
                equipment_info_model.objects.bulk_create(equipment_info_list)

            SageMakerAiObject().upload_image_to_s3(uid, channel, d_push_time, file_list)  # 上传到云端
            if notify:  # 异步APP消息通知
                push_thread = threading.Thread(target=SageMakerAiObject.async_app_msg_push,
                                               kwargs={'uid': uid, 'push_msg_list': push_msg_list})
                push_thread.start()  # APP消息提醒异步推送

            AiController.AiView().save_cloud_ai_tag(uid, d_push_time, event_type, 0)  # 关联AI标签
            return True
        except Exception as e:
            LOGGER.info('***get_table_name***uid={},errLine={errLine}, errMsg={errMsg}'
                        .format(uid, errLine=e.__traceback__.tb_lineno, errMsg=repr(e)))
            return False

    @staticmethod
    def async_app_msg_push(uid, push_msg_list):
        """
        APP消息提醒异步推送
        """
        if not push_msg_list:
            LOGGER.info(f'***uid={uid}APP推送push_info为空***')
        for item in push_msg_list:
            push_type = item['push_type']
            item.pop('push_type')
            SageMakerAiObject.app_user_message_push(push_type, **item)
        LOGGER.info(f'***uid={uid}APP推送完成')

    @staticmethod
    def app_user_message_push(push_type, **kwargs):
        """
            ai识别 app用户消息推送
        """
        uid = kwargs['uid']
        try:
            if push_type == 0:  # ios apns
                PushObject.ios_apns_push(**kwargs)
            elif push_type == 1:  # android gcm
                PushObject.android_fcm_push_v1(**kwargs)
            elif push_type == 2:  # android jpush
                kwargs.pop('uid')
                PushObject.android_jpush(**kwargs)
            elif push_type == 3:
                huawei_push_object = HuaweiPushObject()
                huawei_push_object.send_push_notify_message(**kwargs)
            elif push_type == 4:  # android 小米推送
                PushObject.android_xmpush(**kwargs)
            elif push_type == 5:  # android vivo推送
                PushObject.android_vivopush(**kwargs)
            elif push_type == 6:  # android oppo推送
                PushObject.android_oppopush(**kwargs)
            elif push_type == 7:  # android 魅族推送
                PushObject.android_meizupush(**kwargs)
            else:
                LOGGER.info(f'uid={uid},{push_type}推送类型不存在')
                return False
            return True
        except Exception as e:
            LOGGER.info('ai推送消息异常,uid={},errLine:{}, errMsg:{}'.format(kwargs['uid']
                                                                       , e.__traceback__.tb_lineno, repr(e)))
            return False

    @staticmethod
    def get_push_tag_message(lang, event_type):
        """
        根据语言以及推送类型得到APP通知栏文案
        """
        event_type = str(event_type)
        types = []
        if len(event_type) > 1:
            for i in range(1, len(event_type) + 1):
                types.append(MessageTypeEnum(int(event_type[i - 1:i])))
        else:
            types.append(int(event_type))
        msg_cn = {1: '人', 2: '动物', 3: '车', 4: '包裹'}
        msg_en = {1: 'person', 2: 'animal', 3: 'vehicle', 4: 'package'}
        msg_text = ''
        for item in types:
            if lang == 'cn':
                msg_text += msg_cn.get(item) + ' '
            else:
                msg_text += msg_en.get(item) + ' '
        return msg_text

    @staticmethod
    def upload_image_to_s3(uid, channel, d_push_time, file_list):
        """
        上传图片到S3
        """
        # if CONFIG_INFO == CONFIG_US or CONFIG_INFO == CONFIG_EUR:
        #     # 创建 AWS 访问密钥
        #     aws_key = AWS_ACCESS_KEY_ID[1]
        #     aws_secret = AWS_SECRET_ACCESS_KEY[1]
        #     # 创建会话对象
        #     session = boto3.Session(
        #         aws_access_key_id=aws_key,
        #         aws_secret_access_key=aws_secret,
        #         region_name="us-east-1"
        #     )
        #     s3_resource = session.resource("s3")
        #     bucket_name = "foreignpush"
        # else:
        #     # 存国内
        #     aws_key = AWS_ACCESS_KEY_ID[0]
        #     aws_secret = AWS_SECRET_ACCESS_KEY[0]
        #     session = boto3.Session(
        #         aws_access_key_id=aws_key,
        #         aws_secret_access_key=aws_secret,
        #         region_name="cn-northwest-1")
        #     s3_resource = session.resource("s3")
        #     bucket_name = "push"

        try:
            region = 'eur' if CONFIG_INFO == CONFIG_EUR else 'us'
            oci = OCIObjectStorage(region)
            # 解码base64字符串为图像数据
            for index, pic in enumerate(file_list):
                image_data = base64.b64decode(pic)
                # 指定存储桶名称和对象键
                object_key = f'{uid}/{channel}/{d_push_time}_{index}.jpeg'

                # OCI上传对象
                oci.put_object(PUSH_BUCKET, object_key, image_data, 'image/jpeg')
                # AWS获取指定的存储桶
                # bucket = s3_resource.Bucket(bucket_name)

                # AWS上传图像数据到 S3
                # bucket.put_object(Key=object_key, Body=image_data)
            LOGGER.info(f'uid={uid},base64上传缩略图到S3成功')
        except Exception as e:
            LOGGER.error('***upload_image_to_s3,uid={},errLine:{}, errMsg:{}'
                         .format(uid, e.__traceback__.tb_lineno, repr(e)))

    @staticmethod
    def bird_recognition_demo(resultRsp):
        """
        鸟类识别示例代码
        """
        from PIL import Image
        import base64
        from io import BytesIO
        import json
        import boto3

        image_path = "E:/stephen/AnsjerProject/bird_sagemaker/test.png"
        image = Image.open(image_path).convert("RGB")

        # 将图像转换为 Base64 编码字符串
        image_buffer = BytesIO()
        image.save(image_buffer, format="JPEG")
        image_base64 = base64.b64encode(image_buffer.getvalue()).decode("utf-8")
        input_json = json.dumps({'input': image_base64})

        AWS_SERVER_PUBLIC_KEY = 'AKIARBAGHTGOSK37QB4T'
        AWS_SERVER_SECRET_KEY = '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'
        runtime = boto3.client("sagemaker-runtime",
                               aws_access_key_id=AWS_SERVER_PUBLIC_KEY,
                               aws_secret_access_key=AWS_SERVER_SECRET_KEY,
                               region_name='us-east-1')

        endpoint_name = 'ServerlessClsBirdEndpoint1'
        content_type = "application/json"
        payload = input_json
        session = boto3.Session()
        # runtime = session.client('sagemaker-runtime')
        import time
        start_time = time.time()
        response = runtime.invoke_endpoint(
            EndpointName=endpoint_name,
            ContentType=content_type,
            Body=payload
        )
        end_time = time.time()
        print("耗时: {:.2f}秒".format(end_time - start_time))

        result = json.loads(response['Body'].read().decode())
        # result = json.loads(response['Body'].read().decode())
        print(result)
        return resultRsp.json(0)