Spaces:

Azuredraco
/

audio-denoise

Running on Zero

Despliegue desde GitHub

d903372 verified about 22 hours ago

9.05 kB

"""
metrics.py — Mediciones técnicas de un audio (las que piden las especificaciones de entrega).
Definiciones (todas en dBFS sobre el canal mezclado a mono):
- Pico: máximo absoluto de muestra. Pico real (dBTP): con sobremuestreo ×4.
- RMS de voz (bloque): RMS entre la primera y la última voz detectada (como el analizador v4).
- RMS de voz activa: RMS solo de las tramas con voz (excluye pausas).
- Ruido relativo: max(RMS de los primeros 0,4 s, RMS de los últimos 0,4 s) − RMS de voz (bloque).
- Ruido de fondo: potencia media de las tramas sin voz (excluye silencio digital).
- SNR: voz activa − ruido de fondo.
- Silencio inicial/final: tiempo antes de la primera voz / después de la última.
- Amplitud de voz: muestra más alta dentro de la zona con voz, en unidades de 16 bits (0–32767).
- Ancho de banda: frecuencia más alta con contenido real (no corte de filtro ni silencio digital).
"""

from __future__ import annotations

import numpy as np
import soxr

import dsp

NOISE_EDGE_SEC = 0.4


def _power_db(frames_db: np.ndarray) -> float:
    if frames_db.size == 0:
        return -np.inf
    return float(10 * np.log10(np.mean(10 ** (frames_db / 10))))


def true_peak_db(mono: np.ndarray, sr: int, block_s: float = 30.0) -> float:
    """Pico real (sobremuestreo ×4), por bloques para no disparar la memoria en ficheros largos."""
    blk, pad, peak = int(block_s * sr), 256, 0.0
    for i in range(0, len(mono), blk):
        a, b = max(0, i - pad), min(len(mono), i + blk + pad)
        up = soxr.resample(mono[a:b].astype(np.float32), sr, sr * 4, quality="HQ")
        inner = up[(i - a) * 4 : (i - a + min(blk, len(mono) - i)) * 4]
        if inner.size:
            peak = max(peak, float(np.abs(inner).max()))
    return dsp.db(peak)


SPECTRUM_SECONDS = 120   # para el ancho de banda basta una muestra repartida por todo el audio
CLIFF_DB = 20            # caída que delata un filtro: ≥ 20 dB entre el kHz de antes y el de después


def bandwidth_hz(mono: np.ndarray, sr: int) -> float:
    """
    Frecuencia hasta la que llega el sonido. Un móvil que graba a 48 kHz «de verdad» tiene ruido de micro y
    de sala hasta casi 24 kHz (el filtro del propio conversor actúa por encima de ~22,5 kHz); si el audio
    viene de 44,1 kHz, de un códec (AAC, Opus…) o de un filtro, el espectro cae en picado antes.
    Se busca el primer «acantilado»: una caída de ≥ 20 dB entre el kHz anterior y el siguiente.
    Sin acantilado, el sonido llega hasta arriba (se devuelve la frecuencia máxima evaluable).
    """
    from scipy import signal

    n = len(mono)
    if n < sr // 10:
        return float("nan")
    seg = int(sr)
    if n > int(SPECTRUM_SECONDS * sr):
        starts = np.linspace(0, n - seg, int(SPECTRUM_SECONDS)).astype(int)
        x = np.concatenate([mono[a : a + seg] for a in starts])
    else:
        x = mono
    f, pxx = signal.welch(x.astype(np.float64), sr, nperseg=min(len(x), 4096))
    df = f[1] - f[0]
    step = max(1, int(round(100 / df)))
    nb = len(pxx) // step
    band = 10 * np.log10(np.maximum(pxx[: nb * step].reshape(nb, step).mean(axis=1), 1e-30))
    fb = f[: nb * step].reshape(nb, step).mean(axis=1)
    k = 10  # 10 bandas de 100 Hz = 1 kHz
    last = nb - k - 2
    for i in range(int(3000 / (df * step)), max(last, 0)):
        before = np.median(band[i - k : i])
        after = np.median(band[i + 2 : i + 2 + k])
        if before - after >= CLIFF_DB:
            return float(fb[i])
    return float(fb[last + 1]) if last > 0 else float(fb[-1])


