CAN Bus para MQTT com Raspberry Pi: Leia, Decodifique e Publique Frames

Um gateway CAN bus para MQTT com Raspberry Pi permite pegar sinais que normalmente ficam presos dentro de uma máquina, veículo ou pack de baterias e disponibilizá-los para dashboards, alertas e análises em qualquer lugar. Neste tutorial você vai ligar um HAT CAN baseado no MCP2515, ativar a interface can0, ler frames com python-can, decodificá-los com um arquivo DBC e publicar JSON limpo em um broker MQTT. Os exemplos usam o servidor MQTT público gratuito em broker.connectmq.com, próprio para testes, e depois passam para credenciais privadas em produção.

Como funciona um gateway CAN para MQTT

O CAN é um barramento multimestre em que cada nó transmite frames curtos (até 8 bytes de dados no CAN clássico) identificados por um ID de arbitragem. Não existe o conceito de "topics" ou "leituras": um único frame como 0x1F0 pode empacotar rotação do motor, temperatura do líquido de arrefecimento e um bit de status em bits específicos do seu payload. O significado fica em uma descrição separada, geralmente um arquivo DBC fornecido pelo fabricante do equipamento ou obtido por você via engenharia reversa.

O MQTT é o oposto: topics nomeados, payloads autodescritivos e consumidores que vêm e vão. O trabalho do gateway é traduzir entre os dois mundos:

1. Receber frames brutos do barramento pelo SocketCAN do Linux.
2. Decodificar os frames que interessam em valores físicos nomeados (rpm, graus, volts).
3. Reduzir a taxa de dados, porque um barramento CAN pode transportar milhares de frames por segundo.
4. Publicar os valores em topics bem estruturados em um broker MQTT.

O passo 3 é onde a maioria das primeiras tentativas dá errado. Encaminhar cada frame como uma mensagem MQTT própria inunda a rede e o broker, e ninguém lá na frente precisa da rotação do motor 100 vezes por segundo. Um gateway deve guardar o último valor de cada sinal e publicar snapshots em um intervalo fixo ou quando houver mudança significativa.

Hardware necessário

Ligue CAN_H em CAN_H e CAN_L em CAN_L, além de um terra comum quando o HAT oferecer um. Se o seu Pi fica na extremidade do barramento, habilite o resistor de terminação. Se ele se conecta no meio de um barramento já terminado, deixe a terminação desligada.

Ative a interface can0

Habilite o SPI e o overlay do MCP2515. No Raspberry Pi OS Bookworm o arquivo fica em /boot/firmware/config.txt; versões mais antigas usam /boot/config.txt.

dtparam=spi=on
dtoverlay=mcp2515-can0,oscillator=12000000,interrupt=25

O valor de oscillator precisa corresponder ao cristal do seu HAT (12 MHz em muitas placas Waveshare, 16 MHz na PiCAN2 e outras) e interrupt precisa corresponder ao GPIO que o HAT usa para a linha de interrupção do MCP2515. Um valor de oscilador errado é o motivo mais comum para uma interface que sobe, mas nunca recebe nada. Reinicie e depois ative a interface com o bitrate do barramento:

sudo reboot
# after reboot
dmesg | grep -i mcp251
sudo apt install -y can-utils
sudo ip link set can0 up type can bitrate 500000
ip -details link show can0
candump can0

500 kbit/s é comum em veículos e em muitas máquinas, mas 250 kbit/s (caminhões J1939 e máquinas agrícolas) e 125 kbit/s também aparecem. Se o candump não mostra nada e ip -details -statistics link show can0 reporta erros ou o estado BUS-OFF, o bitrate ou a fiação está errado.

Para que a interface suba automaticamente na inicialização, deixe o systemd-networkd gerenciá-la:

sudo tee /etc/systemd/network/80-can.network > /dev/null <<'EOF'
[Match]
Name=can0

[CAN]
BitRate=500K
EOF
sudo systemctl enable --now systemd-networkd

Instale python-can, cantools e paho-mqtt

python3 -m venv ~/can-gw/.venv
source ~/can-gw/.venv/bin/activate
pip install python-can cantools "paho-mqtt>=2.0"

