Raspberry Pi als Modbus-MQTT-Gateway mit pymodbus (RTU und TCP)

Mit einem Raspberry Pi Modbus MQTT Gateway holen Sie Messwerte aus Energiezählern, Wechselrichtern, Frequenzumrichtern und SPSen, die nur Modbus sprechen, und veröffentlichen sie als sauberes JSON über MQTT. In diesem Tutorial schließen Sie einen USB-RS-485-Adapter für Modbus RTU an, verbinden sich über Ethernet mit Modbus-TCP-Geräten, beschreiben die Register jedes Geräts in einer kleinen Konfiguration, dekodieren skalierte Ganzzahlen und 32-Bit-Floats korrekt und veröffentlichen die Ergebnisse an einen Broker. Das fertige Skript behandelt Timeouts und Wiederverbindungen und kann daher unbeaufsichtigt laufen.

Modbus in zwei Minuten

Modbus ist ein Request-Response-Protokoll: Der Pi (der Client, früher Master genannt) fragt bei einem Gerät (dem Server, früher Slave) einen Registerblock an, und das Gerät antwortet. Topics, Abonnements oder Push gibt es nicht; alles beruht auf Polling. Genau deshalb braucht es ein Gateway, um MQTT-Systeme zu versorgen.

Die Daten liegen in vier Tabellen: Coils und Discrete Inputs (Bits) sowie Input und Holding Register (16-Bit-Wörter). Die meisten Zähler stellen Messwerte als Holding oder Input Register bereit. Werte, die nicht in 16 Bit passen, etwa Energiesummen oder Gleitkomma-Spannungen, belegen zwei aufeinanderfolgende Register, und das Gerätehandbuch nennt die Wortreihenfolge. Eine falsche Reihenfolge ist der häufigste Grund, warum ein Gateway absurde Zahlen veröffentlicht.

Tieferes Protokollwissen zu beiden Varianten finden Sie in den Anleitungen zu Modbus RTU zu MQTT und Modbus TCP zu MQTT.

Hardware und Verdrahtung für Modbus RTU

1. Stecken Sie einen USB-RS-485-Adapter in den Pi. Er erscheint als /dev/ttyUSB0. Mit ls /dev/serial/by-id/ erhalten Sie einen stabilen Namen, der Neustarts und erneutes Einstecken übersteht.
2. Verbinden Sie A mit A (manchmal als D+ beschriftet) und B mit B (D-). Kommen gar keine Antworten, ist das Tauschen von A und B der erste Versuch, denn die Hersteller beschriften uneinheitlich.
3. Verbinden Sie die Signalmasse, falls Adapter und Gerät eine anbieten. Lange Leitungen ohne gemeinsames Bezugspotenzial verursachen sporadische Fehler.
4. Terminieren Sie bei längeren Leitungen die beiden physischen Busenden mit 120-Ohm-Widerständen. Viele Geräte und Adapter haben dafür einen Schalter.
5. Notieren Sie Baudrate, Parität, Stoppbits und Unit ID des Geräts aus Display oder Handbuch. Alle Geräte an einem Bus müssen dieselben seriellen Einstellungen nutzen.

Für Modbus TCP brauchen Sie nur IP-Adresse, Port (meist 502) und Unit ID des Geräts. Hinter einem Modbus-TCP-zu-RTU-Gateway wählt die Unit ID das RTU-Gerät auf der anderen Seite aus.

pymodbus und paho-mqtt installieren

sudo apt update && sudo apt install -y python3-venv
python3 -m venv ~/modbus-env
~/modbus-env/bin/pip install "pymodbus>=3.10" pyserial "paho-mqtt>=2.0"
sudo usermod -aG dialout $USER     # serieller Zugriff, danach ab- und wieder anmelden

pymodbus 3.10 hat das Unit-Argument von slave in device_id umbenannt. Der folgende Code verwendet device_id; wenn Sie an eine ältere 3.x-Version gebunden sind, ändern Sie es zurück in slave.

Ein kurzer Plausibilitätstest in der Python-Shell, bevor Sie etwas Größeres schreiben:

from pymodbus.client import ModbusSerialClient

c = ModbusSerialClient(port="/dev/ttyUSB0", baudrate=9600, parity="N",
                       stopbits=1, bytesize=8, timeout=1)
print(c.connect())
rr = c.read_holding_registers(0, count=2, device_id=1)
print(rr if rr.isError() else rr.registers)
c.close()

