Alertas MQTT de Movimento com Câmera Raspberry Pi e Picamera2

Um detector de movimento MQTT com câmera Raspberry Pi transforma um camera module em uma fonte de eventos: em vez de transmitir vídeo o dia inteiro, o Pi observa um feed de baixa resolução, salva um snapshot quando algo muda e publica um pequeno evento JSON que qualquer coisa pode assinar. Neste tutorial você vai montar esse detector com Picamera2 e Python, ajustá-lo para evitar alarmes falsos e entregar os eventos por meio de um broker MQTT.

Como o detector funciona

O Picamera2 consegue rodar dois streams da câmera ao mesmo tempo: um stream main em resolução total e um minúsculo stream lores. O detector compara frames lores consecutivos em tons de cinza, o que custa pouquíssima CPU mesmo em um Pi Zero 2 W, e só usa o stream main para salvar um JPEG quando encontra movimento.

O fluxo é:

1. Pegar o plano de luminância do frame lores de 320x240.
2. Calcular a diferença quadrática média em relação ao frame anterior.
3. Se a diferença ultrapassar um limite, iniciar um evento de movimento, salvar um snapshot em tamanho total e publicar uma mensagem MQTT com o link dele.
4. Ignorar novos disparos durante um período de cooldown.
5. Após um período sem movimento, publicar um evento de "fim de movimento" e voltar o estado retido para idle.

Enviar imagens pelo MQTT é possível, mas um broker é um péssimo servidor de arquivos: payloads grandes são lentos, e o servidor público de testes limita as mensagens a 64 KB. O padrão robusto é publicar um evento com um link e deixar a imagem em um servidor web ou object storage que você controla.

Hardware e software

O Picamera2 e o NumPy vêm dos pacotes do sistema operacional, então crie o ambiente virtual com acesso a eles:

sudo apt update
sudo apt install -y python3-picamera2 python3-numpy python3-venv
rpicam-hello --list-cameras
mkdir -p ~/motion-mqtt/snapshots && cd ~/motion-mqtt
python3 -m venv --system-site-packages .venv
source .venv/bin/activate
pip install "paho-mqtt>=2.0"

Se rpicam-hello --list-cameras não mostrar nenhuma câmera, desligue, reencaixe o cabo flat com os contatos virados para o lado certo e tente de novo. O manual do Picamera2 documenta a configuração de streams usada abaixo.

Topics de eventos

Um evento de início típico fica assim:

{
  "event": "motion_start",
  "score": 18.4,
  "snapshot": "http://192.168.1.50:8000/20261016-142233.jpg",
  "ts": 1792160553
}

O script de detecção de movimento MQTT com câmera Raspberry Pi

Salve como motion_mqtt.py:

#!/usr/bin/env python3
"""Detect motion with Picamera2 and publish MQTT events with snapshot links."""
import json
import os
import socket
import time
import uuid
from datetime import datetime
from pathlib import Path

import numpy as np
import paho.mqtt.client as mqtt
from picamera2 import Picamera2

BROKER = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "1883"))
USERNAME = os.getenv("MQTT_USERNAME")
PASSWORD = os.getenv("MQTT_PASSWORD")
CLIENT_ID = os.getenv("MQTT_CLIENT_ID", f"cam-{uuid.uuid4().hex[:8]}")
PREFIX = os.getenv("MQTT_PREFIX", "connectmq/demo/e71b04/frontdoor")

THRESHOLD = float(os.getenv("MOTION_THRESHOLD", "12"))
COOLDOWN = 10       # segundos entre eventos de início
QUIET_TIME = 15     # segundos sem movimento antes de "motion_end"
SNAP_DIR = Path(os.getenv("SNAP_DIR", str(Path.home() / "motion-mqtt" / "snapshots")))
HTTP_PORT = int(os.getenv("SNAP_HTTP_PORT", "8000"))
LORES = (320, 240)


def local_ip():
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
        s.connect(("8.8.8.8", 80))
        return s.getsockname()[0]


