210 lines
6.4 KiB
Python
210 lines
6.4 KiB
Python
import os
|
|
import sys
|
|
import time
|
|
|
|
import cv2
|
|
import numpy as np
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
import config
|
|
from camara import CamaraRTSP
|
|
from detector_movimiento import DetectorMovimiento
|
|
from notificador import disparar_webhook_alexa, enviar_telegram
|
|
|
|
APROBADOS = 0
|
|
FALLIDOS = 0
|
|
|
|
|
|
def check(nombre, condicion, detalle=""):
|
|
global APROBADOS, FALLIDOS
|
|
estado = "OK " if condicion else "FALLO"
|
|
if condicion:
|
|
APROBADOS += 1
|
|
else:
|
|
FALLIDOS += 1
|
|
print(f"[{estado}] {nombre} {detalle}")
|
|
|
|
|
|
def frame_estatico():
|
|
img = np.zeros((480, 640, 3), dtype=np.uint8)
|
|
cv2.rectangle(img, (100, 100), (200, 200), (255, 255, 255), -1)
|
|
return img
|
|
|
|
|
|
def frame_movimiento():
|
|
img = frame_estatico()
|
|
cv2.circle(img, (50, 50), 20, (0, 255, 0), -1)
|
|
cv2.putText(img, "MOVIMIENTO", (10, 400),
|
|
cv2.FONT_HERSHEY_SIMPLEX, 1.5, (0, 0, 255), 4)
|
|
return img
|
|
|
|
|
|
def simular_video(frames, nombre="simulado.mp4"):
|
|
h, w = frames[0].shape[:2]
|
|
ruta = os.path.join(config.SNAPSHOT_DIR, nombre)
|
|
writer = cv2.VideoWriter(
|
|
ruta, cv2.VideoWriter_fourcc(*"mp4v"), 10, (w, h)
|
|
)
|
|
for f in frames:
|
|
writer.write(f)
|
|
writer.release()
|
|
return ruta
|
|
|
|
|
|
def test_rtsp_reconexion():
|
|
print("\n--- RF-01/RF-07: Conectividad RTSP + reconexion ---")
|
|
cam = CamaraRTSP("rtsp://127.0.0.1:1/stream1", "Fake")
|
|
t0 = time.time()
|
|
frame = cam.leer_frame()
|
|
check("RTSP invalido: leer_frame devuelve None sin excepcion",
|
|
frame is None, f"({time.time()-t0:.1f}s)")
|
|
|
|
|
|
def test_movimiento():
|
|
print("\n--- RF-02: Deteccion de movimiento ---")
|
|
det = DetectorMovimiento(umbral_pixeles=500)
|
|
est = frame_estatico()
|
|
det.hay_movimiento(est)
|
|
det.hay_movimiento(est)
|
|
det.hay_movimiento(est)
|
|
static_ok = not det.hay_movimiento(est)
|
|
check("Sin movimiento en frame estatico", static_ok)
|
|
|
|
det2 = DetectorMovimiento(umbral_pixeles=500)
|
|
det2.hay_movimiento(frame_estatico())
|
|
det2.hay_movimiento(frame_estatico())
|
|
det2.hay_movimiento(frame_estatico())
|
|
mov_ok = det2.hay_movimiento(frame_movimiento())
|
|
check("Movimiento detectado en frame con objeto", mov_ok)
|
|
|
|
|
|
def test_franja_horaria():
|
|
print("\n--- RF-06: Franja horaria ---")
|
|
from main import dentro_de_franja_horaria
|
|
|
|
orig_i, orig_f = config.HORA_INICIO, config.HORA_FIN
|
|
config.HORA_INICIO, config.HORA_FIN = 8, 17
|
|
check("08-17 activa a las 10", dentro_de_franja_horaria(10) is True)
|
|
check("08-17 inactiva a las 20", dentro_de_franja_horaria(20) is False)
|
|
config.HORA_INICIO, config.HORA_FIN = 22, 6
|
|
check("22-06 activa a las 2", dentro_de_franja_horaria(2) is True)
|
|
check("22-06 inactiva a las 10", dentro_de_franja_horaria(10) is False)
|
|
config.HORA_INICIO, config.HORA_FIN = orig_i, orig_f
|
|
|
|
|
|
def test_telegram_mock():
|
|
print("\n--- RF-04: Notificacion Telegram (mock, sin token real) ---")
|
|
capturas = {}
|
|
|
|
class FakeResponse:
|
|
ok = True
|
|
status_code = 200
|
|
text = ""
|
|
|
|
class FakeRequests:
|
|
def post(self, url, **kwargs):
|
|
capturas["url"] = url
|
|
capturas["data"] = kwargs.get("data", {})
|
|
capturas["files"] = list(kwargs.get("files", {}).keys())
|
|
return FakeResponse()
|
|
|
|
import notificador
|
|
real = notificador.requests
|
|
notificador.requests = FakeRequests()
|
|
orig_token = config.TELEGRAM_TOKEN
|
|
orig_chat = config.TELEGRAM_CHAT_ID
|
|
config.TELEGRAM_TOKEN = "123:FAKE"
|
|
config.TELEGRAM_CHAT_ID = "987654321"
|
|
try:
|
|
ruta = os.path.join(config.SNAPSHOT_DIR, "test_telegram.jpg")
|
|
cv2.imwrite(ruta, frame_movimiento())
|
|
enviar_telegram(ruta, "PRUEBA deteccion")
|
|
check("URL sendPhoto correcta",
|
|
"sendPhoto" in capturas.get("url", ""))
|
|
check("Envia chat_id", capturas.get("data", {}).get("chat_id") == config.TELEGRAM_CHAT_ID)
|
|
check("Adjunta la foto", "photo" in capturas.get("files", []))
|
|
finally:
|
|
notificador.requests = real
|
|
config.TELEGRAM_TOKEN = orig_token
|
|
config.TELEGRAM_CHAT_ID = orig_chat
|
|
|
|
|
|
def test_webhook_mock():
|
|
print("\n--- RF-08: Webhook Alexa (mock) ---")
|
|
llamadas = []
|
|
|
|
class FakeRequests:
|
|
def post(self, url, **kwargs):
|
|
llamadas.append((url, kwargs.get("json", {})))
|
|
|
|
import notificador
|
|
real = notificador.requests
|
|
notificador.requests = FakeRequests()
|
|
orig = config.WEBHOOK_ALEXA
|
|
config.WEBHOOK_ALEXA = "http://localhost:1880/alexa"
|
|
try:
|
|
disparar_webhook_alexa("PRUEBA")
|
|
check("Webhook disparado cuando configurado", len(llamadas) == 1)
|
|
finally:
|
|
config.WEBHOOK_ALEXA = orig
|
|
notificador.requests = real
|
|
|
|
llamadas.clear()
|
|
try:
|
|
disparar_webhook_alexa("PRUEBA")
|
|
check("No dispara si WEBHOOK_ALEXA vacio", len(llamadas) == 0)
|
|
finally:
|
|
notificador.requests = real
|
|
|
|
|
|
def test_yolo():
|
|
print("\n--- RF-02 (Fase 2): DetectorPersona YOLOv8n ---")
|
|
try:
|
|
from detector_persona import DetectorPersona
|
|
except Exception as e:
|
|
check("DetectorPersona importable", False, f"({e})")
|
|
return
|
|
det = DetectorPersona()
|
|
check("Modelo YOLOv8n cargado", det.model is not None)
|
|
res = det.hay_persona(frame_estatico())
|
|
check("Sin persona en frame estatico (falso positivo bajo)",
|
|
res is False)
|
|
|
|
|
|
def test_flujo_completo():
|
|
print("\n--- RF-03/RF-04/RF-05: Flujo completo snapshot+cooldown (mock) ---")
|
|
os.makedirs(config.SNAPSHOT_DIR, exist_ok=True)
|
|
ruta = simular_video([frame_estatico()] * 5 + [frame_movimiento()] * 5)
|
|
cap = cv2.VideoCapture(ruta)
|
|
det = DetectorMovimiento(umbral_pixeles=500)
|
|
det.hay_movimiento(frame_estatico())
|
|
det.hay_movimiento(frame_estatico())
|
|
det.hay_movimiento(frame_estatico())
|
|
detectado = False
|
|
n_snap = 0
|
|
while True:
|
|
ok, fr = cap.read()
|
|
if not ok:
|
|
break
|
|
if det.hay_movimiento(fr):
|
|
detectado = True
|
|
n_snap += 1
|
|
cap.release()
|
|
check("Video simulado: movimiento detectado", detectado)
|
|
check("Snapshot guardado (snapshots/simulado.mp4 existe)",
|
|
os.path.exists(ruta))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_rtsp_reconexion()
|
|
test_movimiento()
|
|
test_franja_horaria()
|
|
test_telegram_mock()
|
|
test_webhook_mock()
|
|
test_yolo()
|
|
test_flujo_completo()
|
|
print(f"\n=== RESULTADO: {APROBADOS} OK, {FALLIDOS} FALLO ===")
|
|
if FALLIDOS:
|
|
sys.exit(1)
|