Werden Registerwerte ausgegeben, stimmen Verdrahtung und Einstellungen. Erscheint ein Timeout, prüfen Sie A/B, Baudrate, Parität und Unit ID in dieser Reihenfolge.

Geräte in einer Registerkarte beschreiben

Fest einprogrammierte Adressen im Python-Code werden schnell unübersichtlich. Legen Sie die Register jedes Geräts stattdessen in einer JSON-Datei ab. Speichern Sie sie als ~/devices.json:

{
  "devices": [
    {
      "name": "meter1",
      "area": "utilities",
      "transport": "rtu",
      "port": "/dev/ttyUSB0",
      "baudrate": 9600,
      "parity": "N",
      "unit": 1,
      "interval": 10,
      "registers": [
        {"field": "voltage_v",  "table": "holding", "address": 0, "type": "uint16",  "scale": 0.1},
        {"field": "current_a",  "table": "holding", "address": 1, "type": "uint16",  "scale": 0.01},
        {"field": "power_w",    "table": "holding", "address": 2, "type": "int32",   "word_order": "big"},
        {"field": "energy_kwh", "table": "input",   "address": 10, "type": "float32", "word_order": "little"}
      ]
    },
    {
      "name": "vfd1",
      "area": "line1",
      "transport": "tcp",
      "host": "192.168.1.70",
      "port": 502,
      "unit": 1,
      "interval": 5,
      "registers": [
        {"field": "speed_rpm",  "table": "holding", "address": 100, "type": "uint16", "scale": 1},
        {"field": "fault_code", "table": "holding", "address": 105, "type": "uint16", "scale": 1}
      ]
    }
  ]
}

Die Adressen oben sind Beispiele. Verwenden Sie die Registertabelle aus dem Handbuch Ihres Geräts und achten Sie auf Handbücher, die Adressen ab 1 oder mit einem Offset von 40001 angeben: pymodbus erwartet die nullbasierte Protokolladresse, Register 40001 ist also Adresse 0.

Das Raspberry Pi Modbus MQTT Gateway-Skript

Speichern Sie es als ~/modbus2mqtt.py:

#!/usr/bin/env python3
"""Poll Modbus RTU/TCP devices and publish readings over MQTT."""
import json
import os
import ssl
import struct
import time

import paho.mqtt.client as mqtt
from pymodbus.client import ModbusSerialClient, ModbusTcpClient
from pymodbus.exceptions import ModbusException

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "1883"))
USERNAME = os.getenv("MQTT_USERNAME")                  # YOUR_USERNAME
PASSWORD = os.getenv("MQTT_PASSWORD")                  # YOUR_PASSWORD
CLIENT_ID = os.getenv("MQTT_CLIENT_ID", "modbus-gw-2f84ad")  # YOUR_CLIENT_ID
PREFIX = os.getenv("MQTT_PREFIX", "connectmq/demo/2f84ad/site1")
CONFIG = os.getenv("DEVICES", os.path.expanduser("~/devices.json"))

SIZES = {"uint16": 1, "int16": 1, "uint32": 2, "int32": 2, "float32": 2}


def decode(regs, rtype, word_order="big"):
    if SIZES[rtype] == 2 and word_order == "little":
        regs = [regs[1], regs[0]]
    raw = b"".join(struct.pack(">H", r) for r in regs)
    fmt = {"uint16": ">H", "int16": ">h", "uint32": ">I",
           "int32": ">i", "float32": ">f"}[rtype]
    return struct.unpack(fmt, raw)[0]


def make_client(dev):
    if dev["transport"] == "rtu":
        return ModbusSerialClient(port=dev["port"], baudrate=dev.get("baudrate", 9600),
                                  parity=dev.get("parity", "N"), stopbits=dev.get("stopbits", 1),
                                  bytesize=8, timeout=1, retries=1)
    return ModbusTcpClient(dev["host"], port=dev.get("port", 502), timeout=2, retries=1)


def read_device(mb, dev):
    values = {}
    for reg in dev["registers"]:
        count = SIZES[reg["type"]]
        fn = mb.read_holding_registers if reg["table"] == "holding" else mb.read_input_registers
        rr = fn(reg["address"], count=count, device_id=dev["unit"])
        if rr.isError():
            raise ModbusException(f"{dev['name']} {reg['field']}: {rr}")
        v = decode(rr.registers, reg["type"], reg.get("word_order", "big"))
        v = v * reg.get("scale", 1)
        values[reg["field"]] = round(v, 4) if isinstance(v, float) else v
    return values


