Compare commits
17 Commits
0.1
...
subscripti
Author | SHA1 | Date | |
---|---|---|---|
72f9d97402 | |||
695594fd1c | |||
10d2ab5268 | |||
980e5c7452 | |||
0aaddc15df | |||
da6574632c | |||
9ce5395547 | |||
00d4004e54 | |||
813470fd03 | |||
578023fbd4 | |||
35087f89ca | |||
d44460bed5 | |||
7fd07bd168 | |||
34adf8bd73 | |||
1e80d00dbf | |||
daa869d1e4 | |||
a0e1ef94cd |
@ -39,13 +39,15 @@ release:
|
|||||||
only:
|
only:
|
||||||
refs:
|
refs:
|
||||||
- tags
|
- tags
|
||||||
|
variables:
|
||||||
|
RELEASE_FILE: opcua2mqtt-bridge-${CI_COMMIT_TAG}.tar.gz
|
||||||
script:
|
script:
|
||||||
- pushd src
|
- 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
|
- popd
|
||||||
- gitlabreleaseuploader.py -p "$PRIVATE_TOKEN" -i $CI_PROJECT_ID -u $CI_PROJECT_URL
|
- gitlabreleaseuploader.py -p "$PRIVATE_TOKEN" -i $CI_PROJECT_ID -u $CI_PROJECT_URL
|
||||||
-I https://devservice.krohne.com/git/
|
-I https://devservice.krohne.com/git/
|
||||||
-C
|
-C
|
||||||
-f opcua2mqtt-bridge-${CI_COMMIT_TAG}.tar-gz
|
-f $RELEASE_FILE
|
||||||
-T $CI_COMMIT_REF_NAME -t "$CI_COMMIT_TAG"
|
-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"
|
||||||
|
141
readme.md
Normal file
141
readme.md
Normal file
@ -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"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
|
@ -4,13 +4,20 @@ from asyncua import Client
|
|||||||
|
|
||||||
|
|
||||||
async def test():
|
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('')
|
# await client.set_security_string('')
|
||||||
async with client:
|
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()
|
value = await node.read_value()
|
||||||
displayName = (await node.read_display_name()).Text
|
displayName = (await node.read_display_name()).Text
|
||||||
|
print(dir(node))
|
||||||
print(f"{displayName=} = {value=}")
|
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())
|
asyncio.run(test())
|
||||||
|
|
||||||
|
32
snippets/test4.py
Normal file
32
snippets/test4.py
Normal file
@ -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())
|
||||||
|
|
@ -1,21 +1,26 @@
|
|||||||
import re
|
import re
|
||||||
import json
|
import json
|
||||||
from AbstractDataObject import AbstractDataObject
|
from AbstractDataObject import AbstractDataObject
|
||||||
|
from datetime import datetime
|
||||||
|
|
||||||
|
|
||||||
class FlatDataObject(AbstractDataObject):
|
class FlatDataObject(AbstractDataObject):
|
||||||
def __init__(self, serverName, nameSpaceIndex, variableName, value):
|
def __init__(self, serverName, nameSpaceIndex, variableName, dataValue):
|
||||||
super().__init__(serverName + '/' + str(nameSpaceIndex) + '/' + variableName)
|
super().__init__(serverName + '/' + str(nameSpaceIndex) + '/' + variableName)
|
||||||
self.serverName = serverName
|
self.serverName = serverName
|
||||||
self.nameSpaceIndex = nameSpaceIndex
|
self.nameSpaceIndex = nameSpaceIndex
|
||||||
self.variableName = variableName
|
self.variableName = variableName
|
||||||
self.value = value
|
self.dataValue = dataValue
|
||||||
|
|
||||||
def getPayload(self):
|
def getPayload(self):
|
||||||
payload = {
|
payload = {
|
||||||
"serverName": self.serverName,
|
"server": self.serverName,
|
||||||
"nameSpaceIndex": self.nameSpaceIndex,
|
"ns": self.nameSpaceIndex,
|
||||||
"variableName": self.variableName,
|
"d": self.variableName,
|
||||||
"value": self.value
|
"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)
|
return json.dumps(payload)
|
||||||
|
@ -11,6 +11,7 @@ LOOP_SLICE = 0.1 # seconds
|
|||||||
class MqttPublish(AbstractMqttPublisher):
|
class MqttPublish(AbstractMqttPublisher):
|
||||||
def __init__(self, config, stats, queue):
|
def __init__(self, config, stats, queue):
|
||||||
super().__init__(config)
|
super().__init__(config)
|
||||||
|
self.name = 'mqtt'
|
||||||
self.queue = queue
|
self.queue = queue
|
||||||
self.stats = stats
|
self.stats = stats
|
||||||
self.topicPre = self.config["publishTopicPrefix"]
|
self.topicPre = self.config["publishTopicPrefix"]
|
||||||
|
@ -17,7 +17,7 @@ class OpcUaRequester(threading.Thread):
|
|||||||
self.name = self.config['name']
|
self.name = self.config['name']
|
||||||
self.url = self.config['url']
|
self.url = self.config['url']
|
||||||
self.nodes = self.config['nodes']
|
self.nodes = self.config['nodes']
|
||||||
self.delay = self.config['delay']
|
self.period = self.config['period']
|
||||||
self.timeout = self.config['timeout']
|
self.timeout = self.config['timeout']
|
||||||
self.dataObjectType = self.config['type']
|
self.dataObjectType = self.config['type']
|
||||||
self.flat = self.dataObjectType == 'flat'
|
self.flat = self.dataObjectType == 'flat'
|
||||||
@ -36,20 +36,21 @@ class OpcUaRequester(threading.Thread):
|
|||||||
try:
|
try:
|
||||||
logger.debug(f"Trying {self.name} {self.url} ns={nodeSpec['ns']};{nodeSpec['n']}")
|
logger.debug(f"Trying {self.name} {self.url} ns={nodeSpec['ns']};{nodeSpec['n']}")
|
||||||
node = client.get_node(f"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
|
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:
|
if self.flat:
|
||||||
self.queue.put(FlatDataObject(self.name, nodeSpec['ns'], displayName, value))
|
self.queue.put(FlatDataObject(self.name, nodeSpec['ns'], displayName, dataValue))
|
||||||
else:
|
else:
|
||||||
dataObject.add(displayName, value)
|
dataObject.add(displayName, dataValue)
|
||||||
self.stats.incOpcUaRequests()
|
self.stats.incOpcUaRequests()
|
||||||
except ua.UaError as e:
|
except ua.UaError as e:
|
||||||
self.stats.incOpcUaErrors()
|
self.stats.incOpcUaErrors()
|
||||||
logger.error(f"UaError in inner OPC-UA loop: {type(e)} {e}")
|
logger.error(f"UaError in inner OPC-UA loop: {type(e)} {e}")
|
||||||
if not self.flat:
|
if not self.flat:
|
||||||
self.queue.put(dataObject)
|
self.queue.put(dataObject)
|
||||||
await asyncio.sleep(self.delay)
|
await asyncio.sleep(self.period)
|
||||||
except asyncio.exceptions.TimeoutError as e:
|
except asyncio.exceptions.TimeoutError as e:
|
||||||
self.stats.incOpcUaTimeouts()
|
self.stats.incOpcUaTimeouts()
|
||||||
logger.error(f"Timeout in inner OPC-UA loop")
|
logger.error(f"Timeout in inner OPC-UA loop")
|
||||||
|
114
src/OpcUaSubscriber.py
Normal file
114
src/OpcUaSubscriber.py
Normal file
@ -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")
|
@ -18,6 +18,7 @@ class StatisticsCollector(threading.Thread):
|
|||||||
def __init__(self, config, queue):
|
def __init__(self, config, queue):
|
||||||
super().__init__()
|
super().__init__()
|
||||||
|
|
||||||
|
self.name = 'stats'
|
||||||
self.config = config['stats']
|
self.config = config['stats']
|
||||||
self.period = self.config['period']
|
self.period = self.config['period']
|
||||||
self.topic = self.config['topic']
|
self.topic = self.config['topic']
|
||||||
|
@ -1,15 +1,22 @@
|
|||||||
import re
|
import re
|
||||||
import json
|
import json
|
||||||
from AbstractDataObject import AbstractDataObject
|
from AbstractDataObject import AbstractDataObject
|
||||||
|
from datetime import datetime
|
||||||
|
|
||||||
|
|
||||||
class StructuredDataObject(AbstractDataObject):
|
class StructuredDataObject(AbstractDataObject):
|
||||||
def __init__(self, topicPart):
|
def __init__(self, topicPart):
|
||||||
super().__init__(topicPart)
|
super().__init__(topicPart)
|
||||||
self.keyValuePairs = []
|
self.keyValuePairs = {}
|
||||||
|
|
||||||
def add(self, key, value):
|
def add(self, key, dataValue):
|
||||||
self.keyValuePairs.append({key: value})
|
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):
|
def getPayload(self):
|
||||||
return json.dumps(self.keyValuePairs)
|
return json.dumps(self.keyValuePairs)
|
||||||
|
@ -1,5 +1,4 @@
|
|||||||
from MqttPublish import MqttPublish
|
from MqttPublish import MqttPublish
|
||||||
from OpcUaRequester import OpcUaRequester
|
|
||||||
from Statistics import StatisticsCollector
|
from Statistics import StatisticsCollector
|
||||||
|
|
||||||
from loguru import logger
|
from loguru import logger
|
||||||
@ -40,24 +39,43 @@ with open(args.config) as f:
|
|||||||
config = json.load(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()
|
queue = queue.Queue()
|
||||||
|
|
||||||
statsThread = StatisticsCollector(config, queue)
|
threads = []
|
||||||
statsThread.start()
|
try:
|
||||||
logger.info("StatisticsCollector started")
|
statsThread = StatisticsCollector(config, queue)
|
||||||
|
statsThread.start()
|
||||||
|
threads.append(statsThread)
|
||||||
|
logger.info("StatisticsCollector started")
|
||||||
|
|
||||||
publishThread = MqttPublish(config, statsThread, queue)
|
publishThread = MqttPublish(config, statsThread, queue)
|
||||||
publishThread.start()
|
publishThread.start()
|
||||||
logger.info("MqttPublish started")
|
threads.append(publishThread)
|
||||||
|
logger.info("MqttPublish started")
|
||||||
|
|
||||||
opcuaThreads = []
|
|
||||||
for o in config['opcua']:
|
opcuaThreads = []
|
||||||
if o['enabled'] != 'true':
|
for o in config['opcua']["servers"]:
|
||||||
continue
|
if o['enabled'] != 'true':
|
||||||
ot = OpcUaRequester(o, statsThread, queue)
|
continue
|
||||||
ot.start()
|
ot = OpcUa(o, statsThread, queue)
|
||||||
logger.info(f"OpcUaRequester thread for {o['name']} started")
|
ot.start()
|
||||||
opcuaThreads.append(ot)
|
logger.info(f"OpcUaRequester thread for {o['name']} started")
|
||||||
|
threads.append(ot)
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"caught exception {type(e)}, {e} during start-up phase")
|
||||||
|
deathBell.set()
|
||||||
|
|
||||||
threading.excepthook = exceptHook
|
threading.excepthook = exceptHook
|
||||||
logger.info("Threading excepthook set")
|
logger.info("Threading excepthook set")
|
||||||
@ -71,24 +89,12 @@ logger.info("opcua2mqtt bridge is running")
|
|||||||
deathBell.wait()
|
deathBell.wait()
|
||||||
logger.error("opcua2mqtt bridge is dying")
|
logger.error("opcua2mqtt bridge is dying")
|
||||||
|
|
||||||
publishThread.stop()
|
for t in threads:
|
||||||
logger.error("publishThread stopped")
|
t.stop()
|
||||||
|
logger.error(f"thread {t.name} stopped")
|
||||||
|
|
||||||
statsThread.stop()
|
for t in threads:
|
||||||
logger.error("statsThread stopped")
|
t.join()
|
||||||
|
logger.error(f"thread {t.name} joined")
|
||||||
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")
|
|
||||||
|
|
||||||
logger.error("opcua2mqtt bridge is terminated")
|
logger.error("opcua2mqtt bridge is terminated")
|
||||||
|
115
src/config.json
115
src/config.json
@ -8,62 +8,65 @@
|
|||||||
"topic": "statistics",
|
"topic": "statistics",
|
||||||
"period": 60
|
"period": 60
|
||||||
},
|
},
|
||||||
"opcua": [
|
"opcua": {
|
||||||
{
|
"type": "subscriber",
|
||||||
"enabled": "true",
|
"servers": [
|
||||||
"type": "flat",
|
{
|
||||||
"url": "opc.tcp://172.16.3.60:4840",
|
"enabled": "true",
|
||||||
"name": "apl",
|
"type": "flat",
|
||||||
"delay": 1.0,
|
"url": "opc.tcp://172.16.3.60:4840",
|
||||||
"timeout": 1.0,
|
"name": "apl",
|
||||||
"nodes": [
|
"period": 1.0,
|
||||||
{ "ns": 0, "n": "i=345", "d": "pv" },
|
"timeout": 1.0,
|
||||||
{ "ns": 0, "n": "i=348", "d": "sv" },
|
"nodes": [
|
||||||
{ "ns": 0, "n": "i=351", "d": "tv" },
|
{ "ns": 0, "n": "i=345", "d": "pv" },
|
||||||
{ "ns": 0, "n": "i=354", "d": "qv" }
|
{ "ns": 0, "n": "i=348", "d": "sv" },
|
||||||
]
|
{ "ns": 0, "n": "i=351", "d": "tv" },
|
||||||
},
|
{ "ns": 0, "n": "i=354", "d": "qv" }
|
||||||
{
|
]
|
||||||
"enabled": "false",
|
},
|
||||||
"type": "flat",
|
{
|
||||||
"url": "opc.tcp://192.168.254.5:4863",
|
"enabled": "false",
|
||||||
"name": "sh",
|
"type": "flat",
|
||||||
"delay": 1.0,
|
"url": "opc.tcp://192.168.254.5:4863",
|
||||||
"timeout": 10.0,
|
"name": "sh",
|
||||||
"nodes": [
|
"period": 1.0,
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A201CD124/MOT_01.AV_Out#Value", "d": "A201CD124" },
|
"timeout": 10.0,
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A201CJ003/PID_01.PV_Out#Value", "d": "A201CJ003" },
|
"nodes": [
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A201CJ004/PID_01.PV_Out#Value", "d": "A201CJ004" },
|
{ "ns": 1, "n": "s=t|SERVER::A201CD124/MOT_01.AV_Out#Value", "d": "A201CD124" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A201CJ011/PID_01.PV_Out#Value", "d": "A201CJ011" },
|
{ "ns": 1, "n": "s=t|SERVER::A201CJ003/PID_01.PV_Out#Value", "d": "A201CJ003" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A201CJ014/PID_01.PV_Out#Value", "d": "A201CJ014" },
|
{ "ns": 1, "n": "s=t|SERVER::A201CJ004/PID_01.PV_Out#Value", "d": "A201CJ004" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A201CJ021/MMON_01.PV_Out#Value", "d": "A201CJ021" },
|
{ "ns": 1, "n": "s=t|SERVER::A201CJ011/PID_01.PV_Out#Value", "d": "A201CJ011" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A201CJ022/MMON_01.PV_Out#Value", "d": "A201CJ022" },
|
{ "ns": 1, "n": "s=t|SERVER::A201CJ014/PID_01.PV_Out#Value", "d": "A201CJ014" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A201CJ023/MMON_01.PV_Out#Value", "d": "A201CJ023" },
|
{ "ns": 1, "n": "s=t|SERVER::A201CJ021/MMON_01.PV_Out#Value", "d": "A201CJ021" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A201CJ024/PID_01.PV_Out#Value", "d": "A201CJ024" },
|
{ "ns": 1, "n": "s=t|SERVER::A201CJ022/MMON_01.PV_Out#Value", "d": "A201CJ022" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A201CJ025/PID_01.PV_Out#Value", "d": "A201CJ025" },
|
{ "ns": 1, "n": "s=t|SERVER::A201CJ023/MMON_01.PV_Out#Value", "d": "A201CJ023" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A201CD123/MOT_01.AV_Out#Value", "d": "A201CD123" },
|
{ "ns": 1, "n": "s=t|SERVER::A201CJ024/PID_01.PV_Out#Value", "d": "A201CJ024" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A201CD121/MOT_01.AV_Out#Value", "d": "A201CD121" },
|
{ "ns": 1, "n": "s=t|SERVER::A201CJ025/PID_01.PV_Out#Value", "d": "A201CJ025" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A212DD110/MOT_01.AV_Out#Value", "d": "A212DD110" },
|
{ "ns": 1, "n": "s=t|SERVER::A201CD123/MOT_01.AV_Out#Value", "d": "A201CD123" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A212DD130/MOT_01.AV_Out#Value", "d": "A212DD130" },
|
{ "ns": 1, "n": "s=t|SERVER::A201CD121/MOT_01.AV_Out#Value", "d": "A201CD121" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A212DD131/MOT_01.AV_Out#Value", "d": "A212DD131" },
|
{ "ns": 1, "n": "s=t|SERVER::A212DD110/MOT_01.AV_Out#Value", "d": "A212DD110" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A212DD111/MOT_01.AV_Out#Value", "d": "A212DD111" },
|
{ "ns": 1, "n": "s=t|SERVER::A212DD130/MOT_01.AV_Out#Value", "d": "A212DD130" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A212DD113/MOT_01.AV_Out#Value", "d": "A212DD113" },
|
{ "ns": 1, "n": "s=t|SERVER::A212DD131/MOT_01.AV_Out#Value", "d": "A212DD131" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A212DJ004/PID_01.PV_Out#Value", "d": "A212DJ004" },
|
{ "ns": 1, "n": "s=t|SERVER::A212DD111/MOT_01.AV_Out#Value", "d": "A212DD111" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A212DJ021/PID_01.PV_Out#Value", "d": "A212DJ021" },
|
{ "ns": 1, "n": "s=t|SERVER::A212DD113/MOT_01.AV_Out#Value", "d": "A212DD113" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A212DJ001/PID_01.PV_Out#Value", "d": "A212DJ001" },
|
{ "ns": 1, "n": "s=t|SERVER::A212DJ004/PID_01.PV_Out#Value", "d": "A212DJ004" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A212DJ011/PID_01.PV_Out#Value", "d": "A212DJ011" },
|
{ "ns": 1, "n": "s=t|SERVER::A212DJ021/PID_01.PV_Out#Value", "d": "A212DJ021" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A212DJ032/MMON_01.PV_Out#Value", "d": "A212DJ032" },
|
{ "ns": 1, "n": "s=t|SERVER::A212DJ001/PID_01.PV_Out#Value", "d": "A212DJ001" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A212DJ031/MMON_01.PV_Out#Value", "d": "A212DJ031" },
|
{ "ns": 1, "n": "s=t|SERVER::A212DJ011/PID_01.PV_Out#Value", "d": "A212DJ011" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A212DJ033/MMON_01.PV_Out#Value", "d": "A212DJ033" },
|
{ "ns": 1, "n": "s=t|SERVER::A212DJ032/MMON_01.PV_Out#Value", "d": "A212DJ032" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A212DJ010/MMON_01.PV_Out#Value", "d": "A212DJ010" },
|
{ "ns": 1, "n": "s=t|SERVER::A212DJ031/MMON_01.PV_Out#Value", "d": "A212DJ031" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A212DJ042/MMON_01.PV_Out#Value", "d": "A212DJ042" },
|
{ "ns": 1, "n": "s=t|SERVER::A212DJ033/MMON_01.PV_Out#Value", "d": "A212DJ033" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A214BJ055/PID_01.PV_Out#Value", "d": "A214BJ055" },
|
{ "ns": 1, "n": "s=t|SERVER::A212DJ010/MMON_01.PV_Out#Value", "d": "A212DJ010" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A214BJ065/PID_01.PV_Out#Value", "d": "A214BJ065" },
|
{ "ns": 1, "n": "s=t|SERVER::A212DJ042/MMON_01.PV_Out#Value", "d": "A212DJ042" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A212BJ010/MMON_01.PV_Out#Value", "d": "A212BJ010" },
|
{ "ns": 1, "n": "s=t|SERVER::A214BJ055/PID_01.PV_Out#Value", "d": "A214BJ055" },
|
||||||
{ "ns": 1, "n": "s=t|SERVER::A212BJ010/MMON_02.PV_Out#Value", "d": "A212BJ010" }
|
{ "ns": 1, "n": "s=t|SERVER::A214BJ065/PID_01.PV_Out#Value", "d": "A214BJ065" },
|
||||||
]
|
{ "ns": 1, "n": "s=t|SERVER::A212BJ010/MMON_01.PV_Out#Value", "d": "A212BJ010" },
|
||||||
}
|
{ "ns": 1, "n": "s=t|SERVER::A212BJ010/MMON_02.PV_Out#Value", "d": "A212BJ010" }
|
||||||
]
|
]
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
20
src/opcua2mqtt-bridge.service
Normal file
20
src/opcua2mqtt-bridge.service
Normal file
@ -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
|
||||||
|
|
||||||
|
|
Reference in New Issue
Block a user