Google Coral Edge TPU con MQTT: clasifica imágenes en una Raspberry Pi y publica resultados

El Google Coral Edge TPU permite que una placa pequeña como una Raspberry Pi ejecute inferencia de redes neuronales mucho más rápido que solo con su CPU, lo que hace viable la clasificación de imágenes en tiempo real en el edge. En este tutorial instalarás el runtime de Coral Edge TPU, ejecutarás un clasificador MobileNet cuantizado y publicarás cada predicción como JSON estructurado por MQTT, para que paneles, herramientas de alertas y otros servicios puedan reaccionar a lo que ve la cámara.

Qué necesitas

La configuración es deliberadamente sencilla para que puedas centrarte en el recorrido de los datos desde el acelerador hasta el broker.

Conecta el acelerador a un puerto USB 3 azul. Un puerto USB 2 funciona, pero ralentiza notablemente las transferencias entre la Pi y el acelerador, lo que oculta gran parte de la ventaja.

Cómo encaja el Coral Edge TPU en un pipeline IoT

El Edge TPU es un ASIC que solo ejecuta modelos TensorFlow Lite totalmente cuantizados a enteros y compilados con el compilador de Edge TPU. Las operaciones que el compilador no puede mapear se ejecutan en la CPU, así que un modelo bien compilado mantiene casi todo en el acelerador.

En una arquitectura IoT, el acelerador es una etapa de un bucle:

1. Llega a la Pi un fotograma de cámara o una imagen.
2. El Edge TPU la clasifica en milisegundos.
3. La Pi publica un resultado JSON compacto (etiquetas, puntuaciones, latencia, ID del dispositivo) en un topic MQTT.
4. Suscriptores como Node-RED, un escritor de base de datos o un script de alertas consumen el resultado.

Solo el resultado de la inferencia sale del dispositivo, nunca la imagen en bruto. Así el ancho de banda se mantiene bajo, se evita enviar a la nube imágenes de personas o instalaciones y se respetan con holgura los límites de los mensajes MQTT. Si todavía estás decidiendo dónde ejecutar la inferencia, nuestra comparativa de edge AI vs IA en la nube analiza en detalle las ventajas e inconvenientes.

Paso 1: instalar el runtime de Edge TPU

Añade el repositorio de paquetes de Coral e instala el runtime de frecuencia estándar:

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

Desconecta y vuelve a conectar el acelerador después de la instalación para que se apliquen las reglas de udev. Comprueba que el dispositivo es visible:

lsusb | grep -Ei "global unichip|google"

Antes de la primera inferencia, el acelerador se identifica como "Global Unichip Corp."; después de que el runtime cargue el firmware, se vuelve a enumerar como "Google Inc.". Ambos casos son normales.

También existe un paquete libedgetpu1-max que hace funcionar el chip a una frecuencia de reloj más alta. Es más rápido, pero el acelerador se calienta, así que empieza con el runtime estándar y cambia solo si has medido que lo necesitas y dispones de ventilación.

Paso 2: crear un entorno de Python compatible

El problema más habitual con Coral es una incompatibilidad de versiones entre Python, el runtime de TensorFlow Lite y la librería de Edge TPU. Los paquetes oficiales de Python de Coral están pensados para Python 3.6 a 3.9, mientras que el Raspberry Pi OS actual incluye un Python más reciente. La vía fiable es un entorno dedicado de Python 3.9, por ejemplo con 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
# añade a ~/.bashrc las líneas de pyenv que imprime y abre una nueva 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"

Instalar pycoral desde el índice de paquetes de Coral también descarga la versión correspondiente de tflite_runtime, que es lo que importa nuestro script. Fijar NumPy por debajo de la versión 2 evita errores de incompatibilidad binaria con estos wheels más antiguos.

Descarga un modelo compilado y sus etiquetas del repositorio de datos de prueba de 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

Paso 3: clasificar imágenes y publicar en MQTT

El script siguiente carga el modelo con el delegate de Edge TPU, clasifica todas las imágenes de una carpeta y publica las tres predicciones principales. Usa paho-mqtt 2.x con la API de callbacks versión 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"
# Sustituye 7f3a9c por tu propia cadena aleatoria para que tus topics sean únicos
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)  # muy por debajo del límite de frecuencia del broker público
finally:
    client.publish(STATUS_TOPIC, "offline", qos=1, retain=True).wait_for_publish()
    client.loop_stop()
    client.disconnect()

Ejecútalo con la imagen del loro descargada:

python classify_mqtt.py ~/coral-mqtt

La primera inferencia es más lenta porque el modelo se carga en el acelerador; las siguientes muestran la latencia en régimen estable.

Paso 4: ver cómo llegan los resultados del Coral Edge TPU

