thread names added

.gitlab-ci.yml

@@ -39,13 +39,15 @@ release:
     only:
       refs:
         - tags
+    variables:
+      RELEASE_FILE: opcua2mqtt-bridge-${CI_COMMIT_TAG}.tar.gz
     script:
         - pushd src
-        - tar -czvf ../opcua2mqtt-bridge-${CI_COMMIT_TAG}.tar.gz --exclude=__pycache__ *
+        - tar -czvf ../$RELEASE_FILE --exclude=__pycache__ *.py *.json opcua2mqtt-bridge.service
         - popd
         - gitlabreleaseuploader.py -p "$PRIVATE_TOKEN" -i $CI_PROJECT_ID -u $CI_PROJECT_URL 
           -I https://devservice.krohne.com/git/
           -C
-          -f opcua2mqtt-bridge-${CI_COMMIT_TAG}.tar-gz
+          -f $RELEASE_FILE
           -T $CI_COMMIT_REF_NAME -t "$CI_COMMIT_TAG"
-          -n "$VERSION" -d "Tarball of $CI_PROJECT_NAME, version $VERSION"
+          -n "${CI_COMMIT_TAG} release" -d "Tarball of $CI_PROJECT_NAME, version $CI_COMMIT_TAG"

readme.md

@@ -0,0 +1,141 @@
+# OPC-UA to MQTT Bridge
+
+## Installation
+
+* Take the tarball from the Release area and unpack it in the desired location on the edge device
+* The systemctl service script assumes to find the application at `/home/alarm/KROHNE/opcua2mqtt-bridge`, so it is a good idea to put it there
+* Create a configuration directory `/etc/opcua2mqtt-bridge` and place the adjusted `config.json` into it
+* Enable the service using `sudo systemctl enable /home/alarm/KROHNE/opcua2mqtt-bridge/opcua2mqtt-bridge.service`
+* Start the service using `sudo systemctl start opcua2mqtt-bridge`
+* Check the log messages using `journalctl -f`
+
+## Configuration
+
+    {
+        "mqtt": {
+            "broker": "172.16.2.16",
+            "port": 1883,
+            "publishTopicPrefix": "opcua"
+        },
+        "stats": {
+            "topic": "statistics",
+            "period": 60
+        },
+        "opcua": [
+            {
+                "enabled": "true",
+                "type": "flat",
+                "url": "opc.tcp://172.16.3.60:4840",
+                "name": "apl",
+                "period": 1.0,
+                "timeout": 1.0,
+                "nodes": [
+                    { "ns": 0, "n": "i=345", "d": "pv" },
+                    { "ns": 0, "n": "i=348", "d": "sv" },
+                    { "ns": 0, "n": "i=351", "d": "tv" },
+                    { "ns": 0, "n": "i=354", "d": "qv" }
+                ]
+            }
+        ]
+    }
+
+The configuration object consists of three parts: `mqtt`, `stats` and `opcua`.
+
+In `mqtt` the access to the broker is configuration and the prefix for the topics used by the statistics module and the actual bridge are defined.
+Besides the above shown attributes the `login` and `password` for authentication at the broker and `ca`, `cert` and `key` for TLS connections to the broker are available. `ca`, `cert` and `key` contain the filenames including complete path of the particular files.
+
+In `stats` the topic suffix and the period of statistics messages are defined. The complete suffix will be `${mqtt.publicTopicPrefix}/${stats.topic}`.
+
+The section `opcua` contains a list of OPC-UA servers to be queried. Each entry can be enabled/disabled using the attribute `enabled`. The attribute `url` obviously has the URL to connect the server, `period` is the period to repeat queries and `timeout` is the timeout when talking to a server.
+`name` identifies the particular server, it becomes part of the topic of the published MQTT messages. The attribute `type` defines whether the individual variables of an OPC-UA server shall be communicated in individual MQTT messages (`flat`) or all variables of a server in a single message (`structured`).
+
+The attribute `nodes` finally contains the list of variables to be queried. It contains the namespace index (`ns`), the node-id (`n`) and an optional descriptive name (`d`). Namespace index and node-id can be determined using for instance UAExpert when browsing the server and navigating to the relevant variables. The descriptive name - if given, otherwise the display name of the variable is used - becomes in `flat` mode part of the topic, in `structured` mode it becomes an attribute name within the message.
+
+
+## Example Output
+
+Besides the value itself the output contains the status of the value and a couple of timestamps (from device `t1`, server `t2` and bridge `t3`).
+
+In `flat` mode the final topic will be `${mqtt.publicTopicPrefix}/${opcua.name}/${opcua.node.ns}/${opcua.node.d}`
+
+An example for the MQTT messages according to the above configuration in `flat` mode is:
+
+    
+    opcua/apl/0/pv {
+        "server": "apl", 
+        "ns": 0, 
+        "d": "pv", 
+        "value": 19.833280563354492, 
+        "status": "Good", 
+        "t1": "2022-02-22 17:03:41.189000", 
+        "t2": "2022-02-22 17:03:41.189000", 
+        "t3": "2022-02-22 17:03:41.575504"
+    }
+    opcua/apl/0/sv {
+        "server": "apl", 
+        "ns": 0, 
+        "d": "sv", 
+        "value": 1706.15771484375, 
+        "status": "Good", 
+        "t1": "2022-02-22 17:03:41.189000", 
+        "t2": "2022-02-22 17:03:41.189000", 
+        "t3": "2022-02-22 17:03:41.625721"
+    }
+    opcua/apl/0/tv {
+        "server": "apl", 
+        "ns": 0, 
+        "d": "tv", 
+        "value": 23.29559326171875, 
+        "status": "Good", 
+        "t1": "2022-02-22 17:03:41.189000", 
+        "t2": "2022-02-22 17:03:41.189000", 
+        "t3": "2022-02-22 17:03:41.675352"
+    }
+    opcua/apl/0/qv {
+        "server": "apl", 
+        "ns": 0, 
+        "d": "qv", 
+        "value": NaN, 
+        "status": "Good", 
+        "t1": "2022-02-22 17:03:41.189000", 
+        "t2": "2022-02-22 17:03:41.189000", 
+        "t3": "2022-02-22 17:03:41.725487"
+    }
+
+
+In `structured` mode the final topic will be `${mqtt.publicTopicPrefix}/${opcua.name}`
+
+An example for the MQTT messages according to the above configuration in `flat` mode is:
+
+    opcua/apl {
+        "pv": {
+            "value": 19.835201263427734, 
+            "status": "Good", 
+            "t1": "2022-02-22 16:58:21.188000", 
+            "t2": "2022-02-22 16:58:21.188000", 
+            "t3": "2022-02-22 16:58:22.071619"
+        }, 
+        "sv": {
+            "value": 1704.4019775390625, 
+            "status": "Good", 
+            "t1": "2022-02-22 16:58:21.188000", 
+            "t2": "2022-02-22 16:58:21.188000", 
+            "t3": "2022-02-22 16:58:22.121559"
+        }, 
+        "tv": {
+            "value": 23.08197021484375, 
+            "status": "Good", 
+            "t1": "2022-02-22 16:58:21.188000", 
+            "t2": "2022-02-22 16:58:21.188000", 
+            "t3": "2022-02-22 16:58:22.171498"
+        }, 
+        "qv": {
+            "value": NaN, 
+            "status": "Good", 
+            "t1": "2022-02-22 16:58:21.188000", 
+            "t2": "2022-02-22 16:58:21.188000", 
+            "t3": "2022-02-22 16:58:22.221639"
+        }
+    }
+
+

