opcua-with-python/opcua2mqtt/bridge.py

from MqttPublish import MqttPublish
from OpcUaRequester import OpcUaRequester
from loguru import logger
import argparse
import json
import threading
import queue
import signal


deathBell = threading.Event()

def exceptHook(args):
    global deathBell
    logger.error("Exception in thread caught: {}".format(args))
    deathBell.set()
    logger.error("rang the death bell")

def terminateHook(sig, frame):
    global deathBell
    logger.error("SIGINT received")
    deathBell.set()
    logger.error("rang the death bell")


logger.info("opcua2mqtt bridge starting")

parser = argparse.ArgumentParser(description="opcua2mqtt")
parser.add_argument('--config', '-f',
                    help='Config file, default is $pwd/config.json',
                    required=False,
                    default='./config.json')
args = parser.parse_args()

with open(args.config) as f:
    config = json.load(f)


queue = queue.Queue()

publishThread = MqttPublish(config, queue)
publishThread.start()
logger.info("MqttPublish started")

opcuaThreads = []
for o in config['opcua']:
    if o['enabled'] != 'true':
        continue
    ot = OpcUaRequester(o, queue)
    ot.start()
    logger.info(f"OpcUaRequester thread for {o['name']} started")
    opcuaThreads.append(ot)

threading.excepthook = exceptHook
logger.info("Threading excepthook set")

signal.signal(signal.SIGINT, terminateHook)
logger.info("SIGINT handler set")

logger.info("opcua2mqtt bridge is running")


deathBell.wait()
logger.error("opcua2mqtt bridge is dying")

publishThread.stop()
logger.error("publishThread stopped")

for ot in opcuaThreads:
    ot.stop()
    logger.error(f"opcua thread {ot.name} stopped")

publishThread.join()
logger.error("publishThread joined")

for ot in opcuaThreads:
    ot.join()
    logger.error(f"opcua thread {ot.name} joined")

logger.error("opcua2mqtt bridge is terminated")