Abre un segundo terminal (en cualquier máquina) y suscríbete con los clientes de Mosquitto:

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

Deberías ver el estado online retenido seguido de un mensaje JSON por imagen, por ejemplo con "macaw" como etiqueta principal para la imagen de prueba del loro. Como broker.connectmq.com es un servidor MQTT público gratuito, pensado para pruebas y no para producción, cualquiera que adivine tu topic puede leerlo, así que envía allí solo datos de demostración. El broker público admite hasta 64 KB por mensaje y 10 mensajes por segundo por cliente, más que suficiente para los resultados de clasificación, pero un motivo para no transmitir fotogramas en bruto.

De las pruebas a producción

Cuando el prototipo funcione, cambia a conexiones autenticadas y cifradas. Crea un broker en el portal de ConnectMQ, copia las credenciales generadas y modifica el bloque de conexión:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # usa el almacén de CA del sistema; el broker tiene un certificado de Let's Encrypt
client.will_set(STATUS_TOPIC, "offline", qos=1, retain=True)
client.connect("YOUR_BROKER_HOST", 8883, keepalive=60)

Un broker MQTT gestionado elimina el trabajo de parches, certificados y planificación de capacidad, de modo que la Pi solo tiene que ocuparse de la inferencia. Con ConnectMQ, el client ID está vinculado a las credenciales emitidas, lo que impide que otro dispositivo suplante a tu nodo de cámara.

Consejos de diseño para proyectos con Edge TPU

Unas pocas decisiones marcan la diferencia entre una demo y un sistema que puedes dejar funcionando:

1. Publica eventos, no cada fotograma. Con una cámara en directo, publica solo cuando cambie la etiqueta principal o se supere un umbral de confianza, más un latido periódico. Así el tráfico es predecible.
2. Incluye contexto en el payload. El ID del dispositivo, el nombre y la versión del modelo, la latencia y la marca de tiempo hacen que los resultados se puedan depurar meses después.
3. Usa un estado retenido con Last Will. El topic online/offline retenido permite que los paneles muestren qué aceleradores están activos.
4. Estructura los topics de forma jerárquica. Algo como site/area/line/camera/classification escala mejor que un espacio de nombres plano y se corresponde limpiamente con los niveles ISA-95.
5. Planifica las actualizaciones del modelo. Compilar un modelo nuevo es fácil; desplegarlo de forma segura en muchos dispositivos, no. Nuestra guía sobre cómo desplegar modelos de ML en dispositivos edge muestra cómo disparar y seguir las actualizaciones por MQTT.

Solución de problemas

La documentación oficial de Coral enumera las plataformas compatibles y los requisitos del compilador de Edge TPU por si quieres compilar tus propios modelos.

Hacia dónde seguir

La clasificación de imágenes es la carga de trabajo más sencilla; el mismo patrón de publicación sirve para la detección de objetos y la estimación de pose. Si necesitas cuadros delimitadores en lugar de etiquetas, consulta detección de objetos con YOLO y MQTT. Para comparar el acelerador Coral con alternativas como los módulos Jetson y los microcontroladores, lee la panorámica de hardware para edge AI, y para una visión de conjunto sobre cómo combinar modelos con dispositivos conectados, empieza por nuestra guía de IA e IoT.

Empieza con ConnectMQ

Los resultados de tu Edge TPU merecen un hogar fiable. ConnectMQ te ofrece un broker MQTT en la nube con TLS, WebSockets y un explorador de topics en directo para ver las clasificaciones a medida que llegan. Crea un broker Serverless gratuito en pocos minutos, sin tarjeta de crédito, y apunta a él tu Raspberry Pi.

Preguntas frecuentes

¿Funciona el Coral Edge TPU con la Raspberry Pi 5?

Sí, el Coral USB Accelerator se conecta por USB y funciona en una Raspberry Pi 5 con un sistema operativo de 64 bits. El principal obstáculo es la versión de Python, por eso este tutorial usa un entorno de Python 3.9 independiente.

¿Puedo ejecutar cualquier modelo de TensorFlow en el Edge TPU?

No. El modelo debe ser un modelo TensorFlow Lite totalmente cuantizado a enteros y compilado con el compilador de Edge TPU. Las operaciones no compatibles se ejecutan en la CPU y reducen la ventaja de velocidad.

¿Debo enviar imágenes de la cámara por MQTT?

Normalmente no. Publica los resultados de la inferencia y envía solo instantáneas ocasionales si hace falta; las imágenes superan enseguida tamaños de mensaje razonables y plantean problemas de privacidad.

¿Qué nivel de QoS debo usar para los resultados de clasificación?

QoS 1 es una buena opción por defecto porque garantiza la entrega al menos una vez sin la sobrecarga de QoS 2. Incluye una marca de tiempo para que los suscriptores puedan ignorar los duplicados.