Google Coral Edge TPU com MQTT: Classifique Imagens em um Raspberry Pi e Publique os Resultados

O Google Coral Edge TPU permite que uma placa pequena como o Raspberry Pi execute inferência de redes neurais muito mais rápido do que apenas com a CPU, o que torna viável a classificação de imagens em tempo real na borda. Neste tutorial você vai instalar o runtime do Coral Edge TPU, executar um classificador MobileNet quantizado e publicar cada previsão como JSON estruturado via MQTT, para que dashboards, ferramentas de alerta e outros serviços possam reagir ao que a câmera vê.

O que você precisa

A configuração é propositalmente simples, para que você possa se concentrar no caminho dos dados do acelerador até o broker.

Conecte o acelerador a uma porta USB 3 azul. Uma porta USB 2 funciona, mas deixa visivelmente mais lentas as transferências entre o Pi e o acelerador, o que esconde boa parte do benefício.

Como o Coral Edge TPU se encaixa em um pipeline IoT

O Edge TPU é um ASIC que só executa modelos TensorFlow Lite totalmente quantizados em inteiros e compilados com o Edge TPU compiler. As operações que o compilador não consegue mapear rodam na CPU, então um modelo bem compilado mantém quase tudo no acelerador.

Em uma arquitetura IoT, o acelerador é uma etapa de um ciclo:

1. Um frame da câmera ou uma imagem chega ao Pi.
2. O Edge TPU a classifica em milissegundos.
3. O Pi publica um resultado JSON compacto (rótulos, scores, latência, ID do dispositivo) em um topic MQTT.
4. Assinantes como Node-RED, um gravador de banco de dados ou um script de alertas consomem o resultado.

Apenas o resultado da inferência sai do dispositivo, nunca a imagem bruta. Isso mantém baixo o consumo de banda, evita enviar fotos de pessoas ou instalações para a nuvem (o que também simplifica a conformidade com a LGPD quando há pessoas nas imagens, por exemplo numa linha de inspeção de um frigorífico) e fica bem dentro dos limites de mensagem do MQTT. Se você ainda está decidindo onde a inferência deve rodar, nossa comparação entre edge AI e IA na nuvem aborda os trade-offs em detalhes.

Passo 1: instale o runtime do Edge TPU

Adicione o repositório de pacotes do Coral e instale o runtime de frequência padrão:

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

Desconecte e reconecte o acelerador após a instalação para que as regras do udev sejam aplicadas. Verifique se o dispositivo está visível:

lsusb | grep -Ei "global unichip|google"

Antes da primeira inferência, o acelerador aparece como "Global Unichip Corp."; depois que o runtime carrega o firmware, ele é reenumerado como "Google Inc.". As duas situações são normais.

Existe também um pacote libedgetpu1-max que roda o chip em um clock mais alto. Ele é mais rápido, mas o acelerador esquenta; então comece com o runtime padrão e só troque se tiver medido que precisa e houver circulação de ar.

Passo 2: crie um ambiente Python compatível

O problema mais comum com o Coral é a incompatibilidade de versões entre o Python, o runtime do TensorFlow Lite e a biblioteca do Edge TPU. Os pacotes Python oficiais do Coral visam o Python 3.6 a 3.9, enquanto o Raspberry Pi OS atual traz um Python mais novo. O caminho confiável é um ambiente Python 3.9 dedicado, por exemplo com 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
# adicione ao ~/.bashrc as linhas do pyenv que ele imprimir e abra um novo 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 o pycoral a partir do índice de pacotes do Coral também traz a build correspondente do tflite_runtime, que é o que o nosso script importa. Fixar o NumPy abaixo da versão 2 evita erros de incompatibilidade binária com essas wheels mais antigas.

Baixe um modelo compilado e seus rótulos do repositório de dados de teste do 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

Passo 3: classifique imagens e publique no MQTT

O script abaixo carrega o modelo com o delegate do Edge TPU, classifica cada imagem de uma pasta e publica as três principais previsões. Ele usa paho-mqtt 2.x com a API de callbacks versão 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"
# Substitua 7f3a9c por uma string aleatória sua para que os seus topics sejam ú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)  # fica bem abaixo do limite de taxa do broker público
finally:
    client.publish(STATUS_TOPIC, "offline", qos=1, retain=True).wait_for_publish()
    client.loop_stop()
    client.disconnect()

Execute-o com a imagem do papagaio que você baixou:

python classify_mqtt.py ~/coral-mqtt

A primeira inferência é mais lenta porque o modelo é carregado no acelerador; as seguintes mostram a latência em regime.

Passo 4: acompanhe a chegada dos resultados do Coral Edge TPU

