Spaces:
Running on Zero
Running on Zero
| """ | |
| pipeline.py — Analizar y corregir ficheros según un perfil de especificación. | |
| Principio: corrección MÍNIMA. Se mide, se ve qué comprobaciones no cumplen y se aplica SOLO | |
| la corrección de cada una (si solo falla el volumen, solo se ajusta el volumen). Después se | |
| vuelve a medir: si una corrección provoca un fallo nuevo (p. ej. al subir el volumen asoma el | |
| ruido), se añade solo la corrección de ese fallo y se repite. Si un audio cumple, no se toca. | |
| Pulido automático (opcional, activado por defecto en la app): además se prueba a quitar ruido de | |
| fondo y retumbes; solo se conserva si la nota de calidad (DNSMOS) sube y la voz no empeora. | |
| Correcciones posibles (se aplican en este orden, solo las necesarias): | |
| mono → frecuencia → ruido (paso alto + DeepFilterNet, escalonado) → silencios (solo el lado | |
| que falla) → volumen (ganancia + limitador) o solo picos (limitador) → formato de escritura. | |
| """ | |
| from __future__ import annotations | |
| import math | |
| from dataclasses import dataclass, field | |
| from pathlib import Path | |
| import numpy as np | |
| import dsp | |
| import metrics | |
| import profiles | |
| from audio_io import AudioMeta, output_format, read_audio, read_meta, write_audio | |
| NOISE_CHECKS = {"Relación señal/ruido (SNR)", "Ruido relativo (inicio/fin)", "Ruido de fondo"} | |
| # Qué corrección arregla cada comprobación del perfil | |
| FIX_FOR = { | |
| "Formato": "formato", "Codificación": "formato", "Frecuencia": "frecuencia", "Canales": "mono", | |
| "Pico máximo": "picos", "Saturación": "picos", "RMS de voz": "volumen", | |
| "Ruido relativo (inicio/fin)": "ruido", "Relación señal/ruido (SNR)": "ruido", | |
| "Silencio inicial": "sil_ini", "Silencio final": "sil_fin", | |
| "Amplitud de la voz": "amplitud", "Ruido de fondo": "ruido", "Zumbido eléctrico": "zumbido", | |
| } | |
| # Niveles de procesado: qué correcciones puede usar cada uno (lo elige el usuario en la app) | |
| MODES = { | |
| "formato": {"formato", "mono", "frecuencia"}, # el sonido no cambia | |
| "permitido": {"formato", "mono", "frecuencia", "volumen", "picos", # + niveles y silencios | |
| "amplitud", "sil_ini", "sil_fin"}, | |
| "ninja": None, # todo, + pulido sutil | |
| "auto": None, # lo mínimo que haga falta, sin pulido | |
| } | |
| MODE_LABEL = {"formato": "Solo formato", "permitido": "Permitido", "ninja": "Ninja", "auto": "Automático"} | |
| # Qué nivel hace falta para cada corrección (para decir «ha hecho falta: Permitido») | |
| LEVEL_OF = {"formato": "formato", "mono": "formato", "frecuencia": "formato", | |
| "volumen": "permitido", "picos": "permitido", "amplitud": "permitido", "sil_ini": "permitido", | |
| "sil_fin": "permitido", "pausas": "permitido", | |
| "ruido": "ninja", "eventos": "ninja", "zumbido": "ninja"} | |
| FIX_WORDS = {"formato": "formato", "mono": "canales (a mono)", "frecuencia": "frecuencia", "volumen": "volumen", | |
| "picos": "picos", "amplitud": "volumen", "sil_ini": "silencio inicial", "sil_fin": "silencio final", | |
| "pausas": "pausas largas", "ruido": "ruido de fondo", "eventos": "ruidos sueltos", | |
| "zumbido": "zumbido eléctrico"} | |
| def level_used(fixes: set[str]) -> str | None: | |
| """Nivel más alto que han necesitado las correcciones aplicadas (None = no se ha tocado).""" | |
| levels = {LEVEL_OF.get(f) for f in fixes} - {None} | |
| for lv in ("ninja", "permitido", "formato"): | |
| if lv in levels: | |
| return lv | |
| return None | |
| def _fixes_for(rows, mode: str = "ninja") -> set[str]: | |
| fixes = {FIX_FOR[r.parametro] for r in rows if r.ok is False and r.parametro in FIX_FOR} | |
| allowed = MODES.get(mode) | |
| return fixes if allowed is None else fixes & allowed | |
| def default_mode(prof: dict) -> str: | |
| return (prof.get("procesado") or {}).get("modo_defecto") or "ninja" | |
| class Settings: | |
| reduccion_db: float | None = 24.0 | |
| paso_alto_hz: float | None = 70.0 | |
| conservar_ultrasonidos: bool = True | |
| forzar_mono: bool = False | |
| usar_gpu: bool = True | |
| escalar_reduccion: bool = True | |
| dnsmos: bool = True # mostrar la nota de calidad antes/después | |
| pulir: bool = True # pulido automático (solo se conserva si mejora la nota) | |
| acortar_pausas: bool = False # acortar pausas internas más largas que profiles.pause_max() | |
| modo: str = "ninja" # formato | permitido | ninja (ver MODES) | |
| quitar_eventos: bool = False # quitar ruidos sueltos en las pausas (cláxones, golpes…) | |
| def from_profile(cls, prof: dict, **overrides) -> "Settings": | |
| p = prof.get("procesado") or {} | |
| s = cls( | |
| reduccion_db=p.get("reduccion_db", 24.0), | |
| paso_alto_hz=p.get("paso_alto_hz", 70.0), | |
| conservar_ultrasonidos=p.get("conservar_ultrasonidos", True), | |
| forzar_mono=(prof.get("formato") or {}).get("canales") == 1, | |
| modo=default_mode(prof), | |
| ) | |
| for k, v in overrides.items(): | |
| if v is not None or k in ("reduccion_db", "paso_alto_hz"): | |
| setattr(s, k, v) | |
| return s | |
| class Analysis: | |
| path: str | |
| meta: AudioMeta | None | |
| m: dict = field(default_factory=dict) | |
| rows: list = field(default_factory=list) | |
| error: str | None = None | |
| def ok(self) -> bool: | |
| return self.error is None and profiles.passes(self.rows) | |
| class Correction: | |
| before: Analysis | |
| after: Analysis | None = None | |
| out_path: str | None = None | |
| notes: list = field(default_factory=list) | |
| error: str | None = None | |
| fixes: set = field(default_factory=set) # correcciones aplicadas de verdad | |
| pulido: bool = False # se ha conservado el pulido automático | |
| eventos: list = field(default_factory=list) # ruidos sueltos quitados: [(inicio_s, duración_s, dB)] | |
| def _needed_reduction(m: dict, prof: dict) -> float: | |
| """dB de reducción que haría falta, con margen, para cumplir los requisitos de ruido del perfil.""" | |
| nz = prof.get("ruido") or {} | |
| need = 0.0 | |
| if nz.get("snr_min_db") is not None and math.isfinite(m.get("snr_db", math.inf)): | |
| need = max(need, nz["snr_min_db"] - m["snr_db"] + 6) | |
| if nz.get("ruido_relativo_max_db") is not None and math.isfinite(m.get("ruido_relativo_db", -math.inf)): | |
| need = max(need, m["ruido_relativo_db"] - nz["ruido_relativo_max_db"] + 6) | |
| if nz.get("ruido_fondo_max_dbfs") is not None and math.isfinite(m.get("ruido_fondo_dbfs", -math.inf)): | |
| need = max(need, m["ruido_fondo_dbfs"] - nz["ruido_fondo_max_dbfs"] + 6) | |
| return need | |
| def analyze(path: str, prof: dict, with_dnsmos: bool = False) -> Analysis: | |
| try: | |
| data, sr, meta = read_audio(path) | |
| if data.shape[0] == 0: | |
| raise ValueError("El fichero está vacío") | |
| m = metrics.measure(data, sr, detector=prof.get("deteccion_voz")) | |
| if with_dnsmos: | |
| import dnsmos | |
| m["dnsmos"] = dnsmos.score(dsp.to_mono(data)[:, 0], sr) | |
| return Analysis(path, meta, m, profiles.check(meta, m, prof)) | |
| except Exception as e: | |
| return Analysis(path, None, error=f"{type(e).__name__}: {e}") | |
| def _voice_rms(a: np.ndarray, sr: int, prof: dict) -> float: | |
| mono = dsp.to_mono(a)[:, 0] | |
| r = dsp.voice_region(mono, sr, prof.get("deteccion_voz")) | |
| return dsp.rms_db(mono[r.start : r.end + 1]) if r.found else dsp.rms_db(mono) | |
| def _ceiling(prof: dict) -> float: | |
| peak_max = (prof.get("niveles") or {}).get("pico_max_dbfs") | |
| return (peak_max if peak_max is not None else -0.1) - 0.05 | |
| def _fix_volume(y: np.ndarray, sr: int, prof: dict, notes: list) -> np.ndarray: | |
| """Ganancia hasta el RMS de voz objetivo; el limitador solo actúa si algún pico pasa del techo.""" | |
| rv = (prof.get("niveles") or {}).get("rms_voz") or {} | |
| target = rv.get("objetivo", (rv["min"] + rv["max"]) / 2) | |
| total, lim = 0.0, None | |
| for _ in range(4): | |
| cur = _voice_rms(y, sr, prof) | |
| if not math.isfinite(cur) or abs(target - cur) < 0.05: | |
| break | |
| y = y * np.float32(10 ** ((target - cur) / 20)) | |
| total += target - cur | |
| y, lim = dsp.peak_limit(y, sr, _ceiling(prof)) | |
| notes.append(f"Volumen de la voz {total:+.1f} dB → {_voice_rms(y, sr, prof):.1f} dBFS (objetivo {target:.1f})".replace("-", "−")) | |
| if lim and lim["muestras_afectadas"]: | |
| notes.append(f"Picos suavizados para no pasar del límite ({lim['muestras_afectadas'] / sr * 1000:.0f} ms en total, " | |
| f"como mucho {lim['reduccion_max_db']:.1f} dB)") | |
| return y | |
| def _fix_peaks(y: np.ndarray, sr: int, prof: dict, notes: list, clipped: bool) -> np.ndarray: | |
| """Solo baja los picos que pasan del techo; el resto del audio no cambia.""" | |
| y, info = dsp.peak_limit(y, sr, _ceiling(prof)) | |
| if info["muestras_afectadas"]: | |
| notes.append(f"Picos suavizados para no pasar de {_ceiling(prof) + 0.05:.1f} dBFS " | |
| f"({info['muestras_afectadas'] / sr * 1000:.0f} ms en total, como mucho " | |
| f"{info['reduccion_max_db']:.1f} dB); el resto del audio no cambia".replace("-", "−")) | |
| if clipped: | |
| notes.append("Había muestras saturadas: se han bajado, pero si se oye distorsión conviene regrabar ese trozo") | |
| return y | |
| def _fix_amplitude(y: np.ndarray, sr: int, prof: dict, notes: list) -> np.ndarray: | |
| """Ganancia para que el pico de la voz quede holgado sobre el mínimo (1,6× el mínimo, ≈ +4 dB); | |
| el limitador solo actúa si algún pico pasa del techo.""" | |
| amin = (prof.get("niveles") or {}).get("amplitud_voz_min") | |
| mono = dsp.to_mono(y)[:, 0] | |
| r = dsp.voice_region(mono, sr, prof.get("deteccion_voz")) | |
| peak = float(np.abs(mono[r.start : r.end + 1] if r.found else mono).max()) | |
| if not amin or peak <= 0: | |
| return y | |
| target = min(1.6 * amin / 32767, 10 ** (_ceiling(prof) / 20)) | |
| g = 20 * math.log10(target / peak) | |
| y, lim = dsp.peak_limit(y * np.float32(10 ** (g / 20)), sr, _ceiling(prof)) | |
| notes.append(f"Volumen subido {g:+.1f} dB para que la voz pase de {amin} (ahora llega a ≈ {target * 32767:.0f})" | |
| .replace("-", "−")) | |
| if lim and lim["muestras_afectadas"]: | |
| notes.append(f"Picos suavizados para no pasar del límite ({lim['muestras_afectadas'] / sr * 1000:.0f} ms en total)") | |
| return y | |
| def _apply(data, sr, meta, prof, settings, fixes, att, expander, cache, progress): | |
| """Aplica al original solo las correcciones de `fixes`. Devuelve (audio, sr, notas).""" | |
| fmt = prof.get("formato") or {} | |
| notes, x = [], data | |
| if "mono" in fixes and x.shape[1] > 1: | |
| x = dsp.to_mono(x) | |
| notes.append(f"{meta.channels} canales → mono") | |
| if "frecuencia" in fixes and fmt.get("frecuencia") and fmt["frecuencia"] != sr: | |
| x = dsp.resample(x, sr, fmt["frecuencia"]) | |
| notes.append(f"Frecuencia de muestreo cambiada de {sr} a {fmt['frecuencia']} Hz") | |
| sr = fmt["frecuencia"] | |
| if "zumbido" in fixes: | |
| zb = metrics.hum(dsp.to_mono(data)[:, 0], meta.samplerate) | |
| if zb: | |
| x = dsp.remove_hum(x, sr, zb["armonicos"]) | |
| notes.append(f"Zumbido eléctrico de {zb['base']} Hz quitado (armónicos en " | |
| f"{', '.join(str(int(f)) for f in zb['armonicos'])} Hz); la voz no cambia") | |
| if "eventos" in fixes: | |
| if "eventos" not in cache: | |
| import vad | |
| if progress: | |
| progress(0.1, "Buscando ruidos sueltos en las pausas") | |
| cache["eventos"] = dsp.remove_events(x, sr, vad.speech_prob(dsp.to_mono(x)[:, 0], sr), vad.HOP_S) | |
| x, ev = cache["eventos"] | |
| if ev: | |
| from audio_io import fmt_duration | |
| lista = ", ".join(f"{fmt_duration(t)} ({d:.1f} s)" for t, d, _ in ev[:12]) + (" …" if len(ev) > 12 else "") | |
| notes.append(f"Ruidos sueltos en las pausas sustituidos por el silencio de la sala: {len(ev)} — {lista}") | |
| else: | |
| notes.append("Ruidos sueltos en las pausas: no se ha encontrado ninguno") | |
| if "ruido" in fixes: | |
| key = (sr, x.shape[1], att, "eventos" in fixes) | |
| if key not in cache: | |
| def prog(f): | |
| if progress: | |
| progress(f, f"Reduciendo ruido ({'sin límite' if att is None else f'{att:.0f} dB'})") | |
| xh = dsp.highpass(x, sr, settings.paso_alto_hz) if settings.paso_alto_hz else x | |
| cache[key] = dsp.denoise(xh, sr, att, keep_ultrasonic=settings.conservar_ultrasonidos, | |
| use_gpu=settings.usar_gpu, progress=prog) | |
| x, info = cache[key] | |
| if settings.paso_alto_hz: | |
| notes.append(f"Retumbes y golpes graves quitados (por debajo de {settings.paso_alto_hz:.0f} Hz)") | |
| notes.append(f"Ruido de fondo reducido ({'al máximo' if att is None else f'hasta {att:.0f} dB'})" | |
| f" · procesado en {info['dispositivo']}") | |
| notes += info["avisos"] | |
| if expander: | |
| x, ex = dsp.pause_expander(x, sr) | |
| notes.append(f"Ruido que quedaba en las pausas bajado 15 dB más ({ex['pausas_pct']:.0f} % del audio son pausas)") | |
| if "pausas" in fixes: | |
| region = dsp.voice_region(dsp.to_mono(x)[:, 0], sr, prof.get("deteccion_voz")) | |
| x, cut = dsp.shorten_pauses(x, sr, region, profiles.pause_max(prof)) | |
| if cut: | |
| notes.append(f"Pausas largas acortadas a {profiles.pause_max(prof):.1f} s: {len(cut)} " | |
| f"(duraban {', '.join(f'{c:.1f}' for c in cut)} s)") | |
| sides = tuple(s for s in ("ini", "fin") if f"sil_{s}" in fixes) | |
| sil = prof.get("silencios") or None | |
| if sides and sil: | |
| # Iterativo: el detector (sobre todo el del analizador v4) estima el ruido con los extremos del | |
| # audio, así que al cambiarlos puede mover un poco dónde «empieza» o «acaba» la voz. | |
| target = sil.get("objetivo_s", 0.6) | |
| first = True | |
| for _ in range(4): | |
| region = dsp.voice_region(dsp.to_mono(x)[:, 0], sr, prof.get("deteccion_voz")) | |
| if not region.found: | |
| notes.append("No se detecta voz: no se pueden ajustar los silencios") | |
| break | |
| head, tail = region.start / sr, (x.shape[0] - 1 - region.end) / sr | |
| pending = tuple(sd for sd in sides if abs((head if sd == "ini" else tail) - target) > 0.02) | |
| if not pending: | |
| break | |
| x, sn = dsp.set_edge_silences(x, sr, region, target, sil.get("relleno", "tono_sala"), pending) | |
| if first: | |
| notes += sn | |
| first = False | |
| if "volumen" in fixes and (prof.get("niveles") or {}).get("rms_voz"): | |
| x = _fix_volume(x, sr, prof, notes) | |
| elif "amplitud" in fixes: | |
| x = _fix_amplitude(x, sr, prof, notes) | |
| elif "picos" in fixes: | |
| x = _fix_peaks(x, sr, prof, notes, clipped=bool((np.abs(data) >= 0.999).any())) | |
| return x, sr, notes | |
| POLISH_MIN_GAIN = 0.05 # el pulido se queda solo si la nota global sube al menos esto… | |
| POLISH_MAX_SIG_LOSS = 0.05 # …y la nota de la voz no baja más que esto | |
| def _solve(data, sr, meta, prof, settings, fixes, container, subtype, before_m, progress): | |
| """ | |
| Aplica `fixes`, vuelve a medir y añade solo lo que haga falta si aparecen fallos nuevos | |
| (o sube la reducción de ruido si el ruido sigue sin cumplir). Devuelve (audio, sr, filas, notas). | |
| """ | |
| fixes = set(fixes) | |
| notes: list[str] = [] | |
| start = settings.reduccion_db | |
| if settings.escalar_reduccion and start is not None and "ruido" in fixes: | |
| need = _needed_reduction(before_m, prof) | |
| if need > start: | |
| start = start + 12 if need <= start + 12 else None | |
| ladder = [(start, False)] | |
| if settings.escalar_reduccion: | |
| if start is not None: | |
| ladder += [(lv, False) for lv in (start + 12, None) if lv != start] | |
| ladder.append((None, True)) | |
| cache: dict = {} | |
| step = 0 | |
| for _ in range(8): | |
| att, expander = ladder[step] | |
| y, out_sr, step_notes = _apply(data, sr, meta, prof, settings, fixes, att, expander, cache, progress) | |
| m = metrics.measure(y, out_sr, with_lufs=False, detector=prof.get("deteccion_voz")) | |
| rows = profiles.check(AudioMeta(meta.name, container, subtype, out_sr, y.shape[1], y.shape[0]), m, prof) | |
| new = _fixes_for(rows, settings.modo) - fixes | |
| noise_left = [r.parametro for r in rows if r.ok is False and r.parametro in NOISE_CHECKS] | |
| if new: | |
| notes.append("Al corregir aparecieron fallos nuevos (" | |
| + ", ".join(r.parametro for r in rows if r.ok is False and FIX_FOR.get(r.parametro) in new) | |
| + "): se corrigen también") | |
| fixes |= new | |
| continue | |
| if noise_left and "ruido" in fixes and step < len(ladder) - 1: | |
| step += 1 | |
| notes.append(f"Seguía sin cumplir «{', '.join(noise_left)}»: se insiste con más reducción de ruido") | |
| continue | |
| break | |
| return y, out_sr, rows, notes + step_notes, fixes, (cache.get("eventos") or (None, []))[1] | |
| def correct(path: str, prof: dict, settings: Settings, out_dir: Path, progress=None, | |
| before: Analysis | None = None) -> Correction: | |
| """ | |
| Corrige `path` según `prof`: | |
| 1. Cumplimiento: corrección mínima (solo lo que no cumple). | |
| 2. Pulido automático (opcional): quitar ruido de fondo y retumbes con criterio de estudio. Solo se | |
| conserva si la nota de calidad (DNSMOS) confirma que suena mejor y no empeora la voz. | |
| `before`: análisis ya hecho del original (se reutiliza para no medir dos veces). | |
| """ | |
| if before is None: | |
| before = analyze(path, prof) | |
| res = Correction(before) | |
| if before.error: | |
| res.error = before.error | |
| return res | |
| try: | |
| import dnsmos | |
| data, sr, meta = read_audio(path) | |
| notes = res.notes | |
| if (settings.dnsmos or settings.pulir) and "dnsmos" not in before.m: | |
| before.m["dnsmos"] = dnsmos.score(dsp.to_mono(data)[:, 0], sr) | |
| if settings.dnsmos: | |
| before.rows = profiles.check(meta, before.m, prof) | |
| fixes = _fixes_for(before.rows, settings.modo) | |
| if settings.modo != "ninja": | |
| settings.pulir = False # «auto» es lo mínimo que haga falta: sin pulido | |
| if settings.forzar_mono and meta.channels > 1: # pedido expresamente aunque el perfil no lo exija | |
| fixes.add("mono") | |
| if settings.acortar_pausas and profiles.long_pauses(before.m, prof): | |
| fixes.add("pausas") | |
| if settings.quitar_eventos: | |
| fixes.add("eventos") | |
| zb = before.m.get("zumbido") | |
| if settings.modo == "ninja" and zb and zb["db"] >= metrics.HUM_MIN_DB: | |
| fixes.add("zumbido") # en Ninja el zumbido se quita siempre que se detecte | |
| failing = profiles.failed(before.rows) | |
| out_dir.mkdir(parents=True, exist_ok=True) | |
| # Formato de escritura: el original salvo que falle el formato o la codificación. | |
| salida = (prof.get("procesado") or {}).get("salida") or {} | |
| if "formato" in fixes: | |
| container, subtype, ext = output_format(meta, salida.get("contenedor"), salida.get("subtipo")) | |
| else: | |
| container, subtype, ext = output_format(meta, None, None) | |
| if failing: | |
| stuck = [r.parametro for r in before.rows if r.ok is False and FIX_FOR.get(r.parametro) not in fixes] | |
| fixable = [f for f in failing if f not in stuck] | |
| if fixable: | |
| notes.append(f"No cumplía: {', '.join(fixable)} → se corrige solo eso") | |
| if stuck: | |
| why = [f for f in stuck if FIX_FOR.get(f)] | |
| notes.append(f"Nivel «{MODE_LABEL.get(settings.modo, settings.modo)}»: no se toca {', '.join(stuck)}" | |
| + (" (se corregiría subiendo el nivel de procesado)" if why else " (no tiene corrección automática)")) | |
| # 1) Cumplimiento | |
| base = _solve(data, sr, meta, prof, settings, fixes, container, subtype, before.m, progress) if fixes else None | |
| # 2) Pulido automático, solo si mejora de verdad | |
| chosen = base | |
| if settings.pulir and "ruido" not in fixes: | |
| if progress: | |
| progress(0.5, "Pulido automático") | |
| cand = _solve(data, sr, meta, prof, settings, fixes | {"ruido"}, container, subtype, before.m, progress) | |
| ref = dsp.to_mono(base[0])[:, 0] if base else dsp.to_mono(data)[:, 0] | |
| ref_sr = base[1] if base else sr | |
| s_ref = dnsmos.score(ref, ref_sr) if base else before.m.get("dnsmos") | |
| s_new = dnsmos.score(dsp.to_mono(cand[0])[:, 0], cand[1]) | |
| ref_ok = profiles.passes(base[2]) if base else profiles.passes(before.rows) | |
| if s_ref and s_new and (profiles.passes(cand[2]) or not ref_ok) \ | |
| and s_new["OVRL"] >= s_ref["OVRL"] + POLISH_MIN_GAIN and s_new["SIG"] >= s_ref["SIG"] - POLISH_MAX_SIG_LOSS: | |
| chosen = cand | |
| notes.append(f"Pulido automático aplicado: suena mejor (nota {s_ref['OVRL']:.2f} → {s_new['OVRL']:.2f})") | |
| elif s_ref and s_new: | |
| notes.append(f"Pulido automático probado y descartado: no mejoraba el sonido " | |
| f"(nota {s_ref['OVRL']:.2f} → {s_new['OVRL']:.2f})") | |
| if chosen is None: # cumple y el pulido no aporta: no se toca | |
| out_path = out_dir / Path(path).name | |
| if Path(path).resolve() != out_path.resolve(): | |
| import shutil | |
| shutil.copyfile(path, out_path) | |
| res.out_path, res.after = str(out_path), before | |
| notes.append("Cumplía todo: no se ha tocado" if before.ok else | |
| "No se ha tocado: lo que falla no se corrige con este nivel de procesado") | |
| return res | |
| y, out_sr, _, step_notes, res.fixes, res.eventos = chosen | |
| res.pulido = chosen is not base | |
| notes += step_notes | |
| out_path = out_dir / (Path(meta.name).stem + ext) | |
| write_audio(out_path, y, out_sr, container, subtype) | |
| res.out_path = str(out_path) | |
| res.after = analyze(str(out_path), prof, with_dnsmos=settings.dnsmos) | |
| if res.after.meta: | |
| notes.append(f"Guardado como {res.after.meta.describe()}") | |
| except Exception as e: | |
| res.error = f"{type(e).__name__}: {e}" | |
| return res | |
| def quick_meta(path: str) -> AudioMeta | None: | |
| try: | |
| return read_meta(path) | |
| except Exception: | |
| return None | |