Gateway Raspberry Pi Modbus MQTT com pymodbus (RTU e TCP)

Um gateway raspberry pi modbus mqtt permite coletar leituras de medidores de energia, inversores, inversores de frequência e CLPs que só falam Modbus e publicá-las como JSON limpo via MQTT. Neste tutorial você vai ligar um adaptador USB RS-485 para Modbus RTU, conectar-se a dispositivos Modbus TCP pela Ethernet, descrever os registradores de cada equipamento em uma pequena configuração, decodificar corretamente inteiros com escala e floats de 32 bits e publicar os resultados em um broker. O script final trata timeouts e reconexões, então pode rodar sem supervisão, seja em uma subestação, em um frigorífico ou em uma planta de manufatura.

Modbus em dois minutos

O Modbus é um protocolo de requisição e resposta: o Pi (o cliente, historicamente chamado de mestre) pede a um dispositivo (o servidor, historicamente escravo) um bloco de registradores, e o dispositivo responde. Não existe a noção de topics, assinaturas ou push; tudo é polling. É exatamente por isso que um gateway é necessário para alimentar sistemas MQTT.

Os dados ficam em quatro tabelas: coils e discrete inputs (bits), e input registers e holding registers (palavras de 16 bits). A maioria dos medidores expõe as medições como holding ou input registers. Valores que não cabem em 16 bits, como totais de energia ou tensões em ponto flutuante, ocupam dois registradores consecutivos, e o manual do equipamento informa a ordem das palavras. Errar essa ordem é o motivo mais comum para um gateway publicar números absurdos.

Para um aprofundamento no protocolo de cada variante, veja os guias de Modbus RTU para MQTT e Modbus TCP para MQTT.

Hardware e ligação para Modbus RTU

1. Conecte um adaptador USB RS-485 ao Pi. Ele aparece como /dev/ttyUSB0. Use ls /dev/serial/by-id/ para obter um nome estável que sobreviva a reboots e reconexões.
2. Ligue A com A (às vezes identificado como D+) e B com B (D-). Se não houver resposta nenhuma, inverter A e B é a primeira coisa a tentar, porque os fabricantes divergem na identificação.
3. Ligue o terra de sinal se o adaptador e o dispositivo tiverem um. Lances longos sem uma referência comum causam erros intermitentes.
4. Termine as duas pontas físicas do barramento com resistores de 120 ohms em lances mais longos. Muitos dispositivos e adaptadores têm uma chave de terminação.
5. Anote baud rate, paridade, stop bits e unit ID do dispositivo a partir do display ou do manual. Todos os dispositivos de um barramento precisam compartilhar as mesmas configurações seriais.

Para Modbus TCP você só precisa do endereço IP do dispositivo, da porta (geralmente 502) e do unit ID. Atrás de um gateway Modbus TCP para RTU, o unit ID seleciona o dispositivo RTU do outro lado.

Instale a pymodbus e a 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     # acesso serial; depois saia e entre na sessão

A pymodbus 3.10 renomeou o argumento de unidade de slave para device_id. O código abaixo usa device_id; se você estiver preso a uma versão 3.x mais antiga, volte para slave.

Faça uma verificação rápida em um shell Python antes de escrever algo maior:

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

Se isso imprimir valores de registradores, a ligação e as configurações estão corretas. Se imprimir um timeout, verifique A/B, baud rate, paridade e unit ID, nessa ordem.

Descreva os dispositivos em um mapa de registradores

Deixar endereços fixos no código Python vira bagunça rapidamente. Coloque os registradores de cada dispositivo em um arquivo JSON. Salve 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}
      ]
    }
  ]
}

Os endereços acima são exemplos. Use a tabela de registradores do manual do seu próprio equipamento e fique atento a manuais que listam endereços a partir de 1 ou com um offset de 40001: a pymodbus espera o endereço de protocolo começando em zero, então o registrador 40001 é o endereço 0.

O script do gateway Raspberry Pi Modbus MQTT

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

# ---------- polling loop ----------
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}"
        # Dispositivos RTU na mesma porta compartilham um único 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 design que vale a pena entender:

• Um cliente por conexão física. Dispositivos RTU na mesma porta RS-485 compartilham um único cliente serial, porque dois processos ou objetos abrindo a mesma porta corrompem os quadros um do outro.
• Topics de status por dispositivo. Cada dispositivo recebe um online ou offline retained, publicado apenas quando o estado muda. Os dashboards veem na hora qual medidor parou de responder, separado do status do próprio gateway.
• Os registradores brutos são decodificados em um só lugar. A função decode trata tipos de 16 e 32 bits e a ordem das palavras com struct, então não depende de APIs auxiliares que mudaram entre versões da pymodbus.
• Intervalos de polling por dispositivo. A velocidade de um inversor a cada 5 segundos e um total de energia a cada 60 é uma combinação sensata; não faça polling mais rápido do que seus consumidores precisam.

Teste a saída Raspberry Pi Modbus MQTT

Rode em primeiro plano:

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

Acompanhe a saída no servidor MQTT público gratuito a partir de outro terminal (use-o só para testes, nunca com dados reais de produção):

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

Saída 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}}

Compare pelo menos um valor com o display do próprio equipamento. Se um float estiver totalmente errado (expoentes enormes ou frações minúsculas), inverta o word_order. Se um valor com escala estiver errado por um fator de dez, corrija o scale.

Ainda não tem hardware Modbus? A pymodbus inclui um simulador, e o guia de Modbus TCP para MQTT mostra como rodar um em um notebook para testes.

Produção: broker privado, TLS e um serviço

Dados de medidores e máquinas são informação operacional, então gateways de produção devem publicar em um broker privado com criptografia. Com o ConnectMQ, um broker MQTT gerenciado, você cria um broker no portal e recebe usuário, senha e client ID. Coloque-os em /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

Depois, crie uma unit do systemd com EnvironmentFile=/etc/modbus2mqtt.env, Restart=always e After=network-online.target, como nos outros tutoriais de Pi desta série. Se você estruturar o prefixo como empresa, site, área e linha, o portal ConnectMQ reconhece a hierarquia ISA-95 e agrupa os medidores na visão de dispositivos, e regras de alerta por limite podem sinalizar valores como um fault_code de inversor diferente de zero. Essa mesma árvore de topics padronizada é uma boa base para alimentar dashboards ou um gêmeo digital da planta.

Mais duas considerações de produção. Primeiro, se o uplink puder ficar fora do ar por longos períodos, adicione um buffer em disco antes da chamada de publish; o guia de gateway IoT com Raspberry Pi inclui uma outbox em SQLite que você pode reutilizar. Segundo, se pretende escrever nos dispositivos a partir da nuvem, adicione validação rigorosa e uma lista de registradores com escrita permitida. Uma escrita Modbus errada pode parar uma máquina.

Solução de problemas

Comece com o ConnectMQ

Envie seus dados Modbus para onde eles sejam úteis: crie um broker em app.connectmq.com no plano Serverless gratuito, sem cartão de crédito, e veja as leituras dos seus medidores chegarem no explorador de topics ao vivo do portal. Para o restante da série sobre o Pi, comece pelo tutorial MQTT com Raspberry Pi.

Perguntas frequentes

Um único Raspberry Pi consegue ler vários dispositivos Modbus RTU?

Sim. Coloque-os no mesmo barramento RS-485 com unit IDs únicos e configurações seriais idênticas, e o gateway os consulta um após o outro por meio de um único cliente serial.

Qual a diferença entre holding registers e input registers?

Input registers são medições somente leitura, enquanto holding registers podem ser lidos e muitas vezes escritos, para configurações e setpoints. O manual do equipamento diz em qual tabela cada valor fica.

Com que rapidez o gateway consegue fazer polling de Modbus RTU?

Depende do baud rate, do número de registradores e do tempo de resposta do dispositivo. Agrupe registradores consecutivos em uma única leitura sempre que possível e faça polling apenas com a frequência de que seus consumidores precisam.

O gateway deve publicar uma mensagem por registrador ou uma por dispositivo?

Uma mensagem JSON por dispositivo e por ciclo de polling costuma ser o melhor. Isso mantém os valores relacionados juntos com um único timestamp e usa muito menos mensagens.