abraham
937bbd91ac
- pet_audio.py: escucha del audio RTSP en un hilo propio con PyAV, remuestreo de 8 kHz a 16 kHz y YAMNet en ONNX sobre ventanas de 0.975 s con medio solape. Dispara por la clase Bark (mejor margen perro/gato que el grupo canino), con anti-repeticion y reconexion automatica. - pet_common.py: utilidades compartidas de .env, construccion de la URL RTSP y enmascarado de credenciales, porque PyAV incluye la contrasena en el texto de sus excepciones. - pet_speed.py: arranca la escucha de audio y la refleja en la linea de estadisticas. - scripts/fetch_models.sh: descarga y verifica por SHA-256 el modelo YAMNet ONNX (16 MB). - models/yamnet.onnx: el fichero versionado no era un modelo, sino una pagina HTML de GitHub. Se saca de git y pasa a descargarse con el script. - Readme.md: seccion de audio, estado de la Fase A y ajuste de umbral.
87 lines
2.7 KiB
Python
87 lines
2.7 KiB
Python
"""Utilidades compartidas por los módulos del Pet Monitor."""
|
|
|
|
import os
|
|
import urllib.parse
|
|
|
|
ENV_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")
|
|
|
|
# Claves que se pueden sobreescribir desde el entorno real (systemd, pruebas
|
|
# puntuales). Las variables de entorno tienen prioridad sobre el fichero .env.
|
|
ENV_KEYS = (
|
|
"CAMERA_IP",
|
|
"CAMERA_USER",
|
|
"CAMERA_PW",
|
|
"CAMERA_STREAM",
|
|
"CAMERA_RTSP_URL",
|
|
"PET_AUDIO",
|
|
"BARK_THRESHOLD",
|
|
"BARK_COOLDOWN",
|
|
)
|
|
|
|
|
|
def load_env(path=ENV_PATH):
|
|
"""Lee un .env sencillo (CLAVE=valor) sin dependencias externas."""
|
|
env = {}
|
|
if os.path.exists(path):
|
|
with open(path) as fh:
|
|
for line in fh:
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
env[key.strip()] = value.strip().strip('"').strip("'")
|
|
for key in ENV_KEYS:
|
|
if os.environ.get(key):
|
|
env[key] = os.environ[key]
|
|
return env
|
|
|
|
|
|
def build_rtsp_url(env, stream):
|
|
"""Construye la URL RTSP de la cámara, escapando usuario y contraseña."""
|
|
if env.get("CAMERA_RTSP_URL"):
|
|
return env["CAMERA_RTSP_URL"]
|
|
ip = env.get("CAMERA_IP")
|
|
user = env.get("CAMERA_USER")
|
|
password = env.get("CAMERA_PW")
|
|
missing = [k for k, v in (("CAMERA_IP", ip), ("CAMERA_USER", user), ("CAMERA_PW", password)) if not v]
|
|
if missing:
|
|
raise SystemExit(f"[ERROR] Faltan variables en {ENV_PATH}: {', '.join(missing)}")
|
|
return "rtsp://%s:%s@%s:554/%s" % (
|
|
urllib.parse.quote(user, safe=""),
|
|
urllib.parse.quote(password, safe=""),
|
|
ip,
|
|
stream,
|
|
)
|
|
|
|
|
|
def mask_url(url):
|
|
"""Oculta las credenciales de la URL para poder mostrarla por consola."""
|
|
parts = urllib.parse.urlsplit(url)
|
|
if not parts.hostname:
|
|
return "***"
|
|
port = f":{parts.port}" if parts.port else ""
|
|
return f"{parts.scheme}://***@{parts.hostname}{port}{parts.path}"
|
|
|
|
|
|
def secrets_of(url):
|
|
"""Devuelve los trozos de una URL que nunca deben aparecer en un log."""
|
|
parts = urllib.parse.urlsplit(url)
|
|
valores = [parts.password, parts.username]
|
|
if parts.password:
|
|
valores.append(urllib.parse.quote(parts.password, safe=""))
|
|
return [v for v in valores if v]
|
|
|
|
|
|
def mask_secrets(text, secrets):
|
|
"""
|
|
Tapa las credenciales dentro de un texto.
|
|
|
|
PyAV incluye la URL completa (con contraseña) en sus excepciones, así que
|
|
todo lo que se imprima desde el hilo de audio tiene que pasar por aquí: si
|
|
no, un fallo de red dejaría la contraseña escrita en journalctl.
|
|
"""
|
|
text = str(text)
|
|
for secret in secrets:
|
|
if secret:
|
|
text = text.replace(secret, "***")
|
|
return text
|