Implementación completa de audio_pipeline y app Gradio

app.py

@@ -1,3 +1,5 @@
+# app.py
+
 import os
 import gradio as gr
 from audio_pipeline import (
@@ -8,22 +10,29 @@ from audio_pipeline import (
 )
 
 def procesar_wav(input_wav_path):
-    separar_audio_demucs(input_wav_path, output_dir="separated")
-    base_name = os.path.splitext(os.path.basename(input_wav_path))[0]
-    stems_dir = os.path.join("separated", base_name)
+    # 1) Separar con Demucs y obtener carpeta de stems
+    stems_dir = separar_audio_demucs(input_wav_path)
+
+    # 2) Limpiar stems (_cleaned.wav)
     limpiar_stems(stems_dir)
+
+    # 3) Combinar stems sin vocales en base_instrumental.wav
     combinar_stems_sin_vocales(stems_dir)
-    wav_base = os.path.join(stems_dir, "base_instrumental.wav")
-    wav_clean = os.path.join(stems_dir, "base_instrumental_clean.wav")
-    reducir_ruido(wav_base, wav_clean)
-    return wav_clean
+
+    # 4) Reducir ruido sobre la base resultante
+    base_instru = os.path.join(stems_dir, "base_instrumental.wav")
+    clean_instru = os.path.join(stems_dir, "base_instrumental_clean.wav")
+    reducir_ruido(base_instru, clean_instru)
+
+    # 5) Devolver la ruta del WAV limpio
+    return clean_instru
 
 demo = gr.Interface(
     fn=procesar_wav,
     inputs=gr.Audio(label="Sube un archivo .wav", type="filepath"),
     outputs=gr.Audio(label="Base instrumental limpia", type="filepath"),
     title="Procesador de WAV a Base Instrumental",
-    description="Sube tu archivo WAV, separo stems, limpio vocales y reduzco ruido."
+    description="Sube tu WAV y obtén su base limpia sin vocales."
 )
 
 if __name__ == "__main__":

audio_pipeline.py

@@ -1,124 +1,87 @@
-import yt_dlp
+# audio_pipeline.py
+
+import os
 import subprocess
 import sys
-import os
+import torch
 import librosa
 import numpy as np
 import soundfile as sf
 import noisereduce as nr
-import scipy.signal as signal
 
-# ###########################################
-# Funciones de Filtros y Efectos
-# ###########################################
+# Directorio base donde guardaremos todos los stems
+BASE_STEMS_DIR = "data/stems"
 
 def highpass_filter(audio, sr, cutoff=80, order=4):
     """Aplica un filtro Butterworth de paso alto al audio."""
     sos = signal.butter(order, cutoff, btype='highpass', fs=sr, output='sos')
-    filtered = signal.sosfilt(sos, audio)
-    return filtered
-
-# ###########################################
-# Etapa 1: Descarga y Separación con Demucs
-# ###########################################
-
-def descargar_audio(url, output_filename='audio'):
-    """Descarga el audio de YouTube y lo convierte a WAV."""
-    opciones = {
-        'format': 'bestaudio/best',
-        'outtmpl': f'{output_filename}.%(ext)s',
-        'postprocessors': [{
-            'key': 'FFmpegExtractAudio',
-            'preferredcodec': 'wav',
-        }],
-    }
-    with yt_dlp.YoutubeDL(opciones) as ydl:
-        ydl.download([url])
-
-
-def separar_audio_demucs(input_file, output_dir="separated"):
-    """Separa stems usando Demucs."""
-    comando = [
-        sys.executable, '-m', 'demucs',
+    return signal.sosfilt(sos, audio)
+
+def separar_audio_demucs(input_file, model="htdemucs"):
+    """
+    Separa stems usando Demucs, los deja en:
+      data/stems/<modelo>/<base_name>/
+    y devuelve la ruta al folder con los stems.
+    """
+    # 1) Crear directorio raíz para ese modelo
+    out_root = os.path.join(BASE_STEMS_DIR, model)
+    os.makedirs(out_root, exist_ok=True)
+
+    # 2) Ejecutar Demucs (en GPU si está disponible)
+    device = "cuda" if torch.cuda.is_available() else "cpu"
+    cmd = [
+        sys.executable, "-m", "demucs",
         input_file,
-        '--out', output_dir
+        "--out", out_root,
+        "--device", device
     ]
-    try:
-        subprocess.run(comando, check=True)
-        print("Separación con Demucs completada.")
-    except subprocess.CalledProcessError as e:
-        print(f"Error durante la separación con Demucs: {e}")
-        sys.exit(1)
+    subprocess.run(cmd, check=True)
 
-# ###########################################
-# Función para limpiar cada stem (reducción de ruido, etc.)
-# ###########################################
+    # 3) Detectar carpeta generada con los stems
+    base = os.path.splitext(os.path.basename(input_file))[0]
+    stems_dir = os.path.join(out_root, base)
+    if not os.path.isdir(stems_dir):
+        raise FileNotFoundError(f"No existe el folder de stems en {stems_dir}")
+    return stems_dir
 
 def limpiar_stems(stems_dir):
-    """Aplica reducción de ruido a cada stem en stems_dir."""
+    """Aplica reducción de ruido a cada stem dentro de stems_dir."""
     for archivo in os.listdir(stems_dir):
-        if archivo.endswith('.wav'):
-            file_path = os.path.join(stems_dir, archivo)
-            y, sr = librosa.load(file_path, sr=None)
+        if archivo.endswith(".wav"):
+            ruta = os.path.join(stems_dir, archivo)
+            y, sr = librosa.load(ruta, sr=None)
             reduced = nr.reduce_noise(y=y, sr=sr)
-            sf.write(file_path.replace('.wav', '_cleaned.wav'), reduced, sr)
-
-# ###########################################
-# Combina stems excluyendo la parte vocal
-# ###########################################
+            sf.write(ruta.replace(".wav", "_cleaned.wav"), reduced, sr)
 
 def combinar_stems_sin_vocales(stems_dir):
-    """Mezcla todos los stems excepto los que contienen vocales."""
-    archivos = [f for f in os.listdir(stems_dir) if f.endswith('.wav')]
-    archivos_clean = [f for f in archivos if 'cleaned' in f.lower() and 'vocal' not in f.lower()]
-    if archivos_clean:
-        lista_archivos = archivos_clean
-    else:
-        lista_archivos = [f for f in archivos if 'vocal' not in f.lower()]
+    """Mezcla todos los stems (limpios) excepto los de vocales."""
+    # Primero, buscar stems *_cleaned.wav que no contengan 'vocal'
+    wavs = [
+        f for f in os.listdir(stems_dir)
+        if f.endswith(".wav") and
+           "cleaned" in f.lower() and
+           "vocal" not in f.lower()
+    ]
+    # Si no hay cleaned, fallback a stems originales sin 'vocal'
+    if not wavs:
+        wavs = [
+            f for f in os.listdir(stems_dir)
+            if f.endswith(".wav") and "vocal" not in f.lower()
+        ]
+
     signals = []
-    for archivo in lista_archivos:
-        file_path = os.path.join(stems_dir, archivo)
-        print(f"Incluyendo: {archivo}")
-        y, sr = librosa.load(file_path, sr=None)
+    for w in wavs:
+        y, sr = librosa.load(os.path.join(stems_dir, w), sr=None)
         signals.append(y)
     if not signals:
-        print("No se encontraron stems para combinar (excluyendo vocales).")
-        return
-    max_len = max(len(s) for s in signals)
-    mezclado = sum(np.pad(s, (0, max_len - len(s))) for s in signals) / len(signals)
-    sf.write(os.path.join(stems_dir, 'base_instrumental.wav'), mezclado, sr)
-
-# ###########################################
-# Reducción de ruido en archivo de audio
-# ###########################################
+        raise RuntimeError("No se encontraron stems para combinar.")
+    maxlen = max(len(s) for s in signals)
+    mix = sum(np.pad(s, (0, maxlen - len(s))) for s in signals) / len(signals)
+    sf.write(os.path.join(stems_dir, "base_instrumental.wav"), mix, sr)
 
 def reducir_ruido(input_file, output_file, noise_duration=0.5):
     """Aplica reducción de ruido basada en los primeros segundos de audio."""
     y, sr = librosa.load(input_file, sr=None)
-    noise = y[:int(noise_duration * sr)]
+    noise = y[:int(sr * noise_duration)]
     reduced = nr.reduce_noise(y=y, sr=sr, y_noise=noise)
     sf.write(output_file, reduced, sr)
-
-# ###########################################
-# Función principal
-# ###########################################
-
-def main():
-    url = input("Introduce la URL de YouTube: ")
-    output_name = input("Nombre base para archivos (sin extensión): ")
-    descargar_audio(url, output_filename=output_name)
-    audio_file = f"{output_name}.wav"
-    separar_audio_demucs(audio_file, output_dir="separated")
-    stems_dir = os.path.join("separated", output_name)
-    limpiar_stems(stems_dir)
-
-    output_base = os.path.join(stems_dir, 'base_instrumental.wav')
-    print("Combinando stems 'cleaned' para generar la base instrumental...")
-    combinar_stems_sin_vocales(stems_dir)
-
-    output_clean = os.path.join(stems_dir, 'base_instrumental_clean.wav')
-    print("Aplicando reducción de ruido...")
-    reducir_ruido(output_base, output_clean, noise_duration=0.5)
-
-    # Se omite la etapa de mastering para mantener la calidad original de la mezcla
-    print("Proceso completado. Revisa el archivo 'base_instrumental_clean.wav' para escuchar el resultado final.")