Abra um segundo terminal (em qualquer máquina) e assine com os clientes do Mosquitto:

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

Você deverá ver o status online retido seguido de uma mensagem JSON por imagem, por exemplo com o rótulo principal "macaw" para a foto de teste do papagaio. Como o broker.connectmq.com é um servidor MQTT público (gratuito), pensado para testes e protótipos, qualquer pessoa que adivinhe o seu topic pode lê-lo, então envie para lá apenas dados de demonstração. O servidor MQTT público permite até 64 KB por mensagem e 10 mensagens por segundo por cliente, o que é mais do que suficiente para resultados de classificação, mas é um motivo para não transmitir frames brutos.

Do teste à produção

Quando o protótipo funcionar, passe a usar conexões autenticadas e criptografadas. Crie um broker no portal do ConnectMQ, copie as credenciais geradas e altere o bloco de conexão:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # usa o repositório de CAs do sistema; o broker tem um certificado Let's Encrypt
client.will_set(STATUS_TOPIC, "offline", qos=1, retain=True)
client.connect("YOUR_BROKER_HOST", 8883, keepalive=60)

Um broker MQTT gerenciado elimina o trabalho de aplicar patches, gerenciar certificados e planejar capacidade, então o Pi só precisa se preocupar com a inferência. No ConnectMQ, o client ID fica vinculado às credenciais emitidas, o que impede que outro dispositivo se passe pelo seu nó de câmera.

Dicas de projeto para soluções com Edge TPU

Algumas decisões fazem a diferença entre uma demonstração e um sistema que você pode deixar rodando:

1. Publique eventos, não todos os frames. Para uma câmera ao vivo, publique apenas quando o rótulo principal mudar ou cruzar um limiar de confiança, além de um heartbeat periódico. Isso mantém o tráfego previsível.
2. Inclua contexto no payload. ID do dispositivo, nome e versão do modelo, latência e timestamp tornam os resultados depuráveis meses depois.
3. Use status retido com Last Will. O topic online/offline retido permite que os dashboards mostrem quais aceleradores estão ativos.
4. Estruture os topics de forma hierárquica. Algo como site/area/line/camera/classification escala melhor do que um namespace plano e se mapeia de forma limpa aos níveis ISA-95.
5. Planeje as atualizações de modelo. Compilar um modelo novo é fácil; distribuí-lo com segurança para muitos dispositivos não é. Nosso guia sobre como implantar modelos de ML em dispositivos de borda mostra como disparar e acompanhar atualizações via MQTT.

Solução de problemas

A documentação oficial do Coral lista as plataformas suportadas e os requisitos do Edge TPU compiler, caso você queira compilar seus próprios modelos.

Para onde ir a seguir

A classificação de imagens é a carga de trabalho mais simples; o mesmo padrão de publicação funciona para detecção de objetos e estimativa de pose. Se você precisa de bounding boxes em vez de rótulos, veja detecção de objetos com YOLO e MQTT. Para comparar o acelerador Coral com alternativas como módulos Jetson e microcontroladores, leia a visão geral de hardware para edge AI e, para o panorama de como combinar modelos com dispositivos conectados, comece pelo nosso guia de IA e IoT.

Comece a usar o ConnectMQ

Os resultados do seu Edge TPU merecem um destino confiável. O ConnectMQ oferece um broker MQTT gerenciado na nuvem com TLS, WebSockets e um explorador de topics em tempo real para acompanhar as classificações à medida que chegam. Crie um broker Serverless gratuito em poucos minutos, sem cartão de crédito, e aponte o seu Raspberry Pi para ele.

Perguntas frequentes

O Coral Edge TPU funciona com o Raspberry Pi 5?

Sim, o Coral USB Accelerator se conecta via USB e funciona em um Raspberry Pi 5 com sistema operacional de 64 bits. O principal obstáculo é a versão do Python, e é por isso que este tutorial usa um ambiente Python 3.9 separado.

Posso rodar qualquer modelo TensorFlow no Edge TPU?

Não. O modelo precisa ser um modelo TensorFlow Lite totalmente quantizado em inteiros e compilado com o Edge TPU compiler. Operações não suportadas voltam para a CPU e reduzem a vantagem de velocidade.

Devo enviar imagens da câmera via MQTT?

Geralmente não. Publique os resultados da inferência e envie apenas snapshots ocasionais, se necessário; as imagens rapidamente excedem tamanhos de mensagem razoáveis e levantam questões de privacidade.

Qual nível de QoS devo usar para os resultados de classificação?

QoS 1 é um bom padrão, porque garante a entrega pelo menos uma vez sem o overhead do QoS 2. Inclua um timestamp para que os assinantes possam ignorar duplicatas.