Detecção de Objetos com YOLO via MQTT: Publique Detecções em JSON

A detecção de objetos com YOLO via MQTT transforma uma câmera em um sensor: em vez de transmitir vídeo, você publica pequenas mensagens JSON que descrevem o que a câmera vê. Neste tutorial, você vai rodar um modelo YOLO sobre uma webcam ou um stream RTSP com Python, publicar as detecções de cada frame, manter contagens de objetos e gerar alertas que qualquer cliente MQTT pode consumir.

Por que enviar detecções em vez de vídeo

Vídeo é pesado, difícil de pesquisar e precisa de uma pessoa para assistir. Detecções são leves, estruturadas e legíveis por máquinas. Uma única mensagem como "3 pessoas, 1 empilhadeira na zona A" tem algumas centenas de bytes, enquanto o frame de onde ela veio tem dezenas ou centenas de kilobytes.

Publicar detecções via MQTT também desacopla a visão de todo o resto:

• Dashboards assinam as contagens e mostram a ocupação em tempo real.
• Um serviço de automação assina os alertas e aciona uma luz de advertência.
• Um historiador armazena as contagens por minuto para relatórios.
• Outro modelo, talvez em outra máquina, combina a visão com outros sensores.

Nenhum desses consumidores precisa saber qual câmera, modelo ou GPU produziu o resultado. É o mesmo acoplamento fraco que torna o MQTT popular para sensores, agora aplicado à visão computacional em dispositivos de borda.

Arquitetura

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)

Pré-requisitos

• Python 3.9 ou mais recente, uma webcam ou uma câmera RTSP
• Uma CPU basta para modelos pequenos a alguns frames por segundo; uma GPU ou uma Jetson permite tempo real
• Pacotes:

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

A primeira execução baixa os pesos pré-treinados yolov8n.pt (o modelo nano treinado nas 80 classes do COCO). Consulte a documentação da Ultralytics para outros tamanhos de modelo e versões mais novas; a API usada abaixo funciona em todos eles.

O publicador de detecções

Salve como yolo_mqtt.py. Ele abre a câmera, roda o YOLO, converte os resultados em JSON, publica detecções no máximo duas vezes por segundo, atualiza as contagens retidas quando elas mudam e gera um alerta quando há pessoas em uma zona restrita.

import json
import time
import uuid

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

SOURCE = 0  # webcam index, or "rtsp://user:pass@192.168.1.50:554/stream1"
PREFIX = "connectmq/demo/yolo-c83f/site1/dock/cam1"
CLASSES = {"person", "car", "truck", "forklift"}  # forklift only with a custom model
MIN_CONF = 0.5
PUBLISH_EVERY = 0.5  # seconds
ZONE = (0.6, 0.0, 1.0, 1.0)  # restricted zone: right 40% of the frame (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)  # reconnect for RTSP drops
            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()  # normalized 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()

Acompanhe a saída em outro terminal:

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

Entre no lado direito do frame e você verá um alerta raised, seguido de cleared quando sair.

Projetando o payload de detecção

Uma mensagem de detecção do script tem esta aparência:

{
  "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]}
  ]
}

Diretrizes que mantêm úteis os payloads de detecção de objetos com YOLO via MQTT:

Contagem e regras de alerta

Contar objetos por frame é fácil; contar objetos que passam é mais difícil, porque a mesma pessoa aparece em muitos frames. Para contagem de fluxo, ative o rastreamento no Ultralytics (model.track(frame, persist=True)), que atribui um ID a cada objeto, e conte cada ID uma única vez quando o seu centro cruzar uma linha. Publique o cruzamento como um evento, e não como uma contagem, e deixe um serviço a jusante agregar por minuto.

As regras de alerta devem ficar perto da câmera quando a latência importa (uma pessoa entrando em uma célula robotizada) e a jusante quando são lógica de negócio (mais de cinco caminhões esperando na doca de um centro de distribuição por dez minutos). Como os resultados estão no MQTT, você pode mover uma regra de um lugar para o outro sem mudar o código da câmera.

Consumindo as detecções a jusante

O valor de publicar os resultados de detecção de objetos com YOLO via MQTT aparece do lado de quem consome. Alguns padrões que as equipes constroem nas primeiras semanas:

