222 lines
7.1 KiB
Python
222 lines
7.1 KiB
Python
#!/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"👻 <b>Barrido nocturno</b> 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()
|