Passerelle Modbus vers MQTT pour automates : un poller Modbus TCP en Python

Une passerelle Modbus vers MQTT lit les registres d'un automate (PLC) en Modbus TCP et les publie sous forme de messages MQTT, pour que tableaux de bord, historiens et services cloud puissent s'abonner aux données de l'usine sans jamais toucher l'automate. Vous pouvez en écrire une fiable en environ 150 lignes de Python avec pymodbus et paho-mqtt.

Ce tutoriel construit cette passerelle étape par étape : une table de tags, une boucle de scrutation, le décodage des valeurs 32 bits, la publication sur changement, un statut online/offline avec Last Will, et une façon de tout tester avant d'approcher une ligne de production.

Où se place la passerelle

Modbus est un protocole requête/réponse : un maître (client) demande des registres à un esclave (serveur), une requête à la fois. MQTT est publish/subscribe : les producteurs publient une fois, un nombre quelconque de consommateurs s'abonnent. La passerelle fait la traduction entre les deux mondes :

Exécutez la passerelle sur un boîtier edge (PC industriel, Raspberry Pi, conteneur sur le serveur de ligne) qui voit les deux réseaux. L'automate ne voit jamais de trafic MQTT et les consommateurs MQTT n'ouvrent jamais de socket Modbus — exactement la séparation que veulent les équipes OT. Pour la vue d'ensemble, voir MQTT pour l'IIoT et l'Industrie 4.0.

1. Décrire vos tags

Modbus n'a pas de noms, seulement des adresses. La première étape est une table de tags qui dit ce que signifie chaque registre. Gardez-la dans un fichier à côté du code pour la relire comme une configuration :

# nom, fonction, adresse, type de donnée, échelle, unité
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,    ""),
]

Les adresses sont ici des adresses protocole commençant à zéro. La documentation des fabricants utilise souvent la numérotation « 40001 » qui commence à un : le holding register 40001 correspond à l'adresse 0, le 40003 à l'adresse 2. Le décalage d'une unité est le bug Modbus le plus fréquent, donc vérifiez une valeur connue avant de faire confiance au reste.

2. Lire et décoder les registres

Un registre Modbus fait 16 bits. Les flottants et les compteurs 32 bits occupent deux registres, et les fabricants ne s'accordent pas sur le mot qui vient en premier. Décoder avec struct rend l'ordre des mots explicite :

import struct
from pymodbus.client import ModbusTcpClient

WORD_ORDER_BIG = True   # mettre False pour les équipements en « word swap » little-endian

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)

En production, lisez des blocs contigus (par exemple les registres 0–5 en une seule requête) plutôt qu'une requête par tag : un automate répond plus vite à une grosse requête qu'à six petites. La version par tag ci-dessus est plus lisible et suffit pour quelques dizaines de tags.

3. Publier vers MQTT sur changement

Combinez maintenant le poller avec un client MQTT. La passerelle publie un topic par tag, n'envoie une valeur que lorsqu'elle varie au-delà d'une bande morte, marque les valeurs comme retained pour que les nouveaux abonnés reçoivent l'état courant immédiatement, et utilise un Last Will pour annoncer sa déconnexion.

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            # secondes, variation relative (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 connecté :", 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)

Quelques choix qui méritent une explication :

• Un topic par tag (.../plc1/temperature) permet aux consommateurs de s'abonner exactement à ce dont ils ont besoin, et des wildcards comme .../plc1/+ leur donnent tout. Testez vos motifs avec le topic matcher.
• Retained + QoS 1 : un tableau de bord ouvert à 3 h du matin affiche immédiatement la dernière valeur connue, et les messages survivent aux courtes coupures réseau. Voir messages retenus.
• Topic de statut avec Last Will : il distingue « la température est stable » de « la passerelle est morte ». Le broker publie offline à votre place si la connexion de la passerelle tombe sans déconnexion propre.
• Horodatage dans le payload : il enregistre quand la valeur a été lue, pas quand elle est arrivée — important pour les historiens.

L'arborescence de topics suit une hiérarchie de type ISA-95 (entreprise / site / ligne / équipement), point de départ d'un Unified Namespace.

4. Tester sans automate de production

Ne développez jamais contre une ligne en fonctionnement. Commencez avec un simulateur Modbus — pymodbus en fournit un, et des outils comme diagslave ou ModbusPal servent des registres configurables sur le port 502 (ou 5020 si vous n'êtes pas root). Pointez PLC_HOST dessus et lancez la passerelle.

Vérifiez ensuite le côté MQTT :

1. Ouvrez le client MQTT en ligne, connectez-vous à broker.connectmq.com et abonnez-vous à testmqtt/acme/plant1/#. Vous devriez voir aussitôt le status = online retenu et un message par tag.
2. Modifiez un registre dans le simulateur. Seul ce tag doit publier un nouveau message ; les tags inchangés restent silencieux.
3. Arrêtez la passerelle avec kill -9 (sans déconnexion propre). Après environ une fois et demie le keep-alive, status doit passer à offline.
4. Arrêtez le simulateur. status doit indiquer plc-unreachable pendant que MQTT reste connecté.

Suivre les topics de la passerelle dans le navigateur

Depuis un terminal, la même vérification tient en une commande :

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

Durcissement pour la production

Modbus n'a aucune authentification, la passerelle est donc souvent la seule frontière contrôlée entre l'automate et tout le reste. Gardez ses droits MQTT restreints et son chemin d'écriture fermé tant que vous n'en avez pas vraiment besoin.

Si vos sites passent plus tard à Sparkplug B ou ont besoin d'une gestion d'état standardisée pour de nombreux équipements, le même cœur de scrutation peut alimenter un edge node Sparkplug B au lieu de simples topics JSON.

Une passerelle Modbus vers MQTT, c'est peu de code pour beaucoup d'effet : elle transforme des registres que seul le programmeur de l'automate comprend en données nommées, horodatées et accessibles par abonnement. Commencez par une table de tags claire, publiez sur changement avec valeurs retenues et Last Will, testez contre un simulateur et le client en ligne, et seulement ensuite raccordez-la à la ligne.

Questions fréquentes