OPC-UA-zu-MQTT-Bridge für Industriedaten

Eine OPC-UA-zu-MQTT-Bridge verbindet sich als Client mit einem OPC-UA-Server, abonniert die Tags, die dich interessieren, und publisht jede Datenänderung erneut als MQTT-Nachricht auf einem sauber benannten Topic. Die Anlage behält OPC UA dort, wo es stark ist – reichhaltige Modelle, Security, Maschinenintegration –, während Dashboards, Historians und Cloud-Dienste einen schlanken Publish/Subscribe-Feed bekommen, den sie frei skalieren können.

Dieses Tutorial baut eine solche Bridge in Python mit asyncua und paho-mqtt, mappt OPC-UA-Node-IDs auf einen Unified-Namespace-Topic-Baum und zeigt, wie du sie mit einem simulierten OPC-UA-Server testest.

Warum bridgen statt direkt verbinden

OPC UA ist Client/Server: Jeder Konsument öffnet eine Session zu jedem Server, den er braucht. Bei fünf Konsumenten und zwanzig Servern sind das hundert Sessions, jede mit eigenen Subscriptions, Zertifikaten und Last auf dem Server. Eine Bridge macht daraus einen Hub:

Standardmäßig ist die Bridge eine Einbahnstraße: OPC UA rein, MQTT raus. So bleibt die OT-Seite geschützt, während die IT die Daten bekommt. Einen breiteren Vergleich beider Protokolle findest du unter MQTT vs. OPC UA.

1. Einen Test-OPC-UA-Server starten

Entwickle nicht gegen einen Produktionsserver. Microsofts Open-Source-Simulator OPC PLC läuft in Docker und liefert sich ändernde Werte:

docker run --rm -p 50000:50000 mcr.microsoft.com/iotedge/opc-plc:latest \
  --pn=50000 --autoaccept --unsecuretransport

Der Endpoint ist opc.tcp://localhost:50000. Durchsuche ihn mit einem GUI-Client wie UaExpert oder mit dem Befehl uals, der mit asyncua kommt, und notiere die Node-IDs, die du bridgen willst. Der Simulator bietet Nodes wie StepUp, AlternatingBoolean und RandomSignedInt32 in seinem eigenen Namespace an; prüfe den Namespace-Index im Browser, denn er kann sich zwischen Versionen unterscheiden.

2. Nodes auf Topics mappen

Das Mapping ist die eigentliche Designarbeit. Ein Topic sollte sagen, wo der Wert in der Anlage lebt – nicht, welcher Server oder welche Node-ID ihn heute zufällig liefert. Halte das Mapping in der Konfiguration:

BASE = "testmqtt/acme/plant1/packaging/line1"

NODES = {
    "ns=3;s=StepUp":             f"{BASE}/filler/counter",
    "ns=3;s=AlternatingBoolean": f"{BASE}/filler/running",
    "ns=3;s=RandomSignedInt32":  f"{BASE}/capper/torque",
}

Wenn du später den simulierten Server durch den echten ersetzt – oder einen Tag auf eine andere SPS verschiebst –, ändert sich nur die linke Seite des Mappings. Alle Subscriber funktionieren weiter. Das ist die Kernidee eines Unified Namespace: Der Topic-Baum ist der Vertrag, die Quellen sind Implementierungsdetails.

3. Die Bridge

Die Subscription-API von asyncua ruft einen Handler auf, sobald sich ein Monitored Item ändert. Der Handler publisht den Wert, seine OPC-UA-Qualität und den Quellzeitstempel:

import asyncio, json, os
import paho.mqtt.client as mqtt
from asyncua import Client
from nodes import BASE, NODES

OPC_URL = os.getenv("OPC_URL", "opc.tcp://localhost:50000")
STATUS = f"{BASE}/_bridge/status"

mq = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="opcua-bridge-line1")
mq.username_pw_set(os.getenv("MQTT_USER"), os.getenv("MQTT_PASS"))
mq.tls_set()
mq.will_set(STATUS, "offline", qos=1, retain=True)
mq.connect("broker.connectmq.com", 8883, keepalive=30)
mq.loop_start()

class Handler:
    def datachange_notification(self, node, val, data):
        dv = data.monitored_item.Value
        ts = dv.SourceTimestamp or dv.ServerTimestamp
        payload = {
            "v": val,
            "good": dv.StatusCode.is_good(),
            "ts": ts.isoformat() if ts else None,
        }
        topic = NODES[node.nodeid.to_string()]
        mq.publish(topic, json.dumps(payload, default=str), qos=1, retain=True)