• Dashboard de ocupação. Assine o topic retido counts de todas as câmeras com um wildcard como site1/+/+/counts. Cada nova sessão de dashboard recebe imediatamente a contagem atual de todas as câmeras e, em seguida, as atualizações ao vivo.
• Estatísticas por minuto. Um pequeno serviço em Python ou Node-RED agrega as contagens por minuto e as grava em um banco de dados de séries temporais. Relatórios de horários de pico ou de utilização da doca saem dali, não do vídeo.
• Intertravamentos e avisos. Um serviço de automação assina o topic de alertas e aciona uma luz de advertência ou envia uma mensagem a um supervisor. Mantenha os intertravamentos críticos de segurança em hardware certificado; os alertas da câmera são um complemento, não um substituto.
• Verificação cruzada com outros sensores. Um alerta de pessoa combinado com uma barreira de luz ou um contato de porta é muito mais confiável do que qualquer um deles sozinho.

Como cada um deles é apenas mais um assinante, adicionar um novo nunca deixa a câmera mais lenta. O broker distribui a mesma mensagem para todos.

Privacidade e minimização de dados

Câmeras que veem pessoas trazem responsabilidades. Publicar apenas rótulos, contagens e caixas normalizadas, nunca rostos ou frames, é a medida de privacidade mais forte que você pode adotar, e ela se encaixa naturalmente nessa arquitetura. Se você precisar de snapshots para verificação, guarde-os localmente com um período de retenção curto ou envie-os a um topic privado que só serviços autorizados possam ler. Documente o que a câmera detecta e por quê, e verifique a legislação local antes de implantar em ambientes de trabalho ou áreas públicas; no Brasil, imagens de pessoas são dados pessoais sob a LGPD.

Solução de problemas

1. Taxa de frames muito baixa na CPU. Use o modelo nano, reduza o tamanho de entrada com imgsz=416 ou processe um frame sim, um não.
2. O stream RTSP congela. O OpenCV armazena frames em buffer; leia continuamente em uma thread separada e processe apenas o frame mais recente.
3. Alertas oscilando. Exija uma detecção em vários frames consecutivos antes de disparar e vários frames vazios antes de encerrar.
4. Mensagens rejeitadas ou descartadas. No servidor MQTT público, fique abaixo de 10 mensagens por segundo por cliente e de 64 KB por mensagem; o limite de taxa do script cuida disso.
5. Classes erradas. O modelo pré-treinado conhece apenas as classes do COCO. Para empilhadeiras, capacetes ou produtos específicos, treine um modelo personalizado com as suas próprias imagens.

Notas para produção

Para câmeras reais, publique em um broker privado via TLS (porta 8883) com credenciais e nunca envie frames de pessoas para um topic público. Um broker MQTT gerenciado como o ConnectMQ cuida de autenticação, certificados e monitoramento, e o seu explorador de topics ao vivo é uma forma rápida de confirmar que as detecções estão fluindo. Use o servidor MQTT público (gratuito) apenas para desenvolvimento com cenas de teste, nunca para dados de produção.

Se a sua câmera roda em uma placa NVIDIA, veja NVIDIA Jetson com MQTT para inferência acelerada por GPU e, para uma variante de baixo custo em um microcontrolador, veja detecção de pessoas com ESP32-CAM. O guia de IA e IoT explica como a visão se encaixa com outras cargas de trabalho de edge AI.

Comece a usar o ConnectMQ

Crie um broker ConnectMQ privado no plano Serverless gratuito em app.connectmq.com, sem cartão de crédito, adicione client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD") e client.tls_set(), mude para a porta 8883 e as detecções da sua câmera estarão fluindo com segurança.

Perguntas frequentes

Posso enviar frames da câmera via MQTT junto com as detecções?

Você pode enviar snapshots JPEG em um topic separado, dentro do limite de tamanho de mensagem do broker, mas mantenha o vídeo contínuo em um protocolo de streaming. As detecções devem continuar pequenas e sem frames.

Qual tamanho de modelo YOLO devo usar?

Comece com o modelo nano para CPUs e placas pequenas e passe para o small ou o medium somente se a precisão não for suficiente. Modelos maiores precisam de uma GPU para taxas em tempo real.

Como contar pessoas em vez de detectá-las frame a frame?

Use o rastreador para atribuir IDs e conte cada ID uma única vez quando ele cruzar uma linha ou entrar em uma zona. Publique cada cruzamento como um evento e agregue a jusante.

O broker roda o modelo YOLO?

Não. O modelo roda no computador da câmera, e o broker apenas transporta as mensagens resultantes para quem as assinar.