Google Coral Edge TPU mit MQTT: Bilder auf einem Raspberry Pi klassifizieren und Ergebnisse veröffentlichen

Mit der Google Coral Edge TPU führt ein kleines Board wie ein Raspberry Pi die Inferenz neuronaler Netze weitaus schneller aus als mit seiner CPU allein, was Bildklassifikation in Echtzeit an der Edge praktikabel macht. In diesem Tutorial installieren Sie die Runtime der Coral Edge TPU, führen einen quantisierten MobileNet-Klassifikator aus und veröffentlichen jede Vorhersage als strukturiertes JSON über MQTT, damit Dashboards, Alarmierungswerkzeuge und andere Dienste auf das reagieren können, was die Kamera sieht.

Was Sie brauchen

Der Aufbau ist bewusst einfach gehalten, damit Sie sich auf den Datenpfad vom Beschleuniger zum Broker konzentrieren können.

Stecken Sie den Beschleuniger in einen blauen USB-3-Anschluss. Ein USB-2-Anschluss funktioniert zwar, verlangsamt aber die Übertragungen zwischen Pi und Beschleuniger spürbar, was einen Großteil des Vorteils zunichtemacht.

Wie die Coral Edge TPU in eine IoT-Pipeline passt

Die Edge TPU ist ein ASIC, der ausschließlich TensorFlow-Lite-Modelle ausführt, die vollständig ganzzahlig quantisiert und mit dem Edge TPU Compiler kompiliert wurden. Operationen, die der Compiler nicht abbilden kann, laufen auf der CPU. Ein gut kompiliertes Modell hält daher fast alles auf dem Beschleuniger.

In einer IoT-Architektur ist der Beschleuniger eine Stufe in einem Kreislauf:

1. Ein Kamerabild oder eine Bilddatei kommt auf dem Pi an.
2. Die Edge TPU klassifiziert es in Millisekunden.
3. Der Pi veröffentlicht ein kompaktes JSON-Ergebnis (Labels, Scores, Latenz, Geräte-ID) auf einem MQTT-Topic.
4. Abonnenten wie Node-RED, ein Datenbank-Writer oder ein Alarmierungsskript konsumieren das Ergebnis.

Nur das Inferenzergebnis verlässt das Gerät, niemals das Rohbild. Das hält die Bandbreite gering, vermeidet, dass Bilder von Personen oder Räumlichkeiten in die Cloud gelangen (was auch die Bewertung nach DSGVO deutlich vereinfacht), und bleibt bequem innerhalb der MQTT-Nachrichtenlimits. In der Fertigung ist das typische Einsatzfeld die visuelle Qualitätsprüfung direkt an der Linie: Die Kamera prüft jedes Teil, und nur das Ergebnis „i. O.“ oder „n. i. O.“ samt Score geht an MES und Qualitätsdashboard. Wenn Sie noch entscheiden, wo die Inferenz laufen soll, behandelt unser Vergleich Edge AI vs. Cloud AI die Zielkonflikte im Detail.

Schritt 1: Die Edge-TPU-Runtime installieren

Fügen Sie das Coral-Paket-Repository hinzu und installieren Sie die Runtime mit Standardtaktfrequenz:

echo "deb [signed-by=/usr/share/keyrings/coral-edgetpu.gpg] https://packages.cloud.google.com/apt coral-edgetpu-stable main" | sudo tee /etc/apt/sources.list.d/coral-edgetpu.list
curl -fsSL https://packages.cloud.google.com/apt/doc/apt-key.gpg | sudo gpg --dearmor -o /usr/share/keyrings/coral-edgetpu.gpg
sudo apt-get update
sudo apt-get install -y libedgetpu1-std

Ziehen Sie den Beschleuniger nach der Installation ab und stecken Sie ihn wieder an, damit die udev-Regeln greifen. Prüfen Sie, ob das Gerät sichtbar ist:

lsusb | grep -Ei "global unichip|google"

Vor der ersten Inferenz meldet sich der Beschleuniger als „Global Unichip Corp.“; nachdem die Runtime die Firmware geladen hat, wird er erneut als „Google Inc.“ erkannt. Beides ist normal.

