Eine Brücke von EtherNet/IP zu MQTT holt Daten aus Rockwell-Logix-Steuerungen, Antrieben und anderen CIP-Geräten in Dashboards, Historians und Cloud-Anwendungen, ohne dass Sie das Steuerungsprogramm anfassen müssen. In diesem Tutorial lernen Sie den Unterschied zwischen Implicit und Explicit Messaging kennen und nutzen anschließend die Open-Source-Bibliothek pycomm3 auf einem Gateway, um Logix-Tags und generische CIP-Attribute zu lesen und sie mit sauberen, selbstbeschreibenden Payloads über MQTT zu veröffentlichen.
EtherNet/IP und CIP in fünf Minuten
EtherNet/IP ist das Common Industrial Protocol (CIP), übertragen über Standard-Ethernet und TCP/UDP. CIP bildet jedes Gerät als Menge von Objekten mit Klassen, Instanzen und Attributen ab. Ein Antrieb besitzt zum Beispiel ein Identity-Objekt (Klasse 0x01), das Hersteller, Produktcode und Seriennummer beschreibt, sowie Applikationsobjekte für Drehzahl, Strom und Status.
CIP-Verkehr gibt es in zwei Varianten:
| Messaging-Typ | Transport | Muster | Typischer Einsatz | Für ein Gateway geeignet? |
|---|---|---|---|---|
| Implicit (I/O) | UDP, Port 2222 | Zyklische, geplante Verbindungen zwischen Scanner und Adapter | Echtzeitsteuerung von I/O und Antrieben | Selten; erfordert, Scanner oder konfigurierter Adapter zu sein |
| Explicit | TCP, Port 44818 | Request/Response | Lesen von Tags, Attributen, Diagnosedaten | Ja, das ist der übliche Ansatz |
Ein Monitoring-Gateway sollte Explicit Messaging verwenden. Es stört die I/O-Verbindungen der Steuerung nicht, erfordert keine Änderungen am SPS-Projekt und lässt sich ohne Stillstand hinzufügen oder entfernen. Der Preis dafür: Es handelt sich um ein Polling-Protokoll, Sie legen also selbst fest, wie oft gelesen wird.
Zwei Arten von Zielgeräten
- Logix-Steuerungen (ControlLogix, CompactLogix, eingeschränkt auch Micro800). Sie unterstützen symbolischen Tag-Zugriff: Sie lesen
Line1_SpeedoderConveyor[3].Statusper Name. Das ist mit Abstand die bequemste Datenquelle. - Generische CIP-Geräte (Antriebe, I/O-Adapter, Scanner, Fremdgeräte). Sie stellen Objektattribute bereit, die Sie über Klassen-, Instanz- und Attributnummer aus der Gerätedokumentation oder der EDS-Datei lesen.
pycomm3 deckt beides ab: LogixDriver für tagbasierte Steuerungen und CIPDriver für generische Explicit Messages. pylogix ist eine weitere beliebte Open-Source-Option für Logix-Tags. Siemens, Beckhoff und andere Plattformen haben eigene Zugriffswege; für Rockwell-spezifische Themen wie die Steuerungskonfiguration lesen Sie unseren Leitfaden Allen-Bradley und MQTT.
Voraussetzungen
- Python 3.10 oder neuer auf einem Gateway, das die Steuerung über TCP 44818 erreicht.
pycomm3undpaho-mqtt2.x.- Die IP-Adresse der Steuerung und bei Chassis-basierten Steuerungen die Steckplatznummer (zum Beispiel
10.0.0.20/0für Slot 0). - Eine mit dem Steuerungstechniker abgestimmte, schreibgeschützte Tag-Liste.
python -m venv .venv && source .venv/bin/activate
pip install pycomm3 "paho-mqtt>=2.0"
Schritt 1: Die Steuerung erkunden
Bevor Sie die Brücke schreiben, prüfen Sie die Verbindung und listen die verfügbaren Tags auf. Dieses Skript gibt Steuerungsinformationen und die ersten Controller-Scope-Tags aus.
# explore_logix.py - Verbindung prüfen und Tags auflisten
from pycomm3 import LogixDriver
with LogixDriver("10.0.0.20/0") as plc: # IP/Slot der Steuerung
print(plc.info) # Hersteller, Produkt, Revision, Name
tags = plc.get_tag_list() # Controller-Scope-Tags
for t in tags[:30]:
print(t["tag_name"], t["data_type_name"])
get_tag_list() lädt die Tag-Definitionen einschließlich benutzerdefinierter Datentypen hoch und kann bei großen Projekten daher einen Moment dauern. Programm-Scope-Tags lesen Sie mit der Syntax Program:MainProgram.TagName.
Schritt 2: Ein generisches CIP-Attribut lesen
Für einen Antrieb oder ein beliebiges CIP-Gerät ohne symbolische Tags verwenden Sie Explicit Messages. Das Beispiel liest den Produktnamen aus dem Identity-Objekt, das jedes EtherNet/IP-Gerät implementiert.
# identity.py - Attribut 7 (Product Name) des Identity-Objekts von einem beliebigen CIP-Gerät lesen
from pycomm3 import CIPDriver, Services, ClassCode, SHORT_STRING
with CIPDriver("10.0.0.31") as dev:
resp = dev.generic_message(
service=Services.get_attribute_single,
class_code=ClassCode.identity_object, # 0x01
instance=1,
attribute=7, # Product Name
data_type=SHORT_STRING,
connected=False,
name="product_name",
)
print(resp.value if not resp.error else resp.error)
Ersetzen Sie Klasse, Instanz, Attribut und Datentyp durch Werte aus dem Gerätehandbuch, um Applikationsdaten wie Ausgangsfrequenz oder Motorstrom zu lesen.
Schritt 3: Die Brücke von EtherNet/IP zu MQTT
Die Brücke liest in jedem Zyklus eine Tag-Liste mit einer einzigen Multi-Tag-Anfrage, veröffentlicht Werte, die sich geändert haben (oder ein Totband überschreiten), und meldet Lesefehler als schlechte Qualität, statt stillschweigend alte Werte zu behalten.
# eip_mqtt_bridge.py - Logix-Tags über EtherNet/IP pollen und an MQTT veröffentlichen
import json, logging, time, uuid
import paho.mqtt.client as mqtt
from pycomm3 import LogixDriver, CommError
logging.basicConfig(level=logging.INFO)
log = logging.getLogger("eip")
PLC_PATH = "10.0.0.20/0"
POLL_SECONDS = 1.0
TOPIC_ROOT = "connectmq/demo/9e6d/plant1/packaging/line2" # Ihre eigene Zufalls-ID
MQTT_HOST, MQTT_PORT = "broker.connectmq.com", 1883 # Produktion: 8883 mit TLS
MQTT_USER, MQTT_PASS = None, None # Produktion: "YOUR_USERNAME", "YOUR_PASSWORD"
CLIENT_ID = f"eip-bridge-{uuid.uuid4().hex[:8]}" # Produktion: "YOUR_CLIENT_ID"
# SPS-Tag -> (Topic-Blatt, Einheit, Totband)
TAGS = {
"Line2_Speed": ("speed", "ppm", 1.0),
"Line2_Running": ("running", "", 0),
"Line2_GoodCount": ("good_count", "count", 0),
"Line2_RejectCount": ("reject_count", "count", 0),
"Filler.MotorCurrent": ("filler_current", "A", 0.2),
}
def mqtt_client():
c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
if MQTT_USER:
c.username_pw_set(MQTT_USER, MQTT_PASS)
c.tls_set()
c.will_set(f"{TOPIC_ROOT}/bridge", "offline", qos=1, retain=True)
c.on_connect = lambda cl, u, f, rc, p: cl.publish(f"{TOPIC_ROOT}/bridge", "online", qos=1, retain=True)
c.connect(MQTT_HOST, MQTT_PORT, keepalive=30)
c.loop_start()
return c
def changed(old, new, deadband):
if old is None:
return True
if isinstance(new, (int, float)) and not isinstance(new, bool) and deadband:
return abs(new - old) >= deadband
return new != old
def main():
mq = mqtt_client()
last = {}
while True:
try:
with LogixDriver(PLC_PATH) as plc:
log.info("connected to %s", plc.info.get("product_name"))
while True:
results = plc.read(*TAGS.keys())
if not isinstance(results, list):
results = [results]
now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
for r in results:
leaf, unit, deadband = TAGS[r.tag]
topic = f"{TOPIC_ROOT}/{leaf}"
if r.error:
if last.get(r.tag, {}).get("quality") != "bad":
mq.publish(topic, json.dumps({"value": None, "quality": "bad",
"error": r.error, "ts": now}), qos=1, retain=True)
last[r.tag] = {"value": None, "quality": "bad"}
continue
prev = last.get(r.tag, {})
if prev.get("quality") == "bad" or changed(prev.get("value"), r.value, deadband):
payload = {"value": r.value, "unit": unit, "type": r.type,
"quality": "good", "ts": now}
mq.publish(topic, json.dumps(payload), qos=1, retain=True)
last[r.tag] = {"value": r.value, "quality": "good"}
time.sleep(POLL_SECONDS)
except CommError as exc:
log.error("EtherNet/IP connection lost: %s - retrying in 5 s", exc)
time.sleep(5)
if __name__ == "__main__":
main()
plc.read() mit mehreren Tag-Namen packt diese in so wenige Anfragen, wie die Verbindungsgröße zulässt. Das belastet die Steuerung deutlich weniger als eine Anfrage pro Tag. Jedes Ergebnis ist ein Tag mit den Feldern tag, value, type und error, sodass die Brücke den Datentyp zusammen mit dem Wert veröffentlichen kann.
Schritt 4: Starten und prüfen
python eip_mqtt_bridge.py
mosquitto_sub -h broker.connectmq.com -p 1883 -i eip-viewer-9e6d -t "connectmq/demo/9e6d/#" -v
connectmq/demo/9e6d/plant1/packaging/line2/speed {"value": 118.0, "unit": "ppm", "type": "REAL", "quality": "good", "ts": "2026-10-16T12:00:01Z"}
connectmq/demo/9e6d/plant1/packaging/line2/running {"value": true, "unit": "", "type": "BOOL", "quality": "good", "ts": "2026-10-16T12:00:01Z"}
Der Topic-Baum folgt einem Aufbau im ISA-95-Stil (Standort, Bereich, Linie, Datenpunkt), sodass dieselben Konsumenten funktionieren, egal ob die Daten aus EtherNet/IP, Modbus oder OPC UA stammen. Genau das ist die Idee eines Unified Namespace (UNS): Die ISA-95-Hierarchie (Enterprise → Site → Area → Line → Cell) wird zur Topic-Struktur auf dem MQTT-Broker, etwa acme/werk-stuttgart/verpackung/linie-2/fueller/motorstrom, und MES, SCADA oder OEE-Dashboards abonnieren die Daten dort, unabhängig davon, ob sie aus einer Rockwell-Steuerung, einer Siemens S7 oder einem OPC-UA-Server stammen. Wie Sparkplug B dazu passt, zeigt der Beitrag zu Sparkplug B und Unified Namespace. Unser Leitfaden zu ISA-95-MQTT-Topics erklärt die Konventionen.
Tags und Polling-Raten wählen
Nicht jeder Tag in einem Logix-Projekt verdient ein eigenes Topic. Gehen Sie von den Fragen der Konsumenten aus: Läuft die Linie, wie schnell, wie viele Gut- und Schlechtteile, welche Alarme sind aktiv und wie viel Energie ziehen die Hauptantriebe? Das sind meist einige Dutzend Werte pro Linie, nicht Tausende. Gruppieren Sie sie danach, wie schnell sie sich ändern. Zustände und Zähler können jede Sekunde gelesen und bei Änderung veröffentlicht werden; langsame Analogwerte wie Tankfüllstände oder Temperaturen alle fünf oder zehn Sekunden mit Totband. Ist ein Wert nur als Schichtsumme nützlich, lassen Sie ihn besser von der SPS aufsummieren und veröffentlichen ihn in einem langsameren Zyklus.
Bewährte Praktiken für den Produktivbetrieb
- Grundsätzlich nur lesend. Fügen Sie einer Monitoring-Brücke keine Schreibfunktion hinzu. Wenn Sie jemals Befehle benötigen, bauen Sie einen separaten Dienst mit expliziter Freigabelogik.
- Einen Schnittstellen-Tag-Satz verwenden. Bitten Sie den Steuerungstechniker, die benötigten Werte in einen eigenen UDT oder ein Array zu kopieren. Dann lesen Sie eine Struktur statt Dutzender verstreuter Tags, und Programmänderungen legen die Brücke nicht lahm.
- Verbindungslimits beachten. Steuerungen unterstützen nur eine begrenzte Zahl von CIP-Verbindungen, die sie mit HMIs, SCADA und Engineering-Tools teilen. Ideal ist eine langlebige Verbindung pro Brücke; vermeiden Sie einen Neuaufbau in jedem Zyklus.
- So schnell pollen wie nötig. Eine Sekunde reicht für Produktionszähler und Zustände völlig aus. Schnelleres Polling erzeugt Last, aber keine zusätzlichen Erkenntnisse.
- Netzwerksegmentierung. Platzieren Sie das Gateway im OT-Netz oder in einer DMZ mit Zugriff nur auf Port 44818 der Steuerungen und ausgehend auf 8883 zum Broker. Unser Artikel zur MQTT-DMZ beschreibt den Aufbau.
- Privater Broker für echte Daten. Der öffentliche MQTT-Broker ist nur für Tests und Prototypen gedacht und für jeden lesbar. Wechseln Sie zu einem verwalteten MQTT-Broker mit TLS und Zugangsdaten pro Broker, bevor Sie eine echte Linie anbinden.
Fehlerbehebung
| Symptom | Wahrscheinliche Ursache | Lösung |
|---|---|---|
CommError beim Verbinden |
Falsche IP, Firewall blockiert 44818, falscher Slot | Steuerung anpingen, Pfad ip/slot prüfen |
Tag-Lesevorgänge liefern Tag doesn't exist |
Programm-Scope-Tag oder Tippfehler | Program:Name.Tag verwenden, get_tag_list() prüfen |
| Läuft, fällt nach Stunden aus | Verbindung von Steuerung oder Switch getrennt | Die Wiederholungsschleife verbindet neu; Verbindungsauslastung der Steuerung prüfen |
| Generische Nachricht liefert CIP-Statusfehler | Falsche Klasse, Instanz, falsches Attribut oder falscher Datentyp | Gerätehandbuch oder EDS-Datei prüfen |
| Werte aktualisieren sich langsam | Große Tag-Liste in viele Anfragen aufgeteilt | Schnittstellen-UDT verwenden, Tags reduzieren |
Wo ConnectMQ ins Spiel kommt
ConnectMQ ist die Cloud-Seite dieser Pipeline: ein verwalteter MQTT-Broker mit TLS auf Port 8883 und WebSockets für Browser-Dashboards. Er erkennt ISA-95- und UNS-Topic-Strukturen, sodass die Geräteansicht Linien und Maschinen automatisch gruppiert, und Schwellwert-Alarmregeln können eine Ausschussmenge oder einen Motorstrom außerhalb der Grenzwerte melden. Wenn verwaltete MQTT-Broker neu für Sie sind, sehen Sie sich an, wie ein Cloud-MQTT-Broker im Vergleich zu einem selbst betriebenen Broker abschneidet; die vollständige Architektur finden Sie im Leitfaden zu Industrial IoT mit MQTT.
Starten Sie mit ConnectMQ
Binden Sie noch heute Ihre erste EtherNet/IP-Linie an. Registrieren Sie sich unter app.connectmq.com, wählen Sie den kostenlosen Serverless-Plan ohne Kreditkarte, erstellen Sie einen Broker und tragen Sie den generierten Benutzernamen, das Passwort und die Client-ID in die Brücke ein. Ihre Logix-Tags erscheinen innerhalb von Sekunden im Topic-Explorer von ConnectMQ.
Häufig gestellte Fragen
Beeinflusst das Lesen von Tags über EtherNet/IP das SPS-Programm?
Explicit Messaging nutzt die Kommunikationsressourcen der Steuerung, nicht ihre I/O-Verbindungen, und verändert das Programm nicht. Halten Sie das Polling moderat und die Zahl der Verbindungen gering, damit HMIs und SCADA nicht beeinträchtigt werden.
Kann ich Implicit-I/O-Messaging für ein MQTT-Gateway verwenden?
Das ist möglich, wenn das Gateway als Adapter im I/O-Baum der SPS konfiguriert ist, erfordert aber eine Projektänderung. Explizite Tag-Lesevorgänge sind für das Monitoring einfacher und greifen nicht in die Steuerung ein.
Funktioniert pycomm3 mit Micro800 und älteren PLC-5- oder SLC-Steuerungen?
pycomm3 zielt auf Logix-Steuerungen und unterstützt Micro800 teilweise; die älteren Familien PLC-5 und SLC verwenden eine andere Adressierung. Prüfen Sie die Dokumentation der Bibliothek für Ihre konkrete Steuerung.
Welchen Port verwendet EtherNet/IP Explicit Messaging?
TCP-Port 44818. Implicit-I/O-Verkehr nutzt UDP-Port 2222.
