Detectar dinámicamente carpeta de stems de Demucs 6stems

audio_pipeline.py

@@ -1,5 +1,3 @@
-# audio_pipeline.py
-
 import os
 import subprocess
 import sys
@@ -20,9 +18,8 @@ BASE_STEMS_DIR = "data/stems"
 def separar_audio_demucs_6stems(input_file, model="htdemucs_6s"):
     """
     Separa 6 stems con Demucs (vocals, drums, bass, guitar, piano, other),
-    los guarda en data/stems/htdemucs_6s/<base>/ y devuelve esa ruta.
+    los guarda en data/stems/<model>/<track_folder>/ y devuelve la ruta de ese folder.
     """
-    base = os.path.splitext(os.path.basename(input_file))[0]
     out_root = os.path.join(BASE_STEMS_DIR, model)
     os.makedirs(out_root, exist_ok=True)
 
@@ -37,10 +34,16 @@ def separar_audio_demucs_6stems(input_file, model="htdemucs_6s"):
     ]
     subprocess.run(cmd, check=True)
 
-    stems_dir = os.path.join(out_root, base)
-    if not os.path.isdir(stems_dir):
-        raise FileNotFoundError(f"No se encontró el folder de stems en {stems_dir}")
-    return stems_dir
+    # Demucs crea un subdirectorio con el nombre de la pista dentro de out_root
+    # Encuentra el primer subdirectorio que contenga archivos .wav
+    for entry in os.listdir(out_root):
+        candidate = os.path.join(out_root, entry)
+        if os.path.isdir(candidate):
+            # Verifica que tenga stems
+            wavs = [f for f in os.listdir(candidate) if f.endswith('.wav')]
+            if wavs:
+                return candidate
+    raise FileNotFoundError(f"No se encontró el folder de stems en {out_root}")
 
 def limpiar_stems(stems_dir):
     """Aplica reducción de ruido a cada stem (_cleaned.wav)."""
@@ -52,19 +55,101 @@ def limpiar_stems(stems_dir):
             sf.write(ruta.replace(".wav", "_cleaned.wav"), reduced, sr)
 
 def combinar_stems_sin_vocales(stems_dir):
-    """Mezcla todos los stems limpios excepto 'vocals' en base_instrumental.wav."""
-    # buscamos *_cleaned.wav excepto vocals
+    """Mezcla todos los stems limpios excepto vocals en base_instrumental.wav."""
     wavs = [
         f for f in os.listdir(stems_dir)
         if f.endswith("_cleaned.wav") and "vocals" not in f.lower()
     ]
     if not wavs:
-        # fallback a stems originales sin vocals
         wavs = [
             f for f in os.listdir(stems_dir)
             if f.endswith(".wav") and "vocals" not in f.lower()
         ]
+    signals = []
+    for w in wavs:
+        y, sr = librosa.load(os.path.join(stems_dir, w), sr=None)
+        signals.append(y)
+    if not signals:
+        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):
+    """Reduce ruido y guarda el resultado."""
+    y, sr = librosa.load(input_file, sr=None)
+    noise = y[:int(sr * noise_duration)]
+    with np.errstate(divide='ignore', invalid='ignore'):
+        reduced = nr.reduce_noise(y=y, sr=sr, y_noise=noise)
+    reduced = np.nan_to_num(reduced)
+    sf.write(output_file, reduced, sr)
+import os
+import subprocess
+import sys
+import warnings
+import torch
+
+import librosa
+import numpy as np
+import soundfile as sf
+import noisereduce as nr
+
+# Suprime warnings de runtime (p.ej. invalid value encountered in divide)
+warnings.filterwarnings("ignore", category=RuntimeWarning)
+
+# Directorio base donde guardaremos todos los stems
+BASE_STEMS_DIR = "data/stems"
+
+def separar_audio_demucs_6stems(input_file, model="htdemucs_6s"):
+    """
+    Separa 6 stems con Demucs (vocals, drums, bass, guitar, piano, other),
+    los guarda en data/stems/<model>/<track_folder>/ y devuelve la ruta de ese folder.
+    """
+    out_root = os.path.join(BASE_STEMS_DIR, model)
+    os.makedirs(out_root, exist_ok=True)
+
+    device = "cuda" if torch.cuda.is_available() else "cpu"
+    cmd = [
+        sys.executable,
+        "-m", "demucs",
+        "-n", model,
+        "--out", out_root,
+        "--device", device,
+        input_file
+    ]
+    subprocess.run(cmd, check=True)
+
+    # Demucs crea un subdirectorio con el nombre de la pista dentro de out_root
+    # Encuentra el primer subdirectorio que contenga archivos .wav
+    for entry in os.listdir(out_root):
+        candidate = os.path.join(out_root, entry)
+        if os.path.isdir(candidate):
+            # Verifica que tenga stems
+            wavs = [f for f in os.listdir(candidate) if f.endswith('.wav')]
+            if wavs:
+                return candidate
+    raise FileNotFoundError(f"No se encontró el folder de stems en {out_root}")
+
+def limpiar_stems(stems_dir):
+    """Aplica reducción de ruido a cada stem (_cleaned.wav)."""
+    for archivo in os.listdir(stems_dir):
+        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(ruta.replace(".wav", "_cleaned.wav"), reduced, sr)
 
+def combinar_stems_sin_vocales(stems_dir):
+    """Mezcla todos los stems limpios excepto vocals en base_instrumental.wav."""
+    wavs = [
+        f for f in os.listdir(stems_dir)
+        if f.endswith("_cleaned.wav") and "vocals" not in f.lower()
+    ]
+    if not wavs:
+        wavs = [
+            f for f in os.listdir(stems_dir)
+            if f.endswith(".wav") and "vocals" not in f.lower()
+        ]
     signals = []
     for w in wavs:
         y, sr = librosa.load(os.path.join(stems_dir, w), sr=None)
@@ -76,7 +161,7 @@ def combinar_stems_sin_vocales(stems_dir):
     sf.write(os.path.join(stems_dir, "base_instrumental.wav"), mix, sr)
 
 def reducir_ruido(input_file, output_file, noise_duration=0.5):
-    """Reduce ruido del WAV de entrada y guarda en output_file."""
+    """Reduce ruido y guarda el resultado."""
     y, sr = librosa.load(input_file, sr=None)
     noise = y[:int(sr * noise_duration)]
     with np.errstate(divide='ignore', invalid='ignore'):