snippets/test2.py

@@ -4,13 +4,20 @@ from asyncua import Client
 
 
 async def test():
-    client = Client(url='opc.tcp://192.168.254.5:4863', timeout=10.0)
+    client = Client(url='opc.tcp://172.16.3.60:4840', timeout=10.0)
     # await client.set_security_string('')
     async with client:
-        node = client.get_node('ns=1;s=t|SERVER::A201CD124/MOT_01.AV_Out#Value')
+        node = client.get_node('ns=0;i=345')
         value = await node.read_value()
         displayName = (await node.read_display_name()).Text
+        print(dir(node))
         print(f"{displayName=} = {value=}")
+        print(f"X1: {await node.read_data_value()}")
+        print(f"X2: {(await node.read_data_value()).Value.Value}")
+        print(f"X3: {dir((await node.read_data_value()).StatusCode)}")
+        print(f"X3: {(await node.read_data_value()).StatusCode.name}")
+        print(f"X4: {(await node.read_data_value()).SourceTimestamp}")
+        print(f"X4: {(await node.read_data_value()).ServerTimestamp}")
 
 asyncio.run(test())
 

snippets/test4.py

@@ -0,0 +1,32 @@
+import asyncio
+from asyncua import Client
+
+
+
+NODE_IDS = [
+    'ns=0;i=345',
+    'ns=0;i=348',
+    'ns=0;i=351',
+    'ns=0;i=354'
+]
+URL = 'opc.tcp://172.16.3.60:4840'
+
+
+class SubscriptionHandler:
+    def datachange_notification(self, node, val, data):
+        print(f"received: {node=}, {val=}, {data=}")
+
+async def test():
+    client = Client(url=URL, timeout=10.0)
+    # await client.set_security_string('')
+    async with client:
+        subscription = await client.create_subscription(500, SubscriptionHandler())
+        nodes = [ client.get_node(n) for n in NODE_IDS ]
+        await subscription.subscribe_data_change(nodes)
+        await asyncio.sleep(300)
+        await subscription.delete()
+
+
+
+asyncio.run(test())
+

