08-ago: [discovery] Lecciones USB Linux ext4 + Realtek 8812AU + scripts kali-todo
This commit is contained in:
minguezsanzjuanjose
221
vigilancia/barrido_nocturno.py
Normal file
221
vigilancia/barrido_nocturno.py
Normal file
@@ -0,0 +1,221 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user