Modbus TCP para MQTT: Lendo Registradores e Publicando Dados com Python

Modbus TCP para MQTT é a ponte mais comum da IoT industrial, porque o Modbus está em toda parte e o MQTT é o caminho pelo qual esses dados chegam a dashboards, a um gêmeo digital da planta e à nuvem. Neste tutorial você vai escrever um poller em Python que lê holding e input registers, decodifica e escala os valores corretamente e publica apenas as mudanças relevantes em um broker MQTT. O resultado é um gateway pequeno e configurável que você pode rodar em qualquer máquina Linux ou PC industrial.

Como o Modbus TCP difere do MQTT

O Modbus é um protocolo de requisição/resposta. Um client (historicamente "mestre") pede a um servidor (historicamente "escravo") um bloco de registradores e os recebe de volta. Nada é enviado espontaneamente: se você quer um valor, precisa fazer polling. O Modbus não tem noção de nomes, unidades ou tipos de dados além de registradores de 16 bits e bits individuais.

O MQTT é o oposto: publish/subscribe orientado a eventos por meio de um broker, com topics que carregam significado. A função da ponte é transformar leituras periódicas em eventos significativos. Isso significa três coisas: conhecer o mapa de registradores, decodificar os registradores brutos em valores de engenharia e decidir quando vale a pena publicar um valor.

Revisão do modelo de dados Modbus

Dois detalhes derrubam quase toda primeira implementação:

• Deslocamento de endereços. A documentação costuma listar os holding registers como 40001, 40002 e assim por diante. No protocolo, 40001 é o endereço 0. As bibliotecas usam o endereço baseado em zero.
• Valores de 32 bits. Floats e inteiros de 32 bits ocupam dois registradores, e os fabricantes discordam sobre a ordem das palavras. Alguns enviam primeiro a palavra mais significativa (big-endian, muitas vezes escrito ABCD), outros primeiro a menos significativa (CDAB). Se a sua temperatura aparecer como 1.2e-38 ou 3.5e+12, tente inverter as palavras.

Pré-requisitos

• Python 3.10 ou mais recente.
• pymodbus 3.8 e paho-mqtt 2.x. Fixe a versão do pymodbus: seus argumentos nomeados mudaram entre versões menores.
• Um dispositivo Modbus TCP ou o simulador abaixo.

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

Passo 1: um simulador Modbus TCP (opcional)

Para desenvolver sem hardware, rode um pequeno servidor que expõe uma temperatura como float de 32 bits nos holding registers 0 e 1, uma pressão como inteiro escalado no input register 0 e um status de funcionamento na coil 0.

# modbus_sim.py - servidor Modbus TCP na porta 5020 com valores variando
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())

A porta 5020 evita a necessidade de privilégios de root para a porta 502. Em um dispositivo real, use a porta 502 e o unit ID indicado no manual.

Passo 2: descreva o mapa de registradores

Mantenha o mapa na configuração, para que adicionar uma tag nunca exija mudanças no código. Cada tag tem tabela, endereço, tipo de dado, escala, unidade, banda morta e caminho 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": "" }
  ]
}

Salve-o como modbus_map.json e substitua b72f por uma string aleatória sua.

Passo 3: a ponte Modbus TCP para MQTT

O poller abaixo implementa report by exception: um valor analógico é publicado quando varia mais do que sua banda morta, um booleano quando muda, e todos os valores são republicados em um intervalo de heartbeat para que os consumidores saibam que a ponte está viva. Saiba mais sobre o padrão no nosso artigo sobre report by exception.

# modbus_mqtt_bridge.py - faz polling do Modbus TCP e publica as mudanças no 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   # produção: 8883 + TLS
MQTT_USER, MQTT_PASS = None, None                     # produção: "YOUR_USERNAME", "YOUR_PASSWORD"
CLIENT_ID = f"modbus-bridge-{uuid.uuid4().hex[:8]}"   # produção: "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()

Passo 4: execute e verifique

python modbus_sim.py            # terminal 1, pule se tiver um 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

Você deverá ver a temperatura apenas quando ela variar meio grau, a pressão apenas quando mudar 0,05 bar e o indicador de funcionamento apenas quando ele alternar. Cada valor também reaparece uma vez por minuto como heartbeat. Isso é uma fração das mensagens que um loop ingênuo de "publicar a cada leitura" enviaria, o que importa no servidor MQTT público, pensado para testes e protótipos e não para dados de produção, onde cada client pode publicar no máximo dez mensagens por segundo, e em qualquer conexão tarifada por uso.

