ASJServer/Object/IOTCore/IotObject.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import json
from abc import ABCMeta,abstractmethod

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


class IOTObject(metaclass=ABCMeta):

    @abstractmethod
    def create_provisioning_claim(self, templateName):
        pass

    @abstractmethod
    def create_keys_and_certificate(self, uid):
        pass


class ChinaIOTClient(IOTObject):

    def __init__(self):
        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)

    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': 'a250bbr0p9u7as-ats.iot.cn-northwest-1.amazonaws.com.cn'
        }
        return res

    def create_keys_and_certificate(self, uid, device_version):
        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': 'a250bbr0p9u7as-ats.iot.cn-northwest-1.amazonaws.com.cn'
        }
        # 根据证书ID注册物品和策略
        templateBody = {
            "Parameters": {
                "ThingName": {
                    "Type": "String"
                },
                "SerialNumber": {
                    "Type": "String"
                },
                "DeviceLocation": {
                    "Type": "String"
                },
                "AWS::IoT::Certificate::Id": {
                    "Type": "String"
                }
            },
            "Resources": {
                "thing": {
                    "Type": "AWS::IoT::Thing",
                    "Properties": {
                        "AttributePayload": {},
                        "ThingGroups": {
                            "Ref": "ThingGroups"
                        },
                        "ThingName": {
                            "Ref": "ThingName"
                        },
                    },
                    "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": "Ansjer_Device_" + uid,
                      "ThingGroups": device_version,
                      "AWS::IoT::Certificate::Id": res['certificateId']}
        self.client.register_thing(
            templateBody=templateBody,
            parameters=parameters
        )
        return res, parameters


class AmericaIOTClient(IOTObject):

    def __init__(self):
        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)

    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': 'a2rqy12o004ad8-ats.iot.us-east-1.amazonaws.com'
        }
        return res

    def create_keys_and_certificate(self, uid, device_version):
        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': 'a2rqy12o004ad8-ats.iot.us-east-1.amazonaws.com'
        }
        # 根据证书ID注册物品和策略
        templateBody = {
            "Parameters": {
                "ThingName": {
                    "Type": "String"
                },
                "SerialNumber": {
                    "Type": "String"
                },
                "DeviceLocation": {
                    "Type": "String"
                },
                "AWS::IoT::Certificate::Id": {
                    "Type": "String"
                }
            },
            "Resources": {
                "thing": {
                    "Type": "AWS::IoT::Thing",
                    "Properties": {
                        "AttributePayload": {},
                        "ThingGroups": {
                            "Ref": "ThingGroups"
                        },
                        "ThingName": {
                            "Ref": "ThingName"
                        },
                    },
                    "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": "Ansjer_Device_" + uid,
                      "ThingGroups": device_version,
                      "AWS::IoT::Certificate::Id": res['certificateId']}
        self.client.register_thing(
            templateBody=templateBody,
            parameters=parameters
        )
        return res, parameters


class AsiaIOTClient(IOTObject):

    def __init__(self):
        self.client = boto3.client('iot', region_name=AWS_IOT_SES_ACCESS_FOREIGN_REGION_ASIA,
                                   aws_access_key_id=AWS_IOT_SES_ACCESS_FOREIGN_ID,
                                   aws_secret_access_key=AWS_IOT_SES_ACCESS_FOREIGN_SECRET)

    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': 'a2rqy12o004ad8-ats.iot.ap-southeast-1.amazonaws.com'
        }
        return res

    def create_keys_and_certificate(self, uid, device_version):
        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': 'a2rqy12o004ad8-ats.iot.ap-southeast-1.amazonaws.com'
        }
        # 根据证书ID注册物品和策略
        templateBody = {
            "Parameters": {
                "ThingName": {
                    "Type": "String"
                },
                "SerialNumber": {
                    "Type": "String"
                },
                "DeviceLocation": {
                    "Type": "String"
                },
                "AWS::IoT::Certificate::Id": {
                    "Type": "String"
                }
            },
            "Resources": {
                "thing": {
                    "Type": "AWS::IoT::Thing",
                    "Properties": {
                        "AttributePayload": {},
                        "ThingGroups": {
                            "Ref": "ThingGroups"
                        },
                        "ThingName": {
                            "Ref": "ThingName"
                        },
                    },
                    "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": "Ansjer_Device_" + uid,
                      "ThingGroups": device_version,
                      "AWS::IoT::Certificate::Id": res['certificateId']}
        self.client.register_thing(
            templateBody=templateBody,
            parameters=parameters
        )
        return res, parameters


class EuropeIOTClient(IOTObject):

    def __init__(self):
        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)

    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': 'a2rqy12o004ad8-ats.iot.eu-west-1.amazonaws.com'
        }
        return res

    def create_keys_and_certificate(self, uid, device_version):
        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': 'a2rqy12o004ad8-ats.iot.eu-west-1.amazonaws.com'
        }
        # 根据证书ID注册物品和策略
        templateBody = {
            "Parameters": {
                "ThingName": {
                    "Type": "String"
                },
                "SerialNumber": {
                    "Type": "String"
                },
                "DeviceLocation": {
                    "Type": "String"
                },
                "AWS::IoT::Certificate::Id": {
                    "Type": "String"
                }
            },
            "Resources": {
                "thing": {
                    "Type": "AWS::IoT::Thing",
                    "Properties": {
                        "AttributePayload": {},
                        "ThingGroups": {
                            "Ref": "ThingGroups"
                        },
                        "ThingName": {
                            "Ref": "ThingName"
                        },
                    },
                    "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": "Ansjer_Device_" + uid,
                      "ThingGroups": device_version,
                      "AWS::IoT::Certificate::Id": res['certificateId']}
        self.client.register_thing(
            templateBody=templateBody,
            parameters=parameters
        )
        return res, parameters