Modbus TCP a MQTT: sondear registros y publicar datos con Python

Modbus TCP a MQTT es el puente más habitual del IoT industrial, porque Modbus está en todas partes y MQTT es la forma en que esos datos llegan a los dashboards y al cloud. En este tutorial escribirás un poller en Python que lee holding e input registers, los decodifica y escala correctamente, y publica solo los cambios relevantes en un broker MQTT. El resultado es un gateway pequeño y configurable que puedes ejecutar en cualquier equipo Linux o PC industrial.

En qué se diferencia Modbus TCP de MQTT

Modbus es un protocolo de petición/respuesta. Un cliente (históricamente "maestro") pide a un servidor (históricamente "esclavo") un bloque de registros y los recibe. No se envía nada por iniciativa propia: si quieres un valor, tienes que sondearlo. Modbus no tiene noción de nombres, unidades ni tipos de datos más allá de registros de 16 bits y bits individuales.

MQTT es lo contrario: publicación/suscripción orientada a eventos a través de un broker, con topics que tienen significado. La tarea del puente es convertir sondeos periódicos en eventos con significado. Eso implica tres cosas: conocer el mapa de registros, decodificar los registros en bruto a valores de ingeniería y decidir cuándo merece la pena publicar un valor.

Repaso del modelo de datos Modbus

Hay dos detalles que hacen tropezar a casi todas las primeras implementaciones:

• Desplazamientos de direccionamiento. La documentación suele listar los holding registers como 40001, 40002, etc. En el cable, 40001 es la dirección 0. Las librerías usan la dirección con base cero.
• Valores de 32 bits. Los floats y los enteros de 32 bits ocupan dos registros, y los fabricantes no se ponen de acuerdo en el orden de las palabras. Unos envían primero la palabra alta (big-endian, a menudo escrito ABCD) y otros primero la baja (CDAB). Si tu temperatura se lee como 1.2e-38 o 3.5e+12, prueba a intercambiar las palabras.

Requisitos previos

• Python 3.10 o posterior.
• pymodbus 3.8 y paho-mqtt 2.x. Fija la versión de pymodbus: sus argumentos con nombre han cambiado entre versiones menores.
• Un dispositivo Modbus TCP, o el simulador que aparece a continuación.

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

Paso 1: un simulador Modbus TCP (opcional)

Para desarrollar sin hardware, ejecuta un pequeño servidor que expone una temperatura como float de 32 bits en los holding registers 0 y 1, una presión como entero escalado en el input register 0 y un estado de marcha en la coil 0.

# modbus_sim.py - servidor Modbus TCP en el puerto 5020 con valores cambiantes
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())

El puerto 5020 evita necesitar privilegios de root para el puerto 502. En un dispositivo real, usa el puerto 502 y el unit ID que indique su manual.

Paso 2: describe el mapa de registros

Mantén el mapa en la configuración para que añadir un tag nunca requiera cambios de código. Cada tag tiene una tabla, una dirección, un tipo de dato, una escala, una unidad, una banda muerta y una ruta 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": "" }
  ]
}

Guárdalo como modbus_map.json y sustituye b72f por tu propia cadena aleatoria.

Paso 3: el puente Modbus TCP a MQTT

El siguiente poller implementa el informe por excepción: un valor analógico se publica cuando se mueve más que su banda muerta, un booleano cuando cambia, y todos los valores se republican con un intervalo de heartbeat para que los consumidores sepan que el puente sigue vivo. Más información sobre el patrón en nuestro artículo sobre el informe por excepción.

# modbus_mqtt_bridge.py - sondear Modbus TCP, publicar los cambios 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   # producción: 8883 + TLS
MQTT_USER, MQTT_PASS = None, None                     # producción: "YOUR_USERNAME", "YOUR_PASSWORD"
CLIENT_ID = f"modbus-bridge-{uuid.uuid4().hex[:8]}"   # producción: "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()

Paso 4: ejecutar y verificar

python modbus_sim.py            # terminal 1, omítelo si tienes un dispositivo real
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

Deberías ver la temperatura solo cuando deriva medio grado, la presión solo cuando se mueve 0,05 bar y el indicador de marcha solo cuando cambia. Además, cada valor reaparece una vez por minuto como heartbeat. Es una fracción de los mensajes que enviaría un bucle ingenuo de "publicar en cada sondeo", algo que importa en el servidor MQTT público gratuito, pensado para pruebas y no para datos de producción, donde cada cliente puede publicar como máximo diez mensajes por segundo, y en cualquier conexión de pago por uso.

Consejos de diseño para el sondeo

