Modbus TCP vers MQTT : lire des registres et publier les données avec Python

Modbus TCP vers MQTT est le pont le plus courant de l'IoT industriel, car Modbus est partout et MQTT est le moyen par lequel ces données atteignent les tableaux de bord et le cloud. Dans ce tutoriel, vous allez écrire un poller Python qui lit les holding registers et input registers, les décode et les met à l'échelle correctement, puis ne publie que les changements significatifs vers un broker MQTT. Le résultat est une petite passerelle configurable que vous pouvez exécuter sur n'importe quelle machine Linux ou PC industriel.

En quoi Modbus TCP diffère de MQTT

Modbus est un protocole requête/réponse. Un client (historiquement « maître ») demande un bloc de registres à un serveur (historiquement « esclave ») et le reçoit en retour. Rien n'est poussé : si vous voulez une valeur, vous devez l'interroger. Modbus n'a aucune notion de noms, d'unités ou de types de données au-delà des registres de 16 bits et des bits isolés.

MQTT est l'inverse : du publish/subscribe piloté par les événements via un broker, avec des topics porteurs de sens. Le rôle du pont est de transformer des interrogations périodiques en événements significatifs. Cela implique trois choses : connaître la table des registres, décoder les registres bruts en valeurs physiques et décider quand une valeur mérite d'être publiée.

Rappel sur le modèle de données Modbus

Deux détails piègent presque toutes les premières implémentations :

• Les décalages d'adressage. La documentation liste souvent les holding registers sous la forme 40001, 40002, etc. Sur le réseau, 40001 correspond à l'adresse 0. Les bibliothèques utilisent l'adresse commençant à zéro.
• Les valeurs 32 bits. Les flottants et entiers 32 bits occupent deux registres, et les fabricants ne s'accordent pas sur l'ordre des mots. Certains envoient le mot de poids fort en premier (big-endian, souvent noté ABCD), d'autres le mot de poids faible en premier (CDAB). Si votre température affiche 1.2e-38 ou 3.5e+12, essayez d'inverser les mots.

Prérequis

• Python 3.10 ou plus récent.
• pymodbus 3.8 et paho-mqtt 2.x. Fixez la version de pymodbus : ses arguments nommés ont changé entre versions mineures.
• Un appareil Modbus TCP, ou le simulateur ci-dessous.

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

Étape 1 : un simulateur Modbus TCP (facultatif)

Pour développer sans matériel, lancez un petit serveur qui expose une température en flottant 32 bits dans les holding registers 0 et 1, une pression en entier mis à l'échelle dans l'input register 0 et un état de marche dans la coil 0.

# modbus_sim.py - serveur Modbus TCP sur le port 5020 avec des valeurs variables
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 registers
        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())

Le port 5020 évite d'avoir besoin des privilèges root requis pour le port 502. Sur un vrai appareil, utilisez le port 502 et l'unit ID indiqué dans son manuel.

Étape 2 : décrire la table des registres

Gardez la table dans la configuration pour qu'ajouter un tag n'exige jamais de modifier le code. Chaque tag a une table, une adresse, un type de données, un facteur d'échelle, une unité, une bande morte et un chemin de topic.

{
  "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": "" }
  ]
}

Enregistrez-la sous modbus_map.json et remplacez b72f par votre propre chaîne aléatoire.

Étape 3 : le pont Modbus TCP vers MQTT

Le poller ci-dessous implémente le report by exception : une valeur analogique est publiée lorsqu'elle varie de plus que sa bande morte, un booléen lorsqu'il change, et chaque valeur est republiée à intervalle de heartbeat pour que les consommateurs sachent que le pont est actif. Pour en savoir plus sur ce modèle, lisez notre article sur le report by exception.

# modbus_mqtt_bridge.py - interroge Modbus TCP, publie les changements en MQTT
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   # production: 8883 + TLS
MQTT_USER, MQTT_PASS = None, None                     # production: "YOUR_USERNAME", "YOUR_PASSWORD"
CLIENT_ID = f"modbus-bridge-{uuid.uuid4().hex[:8]}"   # production: "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()

Étape 4 : exécuter et vérifier

python modbus_sim.py            # terminal 1, à ignorer si vous avez un vrai appareil
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

Vous devriez voir la température uniquement lorsqu'elle dérive d'un demi-degré, la pression uniquement lorsqu'elle varie de 0,05 bar et l'indicateur de marche uniquement lorsqu'il bascule. Chaque valeur réapparaît aussi une fois par minute en guise de heartbeat. C'est une fraction des messages qu'enverrait une boucle naïve « publier à chaque lecture », ce qui compte sur le serveur MQTT public gratuit, où chaque client peut publier au maximum dix messages par seconde, ainsi que sur toute connexion facturée au volume.

Conseils de conception du polling

