profiles.py

"""
profiles.py — Perfiles de especificación (perfiles/*.yaml) y comprobaciones.

Cada comprobación devuelve una fila: parámetro, valor medido, requisito y resultado
(True = cumple, False = no cumple, None = solo informativo).
"""

from __future__ import annotations

import math
import re
from dataclasses import dataclass
from pathlib import Path

import yaml

from audio_io import SUBTYPE_LABEL, AudioMeta, fmt_duration

PROFILES_DIR = Path(__file__).parent / "perfiles"


@dataclass
class Check:
    parametro: str
    medido: str
    requisito: str
    ok: bool | None

    def __post_init__(self):
        # Las medidas son floats de numpy y sus comparaciones dan numpy.bool_, que no es True/False:
        # `numpy.False_ is False` es falso. Se normaliza aquí para que «cumple / no cumple» sea fiable.
        if self.ok is not None:
            self.ok = bool(self.ok)


def load_profiles(directory: Path = PROFILES_DIR) -> dict[str, dict]:
    profiles = {}
    for path in sorted(directory.glob("*.yaml")):
        with open(path, encoding="utf-8") as f:
            data = yaml.safe_load(f) or {}
        data["_id"] = path.stem
        profiles[path.stem] = data
    return dict(sorted(profiles.items(), key=lambda kv: (kv[1].get("orden", 99), kv[1].get("nombre", kv[0]))))


def _sec(section) -> dict:
    return section if isinstance(section, dict) else {}


def fdb(v: float, unit: str = "dBFS") -> str:
    if v is None or (isinstance(v, float) and math.isnan(v)):
        return "—"
    if math.isinf(v):
        return f"{'−' if v < 0 else ''}∞ {unit}"
    return f"{v:.1f} {unit}".replace("-", "−")


def fsec(v: float) -> str:
    return "—" if v is None or math.isnan(v) else f"{v:.2f} s"


def check(meta: AudioMeta, m: dict, prof: dict) -> list[Check]:
    fmt, lvl, nz, sil = _sec(prof.get("formato")), _sec(prof.get("niveles")), _sec(prof.get("ruido")), _sec(prof.get("silencios"))
    rows: list[Check] = []

    if fmt.get("contenedor"):
        rows.append(Check("Formato", meta.container, fmt["contenedor"], meta.container == fmt["contenedor"].upper()))
    if fmt.get("pcm_entero") or fmt.get("bits") or fmt.get("bits_min"):
        bits = meta.bits or 0
        ok = True
        req = []
        if fmt.get("pcm_entero"):
            ok &= meta.is_integer_pcm
            req.append("PCM entero")
        if fmt.get("bits"):
            ok &= bits == fmt["bits"]
            req.append(f"{fmt['bits']} bits")
        elif fmt.get("bits_min"):
            ok &= bits >= fmt["bits_min"]
            req.append(f"≥ {fmt['bits_min']} bits")
        rows.append(Check("Codificación", SUBTYPE_LABEL.get(meta.subtype, meta.subtype), " · ".join(req), ok))
    if fmt.get("frecuencia"):
        rows.append(Check("Frecuencia", f"{meta.samplerate} Hz", f"{fmt['frecuencia']} Hz", meta.samplerate == fmt["frecuencia"]))
    if fmt.get("canales"):
        rows.append(Check("Canales", str(meta.channels), "mono (1)" if fmt["canales"] == 1 else str(fmt["canales"]),
                          meta.channels == fmt["canales"]))

    if lvl.get("pico_max_dbfs") is not None:
        rows.append(Check("Pico máximo", fdb(m["pico_dbfs"]), f"≤ {fdb(lvl['pico_max_dbfs'])}", m["pico_dbfs"] <= lvl["pico_max_dbfs"] + 1e-6))
    rv = _sec(lvl.get("rms_voz"))
    if rv:
        rows.append(Check("RMS de voz", fdb(m["rms_voz_dbfs"]), f"{fdb(rv['min'])} a {fdb(rv['max'])}",
                          rv["min"] <= m["rms_voz_dbfs"] <= rv["max"]))
    if lvl.get("amplitud_voz_min") is not None:
        rows.append(Check("Amplitud de la voz", f"{m['pico_voz_muestras']} (de 32767)",
                          f"> {lvl['amplitud_voz_min']} ({fdb(20 * math.log10(lvl['amplitud_voz_min'] / 32767))})",
                          m["pico_voz_muestras"] > lvl["amplitud_voz_min"]))
    if nz.get("ruido_fondo_max_dbfs") is not None:
        rows.append(Check("Ruido de fondo", fdb(m["ruido_fondo_dbfs"]), f"≤ {fdb(nz['ruido_fondo_max_dbfs'])}",
                          m["ruido_fondo_dbfs"] <= nz["ruido_fondo_max_dbfs"]))
    if fmt.get("ancho_banda_min_hz") and "ancho_banda_hz" in m:
        bw = m["ancho_banda_hz"]
        rows.append(Check("Espectro (agudos)", "—" if math.isnan(bw) else f"hasta {bw / 1000:.1f} kHz",
                          f"más allá de {fmt['ancho_banda_min_hz'] / 1000:.0f} kHz",
                          (not math.isnan(bw)) and bw >= fmt["ancho_banda_min_hz"]))
    zb = m.get("zumbido")
    ztxt = f"{zb['base']} Hz · destaca {zb['db']:.0f} dB" if zb else "no"
    if nz.get("zumbido_max_db") is not None:
        rows.append(Check("Zumbido eléctrico", ztxt, f"≤ {nz['zumbido_max_db']:.0f} dB (sin zumbido)",
                          zb is None or zb["db"] <= nz["zumbido_max_db"]))
    elif "zumbido" in m:
        rows.append(Check("Zumbido eléctrico", ztxt, "informativo", None))
    if nz.get("ruido_relativo_max_db") is not None:
        rows.append(Check("Ruido relativo (inicio/fin)", fdb(m["ruido_relativo_db"], "dB"),
                          f"≤ {fdb(nz['ruido_relativo_max_db'], 'dB')} vs voz", m["ruido_relativo_db"] <= nz["ruido_relativo_max_db"]))
    if nz.get("snr_min_db") is not None:
        rows.append(Check("Relación señal/ruido (SNR)", fdb(m["snr_db"], "dB"), f"≥ {nz['snr_min_db']:.0f} dB",
                          m["snr_db"] >= nz["snr_min_db"]))
    if sil:
        for key, label in (("silencio_ini_s", "Silencio inicial"), ("silencio_fin_s", "Silencio final")):
            v = m[key]
            rows.append(Check(label, fsec(v), f"{sil['min_s']:.2f} a {sil['max_s']:.2f} s",
                              (not math.isnan(v)) and sil["min_s"] - 1e-3 <= v <= sil["max_s"] + 1e-3))
    if prof.get("saturacion_max_muestras") is not None:
        rows.append(Check("Saturación", f"{m['muestras_saturadas']} muestras", f"≤ {prof['saturacion_max_muestras']}",
                          m["muestras_saturadas"] <= prof["saturacion_max_muestras"]))

    # Informativo
    if "pausas_s" in m:
        max_p = pause_max(prof)
        largas = [p for p in m["pausas_s"] if p > max_p + 0.05]
        rows.append(Check(f"Pausas largas (> {max_p:.1f} s)",
                          f"{len(largas)} · la más larga {max(largas):.1f} s" if largas else "ninguna",
                          "informativo (se pueden acortar)", None))
    if nz.get("ruido_fondo_max_dbfs") is None:
        rows.append(Check("Ruido de fondo", fdb(m["ruido_fondo_dbfs"]), "informativo", None))
    if not nz.get("snr_min_db"):
        rows.append(Check("SNR", fdb(m["snr_db"], "dB"), "informativo", None))
    if "lufs" in m:
        rows.append(Check("Sonoridad integrada", fdb(m["lufs"], "LUFS"), "informativo", None))
    rows.append(Check("Pico real", fdb(m["pico_real_dbtp"], "dBTP"), "informativo", None))
    if "dnsmos" in m and m["dnsmos"]:
        d = m["dnsmos"]
        rows.append(Check("Calidad percibida (DNSMOS)", f"{d['OVRL']:.2f} / 5",
                          f"voz {d['SIG']:.2f} · fondo {d['BAK']:.2f}", None))
    if not m.get("voz_detectada", True):
        rows.append(Check("Detección de voz", "no se detecta voz", "—", False))
    return rows


