YOLO-Objekterkennung über MQTT: Detektionen als JSON veröffentlichen

YOLO-Objekterkennung über MQTT macht aus einer Kamera einen Sensor: Statt Video zu streamen, veröffentlichen Sie kleine JSON-Nachrichten, die beschreiben, was die Kamera sieht. In diesem Tutorial führen Sie mit Python ein YOLO-Modell auf einer Webcam oder einem RTSP-Stream aus, veröffentlichen Detektionen pro Frame, führen Objektzählungen und lösen Alarme aus, die jeder MQTT-Client verarbeiten kann.

Warum Detektionen statt Video senden

Video ist schwergewichtig, schwer durchsuchbar und braucht einen Menschen, der es ansieht. Detektionen sind leichtgewichtig, strukturiert und maschinenlesbar. Eine einzelne Nachricht wie „3 Personen, 1 Gabelstapler in Zone A“ umfasst einige hundert Byte, während der zugehörige Frame Dutzende oder Hunderte Kilobyte groß ist.

Werden Detektionen per MQTT veröffentlicht, ist die Bildverarbeitung zudem von allem anderen entkoppelt:

• Dashboards abonnieren Zählwerte und zeigen die Belegung in Echtzeit.
• Ein Automatisierungsdienst abonniert Alarme und schaltet eine Warnleuchte.
• Ein Historian speichert Zählwerte pro Minute für Berichte.
• Ein weiteres Modell, vielleicht auf einem anderen Rechner, kombiniert Bildverarbeitung mit anderen Sensoren.

Keiner dieser Konsumenten muss wissen, welche Kamera, welches Modell oder welche GPU das Ergebnis erzeugt hat. Das ist dieselbe lose Kopplung, die MQTT für Sensoren so beliebt macht – jetzt angewandt auf Computer Vision auf Edge-Geräten.

Architektur

camera (USB / RTSP) -> Python + YOLO -> MQTT broker -> dashboards, alerts, storage

connectmq/demo/<id>/site1/dock/cam1/detections   every processed frame (throttled)
connectmq/demo/<id>/site1/dock/cam1/counts       retained, current count per class
connectmq/demo/<id>/site1/dock/cam1/alerts       event when a rule triggers
connectmq/demo/<id>/site1/dock/cam1/status       retained, online/offline (Last Will)

In einer Fabrik gehören diese Topics in den Unified Namespace (UNS): Die ISA-95-Hierarchie aus Enterprise, Site, Area und Line bildet den Topic-Pfad, und die Kamera hängt sich wie jeder andere Sensor darunter, etwa acme/werk-muenchen/logistik/rampe-1/cam1/counts. So finden MES, Dashboards und Historian Kameradaten an derselben Stelle wie die Maschinendaten der Halle; Details dazu im Beitrag zu Sparkplug B und Unified Namespace.

Voraussetzungen

• Python 3.9 oder neuer, eine Webcam oder eine RTSP-Kamera
• Für kleine Modelle mit wenigen Frames pro Sekunde genügt eine CPU; eine GPU oder ein Jetson ermöglicht Echtzeit
• Pakete:

pip install ultralytics opencv-python "paho-mqtt>=2.0"

Beim ersten Start werden die vortrainierten Gewichte yolov8n.pt heruntergeladen (das Nano-Modell, trainiert auf den 80 COCO-Klassen). Weitere Modellgrößen und neuere Versionen finden Sie in der Ultralytics-Dokumentation; die unten verwendete API funktioniert versionsübergreifend.

Der Detektions-Publisher

Speichern Sie dies als yolo_mqtt.py. Das Skript öffnet die Kamera, führt YOLO aus, wandelt die Ergebnisse in JSON um, veröffentlicht Detektionen höchstens zweimal pro Sekunde, aktualisiert Retained-Zählwerte bei Änderungen und löst einen Alarm aus, wenn sich Personen in einer Sperrzone befinden.

import json
import time
import uuid

import cv2
import paho.mqtt.client as mqtt
from ultralytics import YOLO

SOURCE = 0  # Webcam-Index oder "rtsp://user:pass@192.168.1.50:554/stream1"
PREFIX = "connectmq/demo/yolo-c83f/site1/dock/cam1"
CLASSES = {"person", "car", "truck", "forklift"}  # forklift nur mit eigenem Modell
MIN_CONF = 0.5
PUBLISH_EVERY = 0.5  # Sekunden
ZONE = (0.6, 0.0, 1.0, 1.0)  # Sperrzone: rechte 40 % des Bildes (x1, y1, x2, y2)

model = YOLO("yolov8n.pt")
names = model.names

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"yolo-cam1-{uuid.uuid4().hex[:6]}")
client.will_set(f"{PREFIX}/status", json.dumps({"online": False}), qos=1, retain=True)
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_start()
client.publish(f"{PREFIX}/status", json.dumps({"online": True, "model": "yolov8n"}),
               qos=1, retain=True)

cap = cv2.VideoCapture(SOURCE)
last_pub, last_counts, zone_alert = 0.0, None, False


