Bonus: cámaras IP por RTSP
Las cámaras de vigilancia de verdad, las que cuelgan de una pared y llevan años ahí, casi siempre hablan RTSP. Si vas a llevar este pipeline a un edificio real, este es el protocolo con el que te vas a encontrar.
Y trae una sorpresa desagradable que conviene conocer antes de pelearse con ella.
Por qué RTSP no es como MJPEG
Compara lo que hiciste en el paso 9 con lo que exige RTSP:
MJPEG sobre HTTP manda imágenes JPEG completas, una detrás de otra. Cada frame llega listo para usar y no depende de los anteriores. Por eso pudiste leerlo con aiohttp y un par de readexactly.
RTSP transporta video comprimido de verdad, normalmente H.264. Ahí los frames no son independientes: la mayoría solo guarda las diferencias respecto a otro frame anterior. Para obtener una imagen usable hay que decodificar el flujo, y decodificar H.264 es cómputo pesado, del que ocupa la CPU sin ningún punto donde ceder.
Es decir: RTSP no es un problema de red, es un problema de la trampa del paso 6.
Analogía · Dos formas de mandar un álbum de fotos
MJPEG es mandar cada foto entera, una por una. Ocupa mucho, pero abres cualquiera y ya está.
H.264 es mandar la primera foto entera y después solo las notas de "en esta cambió la esquina de arriba, en la siguiente el señor se movió dos pasos". Ocupa muchísimo menos, y a cambio no puedes ver la foto veinte sin haber reconstruido antes las diecinueve anteriores.
Esa reconstrucción es la decodificación, y alguien tiene que pagarla.
La forma correcta de conectarla
La estructura es la misma del bonus anterior: la parte bloqueante va a un hilo, en el borde, envuelta en la misma interfaz que todas tus fuentes.
Crea un archivo rtsp_source.py:
import asyncio
import cv2
RECONNECT_MAX = 30.0
class RtspSource:
def __init__(self, cam_id, url, jpeg_quality=80):
self.cam_id = cam_id
self.url = url
self.jpeg_quality = jpeg_quality
def _open(self):
cap = cv2.VideoCapture(self.url, cv2.CAP_FFMPEG)
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
return cap if cap.isOpened() else None
def _grab(self, cap):
ok, frame = cap.read()
if not ok:
return None
ok, buf = cv2.imencode(
".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, self.jpeg_quality])
return buf.tobytes() if ok else None
async def frames(self):
backoff = 1.0
while True:
cap = await asyncio.to_thread(self._open)
if cap is None:
print(f"[{self.cam_id}] cannot open, retry in {backoff:.0f}s")
await asyncio.sleep(backoff)
backoff = min(backoff * 2, RECONNECT_MAX)
continue
backoff = 1.0
try:
while True:
jpeg = await asyncio.to_thread(self._grab, cap)
if jpeg is None:
break
yield jpeg
finally:
await asyncio.to_thread(cap.release)
print(f"[{self.cam_id}] stream ended, reconnecting")
Repasa las decisiones, porque cada una viene de algo que ya aprendiste:
cv2.CAP_PROP_BUFFERSIZE, 1 le pide a OpenCV que guarde un solo frame de reserva. Sin esto, la librería acumula un buffer interno y acabas analizando imágenes de hace varios segundos: exactamente el problema del paso 10, pero escondido dentro de una librería donde tu maxsize no llega.
El bucle exterior con backoff es el patrón del paso 14. Una cámara IP se cae, se reinicia, cambia de IP. Aquí no hay excepción que atrapar: cap.read() simplemente empieza a devolver False, así que salir del bucle interior es la señal de "se acabó el stream" y toca reconectar.
El finally con release libera el descriptor pase lo que pase, incluida una cancelación durante el apagado del paso 15.
La interfaz es idéntica a MjpegSource y a la webcam: un generador asíncrono que entrega JPEG. Por eso encaja en el pipeline sin tocar nada más.
Conectarla al pipeline
En main.py, donde creas las tareas de ingesta, agrega tu cámara IP junto a las demás:
RTSP_URL = "rtsp://user:[email protected]:554/stream1"
ingest = [
asyncio.create_task(ingest_source(MjpegSource(cam, session), infer_q, stats))
for cam in cameras]
ingest.append(asyncio.create_task(
ingest_source(RtspSource("ipcam0", RTSP_URL), infer_q, stats)))
Esa es toda la integración: una tarea más apuntando a la misma cola. La cámara IP compite por los mismos trabajadores de inferencia, respeta la misma política de descarte y aparece en el mismo dashboard.
La URL exacta depende del fabricante. Los patrones más habituales son rtsp://usuario:clave@ip:554/stream1, /h264Preview_01_main en Reolink, o /cam/realmonitor?channel=1&subtype=0 en Dahua. Búscala en el manual de tu modelo o en el panel web de la cámara.
Ojo
Casi todas las cámaras IP publican dos streams: uno principal en alta resolución y otro secundario, mucho más pequeño. Para detección, el secundario suele ser suficiente y decodificarlo cuesta una fracción. Si vas a conectar varias cámaras, empieza siempre por el stream secundario: es la optimización más barata de todo este capítulo.
Checkpoint · Si tienes una cámara a mano
Con la cámara conectada a tu red, arranca el pipeline y busca ipcam0 en el panel de eventos del dashboard, junto a las cámaras de archivo.
Y ahora haz la prueba interesante: desconecta la cámara de la corriente o del cable de red. En la terminal vas a ver el mensaje de reconexión con su backoff creciendo, mientras las demás fuentes siguen entregando frames sin inmutarse. Vuelve a conectarla y al cabo de unos segundos reaparece sola.
Eso es el paso 14 funcionando sobre hardware de verdad.
Cuándo esto deja de escalar
Un hilo por cámara IP funciona bien hasta unas pocas fuentes. Pero fíjate en dónde te deja: cada cámara consume un hilo decodificando H.264 de forma continua, y eso es CPU real, no espera.
Con cuatro cámaras, perfecto. Con cuarenta, tu máquina se queda sin CPU decodificando y ninguna cantidad de asyncio lo va a arreglar, porque el problema ya no es la concurrencia.
Ese techo es justamente el tema del último bonus.