"""AudioSR — wzbogacanie krótkich próbek głosu (HF Space, CPU/DARMOWY). UI Lektor AI.
Wejście: gr.Audio (upload) → BEZPOŚREDNIO do enhance (bez State — pewne okablowanie).
Zamiast pustej nuty: WYKRES WIDMA (wierzchołki/góra pasma) jako . Pasek postępu + ETA.
CPU: 1 klip ≈ ~2 min. Model „basic", domyślnie 24 kroki.
Tryb NO-FIDDLE: backend jest w pełni ustawiony (model='basic', guidance=3.5, seed=42,
latent_t_per_second=12.8). UŻYTKOWNIK steruje TYLKO dwoma rzeczami:
• suwak DDIM steps (jakość vs czas),
• przełącznik „Normalizuj" (włącza dwustopniową normalizację LUFS → sufit szczytu).
Całe czyszczenie + normalizacja (force-mono, DC, high-pass 75 Hz, LUFS -18, sufit -1 dB)
żyje w postproc.py (czysty DSP, testowalny lokalnie). app.py tylko orkiestruje."""
import os
import time
import threading
import tempfile
import numpy as np
import soundfile as sf
import gradio as gr
import torch
from audiosr import build_model, super_resolution
import viz
import postproc
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
OUTPUT_SR = 48000
# ── BACKEND HARDCODE (user tego NIE rusza) ───────────────────────────────────
MODEL_NAME = "basic" # „basic" — sprawdzony do tego głosu (vs „speech")
GUIDANCE_SCALE = 3.5 # classifier-free guidance — stały punkt pracy
SEED = 42 # stały seed → powtarzalny wynik dla tej samej próbki
LATENT_T_PER_SECOND = 12.8 # rozdzielczość czasowa latentu AudioSR (zalecana)
CYAN, ORANGE, MUTED = "#00ffff", "#ff9900", "#8a8aa0"
HINT = ("
"
"Wrzuć próbkę, aby zobaczyć widmo (wierzchołki / góra pasma)
")
_CACHE = {}
# ── Hook REALNEGO postępu AudioSR ─────────────────────────────────────────────
# AudioSR liczy dyfuzję w wątku; jedyny wiarygodny sygnał postępu to jego własny
# pasek tqdm „DDIM Sampler". Podpinamy się pod tqdm.update() i czytamy n/total
# TYLKO dla tego paska → UI pokazuje FAKTYCZNY krok zamiast zgadywanej ETA.
# (Na CPU krok ≈ 25 s, więc estymata czasowa była bezużyteczna i pasek „stał" na 95%.)
_SR_PROGRESS = {"frac": 0.0, "step": 0, "total": 0}
try:
from tqdm import std as _tqdm_std
_orig_tqdm_update = _tqdm_std.tqdm.update
def _tqdm_update_hook(self, n=1):
r = _orig_tqdm_update(self, n)
try:
if "DDIM" in (getattr(self, "desc", "") or "") and self.total:
_SR_PROGRESS["step"] = int(self.n)
_SR_PROGRESS["total"] = int(self.total)
_SR_PROGRESS["frac"] = max(0.0, min(0.999, self.n / self.total))
except Exception:
pass
return r
_tqdm_std.tqdm.update = _tqdm_update_hook
except Exception: # pragma: no cover - brak tqdm = zostaje fallback czasowy
pass
def _model(model_name: str):
if model_name not in _CACHE:
_CACHE[model_name] = build_model(model_name=model_name, device=DEVICE)
return _CACHE[model_name]
def on_audio(audio_path):
"""Po wrzuceniu/zmianie pliku → widmo wejścia (lub podpowiedź gdy puste)."""
if not audio_path:
return HINT
try:
return viz.spectrum_html([(audio_path, "wejście", CYAN)], "Widmo wejścia — wierzchołki (góra pasma)")
except Exception as e:
return f"
Nie udało się narysować widma: {e}
"
def enhance(audio_path, ddim_steps, norm, progress=gr.Progress()):
if not audio_path:
raise gr.Error("Najpierw wrzuć plik audio.")
progress(0.03, desc="Ładowanie modelu (pierwszy raz: pobieranie ~GB)…")
mdl = _model(MODEL_NAME)
res = {}
def work():
try:
res["w"] = super_resolution(mdl, audio_path, seed=SEED,
guidance_scale=GUIDANCE_SCALE,
ddim_steps=int(ddim_steps),
latent_t_per_second=LATENT_T_PER_SECOND)
except Exception as e:
res["err"] = e
# reset stanu postępu przed nowym przebiegiem
_SR_PROGRESS["frac"] = 0.0; _SR_PROGRESS["step"] = 0; _SR_PROGRESS["total"] = 0
t = threading.Thread(target=work, daemon=True)
t.start()
# fallback czasowy gdyby hook tqdm zawiódł: CPU-basic ≈ 25-28 s/krok
est = max(60.0, int(ddim_steps) * 28.0)
t0 = time.time()
while t.is_alive():
el = time.time() - t0
real = _SR_PROGRESS["frac"]
if real > 0: # mamy REALNY postęp dyfuzji
frac = real
eta = int(el / real * (1.0 - real))
desc = f"Wzbogacam… krok {_SR_PROGRESS['step']}/{_SR_PROGRESS['total']} · ETA ~{eta} s"
else: # przed startem samplera (ładowanie/inicjalizacja) — pełzanie czasowe
frac = min(0.90, el / est)
desc = f"Inicjalizacja dyfuzji… ETA ~{max(0, int(est - el))} s"
progress(min(0.97, 0.05 + 0.92 * frac), desc=desc)
time.sleep(0.4)
t.join()
if "err" in res:
raise gr.Error(f"Błąd AudioSR: {res['err']}")
progress(0.97, desc="Składanie wyniku…")
wav = np.asarray(res["w"]).squeeze()
if wav.ndim > 1:
wav = wav.T
# Przytnij padding AudioSR do długości wejścia (frames * OUTPUT_SR / src_sr).
# UWAGA: dla mp3 (główne wejście, „spłaszczone mp3") sf.info().frames bywa
# PRZYBLIŻONE lub 0 na niektórych buildach libsndfile → tnij TYLKO gdy wynik jest
# WYRAŹNIE dłuższy (>1%), żeby nie ściąć kilku ważnych próbek ogona; przy frames=0
# warunek `exp` i tak pomija trim.
info = sf.info(audio_path)
exp = int(round(info.frames / info.samplerate * OUTPUT_SR))
if exp and wav.shape[0] > int(exp * 1.01):
wav = wav[:exp]
# Czyszczenie + normalizacja → JEDNO wejście postproc (czysty DSP, testowalny).
# norm=True : mono → DC → high-pass 75 Hz → LUFS -18 → sufit szczytu -1 dB.
# norm=False : mono → dopasowanie RMS do oryginału + bezpiecznik 0.99.
wav, _pp = postproc.process(wav, OUTPUT_SR, audio_path, normalize=bool(norm))
base = os.path.splitext(os.path.basename(audio_path))[0]
out_path = os.path.join(tempfile.gettempdir(), f"{base}_48k.wav")
# Jawny subtype: WAV 16-bit PCM (w spec agy: „16-bit PCM lub 32-bit float").
# Sufit szczytu -1 dB istnieje WŁAŚNIE dlatego, że kontener obcina na ±1.0 —
# nie rozluźniaj go zakładając, że plik jest float.
sf.write(out_path, wav.astype(np.float32), OUTPUT_SR, subtype="PCM_16")
# Anty-silent-skip: jeśli przy normalizacji cel LUFS -18 NIE został osiągnięty
# (pojedynczy głośny transjent — plozja/klik — wymusił zejście sufitem szczytu),
# POWIEDZ to wprost zamiast cicho udawać sukces.
warn = ""
short_db = _pp.get("lufs_shortfall_db", 0.0) if isinstance(_pp, dict) else 0.0
if bool(norm) and short_db > 3.0:
warn = ("
Uwaga: docelowy poziom LUFS -18 nie został "
f"w pełni osiągnięty (brak ~{short_db:.1f} dB) — pojedynczy głośny transjent "
"(plozja/klik) wymusił zejście sufitem szczytu -1 dB. Rozważ ręczne wyciszenie "
"tego transjentu w Audacity i ponów.