Allen-Bradley MQTT: Tags aus ControlLogix und CompactLogix veröffentlichen

Eine Allen-Bradley-MQTT-Integration ist der schnellste Weg, Tag-Daten aus ControlLogix und CompactLogix in Dashboards, Historians und Cloud-Analysen zu bringen, ohne Ihren SCADA-Server zusätzlich zu belasten. In diesem Tutorial bauen Sie ein kleines Python-Gateway, das Logix-Tags mit pycomm3 über EtherNet/IP liest, sie auf ISA-95-Topics abbildet und nur relevante Änderungen an einen MQTT-Broker veröffentlicht. Sie testen zunächst gegen den öffentlichen MQTT-Broker und wechseln dann zu einem privaten mit TLS. So wird die SPS zur Datenquelle eines Unified Namespace (UNS), wie ihn viele Industrie-4.0-Projekte im Maschinenbau und in der Fertigung anstreben.

Wie Allen-Bradley-SPSen zu MQTT passen

Logix-Steuerungen kommunizieren nativ über EtherNet/IP mit CIP (Common Industrial Protocol). Studio-5000-Projekte organisieren Daten als benannte Tags, entweder im Controller- oder im Programm-Scope, mit atomaren Typen (BOOL, DINT, REAL), Arrays und benutzerdefinierten Typen (UDTs). Dieses Modell benannter Tags ist ein Glücksfall für die MQTT-Integration: Anders als bei Modbus brauchen Sie keine Registertabelle, denn der Tag-Name trägt bereits die Bedeutung.

Es gibt drei gängige Wege, diese Tags zu MQTT zu bringen:

Dieses Tutorial wählt den ersten Ansatz, weil er jedes bewegliche Teil zeigt. Dasselbe Topic- und Payload-Design gilt, wenn Sie später zu einem kommerziellen Treiber oder zu Ignition mit MQTT wechseln.

Voraussetzungen

• Eine ControlLogix- oder CompactLogix-Steuerung, die vom Gateway über TCP-Port 44818 erreichbar ist.
• Ein Linux- oder Windows-Rechner mit Python 3.9 oder neuer.
• Einige Tags zum Lesen, zum Beispiel Line1_Temp (REAL), Line1_Speed (DINT), Line1_Running (BOOL) und ein Programm-Scope-Tag Program:MainProgram.Pump1_Pressure (REAL).
• Die Mosquitto-Clients zum Testen.

Installieren Sie die Bibliotheken:

python -m venv abmqtt
source abmqtt/bin/activate
pip install pycomm3 "paho-mqtt>=2.0"

Schritt 1: Die Tag-Liste ermitteln

Bevor Sie das Gateway schreiben, prüfen Sie die Verbindung und sehen nach, wie die Tags genau heißen. pycomm3 lädt beim Verbinden die Tag-Datenbank der Steuerung hoch:

from pycomm3 import LogixDriver

# CompactLogix: nur die IP. ControlLogix mit der CPU in Slot 3: "192.168.1.20/3"
PLC_PATH = "192.168.1.20"

with LogixDriver(PLC_PATH) as plc:
    print(plc.info)                       # Steuerungsname, Revision, Seriennummer
    for name, definition in plc.tags.items():
        print(name, definition["data_type_name"], definition.get("dim", 0))

Programm-Scope-Tags sind standardmäßig nicht in der Controller-Scope-Liste enthalten; adressieren Sie sie beim Lesen als Program:MainProgram.TagName. Bei ControlLogix-Chassis ist der Pfad nach dem Schrägstrich der Steckplatz der Steuerung im Baugruppenträger. Ist er falsch, öffnet sich die Verbindung zum Ethernet-Modul, aber die Tag-Lesevorgänge schlagen fehl.

Die Dokumentation von pycomm3 behandelt Pfadsyntax, Arrays und UDT-Verarbeitung im Detail.

Schritt 2: ISA-95-Topics entwerfen

Bilden Sie jeden Tag auf ein Topic ab, das die Anlage beschreibt, nicht die Steuerung. Eine einfache Zuordnungstabelle im Gateway hält SPS-Namenskonventionen aus Ihrem MQTT-Namensraum heraus:

