Modbus TCP zu MQTT: Register pollen und Daten mit Python veröffentlichen

Modbus TCP zu MQTT ist die häufigste Brücke im Industrial IoT, denn Modbus ist überall, und MQTT ist der Weg, auf dem diese Daten Dashboards und die Cloud erreichen. In diesem Tutorial schreiben Sie einen Python-Poller, der Holding- und Input-Register liest, sie korrekt dekodiert und skaliert und nur relevante Änderungen an einen MQTT-Broker veröffentlicht. Das Ergebnis ist ein kleines, konfigurierbares Gateway, das auf jedem Linux-Rechner oder Industrie-PC läuft.

Wie sich Modbus TCP von MQTT unterscheidet

Modbus ist ein Request/Response-Protokoll. Ein Client (historisch „Master") fragt einen Server (historisch „Slave") nach einem Block von Registern und erhält sie zurück. Nichts wird gepusht: Wenn Sie einen Wert wollen, müssen Sie ihn pollen. Modbus kennt keine Namen, Einheiten oder Datentypen jenseits von 16-Bit-Registern und einzelnen Bits.

MQTT ist das Gegenteil: ereignisgesteuertes Publish/Subscribe über einen Broker, mit Topics, die Bedeutung tragen. Die Aufgabe der Brücke ist es, periodische Abfragen in aussagekräftige Ereignisse zu verwandeln. Das bedeutet drei Dinge: die Registerbelegung kennen, Rohregister in physikalische Werte dekodieren und entscheiden, wann ein Wert veröffentlicht werden sollte.

Auffrischung zum Modbus-Datenmodell

Zwei Details bringen fast jede erste Implementierung ins Stolpern:

• Adressierungsversatz. Dokumentationen führen Holding Register oft als 40001, 40002 und so weiter. Auf der Leitung ist 40001 die Adresse 0. Bibliotheken verwenden die nullbasierte Adresse.
• 32-Bit-Werte. Floats und 32-Bit-Integer erstrecken sich über zwei Register, und Hersteller sind sich bei der Wortreihenfolge uneinig. Manche senden das höherwertige Wort zuerst (Big Endian, oft als ABCD geschrieben), andere das niederwertige (CDAB). Wenn Ihre Temperatur als 1.2e-38 oder 3.5e+12 erscheint, tauschen Sie die Wörter.

Voraussetzungen

• Python 3.10 oder neuer.
• pymodbus 3.8 und paho-mqtt 2.x. Fixieren Sie die pymodbus-Version: Die Schlüsselwortargumente haben sich zwischen Minor-Releases geändert.
• Ein Modbus-TCP-Gerät oder der folgende Simulator.

python -m venv .venv
source .venv/bin/activate
pip install "pymodbus>=3.8,<3.9" "paho-mqtt>=2.0"

Schritt 1: Ein Modbus-TCP-Simulator (optional)

Um ohne Hardware zu entwickeln, starten Sie einen kleinen Server, der eine Temperatur als 32-Bit-Float in den Holding Registern 0 und 1, einen Druck als skalierten Integer im Input Register 0 und einen Laufstatus in Coil 0 bereitstellt.

# modbus_sim.py - Modbus-TCP-Server auf Port 5020 mit sich ändernden Werten
import asyncio, random, struct
from pymodbus.datastore import (ModbusSequentialDataBlock, ModbusServerContext,
                                ModbusSlaveContext)
from pymodbus.server import StartAsyncTcpServer

store = ModbusSlaveContext(
    co=ModbusSequentialDataBlock(0, [0] * 10),
    di=ModbusSequentialDataBlock(0, [0] * 10),
    hr=ModbusSequentialDataBlock(0, [0] * 10),
    ir=ModbusSequentialDataBlock(0, [0] * 10),
)
context = ModbusServerContext(slaves=store, single=True)

async def update():
    while True:
        temp = 60 + random.uniform(-0.3, 0.3)
        hi, lo = struct.unpack(">HH", struct.pack(">f", temp))
        store.setValues(3, 0, [hi, lo])                         # Holding Register
        store.setValues(4, 0, [int(random.uniform(4.9, 5.1) * 100)])  # Input Register, bar x100
        store.setValues(1, 0, [random.random() > 0.1])          # Coil
        await asyncio.sleep(1)

async def main():
    asyncio.create_task(update())
    await StartAsyncTcpServer(context=context, address=("0.0.0.0", 5020))

asyncio.run(main())

