Modbus-zu-MQTT-Gateway für SPS: ein Modbus-TCP-Poller in Python

Ein Modbus-zu-MQTT-Gateway liest Register einer SPS über Modbus TCP und publisht sie als MQTT-Nachrichten. So können Dashboards, Historians und Cloud-Dienste Anlagendaten abonnieren, ohne die SPS je anzufassen. Ein zuverlässiges Gateway schreibst du in etwa 150 Zeilen Python mit pymodbus und paho-mqtt.

Dieses Tutorial baut das Gateway Schritt für Schritt: eine Tag-Map, eine Polling-Schleife, Dekodierung von 32-Bit-Werten, Report-by-Exception, Online/Offline-Status mit Last Will und ein Testweg, bevor irgendetwas in die Nähe einer Produktionslinie kommt.

Wo das Gateway sitzt

Modbus ist ein Request/Response-Protokoll: Ein Master (Client) fragt einen Slave (Server) nach Registern, eine Anfrage nach der anderen. MQTT ist Publish/Subscribe: Produzenten publishen einmal, beliebig viele Konsumenten abonnieren. Das Gateway übersetzt zwischen beiden Welten:

Betreibe das Gateway auf einer Edge-Box (Industrie-PC, Raspberry Pi, Container auf dem Linienserver), die beide Netze erreicht. Die SPS sieht nie MQTT-Traffic, und die MQTT-Konsumenten öffnen nie einen Modbus-Socket – genau die Trennung, die OT-Teams wollen. Das große Bild findest du unter MQTT für IIoT und Industrie 4.0.

1. Tags beschreiben

Modbus kennt keine Namen, nur Adressen. Die erste Aufgabe ist eine Tag-Map, die sagt, was jedes Register bedeutet. Lege sie als Datei neben den Code, damit sie wie Konfiguration reviewt werden kann:

# Name, Funktion, Adresse, Datentyp, Skalierung, Einheit
TAGS = [
    ("temperature",  "holding", 0,  "float32", 1.0,  "degC"),
    ("pressure",     "holding", 2,  "float32", 1.0,  "bar"),
    ("motor_speed",  "holding", 4,  "uint16",  1.0,  "rpm"),
    ("part_count",   "holding", 5,  "uint32",  1.0,  "pcs"),
    ("tank_level",   "input",   0,  "int16",   0.1,  "%"),
    ("running",      "coil",    0,  "bool",    1,    ""),
]

Die Adressen sind hier nullbasierte Protokolladressen. Herstellerdokumentationen nutzen oft die einsbasierte "40001"-Zählung: Holding-Register 40001 ist Adresse 0, 40003 ist Adresse 2. Off-by-one ist der häufigste Modbus-Bug – prüfe also einen bekannten Wert, bevor du dem Rest vertraust.

2. Register lesen und dekodieren

Ein Modbus-Register hat 16 Bit. Floats und 32-Bit-Zähler belegen zwei Register, und Hersteller sind sich uneinig, welches Wort zuerst kommt. Mit struct bleibt die Wortreihenfolge explizit:

import struct
from pymodbus.client import ModbusTcpClient

WORD_ORDER_BIG = True   # auf False setzen bei Geräten mit Little-Endian-Word-Swap

def decode(regs, dtype):
    if dtype == "uint16":
        return regs[0]
    if dtype == "int16":
        return struct.unpack(">h", struct.pack(">H", regs[0]))[0]
    words = regs if WORD_ORDER_BIG else list(reversed(regs))
    raw = struct.pack(">HH", *words)
    return struct.unpack(">f" if dtype == "float32" else ">I", raw)[0]

def read_tag(client, unit, func, addr, dtype):
    count = 2 if dtype in ("float32", "uint32") else 1
    if func == "coil":
        rr = client.read_coils(addr, count=1, device_id=unit)
        return None if rr.isError() else bool(rr.bits[0])
    reader = client.read_holding_registers if func == "holding" else client.read_input_registers
    rr = reader(addr, count=count, device_id=unit)
    return None if rr.isError() else decode(rr.registers, dtype)

In Produktion solltest du zusammenhängende Blöcke lesen (zum Beispiel Register 0–5 in einer Anfrage) statt einer Anfrage pro Tag: Eine SPS beantwortet eine größere Anfrage schneller als sechs kleine. Die Pro-Tag-Variante oben ist leichter lesbar und reicht für ein paar Dutzend Tags.

3. Per Report-by-Exception nach MQTT publishen

Jetzt kombinierst du den Poller mit einem MQTT-Client. Das Gateway publisht ein Topic pro Tag, sendet einen Wert nur, wenn er sich über ein Totband hinaus ändert, markiert Werte als retained, damit neue Subscriber sofort den aktuellen Zustand bekommen, und nutzt ein Last Will, um zu melden, wenn es offline geht.

import json, os, time
import paho.mqtt.client as mqtt
from pymodbus.client import ModbusTcpClient
from modbus_read import read_tag
from tags import TAGS