BASE_URL = os.getenv("SNAP_BASE_URL", f"http://{local_ip()}:{HTTP_PORT}")


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"Connection refused: {reason_code}")
        return
    print(f"Connected as {CLIENT_ID}")
    client.publish(f"{PREFIX}/status", "online", qos=1, retain=True)
    client.publish(f"{PREFIX}/motion/state", "idle", qos=1, retain=True)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
client.on_connect = on_connect
client.will_set(f"{PREFIX}/status", "offline", qos=1, retain=True)
client.reconnect_delay_set(min_delay=1, max_delay=60)
if USERNAME:
    client.username_pw_set(USERNAME, PASSWORD)
if PORT == 8883:
    client.tls_set()
client.connect(BROKER, PORT, keepalive=30)
client.loop_start()

picam2 = Picamera2()
config = picam2.create_video_configuration(
    main={"size": (1280, 720), "format": "RGB888"},
    lores={"size": LORES, "format": "YUV420"},
)
picam2.configure(config)
picam2.start()
time.sleep(2)  # deixa a exposição e o balanço de branco se estabilizarem

SNAP_DIR.mkdir(parents=True, exist_ok=True)
w, h = LORES
previous = None
active = False
last_start = 0.0
last_motion = 0.0


def publish_event(event, score, snapshot=None):
    payload = {"event": event, "score": round(score, 1), "ts": int(time.time())}
    if snapshot:
        payload["snapshot"] = snapshot
    client.publish(f"{PREFIX}/motion", json.dumps(payload), qos=1)
    print(payload)


try:
    while True:
        frame = picam2.capture_buffer("lores")
        current = frame[: w * h].reshape(h, w).astype(np.int16)
        if previous is not None:
            score = float(np.square(current - previous).mean())
            now = time.time()
            if score > THRESHOLD:
                last_motion = now
                if not active and now - last_start > COOLDOWN:
                    name = datetime.now().strftime("%Y%m%d-%H%M%S") + ".jpg"
                    picam2.capture_file(str(SNAP_DIR / name))
                    publish_event("motion_start", score, f"{BASE_URL}/{name}")
                    client.publish(f"{PREFIX}/motion/state", "active", qos=1, retain=True)
                    active = True
                    last_start = now
            elif active and now - last_motion > QUIET_TIME:
                publish_event("motion_end", score)
                client.publish(f"{PREFIX}/motion/state", "idle", qos=1, retain=True)
                active = False
        previous = current
        time.sleep(0.1)
except KeyboardInterrupt:
    pass
finally:
    picam2.stop()
    client.publish(f"{PREFIX}/status", "offline", qos=1, retain=True).wait_for_publish(2)
    client.loop_stop()
    client.disconnect()

A conversão astype(np.int16) é importante: subtrair dois arrays de 8 bits sem sinal dá a volta (overflow) e produz diferenças falsas enormes. Pegar apenas os primeiros w * h bytes do buffer YUV420 dá o plano de luminância em tons de cinza, que é tudo de que a detecção de movimento precisa.

Servindo os snapshots

Para uma configuração local rápida, sirva a pasta de snapshots com o servidor web embutido do Python em um segundo terminal:

python3 -m http.server 8000 --directory ~/motion-mqtt/snapshots

Os links de cada evento passam a abrir em qualquer dispositivo da sua rede. Esse servidor não tem autenticação, então mantenha-o apenas na sua LAN. Para acesso externo, envie os snapshots a partir do seu próprio script para um armazenamento que você controla e defina SNAP_BASE_URL para esse local; o formato do evento MQTT não muda. Adicione também um cron job ou timer do systemd para apagar imagens antigas, ou o cartão SD vai encher.

Testando e ajustando

Acompanhe os eventos de qualquer máquina:

mosquitto_sub -h broker.connectmq.com -p 1883 -i viewer-e71b04 \
  -t 'connectmq/demo/e71b04/frontdoor/#' -v

Passe a mão na frente da lente e você deve ver motion_start com um link de snapshot e depois motion_end após 15 segundos sem movimento. O ajuste é principalmente sobre o limite:

Imprima o score por alguns minutos sem nada se movendo para ver o piso de ruído da sua cena e depois defina o limite com folga acima dele.

Detecção por câmera vs. sensor PIR