src/FlatDataObject.py

@@ -1,21 +1,26 @@
 import re
 import json
 from AbstractDataObject import AbstractDataObject
+from datetime import datetime
 
 
 class FlatDataObject(AbstractDataObject):
-    def __init__(self, serverName, nameSpaceIndex, variableName, value):
+    def __init__(self, serverName, nameSpaceIndex, variableName, dataValue):
         super().__init__(serverName + '/' + str(nameSpaceIndex) + '/' + variableName)
         self.serverName = serverName
         self.nameSpaceIndex = nameSpaceIndex
         self.variableName = variableName
-        self.value = value
+        self.dataValue = dataValue
 
     def getPayload(self):
         payload = {
-            "serverName": self.serverName,
-            "nameSpaceIndex": self.nameSpaceIndex,
-            "variableName": self.variableName,
-            "value": self.value
+            "server": self.serverName,
+            "ns": self.nameSpaceIndex,
+            "d": self.variableName,
+            "value": self.dataValue.Value.Value,
+            "status": self.dataValue.StatusCode.name,
+            "t1": str(self.dataValue.SourceTimestamp),
+            "t2": str(self.dataValue.ServerTimestamp),
+            "t3": str(datetime.utcnow())
         }
         return json.dumps(payload)

src/MqttPublish.py

@@ -11,6 +11,7 @@ LOOP_SLICE = 0.1  # seconds
 class MqttPublish(AbstractMqttPublisher):
     def __init__(self, config, stats, queue):
         super().__init__(config)
+        self.name = 'mqtt'
         self.queue = queue
         self.stats = stats
         self.topicPre = self.config["publishTopicPrefix"]

src/OpcUaRequester.py

@@ -17,7 +17,7 @@ class OpcUaRequester(threading.Thread):
         self.name = self.config['name']
         self.url = self.config['url']
         self.nodes = self.config['nodes']
-        self.delay = self.config['delay']
+        self.period = self.config['period']
         self.timeout = self.config['timeout']
         self.dataObjectType = self.config['type']
         self.flat = self.dataObjectType == 'flat'
