# The MIT License (MIT) # # Copyright (c) 2019 Brent Rubell for Adafruit Industries # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in # all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN # THE SOFTWARE. """ `adafruit_aws_iot` ================================================================================ Amazon AWS IoT MQTT Client for CircuitPython * Author(s): Brent Rubell Implementation Notes -------------------- **Hardware:** **Software and Dependencies:** * Adafruit CircuitPython firmware for the supported boards: https://github.com/adafruit/circuitpython/releases """ import json from adafruit_minimqtt import MMQTTException __version__ = "0.0.0-auto.0" __repo__ = "https://github.com/adafruit/Adafruit_CircuitPython_AWS_IOT.git" class AWS_IOT_ERROR(Exception): """Exception raised on MQTT API return-code errors.""" # pylint: disable=unnecessary-pass pass class MQTT_CLIENT: """Client for interacting with Amazon AWS IoT MQTT API. :param MiniMQTT mmqttclient: Pre-configured MiniMQTT Client object. :param int keep_alive: Optional Keep-alive timer interval, in seconds. Provided interval must be 30 <= keep_alive <= 1200. """ def __init__(self, mmqttclient, keep_alive=30): if "MQTT" in str(type(mmqttclient)): self.client = mmqttclient else: raise TypeError( "This class requires a preconfigured MiniMQTT object, \ please create one." ) # Verify MiniMQTT client object configuration try: self.cid = self.client.client_id assert ( self.cid[0] != "$" ), "Client ID can not start with restricted client ID prefix $." except: raise TypeError( "You must provide MiniMQTT with your AWS IoT Device's Identifier \ as the Client ID." ) # Shadow-interaction topic self.shadow_topic = "$aws/things/{}/shadow".format(self.cid) # keep_alive timer must be between 30 <= keep alive interval <= 1200 seconds # https://docs.aws.amazon.com/general/latest/gr/aws_service_limits.html assert ( 30 <= keep_alive <= 1200 ), "Keep_Alive timer \ interval must be between 30 and 1200 seconds" self.keep_alive = keep_alive # User-defined MQTT callback methods must be init'd to None self.on_connect = None self.on_disconnect = None self.on_message = None self.on_subscribe = None self.on_unsubscribe = None # Connect MiniMQTT callback handlers self.client.on_connect = self._on_connect_mqtt self.client.on_disconnect = self._on_disconnect_mqtt self.client.on_message = self._on_message_mqtt self.client.on_subscribe = self._on_subscribe_mqtt self.client.on_unsubscribe = self._on_unsubscribe_mqtt self.connected_to_aws = False def __enter__(self): return self def __exit__(self, exception_type, exception_value, traceback): self.disconnect() @property def is_connected(self): """Returns if MQTT_CLIENT is connected to AWS IoT MQTT Broker """ return self.connected_to_aws def disconnect(self): """Disconnects from Amazon AWS IoT MQTT Broker and de-initializes the MiniMQTT Client. """ try: self.client.disconnect() except MMQTTException as error: raise AWS_IOT_ERROR("Error disconnecting with AWS IoT: ", error) self.connected_to_aws = False # Reset user-defined callback methods to None self.on_connect = None self.on_disconnect = None self.on_message = None self.on_subscribe = None self.on_unsubscribe = None self.client.deinit() def reconnect(self): """Reconnects to the AWS IoT MQTT Broker """ try: self.client.reconnect() except MMQTTException as error: raise AWS_IOT_ERROR("Error re-connecting to AWS IoT:", error) def connect(self, clean_session=True): """Connects to Amazon AWS IoT MQTT Broker with Client ID. :param bool clean_session: Establishes a clean session with AWS broker. """ try: self.client.connect(clean_session) except MMQTTException as error: raise AWS_IOT_ERROR("Error connecting to AWS IoT: ", error) self.connected_to_aws = True # MiniMQTT Callback Handlers # pylint: disable=not-callable, unused-argument def _on_connect_mqtt(self, client, userdata, flag, ret_code): """Runs when code calls on_connect. :param MiniMQTT client: MiniMQTT client object. :param str user_data: User data from broker :param int flag: QoS flag from broker. :param int ret_code: Return code from broker. """ self.connected_to_aws = True # Call the on_connect callback if defined in code if self.on_connect is not None: self.on_connect(self, userdata, flag, ret_code) # pylint: disable=not-callable, unused-argument def _on_disconnect_mqtt(self, client, userdata, flag, ret_code): """Runs when code calls on_disconnect. :param MiniMQTT client: MiniMQTT client object. :param str user_data: User data from broker :param int flag: QoS flag from broker. :param int ret_code: Return code from broker. """ self.connected_to_aws = False # Call the on_connect callback if defined in code if self.on_connect is not None: self.on_connect(self, userdata, flag, ret_code) # pylint: disable=not-callable def _on_message_mqtt(self, client, topic, payload): """Runs when the client calls on_message. :param MiniMQTT client: MiniMQTT client object. :param str topic: MQTT broker topic. :param str payload: Payload returned by MQTT broker topic """ if self.on_message is not None: self.on_message(self, topic, payload) # pylint: disable=not-callable def _on_subscribe_mqtt(self, client, user_data, topic, qos): """Runs when the client calls on_subscribe. :param MiniMQTT client: MiniMQTT client object. :param str user_data: User data from broker :param str topic: Desired MQTT topic. param int qos: Quality of service level for topic, from broker. """ if self.on_subscribe is not None: self.on_subscribe(self, user_data, topic, qos) # pylint: disable=not-callable def _on_unsubscribe_mqtt(self, client, user_data, topic, pid): """Runs when the client calls on_unsubscribe. """ if self.on_unsubscribe is not None: self.on_unsubscribe(self, user_data, topic, pid) # MiniMQTT Network Control Flow def loop(self): """ Starts a synchronous message loop which maintains connection with AWS IoT. Must be called within the keep_alive timeout specified to init. This method does not handle network connection/disconnection. Example of "pumping" an AWS IoT message loop: ..code-block::python while True: aws_iot.loop() """ if self.connected_to_aws: self.client.loop() def loop_forever(self): """Begins a blocking, asynchronous message loop. This method handles network connection/disconnection. """ if self.connected_to_aws: self.client.loop_forever() @staticmethod def validate_topic(topic): """Validates if user-provided pub/sub topics adhere to AWS Service Limits. https://docs.aws.amazon.com/general/latest/gr/aws_service_limits.html :param str topic: Desired topic to validate """ assert hasattr(topic, "split"), "Topic must be a string" assert len(topic) < 256, "Topic must be less than 256 bytes!" assert len(topic.split("/")) <= 9, "Topics are limited to 7 forward slashes." # MiniMQTT Pub/Sub Methods, for usage with AWS IoT def subscribe(self, topic, qos=1): """Subscribes to an AWS IoT Topic. :param str topic: MQTT topic to subscribe to. :param int qos: Desired topic subscription's quality-of-service. """ assert qos < 2, "AWS IoT does not support subscribing with QoS 2." self.validate_topic(topic) self.client.subscribe(topic, qos) def publish(self, topic, payload, qos=1): """Publishes to a AWS IoT Topic. :param str topic: MQTT topic to publish to. :param str payload: Data to publish to topic. :param int payload: Data to publish to topic. :param float payload: Data to publish to topic. :param json payload: JSON-formatted data to publish to topic. :param int qos: Quality of service level for publishing. """ assert qos < 2, "AWS IoT does not support publishing with QoS 2." self.validate_topic(topic) if isinstance(payload, int or float): payload = str(payload) self.client.publish(topic, payload, qos=qos) # AWS IoT Device Shadow Service def shadow_get_subscribe(self, qos=1): """Subscribes to device's shadow get response. :param int qos: Optional quality of service level. """ self.client.subscribe(self.shadow_topic + "/get/#", qos) def shadow_subscribe(self, qos=1): """Subscribes to all notifications on the device's shadow update topic. :param int qos: Optional quality of service level. """ self.client.subscribe(self.shadow_topic + "/update/#", qos) def shadow_update(self, document): """Publishes a request state document to update the device's shadow. :param json state_document: JSON-formatted state document. """ self.client.publish(self.shadow_topic + "/update", document) def shadow_get(self): """Publishes an empty message to shadow get topic to get the device's shadow. """ self.client.publish( self.shadow_topic + "/get", json.dumps({"message": "ignore"}) ) def shadow_delete(self): """Publishes an empty message to the shadow delete topic to delete a device's shadow """ self.client.publish( self.shadow_topic + "/delete", json.dumps({"message": "delete"}) )