initial

.gitignore

@@ -0,0 +1 @@
+__pycache__/

config.ini

@@ -0,0 +1,6 @@
+[mqtt]
+broker = 172.16.2.16
+
+[master]
+topic = test/master
+dataObjectName = master

src/AbstractMqttHandler.py

@@ -0,0 +1,60 @@
+import paho.mqtt.client as mqtt
+import threading
+from loguru import logger
+
+
+def mqttOnConnectCallback(client, userdata, flags, rc):
+    userdata.onConnect()
+
+def mqttOnMessageCallback(client, userdata, message):
+    userdata.onMessage(message.topic, message.payload)
+
+def mqttOnDisconnectCallback(client, userdata, rc):
+    userdata.onDisconnect(rc)
+
+
+class AbstractMqttHandler(threading.Thread):
+    def __init__(self, config):
+        super().__init__()
+
+        self.config = config
+
+        self.client = mqtt.Client(userdata=self)
+
+        # consider this flag in the localLoop
+        self.killBill = False
+        self.killEvent = threading.Event()
+
+    def run(self):
+        self.client.on_message = mqttOnMessageCallback
+        self.client.on_connect = mqttOnConnectCallback
+        self.client.on_disconnect = mqttOnDisconnectCallback
+
+        if ("login" in self.config["mqtt"]) and ("password" in self.config["mqtt"]):
+            self.client.username_pw_set(self.config["mqtt"]["login"], self.config["mqtt"]["password"])
+        
+        self.client.connect(self.config["mqtt"]["broker"])
+        self.client.loop_start()
+        logger.debug("mqtt loop started")
+
+        self.localLoop()
+    
+    def localLoop(self):
+        raise NotImplementedError()
+
+    def stop(self):
+        self.client.loop_stop()
+        logger.debug("about to stop loop")
+        
+        self.killBill = True
+        logger.debug("kill flag set")
+
+    def onConnect(self):
+        logger.info("mqtt connected")
+    
+    def onDisconnect(self, rc):
+        logger.warning("disconnect, rc: {}".format(rc))
+
+    def onMessage(self, topic, payload):
+        logger.warning("unexpected message received: {} -> {}".format(topic, str(payload)))
+

src/DataObject.py

@@ -0,0 +1,5 @@
+from collections import namedtuple
+
+DataObject = namedtuple("DataObject", ['name', 'topic', 'payload'])
+
+

src/GenericMqttSubscriber.py

@@ -0,0 +1,34 @@
+from json.decoder import JSONDecodeError
+import paho.mqtt.client as mqtt
+import threading
+from loguru import logger
+from time import sleep
+import json
+
+from AbstractMqttHandler import AbstractMqttHandler
+from DataObject import DataObject
+
+class GenericMqttSubscriber(AbstractMqttHandler):
+    def __init__(self, config, configId, queue):
+        super().__init__(config)
+        localConfig = self.config[configId]
+        self.dataObjectName = localConfig["dataObjectName"]
+        self.subscribeTopic = localConfig["topic"]
+        self.queue = queue
+
+    def localLoop(self):
+        while not self.killBill:
+            sleep(float(1.0))
+
+    def onMessage(self, topic, payload):
+        try: 
+            logger.info("message received: {} -> {}".format(topic, str(payload)))
+            dataObject = DataObject(name=self.dataObjectName, topic=topic, payload=json.loads(payload))
+            self.queue.put(dataObject)
+        except JSONDecodeError as e:
+            logger.warning("unable to decode {} to json".format(str(payload)))
+
+    def onConnect(self):
+        logger.debug("connected")
+        self.client.subscribe(self.subscribeTopic)
+        logger.debug(f"subscribed to {self.subscribeTopic}")

src/iiotfeeder2.py

@@ -0,0 +1,55 @@
+from loguru import logger
+import argparse
+import configparser
+import threading
+from queue import Queue
+
+from GenericMqttSubscriber import GenericMqttSubscriber
+
+
+deathBell = threading.Event()
+
+def exceptHook(args):
+    global deathBell
+    logger.error("Exception in thread caught: {}".format(args))
+    deathBell.set()
+    logger.error("rang the death bell")
+
+# ------------------------------------------------------------------------------
+logger.info("IIoTFeeder2 starting")
+
+parser = argparse.ArgumentParser(description="IIoTFeeder2")
+parser.add_argument('--config', '-f',
+                    help='Config file, default is $pwd/config/config.ini',
+                    required=False,
+                    default='./config/config.ini')
+args = parser.parse_args()
+
+config = configparser.ConfigParser()
+config.read(args.config)
+
+# ------------------------------------------------------------------------------
+masterQueue = Queue()
+masterSubscriber = GenericMqttSubscriber(config, "master", masterQueue)
+masterSubscriber.start()
+logger.debug("MasterSubscriber started")
+
+threading.excepthook = exceptHook
+logger.debug("Threading excepthook set")
+
+logger.debug("IIoTFeeder2 running")
+
+# ------------------------------------------------------------------------------
+while not deathBell.wait(1.0):
+    logger.debug("master queue size {}".format(masterQueue.qsize()))
+    
+logger.error("IIoTFeeder2 is dying")
+
+masterSubscriber.stop()
+
+masterSubscriber.join()
+logger.error("masterSubscriber joined")
+
+# ------------------------------------------------------------------------------
+logger.info("IIoTFeeder2 to terminate")
+