RigCmdHandler started

src/AbstractMqttHandler.py

@@ -22,8 +22,7 @@ class AbstractMqttHandler(threading.Thread):
         self.client = mqtt.Client(userdata=self)
 
         # consider this flag in the localLoop
-        self.killBill = False
-        self.killEvent = threading.Event()
+        self.poisonPill = False
 
     def run(self):
         self.client.on_message = mqttOnMessageCallback
@@ -44,9 +43,10 @@ class AbstractMqttHandler(threading.Thread):
 
     def stop(self):
         self.client.loop_stop()
+        self.client.disconnect()
         logger.debug("about to stop loop")
         
-        self.killBill = True
+        self.poisonPill = True
         logger.debug("kill flag set")
 
     def onConnect(self):

src/DummyEGM1.py

@@ -15,6 +15,7 @@ def sendCmd(client, cmd):
     }
     topic = f"{TOPIC_PRE}/rig01/cmd"
     client.publish(topic, json.dumps(msg))
+    logger.debug(topic)
 
 def sendLog(client):
     msg = {
@@ -30,30 +31,24 @@ def sendLog(client):
     }
     topic = f"{TOPIC_PRE}/rig01/log"
     client.publish(topic, json.dumps(msg))
+    logger.debug(topic)
 
 def sendValue(client):
     msg = {
         "ts": str(datetime.now()),
         "dt": 100,
-        "v": sample(range(0xffffffff), 1000)
+        "v": sample(range(0xffffffff), 3000)
     }
     topic = f"{TOPIC_PRE}/rig01/dev05/md"
     client.publish(topic, json.dumps(msg))
+    logger.debug(topic)
 
 
 def perform(client, params):
     logger.info("Start")
     sendCmd(client, "start")
 
-    for i in range(10):
-        sendValue(client)
-        sleep(1.0)
-        sendValue(client)
-        sleep(1.0)
-        sendValue(client)
-        sleep(1.0)
-        sendValue(client)
-        sleep(1.0)
+    for i in range(2):
         sendValue(client)
         sleep(1.0)
         sendValue(client)

src/GenericMqttSubscriber.py

@@ -17,7 +17,7 @@ class GenericMqttSubscriber(AbstractMqttHandler):
         self.queue = queue
 
     def localLoop(self):
-        while not self.killBill:
+        while not self.poisonPill:
             sleep(float(1.0))
 
     def onMessage(self, topic, payload):

src/RigCmdHandler.py

@@ -0,0 +1,53 @@
+import threading
+from loguru import logger
+from time import sleep
+from DataObject import DataObject
+from enum import Enum
+
+
+POISON_PILL = DataObject(name="PoisonPill", topic="kill", payload=None)
+
+class e_State(Enum):
+    IDLE = 0
+    RUNNING = 1
+
+class RigCmdHandler(threading.Thread):
+    def __init__(self, config, inQueue):
+        super().__init__()
+
+        self.config = config
+        self.inQueue = inQueue
+        self.state = e_State.IDLE
+
+    def run(self):
+        logger.debug("RigCmdHandler loop started")
+        while True:
+            dataObject = self.inQueue.get()
+            if (dataObject == POISON_PILL):
+                logger.error("RigCmdHandler swallowed the poison pill")
+                break
+            
+            logger.info(f"State is {self.state}")
+            # Python 3.10 is required!!
+            match self.state:
+                case e_State.IDLE:
+                    if (dataObject.name == 'cmd') and (dataObject.payload["es"] == 'start'):
+                        logger.info("start command received, switch to RUNNING state")
+                        self.state = e_State.RUNNING
+                    elif (dataObject.name == 'cmd'):
+                        logger.error(f"illegal command {dataObject.name} received in IDLE state")
+                    else:
+                        logger.error("illegal message received in IDLE state")
+                case e_State.RUNNING:
+                    if (dataObject.name == 'cmd') and (dataObject.payload["es"] == 'stop'):
+                        logger.info("stop command received, switch to IDLE state")
+                        self.state = e_State.IDLE
+                    elif (dataObject.name == 'cmd'):
+                        logger.error(f"illegal command {dataObject.name} received in RUNNING state")
+                    else:
+                        logger.error("illegal message received in RUNNING state")
+
+
+    def stop(self):
+        self.inQueue.put(POISON_PILL)
+        logger.debug("kill flag set")

src/config/iiotFeederConfig.ini

@@ -0,0 +1,8 @@
+[mqtt]
+login=
+password=
+broker=172.16.2.16
+
+[master]
+topic=rd/+/+/cmd
+dataObjectName=cmd

src/iiotfeeder2.py

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