PLC_HOST, PLC_UNIT = os.getenv("PLC_HOST", "192.168.10.20"), 1
BASE = "testmqtt/acme/plant1/line2/plc1"
POLL_S, DEADBAND = 1.0, 0.05            # Sekunden, relative Änderung (5 %)

mq = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="gw-line2-plc1")
mq.username_pw_set(os.getenv("MQTT_USER"), os.getenv("MQTT_PASS"))
mq.tls_set()
mq.will_set(f"{BASE}/status", "offline", qos=1, retain=True)

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

mq.on_connect = on_connect
mq.connect("broker.connectmq.com", 8883, keepalive=30)
mq.loop_start()

plc = ModbusTcpClient(PLC_HOST, port=502, timeout=2)
last = {}

def changed(name, value):
    old = last.get(name)
    if old is None or isinstance(value, bool):
        return old != value
    return abs(value - old) > abs(old) * DEADBAND

while True:
    if not plc.connected and not plc.connect():
        mq.publish(f"{BASE}/status", "plc-unreachable", qos=1, retain=True)
        time.sleep(5)
        continue
    for name, func, addr, dtype, scale, unit in TAGS:
        value = read_tag(plc, PLC_UNIT, func, addr, dtype)
        if value is None:
            continue
        if not isinstance(value, bool):
            value = round(value * scale, 3)
        if changed(name, value):
            last[name] = value
            payload = {"v": value, "unit": unit, "ts": int(time.time() * 1000)}
            mq.publish(f"{BASE}/{name}", json.dumps(payload), qos=1, retain=True)
    time.sleep(POLL_S)

Ein paar Entscheidungen, die eine Erklärung verdienen:

• Ein Topic pro Tag (.../plc1/temperature) lässt Konsumenten genau das abonnieren, was sie brauchen, und Wildcards wie .../plc1/+ liefern trotzdem alles. Teste Muster mit dem Topic Matcher.
• Retained + QoS 1 heißt: Ein Dashboard, das um 3 Uhr nachts geöffnet wird, zeigt sofort den letzten bekannten Wert, und Nachrichten überstehen kurze Netzaussetzer. Siehe Retained Messages.
• Status-Topic mit Last Will unterscheidet "Temperatur ist stabil" von "Gateway ist tot". Der Broker publisht offline in deinem Namen, wenn die Verbindung des Gateways ohne sauberes Disconnect abbricht.
• Zeitstempel im Payload halten fest, wann der Wert gelesen wurde, nicht wann er ankam – wichtig für Historians.

Das Topic-Layout folgt einer Hierarchie im ISA-95-Stil (Unternehmen / Werk / Linie / Gerät), dem Ausgangspunkt eines Unified Namespace.

4. Ohne Produktions-SPS testen

Entwickle nie gegen eine laufende Linie. Starte mit einem Modbus-Simulator – pymodbus bringt einen mit, und Tools wie diagslave oder ModbusPal stellen konfigurierbare Register auf Port 502 bereit (oder 5020, wenn du nicht root bist). Setze PLC_HOST darauf und starte das Gateway.

Dann prüfst du die MQTT-Seite:

1. Öffne den Online-MQTT-Client, verbinde dich mit broker.connectmq.com und abonniere testmqtt/acme/plant1/#. Du solltest sofort den retained status = online und eine Nachricht pro Tag sehen.
2. Ändere ein Register im Simulator. Nur dieser Tag sollte eine neue Nachricht publishen; unveränderte Tags bleiben still.
3. Stoppe das Gateway mit kill -9 (kein sauberes Disconnect). Nach etwa dem Anderthalbfachen des Keep-Alive sollte status auf offline springen.
4. Stoppe den Simulator. status sollte plc-unreachable melden, während MQTT verbunden bleibt.

Gateway-Topics im Browser beobachten

Im Terminal ist derselbe Check ein einziger Befehl:

mosquitto_sub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t 'testmqtt/acme/plant1/line2/#' -v

Härtung für die Produktion

Modbus hat keinerlei Authentifizierung, daher ist das Gateway oft die einzige kontrollierte Grenze zwischen der SPS und allem anderen. Halte seine MQTT-Rechte eng und den Schreibpfad geschlossen, solange du ihn nicht wirklich brauchst.

Wenn deine Werke später auf Sparkplug B umsteigen oder ein standardisiertes State-Handling für viele Geräte brauchen, kann derselbe Polling-Kern einen Sparkplug-B-Edge-Node speisen statt einfacher JSON-Topics.

Ein Modbus-zu-MQTT-Gateway ist wenig Code mit großer Wirkung: Es verwandelt Register, die nur der SPS-Programmierer versteht, in benannte, zeitgestempelte, abonnierbare Daten. Starte mit einer klaren Tag-Map, publishe bei Änderung mit retained Werten und Last Will, teste gegen einen Simulator und den Online-Client – und schließe es erst dann an die Linie an.

Häufig gestellte Fragen