diff --git a/.gitignore b/.gitignore index 554fa1e..20f869e 100644 --- a/.gitignore +++ b/.gitignore @@ -22,3 +22,5 @@ telegram_mensajes.log # Vigilancia vigilancia/snapshots/ +vigilancia/*.pt +vigilancia/.venv/ diff --git a/vigilancia/main.py b/vigilancia/main.py index ca15932..6ba9c91 100644 --- a/vigilancia/main.py +++ b/vigilancia/main.py @@ -15,8 +15,9 @@ if USAR_YOLO: from detector_persona import DetectorPersona -def dentro_de_franja_horaria() -> bool: - hora_actual = datetime.now().hour +def dentro_de_franja_horaria(hora_actual=None) -> bool: + if hora_actual is None: + hora_actual = datetime.now().hour inicio = config.HORA_INICIO fin = config.HORA_FIN if inicio < fin: diff --git a/vigilancia/test_prueba.py b/vigilancia/test_prueba.py new file mode 100644 index 0000000..21e54e5 --- /dev/null +++ b/vigilancia/test_prueba.py @@ -0,0 +1,209 @@ +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)