Mit einem Präfix für Standort und Bereich wird ein vollständiges Topic zu plant1/assembly/line1/oven/temperature. Solche konsistenten Pfade machen einen werksweiten Namensraum erst möglich; eine tiefere Behandlung von Ebenen und Benennung finden Sie unter ISA-95-MQTT-Topics.

Genau dieses Muster ist der Kern eines UNS: Die ISA-95-Anlagenhierarchie (Enterprise → Site → Area → Line → Cell) wird direkt zur Topic-Struktur auf dem MQTT-Broker, etwa acme/werk-stuttgart/montage/linie-1/zelle-3/temperatur. Jedes Gateway, ob für Allen-Bradley, Siemens S7 oder OPC UA, schreibt in denselben Baum, und MES, ERP oder Historian finden die Daten über den Ort in der Anlage statt über die Steuerung. Wie Sparkplug B dabei hilft, beschreibt der Beitrag zu Sparkplug B und Unified Namespace.

Schritt 3: Das Allen-Bradley-MQTT-Gateway

Das Gateway liest jede Sekunde alle zugeordneten Tags mit einer einzigen Multi-Tag-Anfrage, wendet Totbänder pro Tag an, veröffentlicht Änderungen mit QoS 1, aktualisiert alle fünf Minuten sämtliche Werte und pflegt einen Retained-Online/Offline-Status mit Last Will.

import json
import time
import uuid
from datetime import datetime, timezone

from pycomm3 import LogixDriver
import paho.mqtt.client as mqtt

PLC_PATH = "192.168.1.20"

# Zuerst der öffentliche MQTT-Broker. Er ist für jeden lesbar: nur Testdaten.
BROKER, PORT, USE_TLS = "broker.connectmq.com", 1883, False
USERNAME, PASSWORD = None, None                  # Produktion: "YOUR_USERNAME", "YOUR_PASSWORD"
CLIENT_ID = f"ab-gw-{uuid.uuid4().hex[:8]}"      # Produktion: "YOUR_CLIENT_ID"
BASE = f"connectmq/demo/{uuid.uuid4().hex[:6]}/plant1/assembly/line1"

TAGS = {
    "Line1_Temp": ("oven/temperature", "degC", 0.5),
    "Line1_Speed": ("conveyor/speed", "rpm", 5),
    "Line1_Running": ("line/running", None, 0),
    "Program:MainProgram.Pump1_Pressure": ("pump1/pressure", "bar", 0.05),
}


def on_connect(client, userdata, flags, reason_code, properties):
    print("MQTT connected:", reason_code)
    client.publish(f"{BASE}/gateway/status", "online", qos=1, retain=True)


def should_publish(tag, value, last, deadband):
    if tag not in last:
        return True
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return value != last[tag]
    return abs(value - last[tag]) >= deadband


def main():
    mq = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        mq.username_pw_set(USERNAME, PASSWORD)
    if USE_TLS:
        mq.tls_set()
    mq.will_set(f"{BASE}/gateway/status", "offline", qos=1, retain=True)
    mq.on_connect = on_connect
    mq.connect(BROKER, PORT, keepalive=30)
    mq.loop_start()
    print("Publishing under", BASE)

    plc = LogixDriver(PLC_PATH)
    last, last_full = {}, 0.0

    while True:
        try:
            if not plc.connected:
                plc.open()
            results = plc.read(*TAGS.keys())
            now = time.time()
            force = now - last_full > 300
            ts = datetime.now(timezone.utc).isoformat(timespec="seconds")

            for result in results:
                suffix, unit, deadband = TAGS[result.tag]
                if result.error:
                    mq.publish(f"{BASE}/{suffix}", json.dumps(
                        {"value": None, "quality": "bad", "error": result.error, "ts": ts}), qos=1)
                    continue
                value = result.value
                if isinstance(value, float):
                    value = round(value, 3)
                if force or should_publish(result.tag, value, last, deadband):
                    payload = {"value": value, "quality": "good", "ts": ts}
                    if unit:
                        payload["unit"] = unit
                    mq.publish(f"{BASE}/{suffix}", json.dumps(payload), qos=1)
                    last[result.tag] = value
            if force:
                last_full = now
        except Exception as exc:
            print("PLC communication failed:", exc)
            try:
                plc.close()
            except Exception:
                pass
            time.sleep(5)
        time.sleep(1)