Port 5020 vermeidet, dass für Port 502 Root-Rechte nötig sind. Bei einem echten Gerät verwenden Sie Port 502 und die Unit ID aus dessen Handbuch.

Schritt 2: Die Registerbelegung beschreiben

Halten Sie die Belegung in der Konfiguration, damit ein neuer Tag nie eine Codeänderung erfordert. Jeder Tag hat Tabelle, Adresse, Datentyp, Skalierung, Einheit, Totband und Topic-Pfad.

{
  "modbus": { "host": "127.0.0.1", "port": 5020, "unit_id": 1, "poll_seconds": 1.0 },
  "topic_root": "connectmq/demo/b72f/plant1/utilities/boiler1",
  "heartbeat_seconds": 60,
  "tags": [
    { "name": "temperature", "table": "holding", "address": 0, "type": "float32", "word_order": "big",
      "scale": 1.0, "unit": "degC", "deadband": 0.5 },
    { "name": "pressure", "table": "input", "address": 0, "type": "uint16",
      "scale": 0.01, "unit": "bar", "deadband": 0.05 },
    { "name": "running", "table": "coil", "address": 0, "type": "bool", "unit": "" }
  ]
}

Speichern Sie sie als modbus_map.json und ersetzen Sie b72f durch eine eigene Zufallszeichenfolge.

Schritt 3: Die Brücke von Modbus TCP zu MQTT

Der folgende Poller implementiert Report by Exception: Ein Analogwert wird veröffentlicht, wenn er sich um mehr als sein Totband bewegt, ein Boolean, wenn er sich ändert, und jeder Wert wird in einem Heartbeat-Intervall erneut veröffentlicht, damit Konsumenten wissen, dass die Brücke lebt. Mehr zu diesem Muster lesen Sie in unserem Artikel zu Report by Exception.

# modbus_mqtt_bridge.py - Modbus TCP pollen, Änderungen an MQTT veröffentlichen
import json, logging, struct, time, uuid
import paho.mqtt.client as mqtt
from pymodbus.client import ModbusTcpClient
from pymodbus.exceptions import ModbusException

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("bridge")
CFG = json.load(open("modbus_map.json"))

MQTT_HOST, MQTT_PORT = "broker.connectmq.com", 1883   # Produktion: 8883 + TLS
MQTT_USER, MQTT_PASS = None, None                     # Produktion: "YOUR_USERNAME", "YOUR_PASSWORD"
CLIENT_ID = f"modbus-bridge-{uuid.uuid4().hex[:8]}"   # Produktion: "YOUR_CLIENT_ID"
STATUS = f"{CFG['topic_root']}/bridge_status"


def mqtt_connect():
    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(STATUS, "offline", qos=1, retain=True)
    c.on_connect = lambda cl, ud, fl, rc, pr: cl.publish(STATUS, "online", qos=1, retain=True)
    c.connect(MQTT_HOST, MQTT_PORT, keepalive=30)
    c.loop_start()
    return c


def decode(tag, regs):
    t = tag["type"]
    if t == "uint16":
        raw = regs[0]
    elif t == "int16":
        raw = struct.unpack(">h", struct.pack(">H", regs[0]))[0]
    elif t in ("float32", "int32", "uint32"):
        words = regs[:2] if tag.get("word_order", "big") == "big" else regs[1::-1]
        packed = struct.pack(">HH", *words)
        raw = struct.unpack({"float32": ">f", "int32": ">i", "uint32": ">I"}[t], packed)[0]
    else:
        raise ValueError(f"unsupported type {t}")
    return round(raw * tag.get("scale", 1.0), 4)


def read_tag(client, tag, unit):
    count = 2 if tag["type"] in ("float32", "int32", "uint32") else 1
    table = tag["table"]
    if table == "coil":
        rr = client.read_coils(tag["address"], count=1, slave=unit)
    elif table == "discrete":
        rr = client.read_discrete_inputs(tag["address"], count=1, slave=unit)
    elif table == "holding":
        rr = client.read_holding_registers(tag["address"], count=count, slave=unit)
    else:
        rr = client.read_input_registers(tag["address"], count=count, slave=unit)
    if rr.isError():
        raise ModbusException(f"{tag['name']}: {rr}")
    return bool(rr.bits[0]) if table in ("coil", "discrete") else decode(tag, rr.registers)


