1. import ujson
    2. import utime
    3. import _thread
    4. from umqtt import MQTTClient
    5. from usr.modules.logging import getLogger
    6. from usr.modules.common import CloudObservable
    7. log = getLogger(__name__)
    9. class MqttIot(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, qos, 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, server, qos, port, clean_session, client_id, pub_topic=None, sub_topic=None, life_time=120):
    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.__pk = None
    39.         self.__ps = None
    40.         self.__dk = None
    41.         self.__ds = None
    42.         self.__server = server
    43.         self.__qos = qos
    44.         self.__port = port
    45.         self.__mqtt = None
    46.         self.__clean_session = clean_session
    47.         self.__life_time = life_time
    48.         self.__client_id = client_id
    49.         self.__password = None
    51.         if pub_topic == None:
    52.             self.pub_topic_dict = {"0": "/python/mqtt/pub"}
    53.         else:
    54.             self.pub_topic_dict = pub_topic
    55.         if sub_topic == None:
    56.             self.sub_topic_dict = {"0": "/python/mqtt/sub"}
    57.         else:
    58.             self.sub_topic_dict = sub_topic
    60.     def __subscribe_topic(self):
    61.         for id, usr_sub_topic in self.sub_topic_dict.items():
    62.             if self.__mqtt.subscribe(usr_sub_topic, qos=0) == -1:
    63.                 log.error("Topic [%s] Subscribe Falied." % usr_sub_topic)
    66.     def __sub_cb(self, topic, data):
    67.         """mqtt subscribe topic callback
    69.         Parameter:
    70.             topic: topic info
    71.             data: response dictionary info
    72.         """
    73.         topic = topic.decode()
    74.         try:
    75.             data = ujson.loads(data)
    76.         except:
    77.             pass
    79.         try:
    80.             self.notifyObservers(self, *("raw_data", {"topic":topic, "data":data} ) )
    81.         except Exception as e:
    82.             log.error("{}".format(e))
    84.     def __listen(self):
    85.         while True:
    86.             self.__mqtt.wait_msg()
    87.             utime.sleep_ms(100)
    89.     def __start_listen(self):
    90.         """Start a new thread to listen to the cloud publish 
    91.         """
    92.         _thread.start_new_thread(self.__listen, ())
    94.     def init(self, enforce=False):
    95.         """mqtt connect and subscribe topic
    97.         Parameter:
    98.             enforce:
    99.                 True: enfore cloud connect and subscribe topic
    100.                 False: check connect status, return True if cloud connected
    102.         Return:
    103.             Ture: Success
    104.             False: Failed
    105.         """
    106.         log.debug("[init start] enforce: %s" % enforce)
    107.         if enforce is False and self.__mqtt is not None:
    108.             log.debug("self.get_status(): %s" % self.get_status())
    109.             if self.get_status():
    110.                 return True
    112.         if self.__mqtt is not None:
    113.             self.close()
    115.         log.debug("mqtt init. self.__client_id: %s, self.__password: %s, self.__dk: %s, self.__ds: %s" % (self.__client_id, self.__password, self.__dk, self.__ds))
    116.         self.__mqtt = MQTTClient(client_id=self.__client_id, server=self.__server, port=self.__port,
    117.                               user=self.__dk, password=self.__password, keepalive=self.__life_time, ssl=False)
    118.         try:
    119.             self.__mqtt.connect(clean_session=self.__clean_session)
    120.         except Exception as e:
    121.             log.error("mqtt connect error: %s" % e)
    122.         else:
    123.             self.__mqtt.set_callback(self.__sub_cb)
    124.             self.__subscribe_topic()
    125.             log.debug("mqtt n_subscribe_topic")
    126.             self.__start_listen()
    127.             log.debug("mqtt start.")
    129.         log.debug("self.get_status(): %s" % self.get_status())
    130.         if self.get_status():
    131.             return True
    132.         else:
    133.             return False
    135.     def close(self):
    136.         self.__mqtt.disconnect()
    138.     def get_status(self):
    139.         """Get mqtt connect status
    141.         Return:
    142.             True -- connect success
    143.             False -- connect falied
    144.         """
    145.         try:
    146.             return True if self.__mqtt.get_mqttsta() == 0 else False
    147.         except:
    148.             return False
    150.     def through_post_data(self, data, topic_id):
    151.         try:
    152.             self.__mqtt.publish(self.pub_topic_dict[topic_id], data, self.__qos)
    153.         except Exception:
    154.             log.error("mqtt publish topic %s failed. data: %s" % (self.pub_topic_dict[topic_id], data))
    155.             return False
    156.         else:
    157.             return True
    159.     def post_data(self, data):
    160.         pass
    162.     def ota_request(self):
    163.         pass
    165.     def ota_action(self, action, module=None):
    166.         pass
    168.     def device_report(self):
    169.         pass