@@ -36,20 +36,21 @@ class OpcUaRequester(threading.Thread):
                         try:
                             logger.debug(f"Trying {self.name} {self.url} ns={nodeSpec['ns']};{nodeSpec['n']}")
                             node = client.get_node(f"ns={nodeSpec['ns']};{nodeSpec['n']}")
-                            value = await node.read_value()
+
+                            dataValue = await node.read_data_value()
                             displayName = nodeSpec['d'] if ('d' in nodeSpec) else (await node.read_display_name()).Text
-                            logger.debug(f"Got: {displayName=} = {value=}")
+                            logger.debug(f"Got: {displayName=} = {dataValue.Value.Value=}")
                             if self.flat:
-                                self.queue.put(FlatDataObject(self.name, nodeSpec['ns'], displayName, value))
+                                self.queue.put(FlatDataObject(self.name, nodeSpec['ns'], displayName, dataValue))
                             else:
-                                dataObject.add(displayName, value)
+                                dataObject.add(displayName, dataValue)
                             self.stats.incOpcUaRequests()
                         except ua.UaError as e:
                             self.stats.incOpcUaErrors()
                             logger.error(f"UaError in inner OPC-UA loop: {type(e)} {e}")
                     if not self.flat:
                         self.queue.put(dataObject)
-                await asyncio.sleep(self.delay)
+                await asyncio.sleep(self.period)
             except asyncio.exceptions.TimeoutError as e:
                 self.stats.incOpcUaTimeouts()
                 logger.error(f"Timeout in inner OPC-UA loop")

src/OpcUaSubscriber.py

@@ -0,0 +1,114 @@
+import asyncio
+from sqlite3 import NotSupportedError
+from asyncua import Client, ua
+import threading
+from loguru import logger
+from FlatDataObject import FlatDataObject
+from StructuredDataObject import StructuredDataObject
+
+
+
+class FlatSubscriptionHandler:
+    def __init__(self, serverName, nodes, queue):
+        self.serverName = serverName
+        self.nodes = nodes
+        self.queue = queue
+        self.count = 0
+
+    def getAndResetCount(self):
+        tmp = self.count
+        self.count = 0
+        return tmp
+
+    def datachange_notification(self, node, val, data):
+        logger.info(f"received: {node=}, {val=}, {data=}")
+        self.count += 1
+        match node.nodeid.NodeIdType:
+            case ua.NodeIdType.Numeric:
+                prefix = 'i'
+            case ua.NodeIdType.String:
+                prefix = 's'
+            case _: 
+                prefix = 'x'
+        nodeName = f"{prefix}={node.nodeid.Identifier}"
+        namespaceIndex = node.nodeid.NamespaceIndex
+        displayNames = [ x['d'] for x in self.nodes if x['ns'] == namespaceIndex and x['n'] == nodeName ]
+        name = displayNames[0] if displayNames else nodeName
+        self.queue.put(FlatDataObject(self.serverName, str(namespaceIndex), name, data.monitored_item.Value))
+
+
+class _RenewTriggerException(Exception): pass
+
+class OpcUaSubscriber(threading.Thread):
+    def __init__(self, config, stats, queue):
+        super().__init__()
+
+        self.config = config
+        self.queue = queue
+        self.stats = stats
+
+        self.name = self.config['name']
+        self.url = self.config['url']
+        self.nodes = self.config['nodes']
+        self.period = self.config['period']
+        self.timeout = self.config['timeout']
+        self.dataObjectType = self.config['type']
+        self.flat = self.dataObjectType == 'flat'
+
+        if not self.flat:
+            raise NotImplementedError("Only flat approach supported in OpcUaSubscriber")
+
+        # consider this flag in the localLoop
+        self.killBill = False
+        self.killEvent = asyncio.Event()
+
+    async def opcUaSubscriberInnerLoop(self):
+        while not self.killBill:
+            try:
+                async with Client(url=self.url, timeout=self.timeout) as client:
+                    subscriptionHandler = FlatSubscriptionHandler(self.name, self.nodes, self.queue) 
+                    subscription = await client.create_subscription(self.period * 1000, subscriptionHandler)
+                    nodes = [ client.get_node(f"ns={n['ns']};{n['n']}") for n in self.nodes ]
+                    await subscription.subscribe_data_change(nodes)
+                    logger.info("Subscriptions created, nodes subscribed")                    
+
+                    while True:
+                        try:
+                            await asyncio.wait_for(self.killEvent.wait(), self.period * 10)
+                            logger.info("About to terminate opcUaSubscriber")
+                            break
+                        except asyncio.TimeoutError:
+                            cnt = subscriptionHandler.getAndResetCount()
+                            logger.info(f"receive count: {cnt}")
+                            if cnt == 0:
+                                raise _RenewTriggerException()
+
+                    await subscription.delete()
+                    logger.info("Subscriptions deleted, wait a moment")
+                    await asyncio.sleep(1)
+                    logger.info("opcUaSubscriber terminated")
+            except _RenewTriggerException:
+                logger.error(f"too few data received, renew connection")
+                # continues in the loop
+            except asyncio.exceptions.TimeoutError as e:
+                self.stats.incOpcUaTimeouts()
+                logger.error(f"Timeout in inner OPC-UA loop")
+            except asyncio.exceptions.CancelledError as e:
+                self.stats.incOpcUaErrors()
+                logger.error(f"Cancelled in inner OPC-UA loop")
+            except Exception as e:
+                self.stats.incOpcUaErrors()
+                logger.error(f"Exception in inner OPC-UA loop: {type(e)} {e}")
+
+    def run(self):
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        loop.run_until_complete(self.opcUaSubscriberInnerLoop())
+        loop.close()
+
+    def stop(self):
+        self.killBill = True
+        logger.info("kill flag set")
+
+        self.killEvent.set()
+        logger.info("kill events triggered")