O python-can conversa com o SocketCAN, o cantools interpreta arquivos DBC e decodifica frames, e o paho-mqtt cuida da conexão com o broker. Se você ainda não tem um arquivo DBC, o script abaixo funciona mesmo assim e publica em hexadecimal os frames brutos dos IDs que você liberar, o que é útil enquanto faz a engenharia reversa dos sinais.

O script CAN bus para MQTT com Raspberry Pi

O gateway lê frames em loop, decodifica-os em um dicionário de últimos valores e publica um snapshot uma vez por segundo. Ele também publica um status online/offline retido com Last Will, para que os dashboards saibam quando o gateway desaparece.

import json
import os
import time
import uuid

import can
import cantools
import paho.mqtt.client as mqtt

BROKER = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "1883"))
USERNAME = os.getenv("MQTT_USERNAME")          # defina para um broker privado
PASSWORD = os.getenv("MQTT_PASSWORD")
CLIENT_ID = os.getenv("MQTT_CLIENT_ID", f"can-gw-{uuid.uuid4().hex[:8]}")
PREFIX = os.getenv("TOPIC_PREFIX", "connectmq/demo/k7x2q9/vehicle1")
DBC_PATH = os.getenv("DBC_PATH", "vehicle.dbc")
RAW_IDS = {0x1F0, 0x2A0}                        # encaminhados em hex se não estiverem no DBC
PUBLISH_INTERVAL = 1.0

db = cantools.database.load_file(DBC_PATH) if os.path.exists(DBC_PATH) else None
known_ids = {m.frame_id for m in db.messages} if db else set()

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
if USERNAME:
    client.username_pw_set(USERNAME, PASSWORD)
if PORT == 8883:
    client.tls_set()
client.will_set(f"{PREFIX}/status", "offline", qos=1, retain=True)
client.reconnect_delay_set(min_delay=1, max_delay=60)


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


client.on_connect = on_connect
client.connect_async(BROKER, PORT, keepalive=30)
client.loop_start()

latest = {}
raw = {}
bus = can.Bus(interface="socketcan", channel="can0")
next_publish = time.monotonic() + PUBLISH_INTERVAL

try:
    while True:
        msg = bus.recv(timeout=0.2)
        if msg is not None and not msg.is_error_frame:
            if msg.arbitration_id in known_ids:
                try:
                    decoded = db.decode_message(msg.arbitration_id, msg.data)
                    for name, value in decoded.items():
                        latest[name] = value if isinstance(value, (int, float)) else str(value)
                except Exception as exc:
                    print("decode error", hex(msg.arbitration_id), exc)
            elif msg.arbitration_id in RAW_IDS:
                raw[f"{msg.arbitration_id:03X}"] = msg.data.hex()

        now = time.monotonic()
        if now >= next_publish and client.is_connected():
            next_publish = now + PUBLISH_INTERVAL
            if latest:
                payload = {"ts": round(time.time(), 3), "signals": latest}
                client.publish(f"{PREFIX}/signals", json.dumps(payload), qos=1)
            if raw:
                client.publish(f"{PREFIX}/raw", json.dumps(raw), qos=0)
except KeyboardInterrupt:
    pass
finally:
    client.publish(f"{PREFIX}/status", "offline", qos=1, retain=True)
    client.loop_stop()
    client.disconnect()
    bus.shutdown()

Execute-o e acompanhe a saída em outro terminal:

python gateway.py
mosquitto_sub -h broker.connectmq.com -p 1883 -i can-viewer-k7x2q9 -t "connectmq/demo/k7x2q9/vehicle1/#" -v

Sem um barramento real, você pode testar a cadeia completa com uma interface virtual: sudo modprobe vcan && sudo ip link add dev vcan0 type vcan && sudo ip link set vcan0 up, troque o canal para vcan0 e injete frames com cansend vcan0 1F0#0A1B2C3D.

Limitação de taxa e projeto de topics

O script publica um snapshot combinado por segundo em vez de uma mensagem por frame. Isso respeita o limite de 10 mensagens por segundo por client do servidor MQTT público e também mantém um broker privado tranquilo. Alguns padrões que funcionam bem na prática:

• Topic de snapshot para dashboards: um documento JSON por intervalo contendo todos os sinais decodificados.
• Topics por sinal, como .../signals/EngineSpeed, quando os consumidores assinam valores individuais. Publique apenas quando o valor mudar mais do que uma banda morta.
• Topics de evento para mudanças discretas (códigos de falha, porta aberta, transições de máquina de estados), publicados imediatamente em QoS 1.
• Topic bruto apenas para IDs em investigação, nunca para o barramento inteiro.

Mantenha a hierarquia de topics estável e descritiva. Se o gateway faz parte de uma planta, um caminho no estilo ISA-95 como site/area/line/machine/can se encaixa naturalmente em um Unified Namespace, e um gateway IoT com Raspberry Pi pode combinar dados CAN com outras fontes sob um único prefixo.

Indo para produção

O servidor público é para testes: tudo o que é publicado ali pode ser lido por qualquer pessoa. Para um veículo ou máquina real, crie um broker privado na ConnectMQ, copie do portal o usuário, a senha e o client ID gerados e defina as variáveis de ambiente:

export MQTT_HOST=broker.connectmq.com
export MQTT_PORT=8883
export MQTT_USERNAME=YOUR_USERNAME
export MQTT_PASSWORD=YOUR_PASSWORD
export MQTT_CLIENT_ID=YOUR_CLIENT_ID
export TOPIC_PREFIX=fleet/truck-17/can
python gateway.py

A porta 8883 usa TLS com um certificado Let's Encrypt, então tls_set() com o pacote de CAs do sistema é suficiente. Um broker MQTT gerenciado também elimina a necessidade de operar e atualizar seu próprio broker na borda. Para um gateway de campo, combine isso com buffer offline em SQLite, para que quedas da rede celular não criem lacunas, e execute o script como serviço para que ele inicie após uma falta de energia.

Solução de problemas

Uma observação sobre segurança: este gateway apenas escuta. O MCP2515 ainda vai confirmar (ACK) frames no barramento, o que é comportamento normal do CAN, mas não adicione código que transmita em um barramento de veículo ou máquina a menos que você entenda totalmente as consequências. Se precisar de uma escuta estritamente passiva, coloque o controlador em modo somente escuta com sudo ip link set can0 up type can bitrate 500000 listen-only on.

Para um panorama mais amplo do Pi como client MQTT, comece pelo tutorial de MQTT com Raspberry Pi. A documentação do python-can cobre outras interfaces, como adaptadores USB-CAN, caso um HAT não seja uma opção.

Envie seus dados CAN para a ConnectMQ

Depois que os frames estão decodificados, a parte difícil acabou; o resto é entrega confiável. Crie um broker Serverless gratuito da ConnectMQ em app.connectmq.com, sem cartão de crédito, cole as credenciais no gateway e veja seus sinais chegando no explorador de topics ao vivo. Quando a frota crescer, o mesmo código migra para um plano maior sem alterações.

Perguntas frequentes

Posso usar um adaptador USB-CAN em vez de um HAT?

Sim. Adaptadores suportados pelo kernel Linux, como os que usam o driver gs_usb, aparecem como can0 via SocketCAN, então o script Python funciona sem alterações. Outros adaptadores podem ser usados pelo nome de interface correspondente no python-can.

Isso funciona com CAN FD?

O MCP2515 suporta apenas CAN clássico. Para CAN FD você precisa de um HAT baseado no MCP2518FD e do overlay correspondente; python-can e cantools tratam frames FD.

Como decodifico dados OBD-II ou J1939?

PIDs OBD-II funcionam em requisição/resposta, então você precisa enviar requisições em vez de apenas escutar. O J1939 usa identificadores de 29 bits e PGNs, e arquivos DBC públicos cobrem muitos parâmetros padrão.

Quantas mensagens o gateway deve publicar?

Publique o que os consumidores precisam, não o que o barramento produz. Um snapshot por segundo mais mensagens de evento imediatas é um bom padrão e fica bem dentro dos limites do broker.