Dicas de projeto para o polling

1. Leia blocos contíguos. Uma requisição para os registradores 0 a 19 é muito mais barata do que vinte requisições de um registrador cada. Quando seu mapa crescer, agrupe as tags por tabela e faixa de endereços e decodifique a partir do bloco.
2. Respeite o dispositivo. Muitos CLPs e medidores pequenos suportam apenas algumas requisições por segundo. Comece com polling de um segundo e meça.
3. Use o unit ID correto. Dispositivos Modbus TCP atrás de um gateway usam o unit ID para endereçar dispositivos seriais; dispositivos independentes muitas vezes o ignoram ou esperam 1 ou 255.
4. Separe tags rápidas e lentas. Faça polling rápido de alarmes e estados e lento de contadores de energia.
5. Publique a qualidade explicitamente. Uma leitura com falha deve publicar quality: bad, e não manter silenciosamente o último valor.
6. Nunca escreva a partir do poller por acidente. Deixe o suporte a escrita fora da ponte ou coloque-o atrás de um topic de comando separado e autenticado.

Projetando topics e payloads para dados Modbus

O Modbus bruto entrega endereços, não significado, então a árvore de topics é onde a ponte acrescenta contexto. Dê aos topics nomes de acordo com o que o valor é e onde ele está, nunca pelo número do registrador: plant1/utilities/boiler1/temperature é útil para qualquer pessoa, enquanto device3/hr/0 só é útil para quem escreveu o mapa de registradores. Uma hierarquia no estilo ISA-95, com site, área, linha e equipamento, mantém centenas de dispositivos Modbus navegáveis e permite que os consumidores assinem uma área inteira com um único wildcard.

Mantenha um valor por topic para dados de processo que os consumidores escolhem individualmente e inclua unidade, qualidade e timestamp em cada payload, para que a mensagem se explique sozinha quando aparecer em um log meses depois. Se o dispositivo tem muitos registradores que são sempre usados juntos, como todas as leituras de um medidor, um único objeto JSON por dispositivo por ciclo é uma alternativa razoável e reduz o número de mensagens. Seja qual for a escolha, aplique-a de forma consistente em todas as pontes da planta.

Indo para produção

Para dados reais da planta, mude para um broker privado: defina MQTT_PORT como 8883 e preencha as credenciais do seu broker MQTT gerenciado. No ConnectMQ, cada broker vem com usuário, senha e client ID gerados; o client ID é vinculado a essas credenciais, então use-o em vez de um aleatório. Rode a ponte como um serviço systemd com Restart=always e considere um buffer de store and forward se o link de internet da planta não for confiável, algo comum em usinas e fazendas afastadas.

Se seus dispositivos falam Modbus RTU sobre RS-485 em vez de Ethernet, a mesma ponte funciona com um client serial; nosso tutorial de Modbus RTU para MQTT aborda a fiação e as configurações seriais. Para a arquitetura geral de gateways, brokers e consumidores, veja o guia de IoT industrial com MQTT.

Solução de problemas

Comece com o ConnectMQ

Tire seus dados Modbus da rede da planta sem operar um servidor. Cadastre-se em app.connectmq.com, escolha o plano Serverless gratuito (sem cartão de crédito), crie um broker e cole as credenciais geradas na ponte. Os valores dos seus registradores aparecem imediatamente no explorador de topics ao vivo do ConnectMQ, e regras de alerta por limite podem avisar quando uma temperatura ou pressão sair da faixa.

Perguntas frequentes

Com que frequência devo fazer polling dos registradores Modbus TCP?

Comece com um segundo para valores de processo e mais devagar para contadores, depois ajuste ao que o dispositivo tolera e ao que os consumidores precisam. Publique na mudança, e não a cada leitura.

Por que meus valores float do Modbus parecem errados?

Na maioria das vezes, as duas palavras de 16 bits estão na ordem oposta à que seu código supõe. Inverta a ordem das palavras e verifique se o registrador não é, na verdade, um inteiro escalado em vez de um float.

Uma ponte pode fazer polling de vários dispositivos Modbus TCP?

Sim. Rode um client por dispositivo, ou um client por gateway com unit IDs diferentes, e dê a cada dispositivo seu próprio ramo na árvore de topics.

A ponte deve usar QoS 0 ou QoS 1?

QoS 1 com retained messages é um bom padrão para valores de processo; ele garante a entrega e permite que novos subscribers vejam o estado atual imediatamente.