src/Statistics.py

@@ -18,6 +18,7 @@ class StatisticsCollector(threading.Thread):
     def __init__(self, config, queue):
         super().__init__()
 
+        self.name = 'stats'
         self.config = config['stats']
         self.period = self.config['period']
         self.topic = self.config['topic']

src/StructuredDataObject.py

@@ -1,15 +1,22 @@
 import re
 import json
 from AbstractDataObject import AbstractDataObject
+from datetime import datetime
 
 
 class StructuredDataObject(AbstractDataObject):
     def __init__(self, topicPart):
         super().__init__(topicPart)
-        self.keyValuePairs = []
+        self.keyValuePairs = {}
 
-    def add(self, key, value):
-        self.keyValuePairs.append({key: value})
+    def add(self, key, dataValue):
+        self.keyValuePairs[key] = {
+            "value": dataValue.Value.Value,
+            "status": dataValue.StatusCode.name,
+            "t1": str(dataValue.SourceTimestamp),
+            "t2": str(dataValue.ServerTimestamp),
+            "t3": str(datetime.utcnow())
+        }
 
     def getPayload(self):
         return json.dumps(self.keyValuePairs)

src/bridge.py

@@ -1,5 +1,4 @@
 from MqttPublish import MqttPublish
-from OpcUaRequester import OpcUaRequester
 from Statistics import StatisticsCollector
 
 from loguru import logger
@@ -40,24 +39,43 @@ with open(args.config) as f:
     config = json.load(f)
 
 
+match config['opcua']['type']:
+    case 'requester':
+        logger.info("Loading OpcUaRequester")
+        from OpcUaRequester import OpcUaRequester as OpcUa
+    case 'subscriber':
+        logger.info("Loading OpcUaSubscriber")
+        from OpcUaSubscriber import OpcUaSubscriber as OpcUa
+    case _:
+        raise Exception("unknown OpcUa type")
+
+
 queue = queue.Queue()
 
