peterlu14
84 lines
2.6 KiB
Python
84 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import paho.mqtt.client as mqtt
|
|
import time, random, threading
|
|
import multiprocessing as mp
|
|
import json
|
|
|
|
class MqttThread(threading.Thread):
|
|
def __init__(self, callback_API, controller_id: str, url: str, port: int, client_id: str, name='thread-1', alive=60):
|
|
super(MqttThread, self).__init__(name = name)
|
|
|
|
self.__controller_ID = controller_id
|
|
|
|
self._mqtt_url = url
|
|
self._mqtt_port = port
|
|
self._mqtt_alive = alive
|
|
self._client_id = client_id
|
|
self._mqtt_client = None
|
|
self._connected = False
|
|
|
|
self._handler = callback_API
|
|
|
|
@property
|
|
def mqtt_url(self) -> str:
|
|
return self._mqtt_url
|
|
|
|
@property
|
|
def mqtt_port(self) -> int:
|
|
return self._mqtt_port
|
|
|
|
@property
|
|
def connected(self) -> bool:
|
|
return self._connected
|
|
|
|
def run(self) -> None:
|
|
print('use controller ID', self.__controller_ID)
|
|
|
|
self._mqtt_client = mqtt.Client(self.__controller_ID + '_' + self._client_id)
|
|
self._mqtt_client.connect(self._mqtt_url, self._mqtt_port)
|
|
self._mqtt_client.on_connect = self.on_connect
|
|
self._mqtt_client.on_disconnect = self.on_disconnect
|
|
self._mqtt_client.on_message = self.on_message
|
|
|
|
self.routine()
|
|
|
|
def shutdown(self) -> None:
|
|
if self._mqtt_client != None:
|
|
self._mqtt_client.loop_stop()
|
|
self._mqtt_client.disconnect()
|
|
|
|
def routine(self) -> None:
|
|
self._mqtt_client.loop_start()
|
|
|
|
def on_connect(self, client: mqtt.Client, user_data, flags, rc: int):
|
|
topic = self.__controller_ID + '_user'
|
|
# print('cloud-server subscribe', topic)
|
|
# self._mqtt_client.subscribe(topic)
|
|
self._mqtt_client.subscribe(self.__controller_ID + '/+/+')
|
|
self._connected = True
|
|
|
|
def on_disconnect(self, client: mqtt.Client, user_data, rc):
|
|
print('cloud-server disconnected')
|
|
time.sleep(3)
|
|
self._connected = False
|
|
self._mqtt_client.reconnect()
|
|
|
|
def publish(self, payload: str, wait_for_ack = False) -> int:
|
|
QoS = 2 if wait_for_ack else 0
|
|
message_info = self._mqtt_client.publish(self.__controller_ID + '_user', json.dumps(payload), QoS)
|
|
print(self.__controller_ID + '_user', payload)
|
|
if wait_for_ack:
|
|
print(" > awaiting ACK for {}".format(message_info.mid))
|
|
message_info.wait_for_publish()
|
|
print(" < received ACK for {}".format(message_info.mid))
|
|
|
|
def on_message(self, client: mqtt.Client, user_data, message):
|
|
topic = message.topic.split('/')[1]
|
|
payload = json.loads(message.payload.decode('utf-8'))
|
|
|
|
func = getattr(self._handler, topic, None)
|
|
if func != None:
|
|
return func(payload)
|