# ---------- MQTT ----------
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                     protocol=mqtt.MQTTv5)
if USERNAME:
    client.username_pw_set(USERNAME, PASSWORD)
if PORT == 8883:
    client.tls_set(cert_reqs=ssl.CERT_REQUIRED)
gw_status = f"{PREFIX}/gateway/modbus/status"
client.will_set(gw_status, "offline", qos=1, retain=True)


def on_connect(c, userdata, flags, reason_code, properties):
    if not reason_code.is_failure:
        c.publish(gw_status, "online", qos=1, retain=True)


client.on_connect = on_connect
client.reconnect_delay_set(1, 60)
client.connect_async(HOST, PORT, keepalive=60)
client.loop_start()

# ---------- Polling-Schleife ----------
with open(CONFIG) as f:
    devices = json.load(f)["devices"]

clients = {}
next_poll = {d["name"]: 0.0 for d in devices}
last_ok = {}

while True:
    now = time.monotonic()
    for dev in devices:
        name = dev["name"]
        if now < next_poll[name]:
            continue
        next_poll[name] = now + dev.get("interval", 10)
        base = f"{PREFIX}/{dev['area']}/{name}"
        # RTU-Geräte am selben Port teilen sich einen Client
        key = dev.get("port") if dev["transport"] == "rtu" else f"{dev['host']}:{dev.get('port', 502)}"
        mb = clients.get(key) or clients.setdefault(key, make_client(dev))
        try:
            if not mb.connected:
                mb.connect()
            values = read_device(mb, dev)
            payload = {"ts": int(time.time()), "values": values}
            client.publish(f"{base}/telemetry", json.dumps(payload), qos=1)
            if last_ok.get(name) is not True:
                client.publish(f"{base}/status", "online", qos=1, retain=True)
            last_ok[name] = True
        except (ModbusException, OSError) as exc:
            print(f"{name}: {exc}")
            if last_ok.get(name) is not False:
                client.publish(f"{base}/status", "offline", qos=1, retain=True)
            last_ok[name] = False
            mb.close()
    time.sleep(0.2)

Designentscheidungen, die Sie verstehen sollten:

• Ein Client pro physischer Verbindung. RTU-Geräte am selben RS-485-Port teilen sich einen einzigen seriellen Client, denn zwei Prozesse oder Objekte, die denselben Port öffnen, zerstören gegenseitig ihre Frames.
• Status-Topics pro Gerät. Jedes Gerät erhält ein retained online oder offline, das nur bei einer Zustandsänderung veröffentlicht wird. Dashboards sehen sofort, welcher Zähler nicht mehr antwortet, getrennt vom Status des Gateways selbst.
• Rohregister werden an einer Stelle dekodiert. Die Funktion decode behandelt 16- und 32-Bit-Typen sowie die Wortreihenfolge mit struct und hängt daher nicht von Hilfs-APIs ab, die sich zwischen pymodbus-Versionen geändert haben.
• Polling-Intervalle pro Gerät. Die Drehzahl eines Umrichters alle 5 Sekunden und eine Energiesumme alle 60 Sekunden sind eine sinnvolle Mischung; fragen Sie nicht schneller ab, als Ihre Abnehmer es brauchen.

Die Raspberry Pi Modbus MQTT-Ausgabe testen

Starten Sie das Skript im Vordergrund:

~/modbus-env/bin/python ~/modbus2mqtt.py

Beobachten Sie die Ausgabe in einem zweiten Terminal auf dem kostenlosen öffentlichen MQTT-Broker:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-2f84ad \
  -t 'connectmq/demo/2f84ad/#' -v

Erwartete Ausgabe:

connectmq/demo/2f84ad/site1/gateway/modbus/status online
connectmq/demo/2f84ad/site1/utilities/meter1/status online
connectmq/demo/2f84ad/site1/utilities/meter1/telemetry {"ts": 1760601234, "values": {"voltage_v": 231.4, "current_a": 1.27, "power_w": 287, "energy_kwh": 1532.6875}}

Vergleichen Sie mindestens einen Wert mit dem Display des Geräts. Ist ein Float völlig falsch (riesige Exponenten oder winzige Brüche), drehen Sie word_order um. Liegt ein skalierter Wert um den Faktor zehn daneben, korrigieren Sie scale.

