#!/usr/bin/env python3 """Barrido nocturno del Salon (1:00-6:00). Cada 30 min hace un barrido panoramico L->R y R->L con la camara Tapo del Salon, detectando movimiento comparando frames (PSNR). Si detecta algo irregular guarda las imagenes en una carpeta dedicada, lo anota en el log, y manda un aviso DISCRETO a Telegram (sin alarma/sirena/voz). Independiente del panorama. Config via .env: NOCTURNO_ACTIVO (1/0, default 1) NOCTURNO_INICIO hora inicio (default 1) NOCTURNO_FIN hora fin (default 6) NOCTURNO_INTERVALO minutos entre barridos (default 30) NOCTURNO_PSNR_MIN umbral PSNR (default 25.0; menor = mas diferencia) NOCTURNO_DIR carpeta de evidencias (default salidas_nocturno) NOCTURNO_TAPO_USER / NOCTURNO_TAPO_PASS """ import os import sys import time import json import subprocess from datetime import datetime BASE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, BASE) import config import requests LOG = os.path.join(BASE, "nocturno.log") ACTIVO = config.__dict__.get("NOCTURNO_ACTIVO", "1") try: ACTIVO = int(os.getenv("NOCTURNO_ACTIVO", "1")) except Exception: ACTIVO = 1 INICIO = int(os.getenv("NOCTURNO_INICIO", "1")) FIN = int(os.getenv("NOCTURNO_FIN", "6")) INTERVALO = int(os.getenv("NOCTURNO_INTERVALO", "30")) PSNR_MIN = float(os.getenv("NOCTURNO_PSNR_MIN", "25.0")) DIR = os.getenv("NOCTURNO_DIR", "salidas_nocturno") TAPO_USER = os.getenv("NOCTURNO_TAPO_USER", "") TAPO_PASS = os.getenv("NOCTURNO_TAPO_PASS", "") TAPO_IP = os.getenv("NOCTURNO_TAPO_IP", "192.168.50.227") # RTSP del Salon (para frames) - lo tomamos de config.CAMARAS[0] (Salon) RTSP = None for cam in config.CAMARAS: if cam["nombre"].lower() == "salon": RTSP = cam["rtsp"] break if not RTSP and config.CAMARAS: RTSP = config.CAMARAS[0]["rtsp"] SNAPSHOTS = os.path.join(DIR, "evidencias") def log(linea): linea = f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} {linea}" print(linea) try: with open(LOG, "a") as f: f.write(linea + "\n") except Exception: pass def en_rango_horario(): h = datetime.now().hour if INICIO <= FIN: return INICIO <= h < FIN return h >= INICIO or h < FIN def capturar_frame(ruta, tiempo=6): os.makedirs(os.path.dirname(ruta), exist_ok=True) cmd = ["ffmpeg", "-rtsp_transport", "tcp", "-i", RTSP, "-frames:v", "1", "-q:v", "3", "-y", "-hide_banner", "-loglevel", "error", ruta] try: subprocess.run(cmd, timeout=tiempo, check=True) return os.path.getsize(ruta) > 0 except Exception: return False def psnr(a, b): """Devuelve PSNR entre dos imagenes (mayor = mas similares).""" cmd = ["ffmpeg", "-hide_banner", "-i", a, "-i", b, "-lavfi", "psnr", "-f", "null", "-"] try: r = subprocess.run(cmd, capture_output=True, text=True, timeout=20) txt = r.stderr for part in txt.split(): if part.startswith("average:"): try: return float(part.split(":")[1].split(" ")[0]) except Exception: pass except Exception: pass return None def avisar_telegram(texto): try: requests.post( f"https://api.telegram.org/bot{config.TELEGRAM_TOKEN}/sendMessage", data={"chat_id": config.TELEGRAM_CHAT_ID, "text": texto, "parse_mode": "HTML"}, timeout=10) except Exception as e: log(f"ERR telegram: {e}") def mover_motor(angulo): """Mueve el motor de la camara un paso. Retorna True si ok.""" try: from pytapo import Tapo except Exception as e: log(f"ERR pytapo import para mover: {e}") return False try: t = Tapo(TAPO_IP, TAPO_USER, TAPO_PASS) t.moveMotorStep(angulo) return True except Exception as e: log(f"ERR moviendo motor {angulo}: {e}") return False def barrido(primer_paso, ultimo_paso, paso, prefijo): """Barre desde primer_paso hasta ultimo_paso. En cada posicion: dispara 2 frames c/1.5s y compara PSNR. Devuelve lista de instantes con irregularidad: (angulo, frame_a, frame_b, psnr).""" incidentes = [] ang = primer_paso ruta = 0 while True: ruta += 1 nombre = f"{prefijo}_paso{ruta}_ang{ang}" mover_motor(ang) time.sleep(2) # esperar a que estabilice el giro fa = os.path.join(SNAPSHOTS, f"{nombre}_A.jpg") fb = os.path.join(SNAPSHOTS, f"{nombre}_B.jpg") ok_a = capturar_frame(fa) time.sleep(1.5) ok_b = capturar_frame(fb) if ok_a and ok_b: p = psnr(fa, fb) log(f"[{prefijo}] ang {ang}: PSNR={p}") if p is not None and p < PSNR_MIN: incidentes.append({"angulo": ang, "frame_a": fa, "frame_b": fb, "psnr": p}) else: log(f"[{prefijo}] ang {ang}: fallo captura") if ang == ultimo_paso: break ang += paso if primer_paso < ultimo_paso else -paso return incidentes def ejecutar_barrido(): ts = datetime.now().strftime("%Y%m%d_%H%M%S") log("=" * 50) log(f"Barrido nocturno iniciado {ts}") if not RTSP: log("No hay RTSP del Salon configurado") return if not TAPO_USER or not TAPO_PASS: log("Faltan credenciales Tapo (NOCTURNO_TAPO_USER/PASS)") return # Barrido izquierda -> derecha (por defecto de angulo 20 a 270) inc_a = barrido(20, 270, 50, f"ida_{ts}") # Barrido derecha -> izquierda inc_b = barrido(270, 20, -50, f"vuelta_{ts}") todas = inc_a + inc_b if todas: log(f"*** IRREGULARIDADES detectadas: {len(todas)}") # guardar un resumen JSON resumen = { "timestamp": ts, "psnr_min": PSNR_MIN, "incidentes": [ {"angulo": i["angulo"], "psnr": i["psnr"], "frame_a": i["frame_a"], "frame_b": i["frame_b"]} for i in todas ], } os.makedirs(DIR, exist_ok=True) ruta_json = os.path.join(DIR, f"incidentes_{ts}.json") with open(ruta_json, "w") as f: json.dump(resumen, f, indent=2) log(f"Guardado resumen en {ruta_json}") avisar_telegram( f"👻 Barrido nocturno detectó {len(todas)} instante(s) " f"irregular(es) en el Salón ({ts}).\n" f"Evidencias guardadas. Revisión en la mañana.") else: log("Barrido completado: sin irregularidades") log("=" * 50) def bucle(): log(f"Barrido nocturno activo (rango {INICIO}:00-{FIN}:00, cada {INTERVALO} min)") ultimo = None while True: if en_rango_horario(): ahora = datetime.now() clave = f"{ahora.hour}:{ahora.minute // INTERVALO}" if clave != ultimo: ultimo = clave ejecutar_barrido() time.sleep(60) else: ultimo = None time.sleep(120) if __name__ == "__main__": if not ACTIVO: print("Barrido nocturno desactivado (NOCTURNO_ACTIVO=0)") sys.exit(0) bucle()