if __name__ == "__main__":
    main()

Einige Designpunkte verdienen Beachtung:

• Ein Multi-Tag-Lesevorgang. pycomm3 packt mehrere Tags in so wenige CIP-Anfragen wie möglich. Das schont die Steuerung deutlich mehr, als jeden Tag einzeln zu lesen.
• Fehler pro Tag. Ein Tippfehler in einem Tag-Namen liefert nur für diesen Tag einen Fehler. Das Veröffentlichen von quality: bad macht das Problem sichtbar, statt einen Wert stillschweigend einzufrieren.
• UDTs funktionieren ebenfalls. Das Lesen eines UDT liefert ein Python-Dict, das json.dumps in eine strukturierte Payload verwandelt. Halten Sie UDT-Payloads klein und veröffentlichen Sie keine großen Arrays im Sekundentakt.

Schritt 4: Auf dem öffentlichen MQTT-Broker prüfen

Starten Sie das Gateway und abonnieren Sie sein Präfix:

mosquitto_sub -h broker.connectmq.com -p 1883 -i ab-watch-01 \
  -t 'connectmq/demo/+/plant1/assembly/line1/#' -v

Sie sollten sofort den Retained-Gateway-Status und den ersten vollständigen Wertesatz sehen. Schalten Sie Line1_Running in Studio 5000 um, und die Änderung kommt innerhalb einer Sekunde an. Den öffentlichen MQTT-Broker für diesen Schritt zu nutzen, ist in Ordnung, solange die Werte nicht sensibel sind; für Produktionsdaten ist er nicht gedacht.

Schritt 5: Produktivbetrieb auf einem privaten Broker

Für den Produktivbetrieb erstellen Sie einen privaten Broker im Portal von ConnectMQ und ändern vier Zeilen: Port 8883, USE_TLS = True, den ausgegebenen Benutzernamen und das Passwort sowie die ausgegebene Client-ID. Entfernen Sie das Präfix connectmq/demo/..., damit die Topics wie im UNS üblich auf Unternehmens- oder Standortebene beginnen. Da Client-IDs an ihre Zugangsdaten gebunden sind, geben Sie jedem Gateway einen eigenen Satz, statt einen über mehrere Standorte zu teilen.

Betreiben Sie das Gateway als Dienst, damit es automatisch neu startet:

[Unit]
Description=Allen-Bradley MQTT gateway
After=network-online.target

[Service]
ExecStart=/opt/abmqtt/bin/python /opt/abmqtt/gateway.py
Restart=always
RestartSec=5
User=abmqtt

[Install]
WantedBy=multi-user.target

Sobald Daten fließen, gruppiert die Geräteansicht im ConnectMQ-Portal die ISA-95-Topics nach Standort, Bereich und Linie, und Schwellwert-Alarmregeln können eine Temperatur oder einen Druck außerhalb der Grenzwerte melden. Historians, MES und Dashboards abonnieren wie jeder andere Client des MQTT-Brokers den passenden Ausschnitt des Unified Namespace.

Leseraten und Totbänder abstimmen

Leseintervall und Totbänder bestimmen zusammen, wie viel Verkehr das Gateway erzeugt, sowohl im Steuerungsnetz als auch auf dem Broker. Ein praktischer Weg, sie festzulegen:

• Vom Konsumenten ausgehen. Ein Dashboard, das ein Bediener alle paar Sekunden aktualisiert, braucht keine 100-ms-Daten. Ein Historian für Energieberichte kommt womöglich mit Minutenmittelwerten aus.
• Totbänder nach dem Sensorrauschen bemessen. Beobachten Sie einen Wert einige Minuten in der Trendansicht von Studio 5000. Ein Totband knapp oberhalb des Rauschbands unterdrückt Flattern, ohne echte Änderungen zu verbergen.
• Zustände und Alarme niemals mit Totband versehen. Boolesche Werte, Betriebsartworte und Alarmbits sollten bei jeder Änderung veröffentlicht werden, und ihre Topics eignen sich gut für Retained Messages, damit ein neuer Abonnent sofort den aktuellen Zustand sieht.
• Die periodische Aktualisierung beibehalten. Ohne sie sieht ein Wert, der tatsächlich stundenlang konstant bleibt, genauso aus wie ein Gateway, das stillschweigend ausgefallen ist.