Noch keine Modbus-Hardware? pymodbus enthält einen Simulator, und die Anleitung Modbus TCP zu MQTT zeigt, wie Sie ihn zum Testen auf einem Laptop betreiben.

Produktion: privater Broker, TLS und ein Dienst

Zähler- und Maschinendaten sind betriebliche Informationen, deshalb sollten produktive Gateways verschlüsselt an einen privaten Broker veröffentlichen. Bei ConnectMQ, einem verwalteten MQTT-Broker, legen Sie im Portal einen Broker an und erhalten Benutzername, Passwort und Client ID. Tragen Sie diese in /etc/modbus2mqtt.env ein:

MQTT_HOST=broker.connectmq.com
MQTT_PORT=8883
MQTT_USERNAME=YOUR_USERNAME
MQTT_PASSWORD=YOUR_PASSWORD
MQTT_CLIENT_ID=YOUR_CLIENT_ID
MQTT_PREFIX=acme/plant1
DEVICES=/home/pi/devices.json

Dazu kommt eine systemd-Unit mit EnvironmentFile=/etc/modbus2mqtt.env, Restart=always und After=network-online.target, wie in den anderen Pi-Tutorials dieser Reihe. Strukturieren Sie das Präfix nach Unternehmen, Standort, Bereich und Linie, erkennt das ConnectMQ-Portal die ISA-95-Hierarchie und gruppiert die Zähler in seiner Geräteansicht; Schwellwert-Alarmregeln können Werte wie einen fault_code des Umrichters ungleich null melden. So wird die ISA-95-Gerätehierarchie direkt zur Topic-Struktur eines Unified Namespace (UNS), etwa acme/werk-stuttgart/energie/linie-1/zaehler1/telemetry, in dem MES, SCADA und Energiemonitoring dieselben Daten finden. Mehr dazu im Beitrag Sparkplug B und Unified Namespace.

Zwei weitere Punkte für den Produktivbetrieb. Erstens: Wenn die Uplink-Verbindung länger ausfallen kann, setzen Sie einen Festplattenpuffer vor den Publish-Aufruf; die Anleitung zum Raspberry Pi als IoT-Gateway enthält eine SQLite-Outbox, die Sie wiederverwenden können. Zweitens: Wenn Sie aus der Cloud auf Geräte schreiben wollen, ergänzen Sie eine strenge Validierung und eine Allow-List beschreibbarer Register. Ein falscher Modbus-Schreibzugriff kann eine Maschine stoppen.

Fehlerbehebung

Starten Sie mit ConnectMQ

Schicken Sie Ihre Modbus-Daten dorthin, wo sie Nutzen bringen: Legen Sie unter app.connectmq.com einen Broker im kostenlosen Serverless-Plan an, ohne Kreditkarte, und verfolgen Sie im Live-Topic-Explorer des Portals, wie Ihre Zählerwerte eintreffen. Für den Rest der Pi-Reihe beginnen Sie mit dem Raspberry Pi MQTT Tutorial.

Häufig gestellte Fragen

Kann ein Raspberry Pi mehrere Modbus-RTU-Geräte auslesen?

Ja. Hängen Sie sie mit eindeutigen Unit IDs und identischen seriellen Einstellungen an denselben RS-485-Bus; das Gateway fragt sie nacheinander über einen einzigen seriellen Client ab.

Was ist der Unterschied zwischen Holding und Input Registern?

Input Register sind schreibgeschützte Messwerte, Holding Register lassen sich lesen und oft auch schreiben, etwa für Einstellungen und Sollwerte. Ihr Gerätehandbuch nennt, in welcher Tabelle welcher Wert liegt.

Wie schnell kann das Gateway Modbus RTU abfragen?

Das hängt von Baudrate, Anzahl der Register und Antwortzeit des Geräts ab. Fassen Sie aufeinanderfolgende Register nach Möglichkeit zu einem Lesezugriff zusammen und fragen Sie nur so oft ab, wie Ihre Abnehmer es brauchen.

Sollte das Gateway eine Nachricht pro Register oder pro Gerät veröffentlichen?

Meist ist eine JSON-Nachricht pro Gerät und Abfrage am besten. Zusammengehörige Werte bleiben mit einem Zeitstempel beisammen, und es werden weit weniger Nachrichten erzeugt.