Gateway Modbus para MQTT em CLPs: um poller Modbus TCP em Python

Um gateway Modbus para MQTT lê registradores de um CLP via Modbus TCP e os publica como mensagens MQTT, para que dashboards, historiadores e serviços em nuvem possam assinar os dados da planta sem nunca tocar no CLP. Você consegue construir um gateway confiável em cerca de 150 linhas de Python com pymodbus e paho-mqtt.

Este tutorial monta esse gateway passo a passo: um mapa de tags, um loop de leitura, decodificação de valores de 32 bits, publicação por exceção, status online/offline com Last Will e uma forma de testar tudo antes de chegar perto de uma linha de produção.

Onde o gateway se encaixa

Modbus é um protocolo de requisição/resposta: um mestre (cliente) pede registradores a um escravo (servidor), uma requisição por vez. MQTT é publish/subscribe: produtores publicam uma vez e qualquer número de consumidores assina. O gateway traduz entre os dois mundos:

Rode o gateway em um equipamento edge (PC industrial, Raspberry Pi, um contêiner no servidor da linha) que alcance as duas redes. O CLP nunca vê tráfego MQTT e os consumidores MQTT nunca abrem um socket Modbus — exatamente a separação que os times de TA querem. Para a visão geral, veja MQTT para IIoT e Indústria 4.0.

1. Descreva suas tags

Modbus não tem nomes, só endereços. A primeira tarefa é um mapa de tags que diga o que cada registrador significa. Mantenha-o em um arquivo ao lado do código para revisá-lo como configuração:

# nome, função, endereço, tipo de dado, escala, unidade
TAGS = [
    ("temperature",  "holding", 0,  "float32", 1.0,  "degC"),
    ("pressure",     "holding", 2,  "float32", 1.0,  "bar"),
    ("motor_speed",  "holding", 4,  "uint16",  1.0,  "rpm"),
    ("part_count",   "holding", 5,  "uint32",  1.0,  "pcs"),
    ("tank_level",   "input",   0,  "int16",   0.1,  "%"),
    ("running",      "coil",    0,  "bool",    1,    ""),
]

Os endereços aqui são endereços de protocolo começando em zero. A documentação dos fabricantes costuma usar a numeração "40001", começando em um: o holding register 40001 é o endereço 0 e o 40003 é o endereço 2. Erro de um é o bug Modbus mais comum, então confira um valor conhecido antes de confiar no resto.

2. Leia e decodifique registradores

Um registrador Modbus tem 16 bits. Floats e contadores de 32 bits ocupam dois registradores, e os fabricantes discordam sobre qual palavra vem primeiro. Decodificar com struct deixa a ordem das palavras explícita:

import struct
from pymodbus.client import ModbusTcpClient

WORD_ORDER_BIG = True   # use False em dispositivos com "word swap" little-endian

def decode(regs, dtype):
    if dtype == "uint16":
        return regs[0]
    if dtype == "int16":
        return struct.unpack(">h", struct.pack(">H", regs[0]))[0]
    words = regs if WORD_ORDER_BIG else list(reversed(regs))
    raw = struct.pack(">HH", *words)
    return struct.unpack(">f" if dtype == "float32" else ">I", raw)[0]

def read_tag(client, unit, func, addr, dtype):
    count = 2 if dtype in ("float32", "uint32") else 1
    if func == "coil":
        rr = client.read_coils(addr, count=1, device_id=unit)
        return None if rr.isError() else bool(rr.bits[0])
    reader = client.read_holding_registers if func == "holding" else client.read_input_registers
    rr = reader(addr, count=count, device_id=unit)
    return None if rr.isError() else decode(rr.registers, dtype)

Em produção, leia blocos contíguos (por exemplo os registradores 0–5 em uma única requisição) em vez de uma requisição por tag: um CLP responde mais rápido a uma requisição maior do que a seis pequenas. A versão por tag acima é mais fácil de ler e serve bem para algumas dezenas de tags.

3. Publique no MQTT por exceção

Agora combine o poller com um cliente MQTT. O gateway publica um tópico por tag, envia um valor apenas quando ele muda além de uma banda morta, marca os valores como retained para que novos assinantes recebam o estado atual imediatamente e usa um Last Will para anunciar quando fica offline.

import json, os, time
import paho.mqtt.client as mqtt
from pymodbus.client import ModbusTcpClient
from modbus_read import read_tag
from tags import TAGS