Um sensor infravermelho passivo (PIR) é a outra forma comum de detectar movimento, e vale saber quando cada um se encaixa. Um sensor PIR reage ao calor do corpo, consome quase nada de energia e ignora mudanças de luz, então gera menos alarmes falsos por sombras e nuvens. Mas ele não diz o que se moveu e não tem imagem para anexar. A diferença de frames com a câmera reage a qualquer mudança visual, incluindo uma cortina ao vento, mas dá a você um snapshot e permite restringir a detecção a uma região precisa da imagem.

Os dois se combinam bem. Ligue um sensor PIR a um pino GPIO livre e só aceite um disparo da câmera quando a linha do PIR também estiver em nível alto dentro do mesmo segundo ou dois. A câmera então confirma e documenta o que o PIR detectou, e o evento MQTT pode incluir os dois sinais para que os subscribers decidam quanto confiar nele. Mantenha essa lógica de confirmação no Pi para que o evento publicado continue simples: um motion_start, um motion_end, um link.

Reagindo aos eventos de movimento

Como o detector só publica eventos, a reação fica em outro lugar e pode mudar sem mexer no código da câmera. Um fluxo Node-RED pode enviar uma notificação push; um pequeno subscriber Python pode enviar o snapshot por e-mail, como mostrado em alertas MQTT por e-mail com Python; e um segundo Pi pode acender um refletor usando o controlador de relé com Raspberry Pi via MQTT, assinando o topic motion/state. O Home Assistant também pode consumir o estado retido como um sensor binário. Se você quer um nó de câmera mais barato para uma segunda porta, o guia de ESP32-CAM via MQTT usa a mesma ideia de eventos em um microcontrolador.

Privacidade e configuração de produção

Eventos de movimento revelam quando há pessoas em casa, então nunca devem trafegar por um servidor público em uma instalação real. O servidor MQTT público de testes em broker.connectmq.com pode ser lido por todos e serve apenas para o primeiro teste. Imagens e horários de presença podem ser dados pessoais sob a LGPD, mais um motivo para mantê-los privados. Para uma câmera real, use um broker privado do ConnectMQ com TLS e credenciais geradas:

export MQTT_PORT=8883
export MQTT_USERNAME=YOUR_USERNAME
export MQTT_PASSWORD=YOUR_PASSWORD
export MQTT_CLIENT_ID=YOUR_CLIENT_ID
export MQTT_PREFIX=home/frontdoor
python motion_mqtt.py

Com um broker MQTT gerenciado, seu celular, um servidor doméstico e a câmera conectam todos de saída ao mesmo endpoint TLS, então você não precisa de redirecionamento de portas. Rode o detector e o servidor de snapshots como serviços do systemd para que iniciem após uma queda de energia; o tutorial de MQTT com Raspberry Pi inclui um modelo de serviço pronto.

Comece com o ConnectMQ

Dê à sua câmera um lugar seguro para enviar os alertas. Crie um broker Serverless gratuito no ConnectMQ, sem cartão de crédito, e conecte via TLS na porta 8883. O explorador de topics ao vivo do portal mostra cada evento de movimento e estado retido no momento em que acontecem, o que facilita muito o ajuste dos limites.

Perguntas frequentes

Posso enviar a própria imagem do snapshot via MQTT?

Você pode publicar os bytes do JPEG como payload, mas é ineficiente e atinge rapidamente os limites de tamanho de mensagem. Publicar um link para a imagem e armazenar o arquivo em outro lugar escala muito melhor.

Quanta CPU a detecção de movimento com Picamera2 usa?

Muito pouca, porque a comparação roda sobre um stream de 320x240 em tons de cinza. Até um Raspberry Pi Zero 2 W consegue rodá-la junto com o cliente MQTT.

Por que recebo eventos de movimento quando nada está se mexendo?

Mudanças de autoexposição, luzes cintilando e sombras em movimento alteram os valores dos pixels. Aumente o limite, recorte uma região de interesse ou trave a exposição com os controles do Picamera2.

Isso funciona com o Raspberry Pi Camera Module 3?

Sim. O Picamera2 suporta o Camera Module 2, o Camera Module 3 e a câmera HQ; o código não muda.