async def run():
    while True:
        try:
            async with Client(url=OPC_URL, timeout=10) as opc:
                nodes = [opc.get_node(nid) for nid in NODES]
                sub = await opc.create_subscription(500, Handler())   # 500 ms Publishing-Intervall
                await sub.subscribe_data_change(nodes)
                mq.publish(STATUS, "online", qos=1, retain=True)
                print("Bridge aktiv für", len(nodes), "Nodes")
                while True:
                    await asyncio.sleep(5)
                    await opc.check_connection()
        except Exception as exc:          # Verbindung verloren, Server-Neustart, falsche Node-ID
            print("OPC-UA-Fehler:", exc)
            mq.publish(STATUS, "opcua-unreachable", qos=1, retain=True)
            await asyncio.sleep(5)

asyncio.run(run())

pip install "asyncua>=1.1" "paho-mqtt>=2.0"
python opcua_mqtt_bridge.py

Was dir dieses Design bringt:

• Änderungsgetriebener Traffic. OPC UA sendet Notifications nur bei Wertänderungen, die MQTT-Seite trägt also keine redundanten Samples. Stimme das Publishing-Intervall (hier 500 ms) und das Sampling-Intervall pro Item auf deinen Prozess ab.
• Qualität reist mit dem Wert. Ein schlechter oder unsicherer OPC-UA-Statuscode ist Information; behalte ihn im Payload, statt die Nachricht stillschweigend zu verwerfen.
• Quellzeitstempel. Historians sollten speichern, wann sich der Wert an der Quelle geändert hat, nicht wann die Bridge ihn weitergeleitet hat.
• Retained Werte und ein Status-Topic. Neue Subscriber sehen sofort den aktuellen Zustand, und das Last Will setzt den Bridge-Status auf offline, wenn der Prozess stirbt. Mehr dazu unter Retained Messages.
• Automatische Wiederherstellung. Startet der OPC-UA-Server neu, verbindet die äußere Schleife sich erneut und legt die Subscription neu an. paho verbindet sich nach loop_start() selbstständig wieder mit dem Broker.

4. Die Bridge testen

1. Starte Simulator und Bridge. Die Bridge sollte die Zahl der gebridgten Nodes ausgeben.
2. Öffne den Online-MQTT-Client, verbinde dich mit broker.connectmq.com und abonniere testmqtt/acme/plant1/#. filler/counter sollte hochzählen, filler/running wechseln und _bridge/status online zeigen.
3. Stoppe den Simulator-Container. Innerhalb weniger Sekunden sollte der Status opcua-unreachable werden; starte ihn neu, und die Werte sollten ohne Neustart der Bridge wiederkommen.
4. Beende die Bridge mit kill -9. Nach dem Keep-Alive-Timeout publisht der Broker das Last Will: offline.

Gebridgte OPC-UA-Tags im Browser beobachten

Prüfe mit dem Topic Matcher, ob die Topic-Filter der Konsumenten zu dem passen, was die Bridge publisht, und denk an große strukturierte Werte (Arrays, Extension Objects): Der Paketgrößen-Rechner hilft abzuschätzen, was eine Notification auf der Leitung kostet.

Andere Wege zur Bridge

Eine kleine Python-Bridge ist ideal zum Lernen, für eine Handvoll Server und für volle Kontrolle über das Mapping. Für größere Landschaften lohnen sich:

Welches Werkzeug du auch wählst, es gelten dieselben Regeln: Topics beschreiben die Anlage, Payloads tragen Wert + Qualität + Zeitstempel, die Bridge meldet ihren eigenen Zustand, und Schreibzugriffe zurück nach OPC UA sind ein separater, bewusst abgesicherter Pfad.

Produktions-Checkliste

• Nutze auf echten Servern OPC-UA-Security (Basic256Sha256 mit Signatur und Verschlüsselung) und ein eigenes Anwendungszertifikat für die Bridge; die unsichere Simulator-Einstellung ist nur für lokale Tests.
• Lies die Server-Limits für Monitored Items pro Subscription und verteile große Tag-Listen auf mehrere Subscriptions.
• Betreibe eine Bridge pro Bereich oder Linie, damit ein Ausfall lokal bleibt.
• Versioniere das Node-zu-Topic-Mapping und reviewe es wie Code.
• Überwache das Status-Topic der Bridge und alarmiere, wenn es nicht online ist.

Mit einer OPC-UA-zu-MQTT-Bridge behältst du OPC UA für das, was es am besten kann, und gibst allen anderen Systemen einen einfachen, skalierbaren Feed. Mappe Nodes auf anlagenförmige Topics, publishe Änderungs-Notifications mit Qualität und Zeitstempeln, teste gegen einen Simulator und den Online-Client – und deine OT-Daten werden zu etwas, das das ganze Unternehmen abonnieren kann.

Häufig gestellte Fragen