Jetson Nano MQTT: GPU-Ergebnisse der Edge-Bildverarbeitung sicher veröffentlichen

Mit einer Jetson-Nano-MQTT-Pipeline führen Sie GPU-beschleunigte Bildverarbeitung an der Edge aus und senden nur die Ergebnisse in die Cloud. In diesem Tutorial betreiben Sie Objekterkennung auf einem NVIDIA Jetson, veröffentlichen Detektionen und Gerätezustand als JSON per MQTT mit TLS und machen aus dem Skript einen Dienst, der beim Booten startet und sich selbstständig neu verbindet.

Warum Jetson plus MQTT

NVIDIA-Jetson-Boards kombinieren eine Arm-CPU mit einer CUDA-fähigen GPU in einem kleinen, stromsparenden Modul. Damit sind sie eine beliebte Wahl für Kameras, die Personen, Fahrzeuge oder Defekte in Echtzeit erkennen müssen, ohne Video an ein Rechenzentrum zu senden.

Interessant ist die Ausgabe einer solchen Kamera: Objektlabels, Positionen, Zählwerte und Alarme. MQTT eignet sich ideal, um diese Ausgabe zu übertragen, weil es leichtgewichtig ist, mit unzuverlässigen Verbindungen souverän umgeht und beliebig vielen Konsumenten das Abonnieren erlaubt. Der Jetson veröffentlicht einmal, und Dashboards, Alarmierungswerkzeuge, Historians und andere KI-Dienste erhalten alle dieselben Nachrichten.

Jetson und Software-Stack auswählen

Dieser Versionsunterschied ist der häufigste Stolperstein. paho-mqtt 2.x erfordert Python 3.7 oder neuer. Auf einem originalen Jetson Nano mit JetPack 4.6 installieren Sie daher entweder paho-mqtt 1.6.1 (und lassen das Argument CallbackAPIVersion sowie den Callback-Parameter properties weg) oder führen Ihren Code in einem Container mit neuerem Python aus. Die folgenden Beispiele zielen auf JetPack 5 oder 6 mit paho-mqtt 2.x.

Für die Inferenz verwendet dieses Tutorial NVIDIAs Open-Source-Bibliothek jetson-inference, die TensorRT-optimierte Erkennungsmodelle hinter einer kurzen Python-API kapselt und CSI- sowie USB-Kameras unterstützt. Der MQTT-Teil funktioniert mit jedem anderen Inferenz-Framework genauso, einschließlich YOLO über Ultralytics.

Schritt 1: Den Jetson vorbereiten

Installieren Sie jetson-inference gemäß der Dokumentation (Build aus dem Quellcode oder über den Docker-Container) und ergänzen Sie dann den MQTT-Client:

sudo apt update
sudo apt install -y python3-pip mosquitto-clients
pip3 install "paho-mqtt>=2.0"

Prüfen Sie, ob die Kamera funktioniert. Mit installiertem jetson-inference:

video-viewer csi://0          # CSI-Kamera wie das Raspberry Pi Camera Module v2
video-viewer /dev/video0      # USB-Webcam

Prüfen Sie dann, ob der Jetson den Broker erreicht. Für einen ersten Test nutzen Sie den öffentlichen MQTT-Broker broker.connectmq.com über TLS; er ist für Tests und Prototypen gedacht, nicht für Produktionsdaten:

mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -i jetson-test-7d1e -t "connectmq/demo/jet-7d1e/hello" -m "jetson online"

Schlägt das fehl, prüfen Sie die ausgehenden Firewall-Regeln für Port 8883 und ob die Systemuhr stimmt; TLS scheitert auf einem Jetson, dessen Uhr Jahre zurückliegt.

Schritt 2: Der Detektions-Publisher

Speichern Sie dies als jetson_mqtt.py. Das Skript erfasst Frames, führt detectNet auf der GPU aus und veröffentlicht gedrosselte Detektionen, Retained-Zählwerte pro Klasse sowie einen Heartbeat mit Bildrate und GPU-seitiger Leistung.

import json
import ssl
import time
import uuid

import paho.mqtt.client as mqtt
from jetson_inference import detectNet
from jetson_utils import videoSource

BROKER, PORT = "broker.connectmq.com", 8883
PREFIX = "connectmq/demo/jet-7d1e/site1/gate/cam1"
USERNAME, PASSWORD = None, None  # für einen privaten Broker auf "YOUR_USERNAME", "YOUR_PASSWORD" setzen
CLIENT_ID = f"jetson-cam1-{uuid.uuid4().hex[:6]}"  # auf einem privaten Broker YOUR_CLIENT_ID verwenden
PUBLISH_EVERY = 0.5
HEARTBEAT_EVERY = 30

net = detectNet("ssd-mobilenet-v2", threshold=0.5)
camera = videoSource("csi://0")  # oder "/dev/video0" oder "rtsp://..."


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


def on_disconnect(client, userdata, flags, reason_code, properties):
    print("disconnected:", reason_code)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                     protocol=mqtt.MQTTv5)
