133 lines
4.1 KiB
Python
133 lines
4.1 KiB
Python
# NOTE: does not work, head exploded!
|
|
import json
|
|
import logging
|
|
import random
|
|
import time
|
|
from paho.mqtt.client import Client
|
|
from paho.mqtt.client import CallbackAPIVersion
|
|
|
|
BROKER = 'mqtt.meezenest.nl'
|
|
PORT = 1883
|
|
TOPIC = "python-mqtt/tcp"
|
|
# generate client ID with pub prefix randomly
|
|
CLIENT_ID = f'python-mqtt-tcp-pub-sub-{random.randint(0, 1000)}'
|
|
USERNAME = 'emqx'
|
|
PASSWORD = 'public'
|
|
|
|
FIRST_RECONNECT_DELAY = 1
|
|
RECONNECT_RATE = 2
|
|
MAX_RECONNECT_COUNT = 12
|
|
MAX_RECONNECT_DELAY = 60
|
|
|
|
FLAG_EXIT = False
|
|
|
|
class MqttController(object):
|
|
|
|
def __init__(self, config: dict, message_processor=None):
|
|
self.config = config
|
|
|
|
self.client = Client(
|
|
client_id=self.config['mqtt_client'],
|
|
clean_session=self.config['mqtt_clean_session'],
|
|
userdata={"client": self.config['mqtt_client']},
|
|
)
|
|
|
|
self.client.username_pw_set(self.config['mqtt_username'], self.config['mqtt_password'])
|
|
|
|
if self.config['mqtt_debug']:
|
|
self.client.on_log = self._on_log
|
|
|
|
self.client.on_connect = self._on_connect
|
|
#self.client.on_subscribe = self._on_subscribe
|
|
self.client.on_message = self._on_message
|
|
#self.client.on_publish = self._on_publish
|
|
self.client.on_disconnect = self._on_disconnect
|
|
|
|
self.client.connect(self.config['mqtt_host'], self.config['mqtt_port'], 60)
|
|
|
|
if message_processor:
|
|
self.message_processor = message_processor
|
|
|
|
def _on_connect(self, client, userdata, flags, rc, properties):
|
|
if rc == 0 and self.client.is_connected():
|
|
print("Connected to MQTT Broker!")
|
|
self.client.subscribe(TOPIC)
|
|
else:
|
|
print(f'Failed to connect, return code {rc}')
|
|
|
|
|
|
def _on_disconnect(self, client, userdata, rc):
|
|
logging.info("Disconnected with result code: %s", rc)
|
|
reconnect_count, reconnect_delay = 0, FIRST_RECONNECT_DELAY
|
|
while reconnect_count < MAX_RECONNECT_COUNT:
|
|
logging.info("Reconnecting in %d seconds...", reconnect_delay)
|
|
time.sleep(reconnect_delay)
|
|
|
|
try:
|
|
self.client.reconnect()
|
|
logging.info("Reconnected successfully!")
|
|
return
|
|
except Exception as err:
|
|
logging.error("%s. Reconnect failed. Retrying...", err)
|
|
|
|
reconnect_delay *= RECONNECT_RATE
|
|
reconnect_delay = min(reconnect_delay, MAX_RECONNECT_DELAY)
|
|
reconnect_count += 1
|
|
logging.info("Reconnect failed after %s attempts. Exiting...", reconnect_count)
|
|
global FLAG_EXIT
|
|
FLAG_EXIT = True
|
|
|
|
def _on_message(self, client, userdata, msg):
|
|
print(f'Received `{msg.payload.decode()}` from `{msg.topic}` topic')
|
|
|
|
def loop_start(self):
|
|
self.client.loop_start()
|
|
|
|
def publish(self):
|
|
msg_count = 0
|
|
while not FLAG_EXIT:
|
|
msg_dict = {
|
|
'msg': msg_count
|
|
}
|
|
msg = json.dumps(msg_dict)
|
|
if not self.client.is_connected():
|
|
logging.error("publish: MQTT client is not connected!")
|
|
time.sleep(1)
|
|
continue
|
|
result = self.client.publish(TOPIC, msg)
|
|
# result: [0, 1]
|
|
status = result[0]
|
|
if status == 0:
|
|
print(f'Send `{msg}` to topic `{TOPIC}`')
|
|
else:
|
|
print(f'Failed to send message to topic {TOPIC}')
|
|
msg_count += 1
|
|
time.sleep(1)
|
|
|
|
|
|
def run():
|
|
mqtt_config = {
|
|
"mqtt_client": "MeesElecronics",
|
|
"mqtt_clean_session": "",
|
|
"mqtt_username": "1964",
|
|
"mqtt_password": "NoPasswd",
|
|
"mqtt_debug": False,
|
|
"mqtt_host": "mqtt.meezenest.nl",
|
|
"mqtt_port": 1883
|
|
}
|
|
print(mqtt_config)
|
|
logging.basicConfig(format='%(asctime)s - %(levelname)s: %(message)s',
|
|
level=logging.DEBUG)
|
|
client = MqttController(mqtt_config)
|
|
client.loop_start()
|
|
time.sleep(1)
|
|
client.publish()
|
|
#if client.is_connected():
|
|
# publish()
|
|
#else:
|
|
# client.loop_stop()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
run()
|