Google Coral Edge TPU avec MQTT : classer des images sur un Raspberry Pi et publier les résultats

Le Google Coral Edge TPU permet à une petite carte comme un Raspberry Pi d'exécuter l'inférence de réseaux de neurones bien plus vite que son seul CPU, ce qui rend la classification d'images en temps réel possible en périphérie. Dans ce tutoriel, vous allez installer le runtime Coral Edge TPU, exécuter un classificateur MobileNet quantifié et publier chaque prédiction en JSON structuré via MQTT, pour que les tableaux de bord, les outils d'alerte et d'autres services puissent réagir à ce que voit la caméra.

Ce dont vous avez besoin

La configuration est volontairement simple, pour que vous puissiez vous concentrer sur le chemin des données entre l'accélérateur et le broker.

Branchez l'accélérateur sur un port USB 3 bleu. Un port USB 2 fonctionne mais ralentit nettement les transferts entre le Pi et l'accélérateur, ce qui masque une grande partie du bénéfice.

La place du Coral Edge TPU dans un pipeline IoT

L'Edge TPU est un ASIC qui n'exécute que des modèles TensorFlow Lite entièrement quantifiés en entiers et compilés avec l'Edge TPU compiler. Les opérations que le compilateur ne peut pas mapper s'exécutent sur le CPU ; un modèle bien compilé garde donc presque tout sur l'accélérateur.

Dans une architecture IoT, l'accélérateur est une étape d'une boucle :