HUM_HARMONICS = 10        # se miran los 10 primeros armónicos (hasta 500/600 Hz)
HUM_MIN_DB = 10.0         # un armónico «es zumbido» si destaca ≥ 10 dB sobre las frecuencias vecinas


def hum(mono: np.ndarray, sr: int) -> dict | None:
    """
    Zumbido eléctrico de la red (50 Hz en Europa, 60 Hz en América) y sus armónicos.
    Es un tono fijo: en un espectro de 1 Hz de resolución sobre todo el audio forma picos estrechos,
    mientras que la voz (que cambia de tono continuamente) se reparte. Devuelve
    {'base': 50|60, 'db': prominencia máxima, 'armonicos': [Hz con ≥ 6 dB]} o None.
    """
    from scipy import signal

    n = len(mono)
    if n < 2 * sr:
        return None
    seg = int(2 * sr)
    if n > 120 * sr:
        starts = np.linspace(0, n - seg, 60).astype(int)
        x = np.concatenate([mono[a : a + seg] for a in starts])
    else:
        x = mono
    f, pxx = signal.welch(x.astype(np.float64), sr, nperseg=seg)
    pdb = 10 * np.log10(np.maximum(pxx, 1e-30))
    df = f[1] - f[0]
    best = None
    for base in (50, 60):
        prom = []
        for k in range(1, HUM_HARMONICS + 1):
            fk = base * k
            i = int(round(fk / df))
            if i + 12 >= len(pdb):
                break
            peak = pdb[max(0, i - 1) : i + 2].max()
            neigh = np.concatenate([pdb[max(0, i - 12) : max(0, i - 3)], pdb[i + 4 : i + 13]])
            prom.append((fk, float(peak - np.median(neigh))))
        strong = [fk for fk, d in prom if d >= 6]
        top = max((d for _, d in prom), default=0.0)
        # hace falta un armónico claro y al menos otro: un solo pico aislado puede ser otra cosa
        if top >= HUM_MIN_DB and len(strong) >= 2 and (best is None or top > best["db"]):
            best = {"base": base, "db": top, "armonicos": strong}
    return best


def measure(data: np.ndarray, sr: int, with_lufs: bool = True, detector: str | None = None) -> dict:
    """data: (n, ch) float32. Devuelve un dict de métricas (floats en dB / segundos)."""
    mono = dsp.to_mono(data)[:, 0]
    n = len(mono)
    peak_abs = float(np.abs(data).max()) if n else 0.0
    region = dsp.voice_region(mono, sr, detector)
    voice = mono[region.start : region.end + 1] if region.found else mono

    fdb = region.frame_db
    live = fdb > dsp.DIGITAL_SILENCE_DB
    voiced = fdb >= region.start_th_db
    noise_frames = fdb[live & (fdb < region.noise_db + 6)] if np.isfinite(region.noise_db) else np.array([])
    voice_active_db = _power_db(fdb[voiced]) if voiced.any() else -np.inf
    noise_floor_db = _power_db(noise_frames)

    w = int(NOISE_EDGE_SEC * sr)
    edge_noise = max(dsp.rms_db(mono[: min(w, n)]), dsp.rms_db(mono[max(0, n - w) :]))
    voice_rms = dsp.rms_db(voice)

    m = {
        "duracion_s": n / sr if sr else 0.0,
        "pico_dbfs": dsp.db(peak_abs),
        "pico_real_dbtp": true_peak_db(mono, sr) if n else -np.inf,
        "muestras_saturadas": int((np.abs(data) >= 0.999).any(axis=1).sum()),
        "dc_offset": float(np.mean(mono)) if n else 0.0,
        "rms_voz_dbfs": voice_rms,
        "rms_voz_activa_dbfs": voice_active_db,
        "ruido_fondo_dbfs": noise_floor_db,
        "ruido_relativo_db": edge_noise - voice_rms,
        "snr_db": voice_active_db - noise_floor_db,
        "silencio_ini_s": region.start / sr if region.found else np.nan,
        "silencio_fin_s": (n - 1 - region.end) / sr if region.found else np.nan,
        "voz_detectada": region.found,
        "silencio_digital": bool(not live.any()) or bool(noise_frames.size == 0),
        "pico_voz_muestras": min(32767, int(round(float(np.abs(voice).max()) * 32767))) if voice.size else 0,
        "ancho_banda_hz": bandwidth_hz(mono, sr),
        "zumbido": hum(mono, sr),
        "pausas_s": [(b - a) / sr for a, b in dsp.find_pauses(mono, sr, region)],
    }
    if with_lufs:
        m["lufs"] = lufs(data, sr)
    return m


