Roy
90 lines
2.6 KiB
Python
90 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import paho.mqtt.client as mqtt
|
|
import time, random
|
|
import json
|
|
|
|
class MqttThread():
|
|
def __init__(self, callback_API, controller_id: str, url: str, port: int, client_id: str, name='thread-1', alive=60):
|
|
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 = f'{self.__controller_ID }_data_analysis/#'
|
|
print('subscribe topic', topic)
|
|
self._mqtt_client.subscribe(topic)
|
|
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
|
|
topic = f'{self.__controller_ID}/data_analysis'
|
|
message = json.dumps(payload)
|
|
print('publish', topic, message)
|
|
message_info = self._mqtt_client.publish(topic, message, QoS)
|
|
|
|
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):
|
|
print('on_message', message.topic, message.payload)
|
|
topic = message.topic.split('/')[1]
|
|
payload = None
|
|
|
|
try:
|
|
payload = message.payload.decode('utf-8')
|
|
# print('callback from', topic, message.topic, payload)
|
|
except:
|
|
print('error')
|
|
|
|
func = getattr(self._handler, topic, None)
|
|
if func != None:
|
|
return func(payload)
|