Wächst Ihnen ein einzelnes Gateway später über den Kopf, betreiben Sie eines pro Bereich oder Linie statt eines riesigen Prozesses für das ganze Werk. Ausfälle bleiben begrenzt, und jede Instanz hat eine kleine, überprüfbare Tag-Liste.

Sicherheitsaspekte

CIP über EtherNet/IP hat in den meisten Installationen keine Authentifizierung, das Gateway ist also faktisch ein vertrauenswürdiger Engineering-Client. Platzieren Sie es im Steuerungsnetz hinter einer Firewall, erlauben Sie von dort nur ausgehendes TLS zum Broker, und lassen Sie MQTT-Befehle niemals ohne explizite Allow-List, Bereichsprüfungen und eine Verriegelung in der SPS-Logik in SPS-Tags schreiben. Lesen lässt sich gefahrlos automatisieren; Schreiben verdient ein Design-Review. Der Leitfaden EtherNet/IP zu MQTT behandelt die Netzwerksegmentierung für CIP-Verkehr ausführlicher, und der Leitfaden zu Industrial IoT mit MQTT erklärt, wo Gateways in einer vollständigen Architektur sitzen.

Fehlerbehebung

1. Verbindung läuft in einen Timeout. Prüfen Sie das Routing und ob TCP 44818 zwischen Gateway und Steuerung offen ist.
2. Verbunden, aber jeder Tag schlägt fehl. Falscher Slot in einem ControlLogix-Pfad, oder der Tag liegt im Programm-Scope und braucht das Präfix Program:.
3. REAL-Werte mit langen Nachkommastellen. Das ist die normale Float-Darstellung; runden Sie vor dem Veröffentlichen.
4. MQTT trennt sich alle paar Sekunden. Ein anderer Client verwendet dieselbe Client-ID.
5. Hohe Kommunikationslast der Steuerung. Erhöhen Sie das Leseintervall oder verkleinern Sie die Tag-Liste; die meisten Dashboards benötigen keine Daten im Subsekundenbereich.

Verbinden Sie Ihre Logix-Steuerungen mit ConnectMQ

Ihr Gateway ist bereit; geben Sie ihm einen sicheren Cloud-MQTT-Broker als Ziel. Registrieren Sie sich unter app.connectmq.com für einen kostenlosen Serverless-Plan ohne Kreditkarte, fügen Sie die generierten Zugangsdaten in das Skript ein und sehen Sie Ihre Allen-Bradley-Tags live erscheinen.

Häufig gestellte Fragen

Kann eine ControlLogix-SPS nativ per MQTT veröffentlichen?

Logix-Steuerungen enthalten in der Steuerung selbst keinen universellen MQTT-Client. Üblich ist ein Edge-Gateway, ein Konnektivitätsserver oder ein SCADA-Modul, das Tags über EtherNet/IP liest und veröffentlicht.

Funktioniert pycomm3 mit Micro800 oder SLC 500?

Der LogixDriver von pycomm3 zielt auf ControlLogix, CompactLogix und mit einigen Einschränkungen Micro800; ältere PLC-5 und SLC 500 nutzen einen separaten Treiber. Prüfen Sie in der Dokumentation der Bibliothek die unterstützten Funktionen je Familie.

Wie oft sollte ich Tags lesen?

Eine Sekunde reicht für die meisten Monitoring-Aufgaben völlig aus. Verwenden Sie Totbänder, damit nur echte Änderungen veröffentlicht werden, plus eine periodische vollständige Aktualisierung.

Sollte ich Sparkplug B statt JSON verwenden?

Sparkplug B ergänzt Birth- und Death-Zertifikate sowie eine kompakte Protobuf-Kodierung, was bei vielen Tags, bei der SCADA-Integration und beim Aufbau eines UNS hilft. Reines JSON ist für den Einstieg einfacher und leicht zu inspizieren.