Es gibt außerdem ein Paket libedgetpu1-max, das den Chip mit höherem Takt betreibt. Es ist schneller, doch der Beschleuniger wird heiß. Beginnen Sie daher mit der Standard-Runtime und wechseln Sie nur, wenn Sie gemessen haben, dass Sie sie brauchen, und für Luftzirkulation gesorgt ist.

Schritt 2: Eine kompatible Python-Umgebung anlegen

Das häufigste Coral-Problem ist eine Versionsabweichung zwischen Python, der TensorFlow-Lite-Runtime und der Edge-TPU-Bibliothek. Die offiziellen Coral-Python-Pakete zielen auf Python 3.6 bis 3.9, während das aktuelle Raspberry Pi OS ein neueres Python mitbringt. Der zuverlässige Weg ist eine eigene Python-3.9-Umgebung, zum Beispiel mit pyenv:

sudo apt-get install -y build-essential libssl-dev zlib1g-dev libbz2-dev libreadline-dev libsqlite3-dev libffi-dev liblzma-dev
curl https://pyenv.run | bash
# add the pyenv lines it prints to ~/.bashrc, then open a new shell
pyenv install 3.9.19
pyenv virtualenv 3.9.19 coral
pyenv activate coral
pip install --extra-index-url https://google-coral.github.io/py-repo/ "pycoral~=2.0"
pip install "paho-mqtt>=2.0" pillow "numpy<2"

Wenn Sie pycoral aus dem Coral-Paketindex installieren, wird auch der passende Build von tflite_runtime mitinstalliert, den unser Skript importiert. NumPy unter Version 2 festzuschreiben vermeidet binäre Inkompatibilitätsfehler mit diesen älteren Wheels.

Laden Sie ein kompiliertes Modell und seine Labels aus dem Coral-Repository mit Testdaten herunter:

mkdir -p ~/coral-mqtt && cd ~/coral-mqtt
curl -LO https://github.com/google-coral/test_data/raw/master/mobilenet_v2_1.0_224_quant_edgetpu.tflite
curl -LO https://github.com/google-coral/test_data/raw/master/imagenet_labels.txt
curl -LO https://github.com/google-coral/test_data/raw/master/parrot.jpg

Schritt 3: Bilder klassifizieren und an MQTT veröffentlichen

Das folgende Skript lädt das Modell mit dem Edge-TPU-Delegate, klassifiziert jedes Bild in einem Ordner und veröffentlicht die drei besten Vorhersagen. Es verwendet paho-mqtt 2.x mit der Callback-API Version 2.

import json
import sys
import time
from pathlib import Path

import numpy as np
from PIL import Image
import paho.mqtt.client as mqtt
from tflite_runtime.interpreter import Interpreter, load_delegate

MODEL = "mobilenet_v2_1.0_224_quant_edgetpu.tflite"
LABELS = "imagenet_labels.txt"
BROKER = "broker.connectmq.com"
PORT = 1883
DEVICE_ID = "pi-coral-01"
# Replace 7f3a9c with your own random string so your topics are unique
TOPIC = f"connectmq/demo/7f3a9c/coral/{DEVICE_ID}/classification"
STATUS_TOPIC = f"connectmq/demo/7f3a9c/coral/{DEVICE_ID}/status"


def load_labels(path):
    labels = {}
    with open(path, encoding="utf-8") as f:
        for i, line in enumerate(f):
            parts = line.strip().split(maxsplit=1)
            if len(parts) == 2 and parts[0].isdigit():
                labels[int(parts[0])] = parts[1]
            else:
                labels[i] = line.strip()
    return labels


labels = load_labels(LABELS)
interpreter = Interpreter(
    model_path=MODEL,
    experimental_delegates=[load_delegate("libedgetpu.so.1")],
)
interpreter.allocate_tensors()
inp = interpreter.get_input_details()[0]
out = interpreter.get_output_details()[0]
_, height, width, _ = inp["shape"]


