Alertas MQTT de movimiento con la cámara de Raspberry Pi y Picamera2

Un detector de movimiento MQTT con cámara Raspberry Pi convierte un módulo de cámara en una fuente de eventos: en lugar de transmitir video todo el día, la Pi vigila una imagen de baja resolución, guarda una captura cuando algo cambia y publica un pequeño evento JSON al que cualquier sistema puede suscribirse. En este tutorial construirás ese detector con Picamera2 y Python, lo ajustarás para evitar falsas alarmas y entregarás los eventos a través de un broker MQTT.

Cómo funciona el detector

Picamera2 puede ejecutar dos flujos de la cámara a la vez: un flujo main a resolución completa y un diminuto flujo lores. El detector compara fotogramas lores consecutivos en escala de grises, lo que cuesta muy poca CPU incluso en una Pi Zero 2 W, y solo usa el flujo main para guardar un JPEG cuando detecta movimiento.

El flujo es:

1. Obtener el plano de luminancia del fotograma lores de 320x240.
2. Calcular la diferencia cuadrática media respecto al fotograma anterior.
3. Si la diferencia supera un umbral, iniciar un evento de movimiento, guardar una captura a tamaño completo y publicar un mensaje MQTT con su enlace.
4. Ignorar nuevos disparos durante un periodo de enfriamiento.
5. Tras un periodo de calma, publicar un evento de "fin de movimiento" y devolver el estado retenido a idle.

Enviar imágenes por MQTT es posible, pero un broker es un mal servidor de archivos: los payloads grandes son lentos y el servidor público de pruebas limita los mensajes a 64 KB. El patrón robusto es publicar un evento con un enlace y dejar que la imagen resida en un servidor web o un almacenamiento de objetos que tú controles.

Hardware y software

Picamera2 y NumPy vienen de los paquetes del sistema operativo, así que crea el entorno virtual con acceso a ellos:

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"

Si rpicam-hello --list-cameras no muestra ninguna cámara, apaga la Pi, vuelve a colocar el cable plano con los contactos en la orientación correcta y prueba de nuevo. El manual de Picamera2 documenta la configuración de flujos que se usa a continuación.

Topics de eventos

Un evento de inicio típico tiene este aspecto:

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

El script de detección de movimiento MQTT con cámara Raspberry Pi

Guárdalo 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 inicio
QUIET_TIME = 15     # segundos sin movimiento 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)  # dejar que se estabilicen la exposición y el balance de blancos

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()

La conversión astype(np.int16) es importante: restar dos arrays de 8 bits sin signo provoca desbordamiento y genera enormes diferencias falsas. Tomar solo los primeros w * h bytes del búfer YUV420 te da el plano de luminancia en escala de grises, que es todo lo que necesita la detección de movimiento.

Servir las capturas

Para una configuración local rápida, sirve la carpeta de capturas con el servidor web integrado de Python en una segunda terminal:

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

Los enlaces de cada evento se abren ahora en cualquier dispositivo de tu red. Este servidor no tiene autenticación, así que mantenlo solo en tu LAN. Para acceder desde fuera, sube las capturas desde tu propio script a un almacenamiento que controles y define SNAP_BASE_URL con esa ubicación; el formato del evento MQTT no cambia. Añade también un cron o un temporizador systemd que borre las imágenes antiguas, o la tarjeta SD se llenará.

Pruebas y ajuste

Observa los eventos desde cualquier máquina:

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

Mueve una mano delante del objetivo y deberías ver motion_start con un enlace a la captura y, tras 15 segundos de calma, motion_end. El ajuste consiste sobre todo en el umbral:

Imprime la puntuación durante unos minutos sin que nada se mueva para ver el nivel de ruido de tu escena y fija el umbral holgadamente por encima.

Detección con cámara frente a un sensor PIR

Un sensor de infrarrojos pasivo (PIR) es la otra forma habitual de detectar movimiento, y conviene saber cuándo encaja cada uno. Un sensor PIR reacciona al calor corporal, apenas consume energía e ignora los cambios de luz, así que produce menos falsas alarmas por sombras y nubes. Sin embargo, no puede decirte qué se movió ni aporta una imagen. La diferencia entre fotogramas con la cámara reacciona a cualquier cambio visual, incluida una cortina movida por el viento, pero te da una captura y permite restringir la detección a una región precisa de la imagen.

Ambos se combinan bien. Conecta un sensor PIR a un pin GPIO libre y acepta un disparo de la cámara solo cuando la línea del PIR también esté alta en el mismo segundo o dos. La cámara confirma y documenta entonces lo que detectó el PIR, y el evento MQTT puede incluir ambas señales para que los suscriptores decidan cuánto confiar en él. Mantén esa lógica de confirmación en la Pi para que el evento publicado siga siendo sencillo: un motion_start, un motion_end, un enlace.

Reaccionar a los eventos de movimiento

Como el detector solo publica eventos, la reacción vive en otro lugar y puede cambiar sin tocar el código de la cámara. Un flujo de Node-RED puede enviar una notificación push; un pequeño suscriptor de Python puede enviar la captura por correo, como se muestra en alertas por correo MQTT con Python; y una segunda Pi puede encender un reflector con el controlador de relé MQTT con Raspberry Pi suscribiéndose al topic motion/state. Home Assistant también puede consumir el estado retenido como sensor binario. Si quieres un nodo de cámara más económico para una segunda puerta, la guía de ESP32-CAM por MQTT aplica la misma idea de eventos en un microcontrolador.

Privacidad y configuración de producción

Los eventos de movimiento revelan cuándo hay gente en casa, así que nunca deben viajar por un broker público en una instalación real. El servidor MQTT público gratuito de broker.connectmq.com es legible por cualquiera y está pensado solo para la primera prueba. Para una cámara real, usa un broker privado de ConnectMQ con TLS y credenciales generadas:

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

Con un broker MQTT gestionado, tu teléfono, un servidor doméstico y la cámara se conectan de salida al mismo endpoint TLS, así que no necesitas redirección de puertos. Ejecuta el detector y el servidor de capturas como servicios systemd para que arranquen tras un corte de luz; el tutorial de MQTT con Raspberry Pi incluye una plantilla de servicio lista.

Empieza con ConnectMQ

Dale a tu cámara un lugar seguro al que enviar sus alertas. Crea un broker Serverless gratuito en ConnectMQ, sin tarjeta de crédito, y conéctate con TLS en el puerto 8883. El explorador de topics en vivo del portal muestra cada evento de movimiento y cada estado retenido en el momento en que ocurren, lo que facilita mucho el ajuste de los umbrales.

Preguntas frecuentes

¿Puedo enviar la propia imagen de la captura por MQTT?

Puedes publicar los bytes del JPEG como payload, pero es ineficiente y alcanza rápidamente los límites de tamaño de mensaje. Publicar un enlace a la imagen y guardar el archivo en otro lugar escala mucho mejor.

¿Cuánta CPU usa la detección de movimiento con Picamera2?

Muy poca, porque la comparación se hace sobre un flujo en escala de grises de 320x240. Incluso una Raspberry Pi Zero 2 W puede ejecutarla junto al cliente MQTT.

¿Por qué recibo eventos de movimiento cuando nada se mueve?

Los cambios de autoexposición, las luces que parpadean y las sombras en movimiento modifican los valores de los píxeles. Sube el umbral, recorta a una región de interés o bloquea la exposición con los controles de Picamera2.

¿Funciona con el Raspberry Pi Camera Module 3?

Sí. Picamera2 es compatible con Camera Module 2, Camera Module 3 y la cámara HQ; el código no cambia.