1. Lire des blocs contigus. Une requête pour les registres 0 à 19 coûte bien moins que vingt requêtes d'un registre. Quand votre table grandit, regroupez les tags par table et par plage d'adresses et décodez à partir du bloc.
2. Respecter l'appareil. De nombreux petits automates et compteurs ne supportent que quelques requêtes par seconde. Commencez avec une lecture par seconde et mesurez.
3. Utiliser le bon unit ID. Les appareils Modbus TCP derrière une passerelle utilisent l'unit ID pour adresser les appareils série ; les appareils autonomes l'ignorent souvent ou attendent 1 ou 255.
4. Séparer tags rapides et lents. Interrogez rapidement les alarmes et les états, lentement les compteurs d'énergie.
5. Publier explicitement la qualité. Une lecture échouée doit publier quality: bad, et non conserver silencieusement la dernière valeur.
6. Ne jamais écrire par accident depuis le poller. Tenez l'écriture hors du pont, ou placez-la derrière un topic de commande distinct et authentifié.

Concevoir topics et payloads pour les données Modbus

Modbus brut vous donne des adresses, pas du sens : c'est donc dans l'arborescence des topics que le pont ajoute le contexte. Nommez les topics d'après ce qu'est la valeur et où elle se trouve, jamais d'après le numéro de registre : plant1/utilities/boiler1/temperature est utile à tout le monde, alors que device3/hr/0 n'est utile qu'à celui qui a écrit la table des registres. Une hiérarchie de style ISA-95 (site, zone, ligne, équipement) garde des centaines d'appareils Modbus navigables et permet aux consommateurs de s'abonner à une zone entière avec un seul wildcard.

Gardez une valeur par topic pour les données de procédé que les consommateurs choisissent individuellement, et incluez unité, qualité et horodatage dans chaque payload pour qu'un message soit compréhensible lorsqu'il réapparaît dans un journal des mois plus tard. Si l'appareil comporte de nombreux registres toujours utilisés ensemble, comme toutes les mesures d'un compteur, un objet JSON unique par appareil et par cycle est une alternative raisonnable qui réduit le nombre de messages. Quel que soit votre choix, appliquez-le de manière cohérente à tous les ponts du site.

Passer en production

Pour de vraies données d'usine, passez à un broker privé : réglez MQTT_PORT sur 8883 et renseignez les identifiants de votre broker MQTT managé. Avec ConnectMQ, chaque broker est fourni avec un nom d'utilisateur, un mot de passe et un client ID générés ; le client ID est lié à ces identifiants, utilisez-le donc plutôt qu'un identifiant aléatoire. Exécutez le pont en tant que service systemd avec Restart=always, et envisagez une mise en tampon store and forward si la liaison Internet du site n'est pas fiable.

Si vos appareils parlent Modbus RTU sur RS-485 plutôt qu'Ethernet, le même pont fonctionne avec un client série ; notre tutoriel Modbus RTU vers MQTT couvre le câblage et les paramètres série. Pour l'architecture globale des passerelles, brokers et consommateurs, consultez le guide de l'IoT industriel avec MQTT.

Dépannage

Démarrer avec ConnectMQ

Sortez vos données Modbus du réseau de l'usine sans exploiter de serveur. Inscrivez-vous sur app.connectmq.com, choisissez l'offre Serverless gratuite (sans carte bancaire), créez un broker et collez les identifiants générés dans le pont. Les valeurs de vos registres apparaissent immédiatement dans l'explorateur de topics en temps réel de ConnectMQ, et des règles d'alerte à seuil peuvent vous prévenir lorsqu'une température ou une pression sort de sa plage.

Questions fréquentes

À quelle fréquence faut-il interroger les registres Modbus TCP ?

Commencez à une seconde pour les valeurs de procédé et plus lentement pour les compteurs, puis ajustez selon ce que l'appareil supporte et ce dont les consommateurs ont besoin. Publiez sur changement plutôt qu'à chaque lecture.

Pourquoi mes valeurs flottantes Modbus semblent-elles fausses ?

Le plus souvent, les deux mots de 16 bits sont dans l'ordre inverse de celui supposé par votre code. Inversez l'ordre des mots, et vérifiez si le registre n'est pas en réalité un entier mis à l'échelle plutôt qu'un flottant.

Un même pont peut-il interroger plusieurs appareils Modbus TCP ?

Oui. Utilisez un client par appareil, ou un client par passerelle avec différents unit IDs, et donnez à chaque appareil sa propre branche dans l'arborescence des topics.

Le pont doit-il utiliser la QoS 0 ou la QoS 1 ?

La QoS 1 avec retained messages est un bon choix par défaut pour les valeurs de procédé : elle garantit la livraison et permet aux nouveaux abonnés de voir immédiatement l'état actuel.