def classify(image_path, top_k=3):
    img = Image.open(image_path).convert("RGB").resize((width, height), Image.LANCZOS)
    data = np.expand_dims(np.asarray(img, dtype=np.uint8), axis=0)
    interpreter.set_tensor(inp["index"], data)
    start = time.perf_counter()
    interpreter.invoke()
    latency_ms = (time.perf_counter() - start) * 1000
    raw = interpreter.get_tensor(out["index"])[0].astype(np.float32)
    scale, zero_point = out["quantization"]
    probs = (raw - zero_point) * scale if scale else raw
    top = np.argsort(probs)[::-1][:top_k]
    results = [
        {"label": labels.get(int(i), str(i)), "score": round(float(probs[i]), 4)}
        for i in top
    ]
    return results, latency_ms


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


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"{DEVICE_ID}-7f3a9c")
client.on_connect = on_connect
client.will_set(STATUS_TOPIC, "offline", qos=1, retain=True)
client.connect(BROKER, PORT, keepalive=60)
client.loop_start()

folder = Path(sys.argv[1] if len(sys.argv) > 1 else ".")
try:
    for image_path in sorted(folder.glob("*.jpg")):
        results, latency_ms = classify(image_path)
        payload = {
            "device": DEVICE_ID,
            "image": image_path.name,
            "top": results,
            "latencyMs": round(latency_ms, 2),
            "ts": int(time.time() * 1000),
        }
        info = client.publish(TOPIC, json.dumps(payload), qos=1)
        info.wait_for_publish()
        print(payload)
        time.sleep(1)  # stay well under the public broker's rate limit
finally:
    client.publish(STATUS_TOPIC, "offline", qos=1, retain=True).wait_for_publish()
    client.loop_stop()
    client.disconnect()

Führen Sie es mit dem heruntergeladenen Papageienbild aus:

python classify_mqtt.py ~/coral-mqtt

Die erste Inferenz ist langsamer, weil das Modell auf den Beschleuniger geladen wird; die folgenden zeigen die Latenz im eingeschwungenen Zustand.

Schritt 4: Die Ergebnisse der Coral Edge TPU eintreffen sehen

Öffnen Sie ein zweites Terminal (auf einem beliebigen Rechner) und abonnieren Sie mit den Mosquitto-Clients:

mosquitto_sub -h broker.connectmq.com -p 1883 -i viewer-7f3a9c -t "connectmq/demo/7f3a9c/coral/#" -v

Sie sollten den Retained-Status online sehen, gefolgt von einer JSON-Nachricht pro Bild, zum Beispiel mit dem Top-Label „macaw“ für das Papageien-Testbild. Da broker.connectmq.com ein kostenloser öffentlicher MQTT-Broker ist, kann jeder, der Ihr Topic errät, es lesen. Senden Sie dorthin also nur Demodaten; der öffentliche MQTT-Broker ist für Tests und Prototypen gedacht, nicht für Produktivdaten. Er erlaubt bis zu 64 KB pro Nachricht und 10 Nachrichten pro Sekunde und Client, was für Klassifikationsergebnisse reichlich ist, aber ein Grund, keine Rohbilder zu streamen.

Vom Test in die Produktion

Wenn der Prototyp funktioniert, wechseln Sie zu authentifizierten, verschlüsselten Verbindungen. Erstellen Sie im ConnectMQ-Portal einen Broker, kopieren Sie die generierten Zugangsdaten und ändern Sie den Verbindungsblock:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # uses the system CA store; the broker has a Let's Encrypt certificate
client.will_set(STATUS_TOPIC, "offline", qos=1, retain=True)
client.connect("YOUR_BROKER_HOST", 8883, keepalive=60)

Ein verwalteter MQTT-Broker nimmt Ihnen Patches, Zertifikate und Kapazitätsplanung ab, sodass sich der Pi nur um die Inferenz kümmern muss. Bei ConnectMQ ist die Client ID an die ausgegebenen Zugangsdaten gebunden, was verhindert, dass sich ein anderes Gerät als Ihr Kameraknoten ausgibt.

Designtipps für Edge-TPU-Projekte

Einige Entscheidungen machen den Unterschied zwischen einer Demo und einem System, das Sie dauerhaft laufen lassen können:

1. Ereignisse veröffentlichen, nicht jedes Bild. Bei einer Live-Kamera veröffentlichen Sie nur, wenn sich das Top-Label ändert oder eine Konfidenzschwelle überschreitet, plus einen regelmäßigen Heartbeat. So bleibt der Datenverkehr vorhersehbar.
2. Kontext in den Payload aufnehmen. Geräte-ID, Modellname und -version, Latenz und Zeitstempel machen Ergebnisse auch Monate später nachvollziehbar.
3. Retained-Status mit Last Will verwenden. Das Retained-Topic online/offline ermöglicht Dashboards anzuzeigen, welche Beschleuniger aktiv sind.
4. Topics hierarchisch strukturieren. Etwas wie site/area/line/camera/classification skaliert besser als ein flacher Namensraum und lässt sich sauber auf die ISA-95-Ebenen abbilden. Im Werk wird daraus ein Ast im Unified Namespace (UNS), etwa acme/werk-muenchen/montage/linie-2/kamera-1/classification, sodass Prüfergebnisse neben Maschinen- und Prozessdaten derselben Linie liegen.
5. Modell-Updates planen. Ein neues Modell zu kompilieren ist einfach; es sicher auf viele Geräte auszurollen nicht. Unser Leitfaden, wie Sie ML-Modelle auf Edge-Geräten bereitstellen, zeigt, wie Sie Updates über MQTT auslösen und verfolgen.

Fehlerbehebung

Die offizielle Coral-Dokumentation listet die unterstützten Plattformen und die Anforderungen des Edge TPU Compilers auf, falls Sie eigene Modelle kompilieren möchten.

Wie es weitergeht

Bildklassifikation ist die einfachste Arbeitslast; dasselbe Veröffentlichungsmuster funktioniert für Objekterkennung und Posenschätzung. Wenn Sie Bounding Boxes statt Labels brauchen, lesen Sie YOLO-Objekterkennung mit MQTT. Um den Coral-Beschleuniger mit Alternativen wie Jetson-Modulen und Mikrocontrollern zu vergleichen, lesen Sie die Übersicht zu Edge-AI-Hardware, und für das Gesamtbild, wie Modelle mit vernetzten Geräten zusammenspielen, beginnen Sie mit unserem Leitfaden zu KI und IoT.

Starten Sie mit ConnectMQ

Ihre Edge-TPU-Ergebnisse verdienen ein zuverlässiges Zuhause. ConnectMQ bietet Ihnen einen Cloud-MQTT-Broker mit TLS, WebSockets und einem Live-Topic-Explorer, in dem Sie Klassifikationen beim Eintreffen verfolgen. Erstellen Sie einen kostenlosen Serverless-Broker in wenigen Minuten, ohne Kreditkarte, und richten Sie Ihren Raspberry Pi darauf aus.

Häufig gestellte Fragen

Funktioniert die Coral Edge TPU mit dem Raspberry Pi 5?

Ja, der Coral USB Accelerator wird über USB angeschlossen und funktioniert auf einem Raspberry Pi 5 mit 64-Bit-Betriebssystem. Die größte Hürde ist die Python-Version, weshalb dieses Tutorial eine separate Python-3.9-Umgebung verwendet.

Kann ich jedes TensorFlow-Modell auf der Edge TPU ausführen?

Nein. Das Modell muss ein vollständig ganzzahlig quantisiertes TensorFlow-Lite-Modell sein, das mit dem Edge TPU Compiler kompiliert wurde. Nicht unterstützte Operationen fallen auf die CPU zurück und verringern den Geschwindigkeitsvorteil.

Sollte ich Kamerabilder über MQTT senden?

In der Regel nicht. Veröffentlichen Sie Inferenzergebnisse und senden Sie bei Bedarf nur gelegentlich Schnappschüsse; Bilder überschreiten schnell sinnvolle Nachrichtengrößen und werfen Datenschutzfragen auf.

Welche QoS-Stufe sollte ich für Klassifikationsergebnisse verwenden?

QoS 1 ist ein guter Standard, weil es die Zustellung mindestens einmal garantiert, ohne den Overhead von QoS 2. Fügen Sie einen Zeitstempel hinzu, damit Abonnenten Duplikate ignorieren können.