if USERNAME:
    client.username_pw_set(USERNAME, PASSWORD)
client.tls_set(cert_reqs=ssl.CERT_REQUIRED)
client.will_set(f"{PREFIX}/status", json.dumps({"online": False}), qos=1, retain=True)
client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.reconnect_delay_set(min_delay=1, max_delay=60)
client.connect_async(BROKER, PORT, keepalive=30)
client.loop_start()

last_pub = last_beat = 0.0
last_counts = None
frames = 0

while True:
    img = camera.Capture()
    if img is None:  # Timeout bei der Erfassung
        continue
    detections = net.Detect(img, overlay="none")
    frames += 1

    objects, counts = [], {}
    for d in detections:
        label = net.GetClassDesc(d.ClassID)
        counts[label] = counts.get(label, 0) + 1
        objects.append({
            "label": label,
            "conf": round(d.Confidence, 2),
            "box": [round(d.Left / img.width, 3), round(d.Top / img.height, 3),
                    round(d.Right / img.width, 3), round(d.Bottom / img.height, 3)],
        })

    now = time.time()
    if now - last_pub >= PUBLISH_EVERY:
        last_pub = now
        client.publish(f"{PREFIX}/detections",
                       json.dumps({"ts": now, "objects": objects}), qos=0)

    if counts != last_counts:
        last_counts = counts
        client.publish(f"{PREFIX}/counts", json.dumps({"ts": now, **counts}),
                       qos=1, retain=True)

    if now - last_beat >= HEARTBEAT_EVERY:
        fps = frames / (now - last_beat) if last_beat else 0.0
        client.publish(f"{PREFIX}/health", json.dumps({
            "ts": now, "fps": round(fps, 1),
            "networkFps": round(net.GetNetworkFPS(), 1)}), qos=0)
        last_beat, frames = now, 0

Beobachten Sie die Ergebnisse von Ihrem Laptop aus:

mosquitto_sub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -i jet-watch-7d1e -t "connectmq/demo/jet-7d1e/#" -v

connect_async zusammen mit loop_start bedeutet, dass das Skript die Inferenz auch bei Netzausfall fortsetzt; paho verbindet sich im Hintergrund neu, und das Retained-Status-Topic samt Last Will zeigt allen, ob die Kamera erreichbar ist.

Schritt 3: Gerätezustand ergänzen

Eine Kamera, die unbemerkt überhitzt oder drosselt, ist ein häufiger Ausfall im Feld. Jetson-Boards melden Temperaturen über sysfs, die Sie mit dem Heartbeat veröffentlichen können:

import glob


def read_temps():
    temps = {}
    for zone in glob.glob("/sys/class/thermal/thermal_zone*"):
        try:
            with open(f"{zone}/type") as t, open(f"{zone}/temp") as v:
                temps[t.read().strip()] = int(v.read()) / 1000
        except OSError:
            pass
    return temps

Ergänzen Sie den Health-Payload um "temps": read_temps(). Auf einem privaten Broker warnt Sie eine Schwellenwert-Alarmregel für ein Temperaturfeld, bevor das Board zu drosseln beginnt.

Schritt 4: Als Dienst betreiben

Erstellen Sie /etc/systemd/system/jetson-mqtt.service:

[Unit]
Description=Jetson vision to MQTT
After=network-online.target
Wants=network-online.target

[Service]
User=jetson
WorkingDirectory=/home/jetson/vision
ExecStart=/usr/bin/python3 /home/jetson/vision/jetson_mqtt.py
Restart=always
RestartSec=5
Environment=PYTHONUNBUFFERED=1

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable --now jetson-mqtt
journalctl -u jetson-mqtt -f

Topics für viele Kameras entwerfen

Ein Jetson ist einfach; zwanzig an drei Standorten brauchen einen Plan. Bilden Sie die Standorthierarchie im Topic ab, damit Konsumenten auf der richtigen Ebene abonnieren können: site/area/line-or-gate/camera/.... Ein Sicherheits-Dashboard abonniert site1/+/+/alerts, ein Betriebsbericht abonniert +/+/+/counts, und ein Ingenieur, der eine einzelne Kamera untersucht, abonniert nur diese. Das ist dieselbe Idee wie ein auf ISA-95 basierender Unified Namespace (UNS): In einer Fertigung würde die Kamera zur optischen Qualitätsprüfung etwa unter acme/werk-muenchen/montage/linie-1/zelle-3/cam1/... publizieren, direkt neben den Maschinendaten derselben Zelle. ConnectMQ erkennt solche Topic-Strukturen in seiner Geräteansicht.

Halten Sie die Nachrichtentypen getrennt, wie im Skript gezeigt: hochfrequente Detektionen, änderungsgesteuerte Retained-Zählwerte, niederfrequenter Gerätezustand und ein Retained-Status mit Last Will. Werden sie in einem Topic vermischt, muss jeder Konsument alles parsen und filtern, und es ist unmöglich, unterschiedlichen Nachrichtentypen unterschiedliche QoS-Stufen zuzuweisen.