def should_publish(tag, value, last):
    if last is None:
        return True
    if isinstance(value, bool):
        if value != last["value"]:
            return True
    elif abs(value - last["value"]) >= tag.get("deadband", 0):
        return True
    return time.time() - last["sent"] >= CFG["heartbeat_seconds"]


def main():
    m = CFG["modbus"]
    mq = mqtt_connect()
    mb = ModbusTcpClient(m["host"], port=m["port"], timeout=2)
    last = {}
    while True:
        if not mb.connected and not mb.connect():
            log.warning("Modbus device unreachable, retrying")
            time.sleep(5)
            continue
        for tag in CFG["tags"]:
            topic = f"{CFG['topic_root']}/{tag['name']}"
            try:
                value = read_tag(mb, tag, m["unit_id"])
                quality = "good"
            except (ModbusException, ConnectionError, OSError) as exc:
                log.error("read failed: %s", exc)
                mb.close()
                value, quality = None, "bad"
            prev = last.get(tag["name"])
            if quality == "bad":
                if prev is None or prev.get("quality") != "bad":
                    mq.publish(topic, json.dumps({"value": None, "quality": "bad"}), qos=1, retain=True)
                    last[tag["name"]] = {"value": None, "quality": "bad", "sent": time.time()}
                continue
            if prev is None or prev.get("quality") == "bad" or should_publish(tag, value, prev):
                payload = {"value": value, "unit": tag.get("unit", ""), "quality": quality,
                           "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}
                mq.publish(topic, json.dumps(payload), qos=1, retain=True)
                last[tag["name"]] = {"value": value, "quality": quality, "sent": time.time()}
                log.info("%s = %s", topic, value)
        time.sleep(m["poll_seconds"])


if __name__ == "__main__":
    main()

Schritt 4: Ausführen und prüfen

python modbus_sim.py            # Terminal 1, entfällt bei einem echten Gerät
python modbus_mqtt_bridge.py    # Terminal 2
mosquitto_sub -h broker.connectmq.com -p 1883 -i modbus-viewer-b72f \
  -t "connectmq/demo/b72f/#" -v   # Terminal 3

Sie sollten die Temperatur nur sehen, wenn sie um ein halbes Grad driftet, den Druck nur, wenn er sich um 0,05 bar bewegt, und das Lauf-Flag nur, wenn es umschaltet. Jeder Wert erscheint zusätzlich einmal pro Minute als Heartbeat. Das ist ein Bruchteil der Nachrichten, die eine naive Schleife „bei jeder Abfrage veröffentlichen" senden würde, was auf dem öffentlichen MQTT-Broker, der nur für Tests gedacht ist und wo jeder Client höchstens zehn Nachrichten pro Sekunde veröffentlichen darf, und auf jeder volumenbasiert abgerechneten Verbindung wichtig ist.

Tipps für das Polling-Design

1. Zusammenhängende Blöcke lesen. Eine Anfrage für die Register 0 bis 19 ist weitaus günstiger als zwanzig Anfragen für je ein Register. Wächst Ihre Belegung, gruppieren Sie Tags nach Tabelle und Adressbereich und dekodieren Sie aus dem Block.
2. Das Gerät respektieren. Viele kleine SPSen und Zähler verkraften nur wenige Anfragen pro Sekunde. Beginnen Sie mit einer Abfrage pro Sekunde und messen Sie.
3. Die richtige Unit ID verwenden. Modbus-TCP-Geräte hinter einem Gateway nutzen die Unit ID, um serielle Geräte zu adressieren; eigenständige Geräte ignorieren sie oft oder erwarten 1 oder 255.
4. Schnelle und langsame Tags trennen. Alarme und Zustände schnell pollen, Energiezähler langsam.
5. Qualität explizit veröffentlichen. Ein fehlgeschlagener Lesevorgang sollte quality: bad veröffentlichen und nicht stillschweigend den letzten Wert behalten.
6. Niemals versehentlich aus dem Poller schreiben. Halten Sie Schreibfunktionen aus der Brücke heraus oder legen Sie sie hinter ein separates, authentifiziertes Command-Topic.

Topics und Payloads für Modbus-Daten gestalten

Rohes Modbus liefert Adressen, keine Bedeutung, daher fügt die Brücke im Topic-Baum den Kontext hinzu. Benennen Sie Topics danach, was der Wert ist und wo er sich befindet, niemals nach der Registernummer: plant1/utilities/boiler1/temperature ist für jeden nützlich, device3/hr/0 nur für denjenigen, der die Registerbelegung geschrieben hat.

Den Rahmen dafür liefert ein Unified Namespace (UNS): ein zentraler Topic-Baum auf dem MQTT-Broker, dessen Struktur der Equipment-Hierarchie nach ISA-95 (IEC 62264) folgt, also Enterprise → Site → Area → Line → Cell. Aus dem Demo-Topic wird dann etwa acme/werk-stuttgart/versorgung/kesselhaus/kessel-1/temperatur. Diese Hierarchie hält Hunderte Modbus-Geräte übersichtlich, lässt MES, SCADA und Analytics einen ganzen Bereich mit einer Wildcard abonnieren und bringt Modbus-Daten in dieselbe Struktur wie Werte aus OPC UA oder einer Siemens-S7-SPS. Wie Sparkplug B dazu passt, erklärt der Beitrag zu Sparkplug B und Unified Namespace.

Verwenden Sie ein Topic pro Wert für Prozessdaten, die Konsumenten einzeln auswählen, und nehmen Sie Einheit, Qualität und Zeitstempel in jede Payload auf, damit eine Nachricht auch Monate später in einem Log selbsterklärend ist. Hat das Gerät viele Register, die immer zusammen genutzt werden, etwa alle Messwerte eines Zählers, ist ein einzelnes JSON-Objekt pro Gerät und Zyklus eine sinnvolle Alternative und senkt die Nachrichtenzahl. Wofür Sie sich auch entscheiden: Wenden Sie es einheitlich über alle Brücken des Standorts an.

Der Weg in die Produktion

Für echte Werksdaten wechseln Sie zu einem privaten Broker: Setzen Sie MQTT_PORT auf 8883 und tragen Sie die Zugangsdaten Ihres verwalteten MQTT-Brokers ein. Bei ConnectMQ erhält jeder Broker einen generierten Benutzernamen, ein Passwort und eine Client ID; die Client ID ist an diese Zugangsdaten gebunden, verwenden Sie sie also statt einer zufälligen. Betreiben Sie die Brücke als systemd-Dienst mit Restart=always und ziehen Sie eine Pufferung per Store and Forward in Betracht, wenn die Internetverbindung des Standorts unzuverlässig ist.

Sprechen Ihre Geräte Modbus RTU über RS-485 statt Ethernet, funktioniert dieselbe Brücke mit einem seriellen Client; unser Tutorial Modbus RTU zu MQTT behandelt Verdrahtung und serielle Einstellungen. Zur Gesamtarchitektur aus Gateways, Brokern und Konsumenten lesen Sie den Leitfaden zu Industrial IoT mit MQTT.

Fehlerbehebung

Starten Sie mit ConnectMQ

Bringen Sie Ihre Modbus-Daten aus dem Werksnetz, ohne einen Server zu betreiben. Registrieren Sie sich unter app.connectmq.com, wählen Sie den kostenlosen Serverless-Plan (ohne Kreditkarte), legen Sie einen Broker an und fügen Sie die generierten Zugangsdaten in die Brücke ein. Ihre Registerwerte erscheinen sofort im Live-Topic-Explorer von ConnectMQ, und Regeln für Schwellwert-Alarme können Sie warnen, wenn eine Temperatur oder ein Druck seinen Bereich verlässt.

Häufig gestellte Fragen

Wie oft sollte ich Modbus-TCP-Register pollen?

Beginnen Sie mit einer Sekunde für Prozesswerte und langsamer für Zähler, und passen Sie dann an, was das Gerät verträgt und was die Konsumenten brauchen. Veröffentlichen Sie bei Änderung statt bei jeder Abfrage.

Warum sehen meine Modbus-Float-Werte falsch aus?

Meist stehen die beiden 16-Bit-Wörter in umgekehrter Reihenfolge zu dem, was Ihr Code annimmt. Tauschen Sie die Wortreihenfolge und prüfen Sie, ob das Register nicht eigentlich ein skalierter Integer statt eines Floats ist.

Kann eine Brücke mehrere Modbus-TCP-Geräte pollen?

Ja. Betreiben Sie einen Client pro Gerät oder einen Client pro Gateway mit unterschiedlichen Unit IDs, und geben Sie jedem Gerät einen eigenen Zweig im Topic-Baum.

Sollte die Brücke QoS 0 oder QoS 1 verwenden?

QoS 1 mit Retained Messages ist ein guter Standard für Prozesswerte; es garantiert die Zustellung und lässt neue Subscriber sofort den aktuellen Zustand sehen.