CAN-Bus zu MQTT mit Raspberry Pi: Frames lesen, dekodieren und veröffentlichen

Ein Gateway für CAN-Bus zu MQTT mit Raspberry Pi holt Signale, die normalerweise in einer Maschine, einem Fahrzeug oder einem Batteriepack eingeschlossen bleiben, heraus und macht sie für Dashboards, Alarme und Analysen überall verfügbar. In diesem Tutorial verdrahten Sie einen CAN-HAT auf Basis des MCP2515, aktivieren die Schnittstelle can0, lesen Frames mit python-can, dekodieren sie mit einer DBC-Datei und veröffentlichen sauberes JSON an einen MQTT-Broker. Die Beispiele nutzen den kostenlosen öffentlichen MQTT-Broker unter broker.connectmq.com und wechseln für die Produktion auf private Zugangsdaten.

So funktioniert ein CAN-zu-MQTT-Gateway

CAN ist ein Multi-Master-Bus, auf dem jeder Knoten kurze Frames (bis zu 8 Datenbytes bei klassischem CAN) sendet, die durch eine Arbitrierungs-ID gekennzeichnet sind. Es gibt kein Konzept von „Topics“ oder „Messwerten“: Ein einzelner Frame wie 0x1F0 kann Motordrehzahl, Kühlmitteltemperatur und ein Statusbit in bestimmte Bits seiner Nutzdaten packen. Die Bedeutung steckt in einer separaten Beschreibung, meist einer DBC-Datei, die der Gerätehersteller liefert oder die Sie selbst per Reverse Engineering erstellen.

MQTT ist das Gegenteil: benannte Topics, selbstbeschreibende Payloads und Abnehmer, die kommen und gehen. Aufgabe des Gateways ist es, zwischen beiden Welten zu übersetzen:

1. Rohe Frames über Linux-SocketCAN vom Bus empfangen.
2. Die relevanten Frames in benannte physikalische Werte (U/min, Grad, Volt) dekodieren.
3. Die Datenrate reduzieren, denn ein CAN-Bus kann Tausende Frames pro Sekunde transportieren.
4. Die Werte an gut strukturierte Topics auf einem MQTT-Broker veröffentlichen.

An Schritt 3 scheitern die meisten ersten Versuche. Jeden Frame als eigene MQTT-Nachricht weiterzuleiten, flutet Netzwerk und Broker, und niemand im weiteren Verlauf braucht die Motordrehzahl 100-mal pro Sekunde. Ein Gateway sollte pro Signal den neuesten Wert vorhalten und Momentaufnahmen in festem Intervall oder bei signifikanter Änderung veröffentlichen.

Benötigte Hardware

Verbinden Sie CAN_H mit CAN_H und CAN_L mit CAN_L, dazu eine gemeinsame Masse, sofern der HAT eine bereitstellt. Sitzt Ihr Pi am Ende des Busses, aktivieren Sie seinen Abschlusswiderstand. Greift er in der Mitte eines bereits terminierten Busses ab, lassen Sie die Terminierung aus.

Die Schnittstelle can0 aktivieren

Aktivieren Sie SPI und das MCP2515-Overlay. Unter Raspberry Pi OS Bookworm liegt die Datei unter /boot/firmware/config.txt, ältere Versionen nutzen /boot/config.txt.

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

Der Wert oscillator muss zum Quarz auf Ihrem HAT passen (12 MHz bei vielen Waveshare-Boards, 16 MHz beim PiCAN2 und anderen), und interrupt muss dem GPIO entsprechen, den der HAT für die Interrupt-Leitung des MCP2515 nutzt. Ein falscher Oszillatorwert ist der häufigste Grund für eine Schnittstelle, die zwar hochfährt, aber nie etwas empfängt. Starten Sie neu und aktivieren Sie die Schnittstelle dann mit der Bitrate des Busses:

sudo reboot
# nach dem Neustart
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 sind bei Fahrzeugen und vielen Maschinen üblich, es kommen aber auch 250 kbit/s (J1939 bei Lkw und Landmaschinen) und 125 kbit/s vor. Zeigt candump nichts an und meldet ip -details -statistics link show can0 Fehler oder den Zustand BUS-OFF, stimmen Bitrate oder Verdrahtung nicht.

Damit die Schnittstelle beim Booten automatisch hochfährt, überlassen Sie die Verwaltung systemd-networkd:

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

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

python-can, cantools und paho-mqtt installieren

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

python-can spricht mit SocketCAN, cantools parst DBC-Dateien und dekodiert Frames, und paho-mqtt kümmert sich um die Verbindung zum Broker. Haben Sie noch keine DBC-Datei, funktioniert das folgende Skript trotzdem und veröffentlicht für die freigegebenen IDs rohe Frames in Hex, was beim Reverse Engineering von Signalen hilfreich ist.

Das Skript für CAN-Bus zu MQTT mit Raspberry Pi

Das Gateway liest Frames in einer Schleife, dekodiert sie in ein Dictionary mit den neuesten Werten und veröffentlicht einmal pro Sekunde eine Momentaufnahme. Außerdem veröffentlicht es einen retained Online/Offline-Status mit Last Will, damit Dashboards erkennen, wenn das Gateway verschwindet.

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")          # für einen privaten Broker setzen
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}                        # als Hex weitergeleitet, wenn nicht in der 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()

Starten Sie es und verfolgen Sie die Ausgabe in einem anderen Terminal:

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

