Passerelle Modbus MQTT sur Raspberry Pi avec pymodbus (RTU et TCP)

Une passerelle Modbus MQTT sur Raspberry Pi permet de récupérer les mesures de compteurs d'énergie, d'onduleurs, de variateurs de fréquence et d'automates qui ne parlent que Modbus, puis de les publier en JSON propre via MQTT. Dans ce tutoriel, vous allez câbler un adaptateur USB RS-485 pour Modbus RTU, vous connecter à des équipements Modbus TCP via Ethernet, décrire les registres de chaque équipement dans une petite configuration, décoder correctement les entiers mis à l'échelle et les flottants 32 bits, puis publier les résultats vers un broker. Le script final gère les timeouts et les reconnexions : il peut tourner sans surveillance.

Modbus en deux minutes

Modbus est un protocole requête-réponse : le Pi (le client, historiquement appelé maître) demande un bloc de registres à un équipement (le serveur, historiquement esclave), et l'équipement répond. Il n'y a ni topics, ni abonnements, ni push : tout fonctionne par interrogation (polling). C'est précisément pour cela qu'une passerelle est nécessaire pour alimenter des systèmes MQTT.

Les données sont réparties dans quatre tables : les coils et les entrées discrètes (des bits), ainsi que les input registers et les holding registers (mots de 16 bits). La plupart des compteurs exposent leurs mesures dans des holding ou input registers. Les valeurs qui ne tiennent pas sur 16 bits, comme les totaux d'énergie ou les tensions en virgule flottante, occupent deux registres consécutifs, et le manuel de l'équipement indique l'ordre des mots. Se tromper sur cet ordre est la cause la plus fréquente de valeurs absurdes publiées par une passerelle.

Pour approfondir chaque variante du protocole, consultez les guides Modbus RTU vers MQTT et Modbus TCP vers MQTT.

Matériel et câblage pour Modbus RTU

1. Branchez un adaptateur USB RS-485 sur le Pi. Il apparaît sous /dev/ttyUSB0. Utilisez ls /dev/serial/by-id/ pour obtenir un nom stable qui survit aux redémarrages et aux débranchements.
2. Reliez A à A (parfois noté D+) et B à B (D-). Si vous n'obtenez aucune réponse, inverser A et B est la première chose à essayer, car les fabricants ne s'accordent pas sur l'étiquetage.
3. Reliez la masse de signal si l'adaptateur et l'équipement en disposent. Les longues liaisons sans référence commune provoquent des erreurs intermittentes.
4. Sur les longues liaisons, terminez les deux extrémités physiques du bus avec des résistances de 120 ohms. Beaucoup d'équipements et d'adaptateurs disposent d'un interrupteur de terminaison.
5. Relevez le débit, la parité, les bits de stop et l'unit ID de l'équipement sur son afficheur ou dans son manuel. Tous les équipements d'un même bus doivent partager les mêmes paramètres série.

Pour Modbus TCP, il vous suffit de l'adresse IP de l'équipement, du port (généralement 502) et de l'unit ID. Derrière une passerelle Modbus TCP vers RTU, l'unit ID sélectionne l'équipement RTU situé de l'autre côté.

Installer pymodbus et paho-mqtt

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     # accès au port série, puis déconnexion/reconnexion

pymodbus 3.10 a renommé l'argument d'unité de slave en device_id. Le code ci-dessous utilise device_id ; si vous êtes bloqué sur une version 3.x plus ancienne, revenez à slave.

Petit test de cohérence depuis un shell Python avant d'écrire quelque chose de plus conséquent :

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()

Si des valeurs de registres s'affichent, le câblage et les paramètres sont bons. Si vous obtenez un timeout, vérifiez A/B, le débit, la parité et l'unit ID, dans cet ordre.

Décrire les équipements dans une table de registres

Coder les adresses en dur dans Python devient vite ingérable. Placez plutôt les registres de chaque équipement dans un fichier JSON. Enregistrez-le sous ~/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}
      ]
    }
  ]
}

Les adresses ci-dessus sont des exemples. Utilisez la table de registres du manuel de votre propre équipement, et méfiez-vous des manuels qui numérotent les adresses à partir de 1 ou avec un décalage de 40001 : pymodbus attend l'adresse protocolaire commençant à zéro, donc le registre 40001 correspond à l'adresse 0.

Le script de passerelle Modbus MQTT pour Raspberry Pi

Enregistrez-le sous ~/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()

# ---------- boucle d'interrogation ----------
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}"
        # les équipements RTU sur le même port partagent un seul 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)