1. Lee bloques contiguos. Una petición de los registros 0 a 19 es mucho más barata que veinte peticiones de un registro. Cuando tu mapa crezca, agrupa los tags por tabla y rango de direcciones y decodifica a partir del bloque.
2. Respeta el dispositivo. Muchos PLC pequeños y medidores solo admiten unas pocas peticiones por segundo. Empieza con un sondeo de un segundo y mide.
3. Usa el unit ID correcto. Los dispositivos Modbus TCP detrás de un gateway usan el unit ID para direccionar dispositivos serie; los dispositivos independientes a menudo lo ignoran o esperan 1 o 255.
4. Separa los tags rápidos de los lentos. Sondea rápido las alarmas y los estados, y despacio los contadores de energía.
5. Publica la calidad de forma explícita. Una lectura fallida debe publicar quality: bad, no mantener en silencio el último valor.
6. Nunca escribas desde el poller por accidente. Deja la escritura fuera del puente, o ponla detrás de un topic de comandos separado y autenticado.

Diseño de topics y payloads para datos Modbus

Modbus en bruto te da direcciones, no significado, así que el árbol de topics es donde el puente añade contexto. Nombra los topics según lo que es el valor y dónde está, nunca según el número de registro: plant1/utilities/boiler1/temperature le sirve a cualquiera, mientras que device3/hr/0 solo le sirve a quien escribió el mapa de registros. Una jerarquía de estilo ISA-95 de sitio, área, línea y equipo mantiene navegables cientos de dispositivos Modbus y permite a los consumidores suscribirse a un área entera con un solo comodín.

Usa un valor por topic para los datos de proceso que los consumidores eligen individualmente, e incluye unidad, calidad y marca de tiempo en cada payload para que un mensaje se explique por sí solo cuando aparezca en un log meses después. Si el dispositivo tiene muchos registros que siempre se usan juntos, como todas las lecturas de un medidor, un único objeto JSON por dispositivo y ciclo es una alternativa razonable que reduce el número de mensajes. Elijas lo que elijas, aplícalo de forma coherente en todos los puentes de la planta. Esa misma jerarquía es la que después permite enlazar cada topic con su equipo en un dashboard o en un gemelo digital 3D de la planta.

Paso a producción

Para datos reales de planta, cambia a un broker privado: pon MQTT_PORT a 8883 y rellena las credenciales de tu broker MQTT gestionado. Con ConnectMQ, cada broker incluye un nombre de usuario, una contraseña y un client ID generados; el client ID está vinculado a esas credenciales, así que úsalo en lugar de uno aleatorio. Ejecuta el puente como servicio systemd con Restart=always y considera un búfer store and forward si el enlace a internet de la planta no es fiable.

Si tus dispositivos hablan Modbus RTU sobre RS-485 en lugar de Ethernet, el mismo puente funciona con un cliente serie; nuestro tutorial Modbus RTU a MQTT trata el cableado y los ajustes serie. Para la arquitectura global de gateways, brokers y consumidores, consulta la guía de IoT industrial con MQTT.

Resolución de problemas

Empieza con ConnectMQ

Saca tus datos Modbus de la red de planta sin operar un servidor. Regístrate en app.connectmq.com, elige el plan Serverless gratuito (sin tarjeta de crédito), crea un broker y pega las credenciales generadas en el puente. Los valores de tus registros aparecen de inmediato en el explorador de topics en vivo de ConnectMQ, y las reglas de alerta por umbral pueden avisarte cuando una temperatura o una presión salgan de su rango.

Preguntas frecuentes

¿Con qué frecuencia debo sondear los registros Modbus TCP?

Empieza con un segundo para los valores de proceso y más despacio para los contadores, y después ajústalo a lo que tolere el dispositivo y a lo que necesiten los consumidores. Publica ante cambios en lugar de en cada sondeo.

¿Por qué mis valores float de Modbus parecen incorrectos?

Lo más habitual es que las dos palabras de 16 bits estén en el orden contrario al que supone tu código. Intercambia el orden de las palabras y comprueba si el registro es en realidad un entero escalado en lugar de un float.

¿Puede un puente sondear varios dispositivos Modbus TCP?

Sí. Ejecuta un cliente por dispositivo, o un cliente por gateway con distintos unit IDs, y da a cada dispositivo su propia rama en el árbol de topics.

¿El puente debe usar QoS 0 o QoS 1?

QoS 1 con retained messages es un buen valor por defecto para los valores de proceso; garantiza la entrega y permite que los nuevos suscriptores vean el estado actual al instante.