Tipps zur Performance

Der Großteil der Bildrate auf einem Jetson hängt von der Inferenz-Engine ab, nicht von Python oder MQTT. Einige Gewohnheiten halten die Pipeline schnell:

• Verwenden Sie TensorRT-optimierte Modelle; jetson-inference baut die Engine beim ersten Start und speichert sie zwischen, daher ist der erste Start langsam und spätere Starts sind schnell.
• Verzichten Sie auf gezeichnete Overlays (overlay="none"), wenn niemand das Video ansieht.
• Drosseln Sie das Veröffentlichen wie im Skript; JSON-Kodierung für jeden Frame bei 30 fps erhöht die CPU-Last ohne Nutzen.
• Nutzen Sie sudo jetson_clocks nur bei ausreichender Kühlung und überwachen Sie die Temperaturen über das Health-Topic.

Checkliste zur Fehlerbehebung für Jetson Nano MQTT

1. TypeError bezüglich CallbackAPIVersion. Sie verwenden paho-mqtt 1.x; aktualisieren Sie oder nutzen Sie auf dem originalen Nano die 1.6-API.
2. Fehler beim TLS-Handshake. Prüfen Sie das Systemdatum (timedatectl) und ob ca-certificates installiert ist.
3. Wiederholte Verbindungsabbrüche mit demselben Grund. Zwei Geräte teilen sich eine Client-ID; jeder Client braucht eine eindeutige ID, und bei ConnectMQ ist die Client-ID an die ausgegebenen Zugangsdaten gebunden.
4. Niedrige Bildrate nach einiger Zeit. Prüfen Sie die veröffentlichten Temperaturen; ergänzen Sie einen Lüfter oder wählen Sie mit nvpmodel gezielt einen niedrigeren Energiemodus.
5. Nachrichtenflut. Drosseln Sie die Detektionen; auf dem öffentlichen MQTT-Broker liegt das Limit bei 10 Nachrichten pro Sekunde und Client.
6. Kamera unter systemd nicht gefunden. Der Dienstbenutzer muss Mitglied der Gruppe video sein.

Vom Prototyp zur Flotte

Sobald eine Kamera funktioniert, werden die Fragen betrieblicher Natur: Zugangsdaten pro Gerät, Verschlüsselung und der Überblick, welche Kameras online sind. Wechseln Sie vom öffentlichen MQTT-Broker zu einem privaten, verwalteten MQTT-Broker, sobald echte Bilder oder Standorte im Spiel sind. ConnectMQ gibt jedem Jetson eigene generierte Zugangsdaten, akzeptiert TLS auf Port 8883, listet Geräte nach ISA-95-Ebene auf und zeigt jede Detektion im Live-Topic-Explorer. Wenn viele Jetsons an verschiedenen Standorten an eine zentrale Stelle melden, vermeidet ein Cloud-MQTT-Broker, dass an irgendeinem Standort eingehende Ports geöffnet werden müssen.

Eine modellunabhängige Version dieser Pipeline finden Sie unter YOLO-Objekterkennung über MQTT, eine Alternative im Mikrocontroller-Format unter Personenerkennung mit der ESP32-CAM, und zur Wahl zwischen Jetson, Coral und anderen Beschleunigern lesen Sie Edge-AI-Hardware. Das Gesamtbild beschreibt der Leitfaden zu KI und IoT.

Verbinden Sie Ihren Jetson mit ConnectMQ

Erstellen Sie unter app.connectmq.com einen Broker im kostenlosen Serverless-Tarif, ganz ohne Kreditkarte, und setzen Sie dann BROKER, USERNAME, PASSWORD und CLIENT_ID im Skript auf die generierten Werte. Die Detektionen Ihres Jetson erscheinen innerhalb von Sekunden im Portal.

Häufig gestellte Fragen

Funktioniert paho-mqtt auf dem originalen Jetson Nano?

Ja, doch JetPack 4.x wird mit Python 3.6 ausgeliefert. Verwenden Sie daher paho-mqtt 1.6.1 mit seiner älteren Callback-API oder betreiben Sie für paho-mqtt 2.x einen Container mit Python 3.7 oder neuer.

Sollte der Jetson Video-Frames per MQTT veröffentlichen?

In der Regel nicht. Veröffentlichen Sie Detektionen, Zählwerte und gelegentliche Schnappschüsse innerhalb der maximalen Nachrichtengröße und nutzen Sie ein Streaming-Protokoll, wenn Sie Live-Video benötigen.

Welchen Port sollte ein Jetson für MQTT verwenden?

Verwenden Sie für alles über einen kurzen Test hinaus Port 8883 mit TLS. Port 1883 ist unverschlüsselt und sollte auf die Entwicklung mit synthetischen Daten beschränkt bleiben.

Kann ich MQTT und Inferenz im selben Python-Prozess ausführen?

Ja. Die Netzwerkschleife von paho läuft über loop_start in einem Hintergrund-Thread, sodass die Inferenz in der Hauptschleife nicht durch Netzwerkoperationen blockiert wird.