def lufs(data: np.ndarray, sr: int) -> float:
    """Sonoridad integrada ITU-R BS.1770-4 (ponderación K, bloques de 400 ms, puertas −70 y −10)."""
    if data.shape[0] < int(0.4 * sr):
        return np.nan
    # Filtros K (fórmulas de BS.1770 válidas para cualquier frecuencia de muestreo)
    from scipy import signal

    f0, g, q = 1681.974450955533, 3.999843853973347, 0.7071752369554196
    k = np.tan(np.pi * f0 / sr)
    vh = 10 ** (g / 20)
    vb = vh ** 0.4996667741545416
    a0 = 1 + k / q + k * k
    shelf = [(vh + vb * k / q + k * k) / a0, 2 * (k * k - vh) / a0, (vh - vb * k / q + k * k) / a0,
             1, 2 * (k * k - 1) / a0, (1 - k / q + k * k) / a0]
    f0, q = 38.13547087602444, 0.5003270373238773
    k = np.tan(np.pi * f0 / sr)
    a0 = 1 + k / q + k * k
    hp = [1, -2, 1, 1, 2 * (k * k - 1) / a0, (1 - k / q + k * k) / a0]
    sos = np.array([shelf, hp])
    blk, hop = int(0.4 * sr), int(0.1 * sr)
    power = None
    for c in range(data.shape[1]):
        z = signal.sosfilt(sos, data[:, c]).astype(np.float32) ** 2
        cs = np.concatenate([[0.0], np.cumsum(z, dtype=np.float64)])
        starts = np.arange(0, len(z) - blk + 1, hop)
        ms = (cs[starts + blk] - cs[starts]) / blk
        power = ms if power is None else power + ms
    lk = -0.691 + 10 * np.log10(np.maximum(power, 1e-20))
    gated = power[lk > -70]
    if gated.size == 0:
        return -np.inf
    rel = -0.691 + 10 * np.log10(gated.mean()) - 10
    gated = power[(lk > -70) & (lk > rel)]
    return float(-0.691 + 10 * np.log10(gated.mean())) if gated.size else -np.inf

	"""
	metrics.py — Mediciones técnicas de un audio (las que piden las especificaciones de entrega).

	Definiciones (todas en dBFS sobre el canal mezclado a mono):
	- Pico: máximo absoluto de muestra. Pico real (dBTP): con sobremuestreo ×4.
	- RMS de voz (bloque): RMS entre la primera y la última voz detectada (como el analizador v4).
	- RMS de voz activa: RMS solo de las tramas con voz (excluye pausas).
	- Ruido relativo: max(RMS de los primeros 0,4 s, RMS de los últimos 0,4 s) − RMS de voz (bloque).
	- Ruido de fondo: potencia media de las tramas sin voz (excluye silencio digital).
	- SNR: voz activa − ruido de fondo.
	- Silencio inicial/final: tiempo antes de la primera voz / después de la última.
	- Amplitud de voz: muestra más alta dentro de la zona con voz, en unidades de 16 bits (0–32767).
	- Ancho de banda: frecuencia más alta con contenido real (no corte de filtro ni silencio digital).
	"""

	from __future__ import annotations

	import numpy as np
	import soxr

	import dsp

	NOISE_EDGE_SEC = 0.4


	def _power_db(frames_db: np.ndarray) -> float:
	if frames_db.size == 0:
	return -np.inf
	return float(10 * np.log10(np.mean(10 ** (frames_db / 10))))


	def true_peak_db(mono: np.ndarray, sr: int, block_s: float = 30.0) -> float:
	"""Pico real (sobremuestreo ×4), por bloques para no disparar la memoria en ficheros largos."""
	blk, pad, peak = int(block_s * sr), 256, 0.0
	for i in range(0, len(mono), blk):
	a, b = max(0, i - pad), min(len(mono), i + blk + pad)
	up = soxr.resample(mono[a:b].astype(np.float32), sr, sr * 4, quality="HQ")
	inner = up[(i - a) * 4 : (i - a + min(blk, len(mono) - i)) * 4]
	if inner.size:
	peak = max(peak, float(np.abs(inner).max()))
	return dsp.db(peak)


	SPECTRUM_SECONDS = 120 # para el ancho de banda basta una muestra repartida por todo el audio
	CLIFF_DB = 20 # caída que delata un filtro: ≥ 20 dB entre el kHz de antes y el de después


	def bandwidth_hz(mono: np.ndarray, sr: int) -> float:
	"""
	Frecuencia hasta la que llega el sonido. Un móvil que graba a 48 kHz «de verdad» tiene ruido de micro y
	de sala hasta casi 24 kHz (el filtro del propio conversor actúa por encima de ~22,5 kHz); si el audio
	viene de 44,1 kHz, de un códec (AAC, Opus…) o de un filtro, el espectro cae en picado antes.
	Se busca el primer «acantilado»: una caída de ≥ 20 dB entre el kHz anterior y el siguiente.
	Sin acantilado, el sonido llega hasta arriba (se devuelve la frecuencia máxima evaluable).
	"""
	from scipy import signal

	n = len(mono)
	if n < sr // 10:
	return float("nan")
	seg = int(sr)
	if n > int(SPECTRUM_SECONDS * sr):
	starts = np.linspace(0, n - seg, int(SPECTRUM_SECONDS)).astype(int)
	x = np.concatenate([mono[a : a + seg] for a in starts])
	else:
	x = mono
	f, pxx = signal.welch(x.astype(np.float64), sr, nperseg=min(len(x), 4096))
	df = f[1] - f[0]
	step = max(1, int(round(100 / df)))
	nb = len(pxx) // step
	band = 10 * np.log10(np.maximum(pxx[: nb * step].reshape(nb, step).mean(axis=1), 1e-30))
	fb = f[: nb * step].reshape(nb, step).mean(axis=1)
	k = 10 # 10 bandas de 100 Hz = 1 kHz
	last = nb - k - 2
	for i in range(int(3000 / (df * step)), max(last, 0)):
	before = np.median(band[i - k : i])
	after = np.median(band[i + 2 : i + 2 + k])
	if before - after >= CLIFF_DB:
	return float(fb[i])
	return float(fb[last + 1]) if last > 0 else float(fb[-1])


	HUM_HARMONICS = 10 # se miran los 10 primeros armónicos (hasta 500/600 Hz)
	HUM_MIN_DB = 10.0 # un armónico «es zumbido» si destaca ≥ 10 dB sobre las frecuencias vecinas


	def hum(mono: np.ndarray, sr: int) -> dict \| None:
	"""
	Zumbido eléctrico de la red (50 Hz en Europa, 60 Hz en América) y sus armónicos.
	Es un tono fijo: en un espectro de 1 Hz de resolución sobre todo el audio forma picos estrechos,
	mientras que la voz (que cambia de tono continuamente) se reparte. Devuelve
	{'base': 50\|60, 'db': prominencia máxima, 'armonicos': [Hz con ≥ 6 dB]} o None.
	"""
	from scipy import signal

	n = len(mono)
	if n < 2 * sr:
	return None
	seg = int(2 * sr)
	if n > 120 * sr:
	starts = np.linspace(0, n - seg, 60).astype(int)
	x = np.concatenate([mono[a : a + seg] for a in starts])
	else:
	x = mono
	f, pxx = signal.welch(x.astype(np.float64), sr, nperseg=seg)
	pdb = 10 * np.log10(np.maximum(pxx, 1e-30))
	df = f[1] - f[0]
	best = None
	for base in (50, 60):
	prom = []
	for k in range(1, HUM_HARMONICS + 1):
	fk = base * k
	i = int(round(fk / df))
	if i + 12 >= len(pdb):
	break
	peak = pdb[max(0, i - 1) : i + 2].max()
	neigh = np.concatenate([pdb[max(0, i - 12) : max(0, i - 3)], pdb[i + 4 : i + 13]])
	prom.append((fk, float(peak - np.median(neigh))))
	strong = [fk for fk, d in prom if d >= 6]
	top = max((d for _, d in prom), default=0.0)
	# hace falta un armónico claro y al menos otro: un solo pico aislado puede ser otra cosa
	if top >= HUM_MIN_DB and len(strong) >= 2 and (best is None or top > best["db"]):
	best = {"base": base, "db": top, "armonicos": strong}
	return best


	def measure(data: np.ndarray, sr: int, with_lufs: bool = True, detector: str \| None = None) -> dict:
	"""data: (n, ch) float32. Devuelve un dict de métricas (floats en dB / segundos)."""
	mono = dsp.to_mono(data)[:, 0]
	n = len(mono)
	peak_abs = float(np.abs(data).max()) if n else 0.0
	region = dsp.voice_region(mono, sr, detector)
	voice = mono[region.start : region.end + 1] if region.found else mono

	fdb = region.frame_db
	live = fdb > dsp.DIGITAL_SILENCE_DB
	voiced = fdb >= region.start_th_db
	noise_frames = fdb[live & (fdb < region.noise_db + 6)] if np.isfinite(region.noise_db) else np.array([])
	voice_active_db = _power_db(fdb[voiced]) if voiced.any() else -np.inf
	noise_floor_db = _power_db(noise_frames)

	w = int(NOISE_EDGE_SEC * sr)
	edge_noise = max(dsp.rms_db(mono[: min(w, n)]), dsp.rms_db(mono[max(0, n - w) :]))
	voice_rms = dsp.rms_db(voice)

	m = {
	"duracion_s": n / sr if sr else 0.0,
	"pico_dbfs": dsp.db(peak_abs),
	"pico_real_dbtp": true_peak_db(mono, sr) if n else -np.inf,
	"muestras_saturadas": int((np.abs(data) >= 0.999).any(axis=1).sum()),
	"dc_offset": float(np.mean(mono)) if n else 0.0,
	"rms_voz_dbfs": voice_rms,
	"rms_voz_activa_dbfs": voice_active_db,
	"ruido_fondo_dbfs": noise_floor_db,
	"ruido_relativo_db": edge_noise - voice_rms,
	"snr_db": voice_active_db - noise_floor_db,
	"silencio_ini_s": region.start / sr if region.found else np.nan,
	"silencio_fin_s": (n - 1 - region.end) / sr if region.found else np.nan,
	"voz_detectada": region.found,
	"silencio_digital": bool(not live.any()) or bool(noise_frames.size == 0),
	"pico_voz_muestras": min(32767, int(round(float(np.abs(voice).max()) * 32767))) if voice.size else 0,
	"ancho_banda_hz": bandwidth_hz(mono, sr),
	"zumbido": hum(mono, sr),
	"pausas_s": [(b - a) / sr for a, b in dsp.find_pauses(mono, sr, region)],
	}
	if with_lufs:
	m["lufs"] = lufs(data, sr)
	return m


	def lufs(data: np.ndarray, sr: int) -> float:
	"""Sonoridad integrada ITU-R BS.1770-4 (ponderación K, bloques de 400 ms, puertas −70 y −10)."""
	if data.shape[0] < int(0.4 * sr):
	return np.nan
	# Filtros K (fórmulas de BS.1770 válidas para cualquier frecuencia de muestreo)
	from scipy import signal

	f0, g, q = 1681.974450955533, 3.999843853973347, 0.7071752369554196
	k = np.tan(np.pi * f0 / sr)
	vh = 10 ** (g / 20)
	vb = vh ** 0.4996667741545416
	a0 = 1 + k / q + k * k
	shelf = [(vh + vb * k / q + k * k) / a0, 2 * (k * k - vh) / a0, (vh - vb * k / q + k * k) / a0,
	1, 2 * (k * k - 1) / a0, (1 - k / q + k * k) / a0]
	f0, q = 38.13547087602444, 0.5003270373238773
	k = np.tan(np.pi * f0 / sr)
	a0 = 1 + k / q + k * k
	hp = [1, -2, 1, 1, 2 * (k * k - 1) / a0, (1 - k / q + k * k) / a0]
	sos = np.array([shelf, hp])
	blk, hop = int(0.4 * sr), int(0.1 * sr)
	power = None
	for c in range(data.shape[1]):
	z = signal.sosfilt(sos, data[:, c]).astype(np.float32) ** 2
	cs = np.concatenate([[0.0], np.cumsum(z, dtype=np.float64)])
	starts = np.arange(0, len(z) - blk + 1, hop)
	ms = (cs[starts + blk] - cs[starts]) / blk
	power = ms if power is None else power + ms
	lk = -0.691 + 10 * np.log10(np.maximum(power, 1e-20))
	gated = power[lk > -70]
	if gated.size == 0:
	return -np.inf
	rel = -0.691 + 10 * np.log10(gated.mean()) - 10
	gated = power[(lk > -70) & (lk > rel)]
	return float(-0.691 + 10 * np.log10(gated.mean())) if gated.size else -np.inf