Ohne echten Bus können Sie die gesamte Kette mit einer virtuellen Schnittstelle testen: sudo modprobe vcan && sudo ip link add dev vcan0 type vcan && sudo ip link set vcan0 up, den Kanal auf vcan0 ändern und Frames mit cansend vcan0 1F0#0A1B2C3D einspeisen.

Ratenbegrenzung und Topic-Design

Das Skript veröffentlicht pro Sekunde eine kombinierte Momentaufnahme statt einer Nachricht pro Frame. Das passt zum Limit des öffentlichen MQTT-Brokers von 10 Nachrichten pro Sekunde und Client und hält auch einen privaten Broker ruhig. Einige Muster, die sich in der Praxis bewähren:

• Snapshot-Topic für Dashboards: ein JSON-Dokument pro Intervall mit allen dekodierten Signalen.
• Topics pro Signal wie .../signals/EngineSpeed, wenn Abnehmer einzelne Werte abonnieren. Nur veröffentlichen, wenn sich der Wert um mehr als ein Totband ändert.
• Ereignis-Topics für diskrete Änderungen (Fehlercodes, Tür offen, Zustandswechsel), sofort mit QoS 1 veröffentlicht.
• Raw-Topic nur für IDs, die gerade untersucht werden, niemals für den ganzen Bus.

Halten Sie die Topic-Hierarchie stabil und aussagekräftig. Ist das Gateway Teil eines Werks, fügt sich ein Pfad im ISA-95-Stil wie site/area/line/machine/can nahtlos in einen Unified Namespace (UNS) ein, in dem die ISA-95-Ebenen die Topic-Struktur bilden, und ein Raspberry-Pi-IoT-Gateway kann CAN-Daten mit anderen Quellen unter einem Präfix zusammenführen.

Der Weg in die Produktion

Der öffentliche MQTT-Broker ist zum Testen gedacht: Alles, was dort veröffentlicht wird, kann jeder lesen. Für ein echtes Fahrzeug oder eine echte Maschine erstellen Sie einen privaten Broker bei ConnectMQ, kopieren den generierten Benutzernamen, das Passwort und die Client-ID aus dem Portal und setzen die Umgebungsvariablen:

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

Port 8883 nutzt TLS mit einem Let's-Encrypt-Zertifikat, daher genügt tls_set() mit dem CA-Bundle des Systems. Ein vollständig verwalteter MQTT-Broker erspart Ihnen außerdem, am Edge einen eigenen Broker zu betreiben und zu patchen. Für ein Feld-Gateway kombinieren Sie das mit Offline-Pufferung in SQLite, damit Mobilfunkabbrüche keine Lücken erzeugen, und betreiben das Skript als Dienst, damit es nach einem Stromausfall startet.

Fehlerbehebung

Ein Hinweis zur Sicherheit: Dieses Gateway hört nur zu. Der MCP2515 bestätigt dennoch Frames auf dem Bus, was normales CAN-Verhalten ist, doch fügen Sie keinen Code hinzu, der auf einen Fahrzeug- oder Maschinenbus sendet, solange Sie die Folgen nicht vollständig verstehen. Benötigen Sie einen streng passiven Abgriff, versetzen Sie den Controller mit sudo ip link set can0 up type can bitrate 500000 listen-only on in den Listen-only-Modus.

Allgemeine Grundlagen zum Pi als MQTT-Client bietet das Raspberry-Pi-MQTT-Tutorial. Die python-can-Dokumentation behandelt weitere Schnittstellen wie USB-CAN-Adapter, falls ein HAT nicht infrage kommt.

Streamen Sie Ihre CAN-Daten zu ConnectMQ

Sind die Frames dekodiert, ist der schwierige Teil erledigt; der Rest ist zuverlässige Zustellung. Erstellen Sie unter app.connectmq.com kostenlos und ohne Kreditkarte einen ConnectMQ-Serverless-Broker, tragen Sie die Zugangsdaten im Gateway ein und verfolgen Sie, wie Ihre Signale im Live-Topic-Explorer eintreffen. Wächst die Flotte, zieht derselbe Code ohne Änderungen auf einen größeren Plan um.

Häufig gestellte Fragen

Kann ich statt eines HATs einen USB-CAN-Adapter verwenden?

Ja. Vom Linux-Kernel unterstützte Adapter, etwa solche mit dem Treiber gs_usb, erscheinen über SocketCAN als can0, sodass das Python-Skript unverändert funktioniert. Andere Adapter lassen sich über ihren python-can-Schnittstellennamen nutzen.

Funktioniert das mit CAN FD?

Der MCP2515 unterstützt nur klassisches CAN. Für CAN FD benötigen Sie einen HAT auf Basis des MCP2518FD und das passende Overlay; python-can und cantools verarbeiten beide FD-Frames.

Wie dekodiere ich OBD-II- oder J1939-Daten?

OBD-II-PIDs arbeiten nach dem Request/Response-Prinzip, Sie müssen also Anfragen senden, statt nur zuzuhören. J1939 nutzt 29-Bit-Identifier und PGNs, und öffentliche DBC-Dateien decken viele Standardparameter ab.

Wie viele Nachrichten sollte das Gateway veröffentlichen?

Veröffentlichen Sie, was die Abnehmer brauchen, nicht, was der Bus erzeugt. Eine Momentaufnahme pro Sekunde plus sofortige Ereignisnachrichten ist ein guter Standard und bleibt deutlich innerhalb der Broker-Limits.