def pause_max(prof: dict) -> float:
    """Pausa interna máxima para informar y acortar: la elegida en la app o la del perfil (2 s por defecto)."""
    return float(prof.get("_pausa_max_s") or (prof.get("pausas") or {}).get("max_s") or 2.0)


def long_pauses(m: dict, prof: dict) -> list[float]:
    return [p for p in m.get("pausas_s", []) if p > pause_max(prof) + 0.05]


def passes(rows: list[Check]) -> bool:
    return all(r.ok is not False for r in rows)


def failed(rows: list[Check]) -> list[str]:
    return [r.parametro for r in rows if r.ok is False]


# ── Entregables ──────────────────────────────────────────────────────────────

def _norm(name: str) -> str:
    stem = Path(name).stem.lower()
    return re.sub(r"[\s_\-]+", " ", stem).strip()


def deliverables(prof: dict, files: list[tuple[str, float]]) -> dict | None:
    """files: [(nombre, duración_s)]. Devuelve resumen de entregables o None si el perfil no los define."""
    ent = _sec(prof.get("entregables"))
    if not ent:
        return None
    total = sum(d for _, d in files)
    out = {"total_s": total, "objetivo_s": ent.get("objetivo_total_s"), "ficheros": []}
    by_name = {_norm(n): (n, d) for n, d in files}
    for spec in ent.get("ficheros") or []:
        found = by_name.get(_norm(spec["nombre"]))
        row = {"esperado": spec["nombre"], "fichero": found[0] if found else None,
               "duracion_s": found[1] if found else None, "ok": None, "nota": "", "estado": "recibido"}
        if not found:
            row["ok"], row["nota"], row["estado"] = False, "aún no subido", "pendiente"
        elif spec.get("duracion_min_s"):
            row["ok"] = found[1] >= spec["duracion_min_s"] - 0.05
            row["nota"] = f"mín. {fmt_duration(spec['duracion_min_s'])}"
            if not row["ok"]:
                row["estado"] = "corto"
        elif spec.get("duracion_aprox_s"):
            row["nota"] = f"≈ {fmt_duration(spec['duracion_aprox_s'])}"
        out["ficheros"].append(row)
    return out