def in_zone(cx, cy):
    x1, y1, x2, y2 = ZONE
    return x1 <= cx <= x2 and y1 <= cy <= y2


try:
    while True:
        ok, frame = cap.read()
        if not ok:
            time.sleep(1)
            cap.open(SOURCE)  # Neuverbindung bei RTSP-Abbrüchen
            continue
        h, w = frame.shape[:2]
        result = model(frame, conf=MIN_CONF, verbose=False)[0]

        objects, counts, people_in_zone = [], {}, 0
        for box in result.boxes:
            label = names[int(box.cls)]
            if label not in CLASSES:
                continue
            x1, y1, x2, y2 = box.xyxyn[0].tolist()  # normalisiert 0..1
            cx, cy = (x1 + x2) / 2, (y1 + y2) / 2
            objects.append({"label": label, "conf": round(float(box.conf), 2),
                            "box": [round(v, 3) for v in (x1, y1, x2, y2)]})
            counts[label] = counts.get(label, 0) + 1
            if label == "person" and in_zone(cx, cy):
                people_in_zone += 1

        now = time.time()
        if now - last_pub >= PUBLISH_EVERY:
            last_pub = now
            client.publish(f"{PREFIX}/detections", json.dumps(
                {"ts": now, "frame": [w, h], "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 people_in_zone and not zone_alert:
            zone_alert = True
            client.publish(f"{PREFIX}/alerts", json.dumps(
                {"ts": now, "rule": "person-in-zone", "state": "raised",
                 "count": people_in_zone}), qos=1)
        elif not people_in_zone and zone_alert:
            zone_alert = False
            client.publish(f"{PREFIX}/alerts", json.dumps(
                {"ts": now, "rule": "person-in-zone", "state": "cleared"}), qos=1)
finally:
    cap.release()
    client.publish(f"{PREFIX}/status", json.dumps({"online": False}), qos=1, retain=True)
    client.loop_stop()
    client.disconnect()

Beobachten Sie die Ausgabe in einem anderen Terminal:

mosquitto_sub -h broker.connectmq.com -p 1883 -i yolo-watch-c83f -t "connectmq/demo/yolo-c83f/#" -v

Gehen Sie in die rechte Bildhälfte, und Sie sehen einen raised-Alarm, gefolgt von cleared, sobald Sie den Bereich verlassen.

Den Detektions-Payload gestalten

Eine Detektionsnachricht des Skripts sieht so aus:

{
  "ts": 1792140112.42,
  "frame": [1280, 720],
  "objects": [
    {"label": "person", "conf": 0.88, "box": [0.612, 0.204, 0.731, 0.958]},
    {"label": "car", "conf": 0.74, "box": [0.051, 0.402, 0.338, 0.811]}
  ]
}

Richtlinien, die Payloads der YOLO-Objekterkennung über MQTT nützlich halten:

Zählung und Alarmregeln

Objekte pro Frame zu zählen ist einfach; vorbeilaufende Objekte zu zählen ist schwieriger, weil dieselbe Person in vielen Frames erscheint. Für Durchsatzzählungen aktivieren Sie das Tracking in Ultralytics (model.track(frame, persist=True)), das jedem Objekt eine ID zuweist, und zählen eine ID einmal, wenn ihr Mittelpunkt eine Linie überquert. Veröffentlichen Sie die Überquerung als Ereignis statt als Zählwert und lassen Sie einen nachgelagerten Dienst pro Minute aggregieren.

Alarmregeln gehören nah an die Kamera, wenn Latenz zählt (eine Person betritt eine Roboterzelle), und in nachgelagerte Dienste, wenn es sich um Geschäftslogik handelt (mehr als fünf Lkw warten zehn Minuten lang an der Rampe). Da die Ergebnisse auf MQTT liegen, können Sie eine Regel von einem Ort an den anderen verschieben, ohne den Kameracode zu ändern.

Detektionen nachgelagert verarbeiten

Der Nutzen, Ergebnisse der YOLO-Objekterkennung per MQTT zu veröffentlichen, zeigt sich auf der Konsumentenseite. Einige Muster, die Teams in den ersten Wochen umsetzen:

• Belegungs-Dashboard. Abonnieren Sie das Retained-Topic counts jeder Kamera mit einer Wildcard wie site1/+/+/counts. Jede neue Dashboard-Sitzung erhält sofort den aktuellen Zählwert jeder Kamera und danach Live-Updates.
• Statistiken pro Minute. Ein kleiner Python- oder Node-RED-Dienst aggregiert die Zählwerte pro Minute und schreibt sie in eine Zeitreihendatenbank. Berichte über Spitzenzeiten oder die Auslastung der Rampen entstehen dort, nicht aus Video.
• Verriegelungen und Warnungen. Ein Automatisierungsdienst abonniert das Alarm-Topic und schaltet eine Warnleuchte oder benachrichtigt eine Aufsichtsperson. Sicherheitskritische Verriegelungen gehören in zertifizierte Hardware; Kameraalarme sind eine Ergänzung, kein Ersatz.
• Abgleich mit anderen Sensoren. Ein Personenalarm in Kombination mit einer Lichtschranke oder einem Türkontakt ist weitaus zuverlässiger als jedes Signal für sich.

Da jeder dieser Konsumenten nur ein weiterer Subscriber ist, verlangsamt ein zusätzlicher niemals die Kamera. Der Broker verteilt dieselbe Nachricht an alle.

Datenschutz und Datenminimierung

Kameras, die Menschen erfassen, bringen Verantwortung mit sich. Nur Labels, Zählwerte und normalisierte Rahmen zu veröffentlichen, niemals Gesichter oder Frames, ist die wirksamste Datenschutzmaßnahme, die Sie ergreifen können, und sie passt ganz natürlich zu dieser Architektur. Benötigen Sie Schnappschüsse zur Überprüfung, speichern Sie sie lokal mit kurzer Aufbewahrungsfrist oder senden Sie sie an ein privates Topic, das nur autorisierte Dienste lesen können. Dokumentieren Sie, was die Kamera erkennt und warum, und prüfen Sie die lokalen Vorschriften, bevor Sie sie an Arbeitsplätzen oder in öffentlichen Bereichen einsetzen. In der EU ist das vor allem die DSGVO; in Deutschland ist bei Kameras am Arbeitsplatz zudem in der Regel der Betriebsrat einzubinden.

Fehlerbehebung

1. Sehr niedrige Bildrate auf der CPU. Verwenden Sie das Nano-Modell, verringern Sie die Eingabegröße mit imgsz=416 oder verarbeiten Sie nur jeden zweiten Frame.
2. Der RTSP-Stream friert ein. OpenCV puffert Frames; lesen Sie kontinuierlich in einem separaten Thread und verarbeiten Sie nur den jeweils neuesten Frame.
3. Flackernde Alarme. Verlangen Sie eine Detektion in mehreren aufeinanderfolgenden Frames, bevor ein Alarm ausgelöst wird, und mehrere leere Frames, bevor er zurückgesetzt wird.
4. Nachrichten werden abgelehnt oder verworfen. Bleiben Sie auf dem öffentlichen MQTT-Broker unter 10 Nachrichten pro Sekunde und Client sowie 64 KB pro Nachricht; die Drosselung sorgt dafür.
5. Falsche Klassen. Das vortrainierte Modell kennt nur die COCO-Klassen. Für Gabelstapler, Helme oder bestimmte Produkte trainieren Sie ein eigenes Modell mit Ihren eigenen Bildern.

Hinweise für den Produktivbetrieb

Veröffentlichen Sie bei echten Kameras an einen privaten Broker über TLS (Port 8883) mit Zugangsdaten und senden Sie niemals Frames mit Personen an ein öffentliches Topic. Ein verwalteter MQTT-Broker wie ConnectMQ übernimmt Authentifizierung, Zertifikate und Monitoring, und sein Live-Topic-Explorer ist ein schneller Weg, um zu prüfen, ob Detektionen fließen. Verwenden Sie den öffentlichen MQTT-Broker nur für die Entwicklung mit Testszenen, nie für Produktionsdaten.

Läuft Ihre Kamera auf einem NVIDIA-Board, finden Sie unter NVIDIA Jetson mit MQTT GPU-beschleunigte Inferenz; eine kostengünstige Variante auf einem Mikrocontroller beschreibt Personenerkennung mit der ESP32-CAM. Der Leitfaden zu KI und IoT erklärt, wie Bildverarbeitung zu anderen Edge-AI-Workloads passt.

Mit ConnectMQ starten

Erstellen Sie unter app.connectmq.com einen privaten ConnectMQ-Broker im kostenlosen Serverless-Tarif, ganz ohne Kreditkarte, ergänzen Sie client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD") und client.tls_set(), wechseln Sie auf Port 8883, und die Detektionen Ihrer Kamera fließen sicher.

Häufig gestellte Fragen

Kann ich Kamera-Frames zusammen mit Detektionen per MQTT senden?

Sie können JPEG-Schnappschüsse auf einem separaten Topic innerhalb der maximalen Nachrichtengröße des Brokers senden, sollten kontinuierliches Video aber über ein Streaming-Protokoll übertragen. Detektionen sollten klein und ohne Frames bleiben.

Welche YOLO-Modellgröße sollte ich verwenden?

Beginnen Sie auf CPUs und kleinen Boards mit dem Nano-Modell und wechseln Sie nur dann zu Small oder Medium, wenn die Genauigkeit nicht ausreicht. Größere Modelle benötigen für Echtzeitraten eine GPU.

Wie zähle ich Personen, statt sie nur pro Frame zu erkennen?

Nutzen Sie den Tracker, um IDs zu vergeben, und zählen Sie jede ID einmal, wenn sie eine Linie überquert oder eine Zone betritt. Veröffentlichen Sie jede Überquerung als Ereignis und aggregieren Sie nachgelagert.

Führt der Broker das YOLO-Modell aus?

Nein. Das Modell läuft auf dem Rechner Ihrer Kamera, und der Broker transportiert lediglich die resultierenden Nachrichten an alle Abonnenten.