-statsThread = StatisticsCollector(config, queue)
-statsThread.start()
-logger.info("StatisticsCollector started")
+threads = []
+try:
+    statsThread = StatisticsCollector(config, queue)
+    statsThread.start()
+    threads.append(statsThread)
+    logger.info("StatisticsCollector started")
 
-publishThread = MqttPublish(config, statsThread, queue)
-publishThread.start()
-logger.info("MqttPublish started")
+    publishThread = MqttPublish(config, statsThread, queue)
+    publishThread.start()
+    threads.append(publishThread)
+    logger.info("MqttPublish started")
             
-opcuaThreads = []
-for o in config['opcua']:
+
+    opcuaThreads = []
+    for o in config['opcua']["servers"]:
         if o['enabled'] != 'true':
             continue
-    ot = OpcUaRequester(o, statsThread, queue)
+        ot = OpcUa(o, statsThread, queue)
         ot.start()
         logger.info(f"OpcUaRequester thread for {o['name']} started")
-    opcuaThreads.append(ot)
+        threads.append(ot)
+except Exception as e:
+    logger.error(f"caught exception {type(e)}, {e} during start-up phase")
+    deathBell.set()
 
 threading.excepthook = exceptHook
 logger.info("Threading excepthook set")
@@ -71,24 +89,12 @@ logger.info("opcua2mqtt bridge is running")
 deathBell.wait()
 logger.error("opcua2mqtt bridge is dying")
 
-publishThread.stop()
-logger.error("publishThread stopped")
+for t in threads:
+    t.stop()
+    logger.error(f"thread {t.name} stopped")
 
-statsThread.stop()
-logger.error("statsThread stopped")
-
-for ot in opcuaThreads:
-    ot.stop()
-    logger.error(f"opcua thread {ot.name} stopped")
-
-publishThread.join()
-logger.error("publishThread joined")
-
-statsThread.join()
-logger.error("statsThread joined")
-
-for ot in opcuaThreads:
-    ot.join()
-    logger.error(f"opcua thread {ot.name} joined")
+for t in threads:
+    t.join()
+    logger.error(f"thread {t.name} joined")
 
 logger.error("opcua2mqtt bridge is terminated")

src/config.json

@@ -8,13 +8,15 @@
         "topic": "statistics",
         "period": 60
     },
-    "opcua": [
+    "opcua": {
+        "type": "subscriber",
+        "servers": [
             {
                 "enabled": "true",
                 "type": "flat",
                 "url": "opc.tcp://172.16.3.60:4840",
                 "name": "apl",
-            "delay": 1.0,
+                "period": 1.0,
                 "timeout": 1.0,
                 "nodes": [
                     { "ns": 0, "n": "i=345", "d": "pv" },
@@ -28,7 +30,7 @@
                 "type": "flat",
                 "url": "opc.tcp://192.168.254.5:4863",
                 "name": "sh",
-            "delay": 1.0,
+                "period": 1.0,
                 "timeout": 10.0,
                 "nodes": [
                     { "ns": 1, "n": "s=t|SERVER::A201CD124/MOT_01.AV_Out#Value", "d": "A201CD124" },
@@ -64,6 +66,7 @@
                 ]
             }
         ]
+    }
 }
 
 

src/opcua2mqtt-bridge.service

@@ -0,0 +1,20 @@
+[Unit]
+Description=OPCUA2MQTT-Bridge
+Wants=network-online.target
+After=network-online.target
+
+
+[Service]
+Type=simple
+GuessMainPID=yes
+ExecStart=/usr/bin/python bridge.py -f /etc/opcua2mqtt-bridge/config.json
+ExecStop=kill -SIGINT $mainpid
+Restart=on-failure
+User=alarm
+WorkingDirectory=/home/alarm/KROHNE/opcua2mqtt-bridge
+
+[Install]
+Alias=OPCUA2MQTT-Bridge
+WantedBy=multi-user.target
+
+