Quelques choix de conception à bien comprendre :

• Un client par connexion physique. Les équipements RTU d'un même port RS-485 partagent un seul client série, car deux processus ou objets qui ouvrent le même port corrompent mutuellement leurs trames.
• Des topics de statut par équipement. Chaque équipement reçoit un online ou offline en retained message, publié uniquement lors d'un changement d'état. Les tableaux de bord voient immédiatement quel compteur ne répond plus, indépendamment du statut de la passerelle elle-même.
• Les registres bruts sont décodés en un seul endroit. La fonction decode gère les types 16 et 32 bits ainsi que l'ordre des mots avec struct ; elle ne dépend donc pas d'API utilitaires qui ont changé d'une version de pymodbus à l'autre.
• Un intervalle d'interrogation par équipement. La vitesse d'un variateur toutes les 5 secondes et un total d'énergie toutes les 60 secondes forment un mélange raisonnable ; n'interrogez pas plus vite que ce dont vos consommateurs ont besoin.

Tester la sortie Modbus MQTT du Raspberry Pi

Lancez le script au premier plan :

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

Observez la sortie sur le serveur MQTT public gratuit depuis un autre terminal :

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

Sortie attendue :

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}}

Comparez au moins une valeur avec l'afficheur de l'équipement. Si un flottant est complètement faux (exposants énormes ou fractions minuscules), inversez word_order. Si une valeur mise à l'échelle est décalée d'un facteur dix, corrigez scale.

Pas encore de matériel Modbus ? pymodbus inclut un simulateur, et le guide Modbus TCP vers MQTT montre comment en lancer un sur un ordinateur portable pour vos tests.

Production : broker privé, TLS et service

Les données de compteurs et de machines sont des informations opérationnelles : en production, les passerelles doivent publier vers un broker privé avec chiffrement. Avec ConnectMQ, un broker MQTT managé, vous créez un broker dans le portail et recevez un nom d'utilisateur, un mot de passe et un client ID. Placez-les dans /etc/modbus2mqtt.env :

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

Ajoutez ensuite une unité systemd avec EnvironmentFile=/etc/modbus2mqtt.env, Restart=always et After=network-online.target, comme dans les autres tutoriels Pi de cette série. Si vous structurez le préfixe en entreprise, site, zone et ligne, le portail ConnectMQ reconnaît la hiérarchie ISA-95 et regroupe les compteurs dans sa vue des équipements, et des règles d'alerte à seuil peuvent signaler des valeurs comme un fault_code de variateur différent de zéro.

Deux autres points à considérer en production. D'abord, si la liaison montante peut être coupée longtemps, ajoutez un tampon sur disque avant l'appel de publication ; le guide de la passerelle IoT Raspberry Pi inclut une file d'attente SQLite réutilisable. Ensuite, si vous prévoyez d'écrire dans les équipements depuis le cloud, ajoutez une validation stricte et une liste blanche des registres modifiables. Une mauvaise écriture Modbus peut arrêter une machine.

Dépannage

Démarrer avec ConnectMQ

Envoyez vos données Modbus là où elles sont utiles : créez un broker sur app.connectmq.com avec l'offre gratuite Serverless (Free), sans carte bancaire, et regardez vos relevés de compteurs arriver dans l'explorateur de topics en direct du portail. Pour la suite de la série Pi, commencez par le tutoriel MQTT sur Raspberry Pi.

Questions fréquentes

Un seul Raspberry Pi peut-il lire plusieurs équipements Modbus RTU ?

Oui. Placez-les sur le même bus RS-485 avec des unit ID uniques et des paramètres série identiques, et la passerelle les interroge l'un après l'autre via un seul client série.

Quelle est la différence entre holding registers et input registers ?

Les input registers sont des mesures en lecture seule, tandis que les holding registers peuvent être lus et souvent écrits, pour les réglages et les consignes. Le manuel de votre équipement indique dans quelle table se trouve chaque valeur.

À quelle vitesse la passerelle peut-elle interroger Modbus RTU ?

Cela dépend du débit, du nombre de registres et du temps de réponse de l'équipement. Regroupez si possible les registres consécutifs en une seule lecture et n'interrogez pas plus souvent que ce dont vos consommateurs ont besoin.

La passerelle doit-elle publier un message par registre ou un par équipement ?

Un message JSON par équipement et par cycle d'interrogation est généralement préférable. Les valeurs liées restent ensemble avec un seul horodatage, et le nombre de messages est bien plus faible.