PLC_HOST, PLC_UNIT = os.getenv("PLC_HOST", "192.168.10.20"), 1
BASE = "testmqtt/acme/plant1/line2/plc1"
POLL_S, DEADBAND = 1.0, 0.05            # segundos, mudança relativa (5 %)

mq = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="gw-line2-plc1")
mq.username_pw_set(os.getenv("MQTT_USER"), os.getenv("MQTT_PASS"))
mq.tls_set()
mq.will_set(f"{BASE}/status", "offline", qos=1, retain=True)

def on_connect(client, userdata, flags, reason_code, properties):
    print("MQTT conectado:", reason_code)
    client.publish(f"{BASE}/status", "online", qos=1, retain=True)

mq.on_connect = on_connect
mq.connect("broker.connectmq.com", 8883, keepalive=30)
mq.loop_start()

plc = ModbusTcpClient(PLC_HOST, port=502, timeout=2)
last = {}

def changed(name, value):
    old = last.get(name)
    if old is None or isinstance(value, bool):
        return old != value
    return abs(value - old) > abs(old) * DEADBAND

while True:
    if not plc.connected and not plc.connect():
        mq.publish(f"{BASE}/status", "plc-unreachable", qos=1, retain=True)
        time.sleep(5)
        continue
    for name, func, addr, dtype, scale, unit in TAGS:
        value = read_tag(plc, PLC_UNIT, func, addr, dtype)
        if value is None:
            continue
        if not isinstance(value, bool):
            value = round(value * scale, 3)
        if changed(name, value):
            last[name] = value
            payload = {"v": value, "unit": unit, "ts": int(time.time() * 1000)}
            mq.publish(f"{BASE}/{name}", json.dumps(payload), qos=1, retain=True)
    time.sleep(POLL_S)

Algumas decisões que valem explicação:

• Um tópico por tag (.../plc1/temperature) permite que os consumidores assinem exatamente o que precisam, e wildcards como .../plc1/+ ainda entregam tudo. Teste os padrões com o topic matcher.
• Retained + QoS 1 significa que um dashboard aberto às 3 da manhã mostra o último valor conhecido na hora, e que as mensagens sobrevivem a quedas curtas de rede. Veja mensagens retidas.
• Tópico de status com Last Will diferencia "a temperatura está estável" de "o gateway morreu". O broker publica offline em seu nome se a conexão do gateway cair sem um disconnect limpo.
• Timestamps no payload registram quando o valor foi lido, não quando chegou — importante para historiadores.

A estrutura de tópicos segue uma hierarquia no estilo ISA-95 (empresa / planta / linha / dispositivo), que é o ponto de partida de um Unified Namespace.

4. Teste sem um CLP de produção

Nunca desenvolva contra uma linha em operação. Comece com um simulador Modbus — o pymodbus traz um, e ferramentas como diagslave ou ModbusPal servem registradores configuráveis na porta 502 (ou 5020 se você não for root). Aponte PLC_HOST para ele e rode o gateway.

Depois confira o lado MQTT:

1. Abra o cliente MQTT online, conecte em broker.connectmq.com e assine testmqtt/acme/plant1/#. Você deve ver na hora o status = online retido e uma mensagem por tag.
2. Altere um registrador no simulador. Só essa tag deve publicar uma nova mensagem; as que não mudaram ficam em silêncio.
3. Pare o gateway com kill -9 (sem disconnect limpo). Depois de cerca de uma vez e meia o keep-alive, status deve mudar para offline.
4. Pare o simulador. status deve indicar plc-unreachable enquanto o MQTT continua conectado.

Acompanhe os tópicos do gateway no navegador

No terminal, a mesma verificação é um único comando:

mosquitto_sub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t 'testmqtt/acme/plant1/line2/#' -v

Preparando para produção

Modbus não tem autenticação nenhuma, então o gateway costuma ser a única fronteira controlada entre o CLP e todo o resto. Mantenha as permissões MQTT dele estreitas e o caminho de escrita fechado, a menos que você realmente precise.

Se suas plantas migrarem depois para Sparkplug B ou precisarem de gestão de estado padronizada para muitos dispositivos, o mesmo núcleo de leitura pode alimentar um edge node Sparkplug B em vez de tópicos JSON simples.

Um gateway Modbus para MQTT é pouco código com muito efeito: transforma registradores que só o programador do CLP entende em dados nomeados, com timestamp e que qualquer um pode assinar. Comece com um mapa de tags claro, publique na mudança com valores retidos e Last Will, teste contra um simulador e o cliente online, e só então conecte à linha.

Perguntas frequentes