1. import ujson,utime,_thread,sys_bus
    2. from umqtt import MQTTClient
    3. from usr.modules.logging import getLogger
    4. from usr.modules.common import numiter, option_lock, CloudObservable,UtilsDataEncryption,TimeResolver
    5. from usr.modules.common import Singleton
    7. log = getLogger("TimerService")
    9. class CloudIOT(CloudObservable):
    10.     """This is a class for universal mqtt iot.
    12.     This class extend CloudObservable.
    14.     This class has the following functions:
    15.         1. Cloud connect and disconnect
    16.         2. Publish data to cloud
    17.         3. Subscribe data from cloud
    19.     Attribute:
    20.         pub_topic_dict: topic dict for publish dtu through data
    21.         sub_topic_dict: topic dict for subscribe cloud through data
    22.         conn_type:cloud name
    24.     Run step:
    25.         1. cloud = MqttIot(server, port, clean_session, client_id, pub_topic, sub_topic)
    26.         2. cloud.addObserver(RemoteSubscribe)
    27.         3. cloud.init()
    28.         4. cloud.post_data(data)
    29.         5. cloud.close()
    30.     """
    31.     def __init__(self, client_id,settings):
    32.         """
    33.         1. Init parent class CloudObservable
    34.         2. Init cloud connect params and topic
    35.         """
    36.         super().__init__()
    37.         self.conn_type = "mqtt"
    38.         self.__settings = settings
    40.         cloudConfig = self.__settings.current_settings.get("cloud")
    42.         self.__productCode = self.__settings.current_settings.get("PRODUCT_CODE")
    43.         self.__server = cloudConfig.get("host",None)
    44.         self.__port = cloudConfig.get("port",1883)
    45.         self.__mqtt = None
    46.         self.__clean_session = cloudConfig.get("cleanSession",False)
    47.         self.__life_time = cloudConfig.get("lifeTime",120)
    48.         self.__client_id = client_id
    49.         self.__key = cloudConfig.get("userName",None)
    50.         self.__secret = cloudConfig.get("secret",None)
    51.         self.__pubTopic = cloudConfig.get("pubTopic").format(self.__productCode ,client_id)
    52.         self.__subTopic = cloudConfig.get("subTopic").format(self.__productCode ,client_id)
    53.         self.__qos = cloudConfig.get("qos",0)
    56.         ts = utime.mktime(utime.localtime())+(utime.getTimeZone())*60*60
    57.         #cloud userName
    58.         self.__userName = "{}_{}_{}_{}".format(client_id,self.__productCode,self.__key,ts)
    59.         self.__password = UtilsDataEncryption.md5_32_U("{}_{}".format(self.__userName,self.__secret))
    61.     def init(self, enforce=False):
    62.         """mqtt connect and subscribe topic
    64.         Parameter:
    65.             enforce:
    66.                 True: enfore cloud connect and subscribe topic
    67.                 False: check connect status, return True if cloud connected
    69.         Return:
    70.             Ture: Success
    71.             False: Failed
    72.         """
    73.         log.debug("[init start] enforce: %s" % enforce)
    74.         if enforce is False and self.__mqtt is not None:
    75.             log.debug("self.get_status(): %s" % self.get_status())
    76.             if self.get_status():
    77.                 return True
    79.         if self.__mqtt is not None:
    80.             self.close()
    82.         log.debug("mqtt init. self.__client_id: %s, self.__userName: %s, self.__password: %s" % (self.__client_id, self.__userName, self.__password))
    83.         self.__mqtt = MQTTClient(client_id=self.__client_id, server=self.__server, port=self.__port,
    84.                               user=self.__userName, password=self.__password, keepalive=self.__life_time, ssl=False)
    85.         try:
    86.             self.__mqtt.connect(clean_session=self.__clean_session)
    87.         except Exception as e:
    88.             log.error("mqtt connect error: %s" % e)
    89.         else:
    90.             self.__mqtt.set_callback(self.__sub_cb)
    91.             self.__subscribe_topic()
    92.             log.debug("========mqtt n_subscribe_topic")
    93.             self.__start_listen()
    94.             log.debug("mqtt start.")
    96.         log.debug("self.get_status(): %s" % self.get_status())
    97.         if self.get_status():
    98.             return True
    99.         else:
    100.             return False
    102.     def __subscribe_topic(self):
    103.         if self.__mqtt.subscribe(self.__subTopic, qos=0) == -1:
    104.             log.error("Topic [%s] Subscribe Falied." % self.__subTopic)
    106.     def __sub_cb(self, topic, data):
    107.         """mqtt subscribe topic callback
    109.         Parameter:
    110.             topic: topic info
    111.             data: response dictionary info
    112.         """
    113.         topic = topic.decode()
    114.         try:
    115.             log.debug("Cloud收到数据：{}\r\n{}".format(topic,data))
    116.             data = ujson.loads(data)
    117.         except:
    118.             pass
    120.         try:
    121.             sys_bus.publish("CloudDataProcess",data)
    122.         except Exception as e:
    123.             log.error("{}".format(e))
    125.     def __listen(self):
    126.         while True:
    127.             self.__mqtt.wait_msg()
    128.             utime.sleep_ms(100)
    130.     def __start_listen(self):
    131.         """Start a new thread to listen to the cloud publish 
    132.         """
    133.         _thread.start_new_thread(self.__listen, ())
    135.     def get_status(self):
    136.         """Get mqtt connect status
    138.         Return:
    139.             True -- connect success
    140.             False -- connect falied
    141.         """
    142.         try:
    143.             return True if self.__mqtt.get_mqttsta() == 0 else False
    144.         except:
    145.             return False
    147.     def close(self):
    148.         self.__mqtt.disconnect()
    150.     def post_data(self, data):
    151.         try:
    152.             log.info("up,{},{}".format(self.__pubTopic,data))
    153.             self.__mqtt.publish(self.__pubTopic, data, self.__qos)
    154.         except Exception:
    155.             log.error("mqtt publish topic %s failed. data: %s" % (self.__pubTopic, data))
    156.             return False
    157.         else:
    158.             return True
    160.     def post_data2(self, serviceName,jsonData):
    161.         return self.post_data(self.UpdataPack(serviceName,jsonData))
    163.     def ota_request(self):
    164.         pass
    166.     def ota_action(self, action, module=None):
    167.         pass
    169.     def device_report(self):
    170.         pass
    172.     def UpdataPack(self,serviceName,jsonData):
    173.         msg = configCloudiot.current_settings.get("Cloud")
    174.         msg["service_id"] = serviceName
    175.         msg["data"] = jsonData
    176.         msg["event_time"] = TimeResolver.getUtcTime()
    177.         return "[{}]".format(msg)
    179.     def UploadData(self):
    180.         self.UploadData_GatewayBasicInfo()
    182.     def UploadData_GatewayBasicInfo(self):
    183.         server = ModelGateway()
    184.         er = server.current_settings.get("Gateway_BasicInfo")
    185.         er["currentDateTime"] = TimeResolver.getUtcTime()
    186.         return self.post_data2(server.current_settings.get("cloud_service_id"),er)
    188. class CloudOTA(object):
    189.     def __init__(self) -> None:
    190.         pass