Gateway Modbus MQTT con Raspberry Pi y pymodbus (RTU y TCP)

Un gateway Raspberry Pi Modbus MQTT te permite tomar lecturas de medidores de energía, inversores, variadores de frecuencia y PLC que solo hablan Modbus y publicarlas como JSON limpio por MQTT. En este tutorial cablearás un adaptador USB RS-485 para Modbus RTU, te conectarás a equipos Modbus TCP por Ethernet, describirás los registros de cada equipo en una pequeña configuración, decodificarás correctamente enteros escalados y floats de 32 bits y publicarás los resultados en un broker. El script final gestiona timeouts y reconexiones, así que puede funcionar desatendido.

Modbus en dos minutos

Modbus es un protocolo de petición y respuesta: la Pi (el cliente, históricamente llamado maestro) pide a un equipo (el servidor, históricamente esclavo) un bloque de registros, y el equipo responde. No tiene concepto de topics, suscripciones ni envío push; todo es polling. Precisamente por eso hace falta un gateway para alimentar sistemas MQTT.

Los datos viven en cuatro tablas: coils y entradas discretas (bits), y registros de entrada y registros holding (palabras de 16 bits). La mayoría de los medidores exponen sus mediciones como registros holding o de entrada. Los valores que no caben en 16 bits, como los totales de energía o las tensiones en coma flotante, ocupan dos registros consecutivos, y el manual del equipo indica el orden de las palabras. Equivocarse con ese orden es la razón más común por la que un gateway publica números absurdos.

Para profundizar en cada variante del protocolo, consulta las guías de Modbus RTU a MQTT y Modbus TCP a MQTT.

Hardware y cableado para Modbus RTU

1. Conecta un adaptador USB RS-485 a la Pi. Aparece como /dev/ttyUSB0. Usa ls /dev/serial/by-id/ para obtener un nombre estable que sobreviva a reinicios y reconexiones.
2. Conecta A con A (a veces rotulado D+) y B con B (D-). Si no ves ninguna respuesta, lo primero que debes probar es intercambiar A y B, porque los fabricantes no se ponen de acuerdo en el rotulado.
3. Conecta la tierra de señal si el adaptador y el equipo la ofrecen. Los tendidos largos sin una referencia común provocan errores intermitentes.
4. Termina los dos extremos físicos del bus con resistencias de 120 ohmios en tendidos largos. Muchos equipos y adaptadores tienen un interruptor de terminación.
5. Anota la velocidad en baudios, la paridad, los bits de parada y el unit ID del equipo desde su pantalla o su manual. Todos los equipos de un mismo bus deben compartir la misma configuración serie.

Para Modbus TCP solo necesitas la dirección IP del equipo, el puerto (normalmente 502) y el unit ID. Detrás de un gateway Modbus TCP a RTU, el unit ID selecciona el equipo RTU del otro lado.

Instala pymodbus y 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     # acceso al puerto serie; después cierra sesión y vuelve a entrar

pymodbus 3.10 cambió el nombre del argumento de unidad de slave a device_id. El código de abajo usa device_id; si estás atado a una versión 3.x anterior, vuelve a cambiarlo a slave.

Haz una comprobación rápida desde una consola de Python antes de escribir algo más grande:

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 imprime valores de registros, el cableado y la configuración son correctos. Si imprime un timeout, revisa A/B, baudios, paridad y unit ID, en ese orden.

Describe los equipos en un mapa de registros

Escribir direcciones a mano en Python se vuelve un lío enseguida. Pon los registros de cada equipo en un archivo JSON. Guárdalo como ~/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}
      ]
    }
  ]
}

Las direcciones anteriores son ejemplos. Usa la tabla de registros del manual de tu propio equipo y presta atención a los manuales que numeran las direcciones empezando en 1 o con un desplazamiento de 40001: pymodbus espera la dirección de protocolo con base cero, así que el registro 40001 es la dirección 0.

El script del gateway Raspberry Pi Modbus MQTT

Guárdalo como ~/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()

# ---------- bucle de polling ----------
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}"
        # Los equipos RTU del mismo puerto comparten un solo cliente
        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)

Notas de diseño que conviene entender:

• Un cliente por conexión física. Los equipos RTU del mismo puerto RS-485 comparten un único cliente serie, porque dos procesos u objetos que abren el mismo puerto corrompen las tramas del otro.
• Topics de estado por equipo. Cada equipo recibe un online u offline retained, publicado solo cuando cambia el estado. Los dashboards ven al instante qué medidor ha dejado de responder, por separado del estado del propio gateway.
• Los registros en bruto se decodifican en un solo lugar. La función decode gestiona tipos de 16 y 32 bits y el orden de palabras con struct, así que no depende de APIs auxiliares que han cambiado entre versiones de pymodbus.
• Intervalos de polling por equipo. La velocidad de un variador cada 5 segundos y un total de energía cada 60 es una combinación sensata; no consultes más rápido de lo que necesitan tus consumidores.

Prueba la salida Raspberry Pi Modbus MQTT

Ejecútalo en primer plano:

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

Observa la salida en el servidor MQTT público gratuito desde otra terminal (solo para pruebas, no para datos de producción):

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

Salida esperada:

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

Compara al menos un valor con la pantalla del propio equipo. Si un float sale disparatado (exponentes enormes o fracciones diminutas), invierte word_order. Si un valor escalado está desviado por un factor de diez, corrige scale.

¿Aún no tienes hardware Modbus? pymodbus incluye un simulador, y la guía de Modbus TCP a MQTT muestra cómo ejecutarlo en un portátil para hacer pruebas.

Producción: broker privado, TLS y un servicio

Los datos de medidores y máquinas son información operativa, así que los gateways de producción deberían publicar en un broker privado con cifrado. Con ConnectMQ, un broker MQTT gestionado, creas un broker en el portal y recibes un nombre de usuario, una contraseña y un client ID. Colócalos en /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

Después crea una unidad systemd con EnvironmentFile=/etc/modbus2mqtt.env, Restart=always y After=network-online.target, como en los demás tutoriales de Pi de esta serie. Si estructuras el prefijo como empresa, sitio, área y línea, el portal de ConnectMQ reconoce la jerarquía ISA-95 y agrupa los medidores en su vista de dispositivos, y las reglas de alerta por umbral pueden señalar valores como un fault_code de un variador distinto de cero. Esa misma jerarquía es una buena base si más adelante quieres visualizar la planta como un gemelo digital 3D.

Dos consideraciones más para producción. Primero, si el enlace puede caerse durante periodos largos, añade un buffer en disco delante de la llamada de publicación; la guía del gateway IoT con Raspberry Pi incluye una bandeja de salida en SQLite que puedes reutilizar. Segundo, si piensas escribir en los equipos desde la nube, añade una validación estricta y una lista blanca de registros escribibles. Una escritura Modbus errónea puede parar una máquina.

Solución de problemas

Empieza con ConnectMQ

Envía tus datos Modbus a un lugar útil: crea un broker en app.connectmq.com con el plan gratuito Serverless, sin tarjeta de crédito, y observa cómo llegan las lecturas de tus medidores al explorador de topics en vivo del portal. Para el resto de la serie sobre Pi, empieza por el tutorial de MQTT con Raspberry Pi.

Preguntas frecuentes

¿Puede una Raspberry Pi leer varios equipos Modbus RTU?

Sí. Ponlos en el mismo bus RS-485 con unit IDs únicos y la misma configuración serie, y el gateway los consultará uno tras otro a través de un único cliente serie.

¿Qué diferencia hay entre registros holding y registros de entrada?

Los registros de entrada son mediciones de solo lectura, mientras que los registros holding se pueden leer y a menudo escribir, para configuraciones y consignas. El manual de tu equipo indica en qué tabla está cada valor.

¿A qué velocidad puede consultar el gateway por Modbus RTU?

Depende de los baudios, del número de registros y del tiempo de respuesta del equipo. Agrupa registros consecutivos en una sola lectura cuando sea posible y consulta solo con la frecuencia que necesiten tus consumidores.

¿El gateway debe publicar un mensaje por registro o uno por equipo?

Normalmente lo mejor es un mensaje JSON por equipo y por consulta. Mantiene juntos los valores relacionados con una sola marca de tiempo y usa muchos menos mensajes.