|
| 1 | +import time |
| 2 | +from json import loads, decoder |
| 3 | +from ssl import PROTOCOL_TLSv1_2 |
| 4 | +import paho.mqtt.client as mqtt |
| 5 | +from myDevices.utils.logger import debug, error, exception, info, logJson, warn |
| 6 | + |
| 7 | +# Topics |
| 8 | +DATA_TOPIC = 'data/json' |
| 9 | +COMMAND_TOPIC = 'cmd' |
| 10 | +COMMAND_JSON_TOPIC = 'cmd.json' |
| 11 | +COMMAND_RESPONSE_TOPIC = 'response' |
| 12 | + |
| 13 | +# Data Channels |
| 14 | +SYS_HARDWARE_MAKE = 'sys:hw:make' |
| 15 | +SYS_HARDWARE_MODEL = 'sys:hw:model' |
| 16 | +SYS_OS_NAME = 'sys:os:name' |
| 17 | +SYS_OS_VERSION = 'sys:os:version' |
| 18 | +SYS_NET = 'sys:net' |
| 19 | +SYS_STORAGE = 'sys:storage' |
| 20 | +SYS_RAM = 'sys:ram' |
| 21 | +SYS_CPU = 'sys:cpu' |
| 22 | +SYS_I2C = 'sys:i2c' |
| 23 | +SYS_SPI = 'sys:spi' |
| 24 | +SYS_UART = 'sys:uart' |
| 25 | +SYS_ONEWIRE = 'sys:1wire' |
| 26 | +SYS_DEVICETREE = 'sys:devicetree' |
| 27 | +SYS_GPIO = 'sys:gpio' |
| 28 | +SYS_POWER_RESET = 'sys:pwr:reset' |
| 29 | +SYS_POWER_HALT = 'sys:pwr:halt' |
| 30 | +AGENT_VERSION = 'agent:version' |
| 31 | +AGENT_DEVICES = 'agent:devices' |
| 32 | +AGENT_MANAGE = 'agent:manage' |
| 33 | +DEV_SENSOR = 'dev' |
| 34 | + |
| 35 | +# Channel Suffixes |
| 36 | +IP = 'ip' |
| 37 | +SPEEDTEST = 'speedtest' |
| 38 | +SSID = 'ssid' |
| 39 | +USAGE = 'usage' |
| 40 | +CAPACITY = 'capacity' |
| 41 | +LOAD = 'load' |
| 42 | +TEMPERATURE = 'temp' |
| 43 | +VALUE = 'value' |
| 44 | +FUNCTION = 'function' |
| 45 | + |
| 46 | + |
| 47 | +class DataChannel: |
| 48 | + @staticmethod |
| 49 | + def add(data_list, prefix, channel=None, suffix=None, value=None, type=None, unit=None, name=None): |
| 50 | + """Create data channel dict and append it to a list""" |
| 51 | + data_channel = prefix |
| 52 | + if channel is not None: |
| 53 | + data_channel += ':' + str(channel) |
| 54 | + if suffix is not None: |
| 55 | + data_channel += ';' + str(suffix) |
| 56 | + data = {} |
| 57 | + data['channel'] = data_channel |
| 58 | + data['value'] = value |
| 59 | + if type is not None: |
| 60 | + data['type'] = type |
| 61 | + if unit is not None: |
| 62 | + data['unit'] = unit |
| 63 | + if name is not None: |
| 64 | + data['name'] = name |
| 65 | + data_list.append(data) |
| 66 | + |
| 67 | + |
| 68 | +class CayenneMQTTClient: |
| 69 | + """Cayenne MQTT Client class. |
| 70 | + |
| 71 | + This is the main client class for connecting to Cayenne and sending and recFUeiving data. |
| 72 | + |
| 73 | + Standard usage: |
| 74 | + * Set on_message callback, if you are receiving data. |
| 75 | + * Connect to Cayenne using the begin() function. |
| 76 | + * Call loop() at intervals (or loop_forever() once) to perform message processing. |
| 77 | + * Send data to Cayenne using write functions: virtualWrite(), celsiusWrite(), etc. |
| 78 | + * Receive and process data from Cayenne in the on_message callback. |
| 79 | +
|
| 80 | + The on_message callback can be used by creating a function and assigning it to CayenneMQTTClient.on_message member. |
| 81 | + The callback function should have the following signature: on_message(topic, message) |
| 82 | + If it exists this callback is used as the default message handler. |
| 83 | + """ |
| 84 | + client = None |
| 85 | + root_topic = "" |
| 86 | + connected = False |
| 87 | + on_message = None |
| 88 | + |
| 89 | + def begin(self, username, password, clientid, hostname='mqtt.mydevices.com', port=8883): |
| 90 | + """Initializes the client and connects to Cayenne. |
| 91 | + |
| 92 | + username is the Cayenne username. |
| 93 | + password is the Cayenne password. |
| 94 | + clientid is the Cayennne client ID for the device. |
| 95 | + hostname is the MQTT broker hostname. |
| 96 | + port is the MQTT broker port. |
| 97 | + """ |
| 98 | + self.root_topic = 'v1/{}/things/{}'.format(username, clientid) |
| 99 | + self.client = mqtt.Client(client_id=clientid, clean_session=True, userdata=self) |
| 100 | + self.client.on_connect = self.connect_callback |
| 101 | + self.client.on_disconnect = self.disconnect_callback |
| 102 | + self.client.on_message = self.message_callback |
| 103 | + self.client.username_pw_set(username, password) |
| 104 | + if port != 1883: |
| 105 | + self.client.tls_set(ca_certs='/etc/ssl/certs/ca-certificates.crt', tls_version=PROTOCOL_TLSv1_2) |
| 106 | + self.client.connect(hostname, port, 60) |
| 107 | + info('Connecting to {}:{}'.format(hostname, port)) |
| 108 | + |
| 109 | + def connect_callback(self, client, userdata, flags, rc): |
| 110 | + """The callback for when the client connects to the server. |
| 111 | +
|
| 112 | + client is the client instance for this callback. |
| 113 | + userdata is the private user data as set in Client() or userdata_set(). |
| 114 | + flags are the response flags sent by the broker. |
| 115 | + rc is the connection result. |
| 116 | + """ |
| 117 | + if rc != 0: |
| 118 | + # MQTT broker error codes |
| 119 | + broker_errors = { |
| 120 | + 1 : 'unacceptable protocol version', |
| 121 | + 2 : 'identifier rejected', |
| 122 | + 3 : 'server unavailable', |
| 123 | + 4 : 'bad user name or password', |
| 124 | + 5 : 'not authorized', |
| 125 | + } |
| 126 | + raise Exception("Connection failed, " + broker_errors.get(rc, "result code " + str(rc))) |
| 127 | + else: |
| 128 | + info("Connected with result code "+str(rc)) |
| 129 | + self.connected = True |
| 130 | + # Subscribing in on_connect() means that if we lose the connection and |
| 131 | + # reconnect then subscriptions will be renewed. |
| 132 | + client.subscribe(self.get_topic_string(COMMAND_TOPIC, True)) |
| 133 | + client.subscribe(self.get_topic_string(COMMAND_JSON_TOPIC, False)) |
| 134 | + |
| 135 | + def disconnect_callback(self, client, userdata, rc): |
| 136 | + """The callback for when the client disconnects from the server. |
| 137 | +
|
| 138 | + client is the client instance for this callback. |
| 139 | + userdata is the private user data as set in Client() or userdata_set(). |
| 140 | + rc is the connection result. |
| 141 | + """ |
| 142 | + info("Disconnected with result code "+str(rc)) |
| 143 | + self.connected = False |
| 144 | + reconnected = False |
| 145 | + while not reconnected: |
| 146 | + try: |
| 147 | + self.client.reconnect() |
| 148 | + reconnected = True |
| 149 | + except: |
| 150 | + print("Reconnect failed, retrying") |
| 151 | + time.sleep(5) |
| 152 | + |
| 153 | + def message_callback(self, client, userdata, msg): |
| 154 | + """The callback for when a message is received from the server. |
| 155 | +
|
| 156 | + client is the client instance for this callback. |
| 157 | + userdata is the private user data as set in Client() or userdata_set(). |
| 158 | + msg is the received message. |
| 159 | + """ |
| 160 | + try: |
| 161 | + message = {} |
| 162 | + if msg.topic[-len(COMMAND_JSON_TOPIC):] == COMMAND_JSON_TOPIC: |
| 163 | + payload = loads(msg.payload.decode()) |
| 164 | + message['payload'] = payload['value'] |
| 165 | + message['cmdId'] = payload['cmdId'] |
| 166 | + channel = payload['channel'].split('/')[-1].split(';') |
| 167 | + else: |
| 168 | + payload = msg.payload.decode().split(',') |
| 169 | + if len(payload) > 1: |
| 170 | + message['cmdId'] = payload[0] |
| 171 | + message['payload'] = payload[1] |
| 172 | + else: |
| 173 | + message['payload'] = payload[0] |
| 174 | + channel = msg.topic.split('/')[-1].split(';') |
| 175 | + message['channel'] = channel[0] |
| 176 | + if len(channel) > 1: |
| 177 | + message['suffix'] = channel[1] |
| 178 | + debug('message_callback: {}'.format(message)) |
| 179 | + if self.on_message: |
| 180 | + self.on_message(message) |
| 181 | + except: |
| 182 | + exception('Error processing message: {} {}'.format(msg.topic, str(msg.payload))) |
| 183 | + |
| 184 | + def get_topic_string(self, topic, append_wildcard=False): |
| 185 | + """Return a topic string. |
| 186 | + |
| 187 | + topic: the topic substring |
| 188 | + append_wildcard: if True append the single level topics wildcard (+)""" |
| 189 | + if append_wildcard: |
| 190 | + return '{}/{}/+'.format(self.root_topic, topic) |
| 191 | + else: |
| 192 | + return '{}/{}'.format(self.root_topic, topic) |
| 193 | + |
| 194 | + def disconnect(self): |
| 195 | + """Disconnect from Cayenne. |
| 196 | + """ |
| 197 | + self.client.disconnect() |
| 198 | + |
| 199 | + def loop(self, timeout=1.0): |
| 200 | + """Process Cayenne messages. |
| 201 | + |
| 202 | + This should be called regularly to ensure Cayenne messages are sent and received. |
| 203 | + |
| 204 | + timeout: The time in seconds to wait for incoming/outgoing network |
| 205 | + traffic before timing out and returning. |
| 206 | + """ |
| 207 | + self.client.loop(timeout) |
| 208 | + |
| 209 | + def loop_start(self): |
| 210 | + """This is part of the threaded client interface. Call this once to |
| 211 | + start a new thread to process network traffic. This provides an |
| 212 | + alternative to repeatedly calling loop() yourself. |
| 213 | + """ |
| 214 | + self.client.loop_start() |
| 215 | + |
| 216 | + def loop_stop(self): |
| 217 | + """This is part of the threaded client interface. Call this once to |
| 218 | + stop the network thread previously created with loop_start(). This call |
| 219 | + will block until the network thread finishes. |
| 220 | + """ |
| 221 | + self.client.loop_stop() |
| 222 | + |
| 223 | + def publish_packet(self, topic, packet, qos=0, retain=False): |
| 224 | + """Publish a packet. |
| 225 | + |
| 226 | + topic: topic substring. |
| 227 | + packet: JSON packet to publish. |
| 228 | + qos: quality of service level to use. |
| 229 | + retain: if True, the message will be set as the "last known good"/retained message for the topic. |
| 230 | + """ |
| 231 | + debug('Publish to {}'.format(self.get_topic_string(topic))) |
| 232 | + self.client.publish(self.get_topic_string(topic), packet, qos, retain) |
| 233 | + |
| 234 | + def publish_response(self, msg_id, error_message=None): |
| 235 | + """Send a command response to Cayenne. |
| 236 | + |
| 237 | + This should be sent when a command message has been received. |
| 238 | + msg_id is the ID of the message received. |
| 239 | + error_message is the error message to send. This should be set to None if there is no error. |
| 240 | + """ |
| 241 | + topic = self.get_topic_string(COMMAND_RESPONSE_TOPIC) |
| 242 | + if error_message: |
| 243 | + payload = "error,%s=%s" % (msg_id, error_message) |
| 244 | + else: |
| 245 | + payload = "ok,%s" % (msg_id) |
| 246 | + self.client.publish(topic, payload) |
0 commit comments