ASJServer/Object/IOTCore/IotObject.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import json

import boto3

from Ansjer.config import AWS_IOT_SES_ACCESS_CHINA_REGION, AWS_IOT_SES_ACCESS_CHINA_ID, AWS_IOT_SES_ACCESS_CHINA_SECRET, \
    AWS_IOT_SES_ACCESS_FOREIGN_REGION_AMERICA, AWS_IOT_SES_ACCESS_FOREIGN_ID, AWS_IOT_SES_ACCESS_FOREIGN_SECRET, \
    AWS_IOT_SES_ACCESS_FOREIGN_REGION_EUROPE, AWS_IOT_SES_ACCESS_FOREIGN_REGION_ASIA, AWS_IOT_SES_ACCESS_CHINA_ROLE, \
    AWS_IOT_SES_ACCESS_FOREIGN_ROLE


class IOTClient:
    def __init__(self, region_id=1):
        if region_id == 1 or region_id == 5:
            # 中国宁夏
            self.client = boto3.client('iot', region_name=AWS_IOT_SES_ACCESS_CHINA_REGION,
                                       aws_access_key_id=AWS_IOT_SES_ACCESS_CHINA_ID,
                                       aws_secret_access_key=AWS_IOT_SES_ACCESS_CHINA_SECRET)
            # 终端节点: https://cn-northwest-1.console.amazonaws.cn/iot/home?region=cn-northwest-1#/settings
            self.endpoint = 'a250bbr0p9u7as-ats.iot.cn-northwest-1.amazonaws.com.cn'
            self.iot_role = AWS_IOT_SES_ACCESS_CHINA_ROLE

        elif region_id == 2 or region_id == 3:
            # 美东弗吉尼亚
            self.client = boto3.client('iot', region_name=AWS_IOT_SES_ACCESS_FOREIGN_REGION_AMERICA,
                                       aws_access_key_id=AWS_IOT_SES_ACCESS_FOREIGN_ID,
                                       aws_secret_access_key=AWS_IOT_SES_ACCESS_FOREIGN_SECRET)
            self.endpoint = 'a2rqy12o004ad8-ats.iot.us-east-1.amazonaws.com'
            self.iot_role = AWS_IOT_SES_ACCESS_FOREIGN_ROLE

        elif region_id == 4:
            # 西欧爱尔兰
            self.client = boto3.client('iot', region_name=AWS_IOT_SES_ACCESS_FOREIGN_REGION_EUROPE,
                                       aws_access_key_id=AWS_IOT_SES_ACCESS_FOREIGN_ID,
                                       aws_secret_access_key=AWS_IOT_SES_ACCESS_FOREIGN_SECRET)
            self.endpoint = 'a2rqy12o004ad8-ats.iot.eu-west-1.amazonaws.com'
            self.iot_role = AWS_IOT_SES_ACCESS_FOREIGN_ROLE

    def create_provisioning_claim(self, templateName):

        result = self.client.create_provisioning_claim(templateName=templateName)

        res = {
            'certificateId': result['certificateId'],
            'certificatePem': result['certificatePem'],
            'publicKey': result['keyPair']['PublicKey'],
            'privateKey': result['keyPair']['PrivateKey'],
            'endpoint': self.endpoint
        }
        return res

    def register_to_iot_core(self, ThingName, thingGroup, response):
        try:
            result = self.client.create_keys_and_certificate(setAsActive=True)
            res = {
                'certificateId': result['certificateId'],
                'certificatePem': result['certificatePem'],
                'publicKey': result['keyPair']['PublicKey'],
                'privateKey': result['keyPair']['PrivateKey'],
                'endpoint': self.endpoint
            }
            # 搜索是否存在该物品组
            thing_groups_res = self.client.list_thing_groups(nextToken='', maxResults=1,
                                                             namePrefixFilter=thingGroup, recursive=False)
            if thing_groups_res['thingGroups']:
                thingGroupName = thing_groups_res['thingGroups'][0]['groupName']  # 获取物品组名称
            else:
                attributes = {
                    "update_time": "0"
                }
                thingGroupProperties = {
                    "thingGroupDescription": "OTA",
                    "attributePayload": {
                        "attributes": attributes,
                        "merge": False  # 更新时覆盖掉而不是合并
                    }
                }
                create_thing_group_res = self.client.create_thing_group(thingGroupName=thingGroup,
                                                                        thingGroupProperties=thingGroupProperties)
                thingGroupName = create_thing_group_res['thingGroupName']  # 获取物品组名称
            print('物品组:', thingGroupName)
            # 根据证书ID注册物品和策略
            templateBody = {
                "Parameters": {
                    "ThingName": {
                        "Type": "String"
                    },
                    "SerialNumber": {
                        "Type": "String"
                    },
                    "thingGroupName": {
                        "Type": "String"
                    },
                    "AWS::IoT::Certificate::Id": {
                        "Type": "String"
                    }
                },
                "Resources": {
                    "thing": {
                        "Type": "AWS::IoT::Thing",
                        "Properties": {
                            "AttributePayload": {},
                            # "ThingGroups" : ["v1-lightbulbs", {"Ref" : "DeviceLocation"}],
                            "ThingName": {
                                "Ref": "ThingName"
                            },
                            "ThingGroups": [{"Ref": "thingGroupName"}]
                        },
                        "OverrideSettings": {
                            "AttributePayload": "MERGE",
                            "ThingTypeName": "REPLACE",
                            "ThingGroups": "DO_NOTHING"
                        }
                    },
                    "certificate": {
                        "Type": "AWS::IoT::Certificate",
                        "Properties": {
                            "CertificateId": {"Ref": "AWS::IoT::Certificate::Id"},
                            "Status": "Active"
                        }
                    },
                    "policy": {
                        "Properties": {
                            "PolicyName": "My_Iot_Policy"
                        },
                        "Type": "AWS::IoT::Policy"
                    },
                }
            }

            templateBody = json.dumps(templateBody)
            parameters = {"ThingName": ThingName,
                          "thingGroupName": thingGroupName,
                          "AWS::IoT::Certificate::Id": res['certificateId']}
            self.client.register_thing(
                templateBody=templateBody,
                parameters=parameters
            )
            return res, parameters
        except Exception as e:
            print(e)
            return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))