1. Une image de caméra ou un fichier image arrive sur le Pi.
2. L'Edge TPU la classe en quelques millisecondes.
3. Le Pi publie un résultat JSON compact (labels, scores, latence, identifiant d'appareil) sur un topic MQTT.
4. Des abonnés comme Node-RED, un processus d'écriture en base ou un script d'alerte consomment le résultat.

Seul le résultat d'inférence quitte l'appareil, jamais l'image brute. Cela limite la bande passante, évite d'envoyer dans le cloud des photos de personnes ou de locaux et respecte largement les limites de taille des messages MQTT. Si vous hésitez encore sur l'endroit où exécuter l'inférence, notre comparaison edge AI vs IA cloud détaille les compromis.

Étape 1 : installer le runtime Edge TPU

Ajoutez le dépôt de paquets Coral et installez le runtime à fréquence standard :

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

Débranchez puis rebranchez l'accélérateur après l'installation pour que les règles udev s'appliquent. Vérifiez que l'appareil est visible :

lsusb | grep -Ei "global unichip|google"

Avant la première inférence, l'accélérateur s'identifie comme « Global Unichip Corp. » ; une fois que le runtime a chargé le firmware, il se réénumère en « Google Inc. ». Les deux sont normaux.

Il existe aussi un paquet libedgetpu1-max qui fait tourner la puce à une fréquence plus élevée. Il est plus rapide mais l'accélérateur chauffe : commencez donc avec le runtime standard et ne changez que si vous avez mesuré que c'est nécessaire et que vous disposez d'une bonne ventilation.

Étape 2 : créer un environnement Python compatible

Le problème Coral le plus fréquent est une incompatibilité de versions entre Python, le runtime TensorFlow Lite et la bibliothèque Edge TPU. Les paquets Python officiels de Coral ciblent Python 3.6 à 3.9, alors que Raspberry Pi OS embarque aujourd'hui un Python plus récent. La voie fiable est un environnement Python 3.9 dédié, par exemple avec 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"

Installer pycoral depuis l'index de paquets Coral récupère aussi la version correspondante de tflite_runtime, que notre script importe. Fixer NumPy en dessous de la version 2 évite les erreurs d'incompatibilité binaire avec ces anciens wheels.

Téléchargez un modèle compilé et ses labels depuis le dépôt de données de test Coral :

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

Étape 3 : classer des images et publier vers MQTT

Le script ci-dessous charge le modèle avec le delegate Edge TPU, classe chaque image d'un dossier et publie les trois meilleures prédictions. Il utilise paho-mqtt 2.x avec l'API de callbacks 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()

Exécutez-le sur l'image de perroquet téléchargée :

python classify_mqtt.py ~/coral-mqtt

La première inférence est plus lente, car le modèle est chargé sur l'accélérateur ; les suivantes affichent la latence en régime établi.

Étape 4 : regarder arriver les résultats du Coral Edge TPU

Ouvrez un second terminal (sur n'importe quelle machine) et abonnez-vous avec les clients Mosquitto :

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

Vous devriez voir le statut online en retained, suivi d'un message JSON par image, par exemple « macaw » comme label principal pour la photo de perroquet de test. Comme broker.connectmq.com est le serveur MQTT public gratuit, réservé aux tests et aux prototypes, quiconque devine votre topic peut le lire : n'y envoyez que des données de démonstration. Le serveur public autorise jusqu'à 64 Ko par message et 10 messages par seconde et par client, ce qui est largement suffisant pour des résultats de classification, mais constitue une raison de ne pas diffuser des images brutes.

Passer du test à la production

Une fois le prototype fonctionnel, passez à des connexions authentifiées et chiffrées. Créez un broker dans le portail ConnectMQ, copiez les identifiants générés et modifiez le bloc de connexion :

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)

Un broker MQTT managé vous décharge des correctifs, des certificats et de la planification de capacité : le Pi n'a plus qu'à s'occuper de l'inférence. Avec ConnectMQ, le client ID est lié aux identifiants attribués, ce qui empêche un autre appareil d'usurper l'identité de votre nœud caméra.

Conseils de conception pour les projets Edge TPU

Quelques décisions font la différence entre une démo et un système que vous pouvez laisser tourner :

1. Publiez des événements, pas chaque image. Pour une caméra en direct, ne publiez que lorsque le label principal change ou franchit un seuil de confiance, avec un heartbeat périodique. Le trafic reste ainsi prévisible.
2. Ajoutez du contexte dans le payload. L'identifiant de l'appareil, le nom et la version du modèle, la latence et l'horodatage rendent les résultats exploitables pour le débogage des mois plus tard.
3. Utilisez un statut en retained avec un Last Will. Le topic online/offline en retained permet aux tableaux de bord d'afficher quels accélérateurs sont actifs.
4. Structurez les topics de manière hiérarchique. Une forme comme site/area/line/camera/classification passe mieux à l'échelle qu'un espace de noms plat et correspond proprement aux niveaux ISA-95.
5. Planifiez les mises à jour du modèle. Compiler un nouveau modèle est facile ; le déployer en toute sécurité sur de nombreux appareils, beaucoup moins. Notre guide sur le déploiement de modèles ML sur des appareils edge montre comment déclencher et suivre les mises à jour via MQTT.

Dépannage

La documentation officielle de Coral liste les plateformes prises en charge et les exigences de l'Edge TPU compiler si vous souhaitez compiler vos propres modèles.

Pour aller plus loin

La classification d'images est la charge de travail la plus simple ; le même schéma de publication fonctionne pour la détection d'objets et l'estimation de pose. Si vous avez besoin de boîtes englobantes plutôt que de labels, consultez la détection d'objets YOLO avec MQTT. Pour comparer l'accélérateur Coral à d'autres options comme les modules Jetson et les microcontrôleurs, lisez la présentation du matériel edge AI, et pour une vue d'ensemble de l'association entre modèles et appareils connectés, commencez par notre guide IA et IoT.

Démarrer avec ConnectMQ

Les résultats de votre Edge TPU méritent un hébergement fiable. ConnectMQ vous offre un broker MQTT cloud avec TLS, WebSockets et un explorateur de topics en direct pour suivre les classifications dès leur arrivée. Créez un broker Serverless (Free) en quelques minutes, sans carte bancaire, et dirigez-y votre Raspberry Pi.

Questions fréquentes

Le Coral Edge TPU fonctionne-t-il avec le Raspberry Pi 5 ?

Oui, le Coral USB Accelerator se connecte en USB et fonctionne sur un Raspberry Pi 5 avec un OS 64 bits. Le principal obstacle est la version de Python, c'est pourquoi ce tutoriel utilise un environnement Python 3.9 séparé.

Puis-je exécuter n'importe quel modèle TensorFlow sur l'Edge TPU ?

Non. Le modèle doit être un modèle TensorFlow Lite entièrement quantifié en entiers et compilé avec l'Edge TPU compiler. Les opérations non prises en charge retombent sur le CPU et réduisent le gain de vitesse.

Faut-il envoyer les images de la caméra via MQTT ?

En général, non. Publiez les résultats d'inférence et n'envoyez que des instantanés occasionnels si nécessaire ; les images dépassent rapidement des tailles de message raisonnables et posent des questions de confidentialité.

Quel niveau de QoS utiliser pour les résultats de classification ?

La QoS 1 est un bon choix par défaut, car elle garantit une livraison au moins une fois sans la surcharge de la QoS 2. Ajoutez un horodatage pour que les abonnés puissent ignorer les doublons.