#!/usr/bin/env python3 # ============================================================================ # MOLE · prepara stems para el XDJ-RX3 # https://get.molemusic.es # # Convierte una voz ya separada (UVR, MDX, el programa que uses) en el # fichero .rx3stem que entiende la mesa, con el nombre exacto que espera # y directamente en la carpeta RX3_STEMS del USB. # # Uso: python mole-stems.py voz.wav "E:\Contents\cancion.mp3" # Ayuda: python mole-stems.py --help # # La forma cómoda de hacer todo esto es MOLE Studio, que separa, nombra y # coloca cada stem por ti: https://get.molemusic.es # # © 2026 MOLE (molemusic.es) — uso libre para clientes de MOLE. # Se entrega tal cual, sin garantías y sin soporte oficial. # ============================================================================ """Prepara un .rx3stem a partir de una voz ya separada y la canción original.""" from __future__ import annotations import argparse import math import os import pathlib import re import shutil import struct import subprocess import sys import tempfile import unicodedata from dataclasses import dataclass # --------------------------------------------------------------------------- # El formato .rx3stem: una cabecera fija y el audio en crudo. # --------------------------------------------------------------------------- CABECERA = struct.Struct("<8sIIIIQ32s") MAGIA = b"RX3STM1\0" FRECUENCIA = 44100 # la mesa reproduce a 44.1 kHz, siempre CANALES = 2 # estéreo ID_FORMATO = 2 # int16 intercalado: el único formato en circulación FORMATO_FFMPEG = "s16le" BYTES_POR_FRAME = 4 # 2 canales × 2 bytes # El pico se mide en coma flotante ANTES de convertir a int16: si se midiera # después, las muestras ya llegarían recortadas al máximo y el exceso que # queremos medir quedaría escondido. MEDIDOR = "aformat=sample_fmts=fltp,astats=reset=0" # astats da una cifra por canal y otra global; nos quedamos con la más alta. NIVEL_PICO = re.compile(r"Peak level dB:\s*(-?\d+(?:\.\d+)?|-?inf)") MUESTRAS_RECORTADAS = re.compile(r"Number of clipped samples:\s*(\d+)") # Un MP3 o AAC declara las muestras de relleno que metió su codificador y # ffmpeg las quita al decodificar. La mesa no las quita: su posición cero es # la primera muestra que emite su propio decodificador, relleno incluido. Una # voz cortada sobre la rejilla de ffmpeg quedaría adelantada respecto a lo que # suena en la mesa, y con 25 ms de desfase la resta deja la voz entera de # fondo. Este flag pide la decodificación SIN quitar el relleno, que es la # rejilla en la que trabaja la mesa. SIN_RECORTE = ("-flags2", "+skip_manual") # Ningún codificador mete ni de lejos un segundo de relleno; un desfase mayor # es una medición fallida, no relleno. RETARDO_MAXIMO = FRECUENCIA # Patrón largo para que no se repita por casualidad, corto para leerlo rápido. FRAMES_PATRON = 4096 # Puntos de la canción donde buscar ese patrón. El silencio encaja en # cualquier parte, así que un patrón que no sea único se descarta y se prueba # en el siguiente punto. POSICIONES_PATRON = (0.5, 0.25, 0.75, 0.1) # Los procesos hijos van sin consola propia (en Windows evita ventanas que # parpadean) y con la entrada cerrada, para que nada se quede esperando. PROCESO_SILENCIOSO: dict = {"stdin": subprocess.DEVNULL} if sys.platform == "win32": PROCESO_SILENCIOSO["creationflags"] = subprocess.CREATE_NO_WINDOW # --------------------------------------------------------------------------- # El nombre de fichero, calculado igual que lo hace MOLE Studio. # --------------------------------------------------------------------------- # Rekordbox recorta el nombre a 44 caracteres al exportar al USB, y la mesa # busca el .rx3stem con el nombre YA recortado. Los espacios finales que deja # el corte se respetan: después de cortar no se limpia nada. LIMITE_NOMBRE = 44 # Carpetas que delatan la raíz de un USB preparado para la mesa. MARCAS_USB = ("RX3_STEMS", "Contents", "PIONEER") class MoleError(Exception): """Un problema esperado, contado en cristiano y sin traza.""" def _nombre_base(valor: str) -> str: """Reduce un nombre a lo que la mesa es capaz de emparejar.""" valor = unicodedata.normalize("NFC", valor).replace("/", "_").replace(":", "_") valor = "".join(letra for letra in valor if ord(letra) >= 32).strip(" .") return valor[:180] or "track" def nombre_rx3stem(audio: pathlib.Path) -> str: """El nombre de .rx3stem que la mesa espera para este audio.""" return (_nombre_base(audio.stem)[:LIMITE_NOMBRE] or "track") + ".rx3stem" def _raiz_usb(cancion: pathlib.Path) -> pathlib.Path | None: """La raíz del USB donde vive la canción, si se puede deducir sola.""" # Primero, una carpeta madre con pinta de USB de Rekordbox. for carpeta in cancion.parents: if any((carpeta / marca).is_dir() for marca in MARCAS_USB): return carpeta if sys.platform == "win32": # Una unidad con letra que no sea la del sistema (C:) cuenta como USB. ancla = cancion.anchor if not ancla: return None sistema = os.environ.get("SystemDrive", "C:").rstrip("\\/").upper() if ancla.rstrip("\\/").upper() == sistema: return None return pathlib.Path(ancla) # macOS monta en /Volumes/; Linux en /media, /run/media o /mnt. for carpeta in cancion.parents: padre = str(carpeta.parent) if padre in ("/Volumes", "/media", "/mnt"): return carpeta abuelo = str(carpeta.parent.parent) if abuelo in ("/media", "/run/media"): return carpeta return None # --------------------------------------------------------------------------- # El trabajo con ffmpeg. # --------------------------------------------------------------------------- def _localizar_ffmpeg(indicado: str) -> str: encontrado = shutil.which(str(indicado)) if encontrado: return encontrado if pathlib.Path(indicado).is_file(): return str(indicado) raise MoleError( "no encuentro ffmpeg (el conversor de audio que hace el trabajo duro).\n" " · Windows: instálalo con «winget install ffmpeg» o descarga el\n" " «release full» de https://www.gyan.dev/ffmpeg/builds/ y abre\n" " una terminal nueva.\n" " · Mac: «brew install ffmpeg» (https://brew.sh)\n" " · Todas las descargas oficiales: https://ffmpeg.org/download.html\n" " · Si ya lo tienes descargado, dime dónde está:\n" " --ffmpeg \"C:\\ruta\\a\\ffmpeg.exe\"" ) def _ffmpeg_pasada(ffmpeg: str, argumentos: list, con_informe: bool = False) -> str: """Una pasada de ffmpeg; devuelve su registro cuando hay que medir algo.""" nivel = "info" if con_informe else "error" orden = [str(ffmpeg), "-hide_banner", "-loglevel", nivel, "-y", *argumentos] try: proceso = subprocess.run( orden, capture_output=True, text=True, **PROCESO_SILENCIOSO ) except OSError as fallo: raise MoleError(f"no he podido arrancar ffmpeg: {fallo}") from None if proceso.returncode: detalle = " ".join((proceso.stderr or proceso.stdout).split())[-260:] raise MoleError( "ffmpeg no ha podido con uno de los audios " f"(código {proceso.returncode}). Su explicación: {detalle}" ) return proceso.stderr or "" def _pico_medido(informe: str) -> float | None: """La muestra más alta que vio astats, o None si no dijo nada. Se mide antes de la conversión final, así que un audio con pérdida que se reconstruye por encima del máximo también se detecta. """ niveles = [ -math.inf if valor.endswith("inf") else float(valor) for valor in NIVEL_PICO.findall(informe) ] if not niveles: return None mayor = max(niveles) return 0.0 if mayor == -math.inf else 10.0 ** (mayor / 20.0) def _colchon_codificador(recortado: pathlib.Path, entero: pathlib.Path) -> int | None: """Frames de relleno que ffmpeg quitó al principio, o None si no se ve. Los dos ficheros llevan la misma decodificación, uno con el relleno declarado quitado y otro sin quitar, así que el desfase se encuentra buscando un trozo del primero dentro del segundo en vez de fiarse de una marca de tiempo. Los primeros frames difieren en cualquier caso, por eso el patrón se toma de más adentro. """ frames_recortado = recortado.stat().st_size // 4 extra = entero.stat().st_size // 4 - frames_recortado if extra == 0: return 0 if extra < 0: return None limite = min(extra, RETARDO_MAXIMO) with recortado.open("rb") as origen, entero.open("rb") as referencia: for fraccion in POSICIONES_PATRON: posicion = int(frames_recortado * fraccion) if posicion + FRAMES_PATRON > frames_recortado: continue origen.seek(posicion * 4) patron = origen.read(FRAMES_PATRON * 4) referencia.seek(posicion * 4) ventana = referencia.read((limite + FRAMES_PATRON) * 4) desfase = ventana.find(patron) # Un desfase fuera de la rejilla de frames, o un patrón que se # repite, no demuestran nada; el silencio provoca las dos cosas. if desfase < 0 or desfase % 4 or ventana.find(patron, desfase + 4) >= 0: continue return desfase // 4 return None @dataclass(frozen=True) class Resultado: salida: pathlib.Path frames: int segundos: float bytes_audio: int # Factor aplicado para devolver la voz al volumen del fichero original. ganancia: float = 1.0 # Muestras que no cupieron una vez aplicado ese factor. recortadas: int = 0 # Relleno del codificador que hubo que devolver al principio de la voz. colchon: int = 0 # False cuando ese relleno no se pudo medir y la voz quedó en la rejilla # del separador (correcto solo para audios sin relleno, como WAV o FLAC). alineado: bool = True def escribir_rx3stem( voz: pathlib.Path, salida: pathlib.Path, cancion: pathlib.Path, *, ffmpeg: str = "ffmpeg", pico_normalizado: float | None = None, avisa=lambda texto: None, ) -> Resultado: """Convierte la voz al terreno de la mesa y escribe el .rx3stem. La voz se alinea SIEMPRE contra la canción original tal como la decodifica la propia mesa: el número de frames sale de esa decodificación, y el relleno de codificador que ffmpeg quitó camino del separador se devuelve al principio de la voz. `pico_normalizado` es el pico al que el separador normalizó la canción antes de separar, cuando lo hace (UVR y MDX suelen). La voz que devuelve arrastra ese mismo cambio de volumen, pero la mesa la resta de la canción SIN tocar; deshacerlo aquí deja a las dos en el mismo volumen. Se mide en vez de suponer, porque hay audios que decodifican por encima del máximo y son justo los que el separador reescala. """ with tempfile.TemporaryDirectory(prefix="mole-stems-") as temporal: mesa_de_trabajo = pathlib.Path(temporal) ganancia = 1.0 colchon = 0 alineado = True avisa("Leyendo la canción original (es la que alinea la voz)…") def decodifica_cancion(nombre: str, entera: bool) -> tuple: crudo = mesa_de_trabajo / nombre # El pico se mide después del cambio de frecuencia y antes de la # conversión a int16, que taparía cualquier exceso. informe = _ffmpeg_pasada(ffmpeg, [ *(SIN_RECORTE if entera else ()), "-i", str(cancion), "-map", "0:a:0", "-vn", "-af", f"aresample={FRECUENCIA},{MEDIDOR}", "-ar", str(FRECUENCIA), "-ac", str(CANALES), "-f", "s16le", str(crudo), ], con_informe=True) talla = crudo.stat().st_size if not talla or talla % 4: raise MoleError( "la canción original decodifica vacía o rota; " "comprueba que es un fichero de audio normal" ) return crudo, informe # La rejilla del separador (por la que volvió la voz) y la rejilla de # la mesa (contra la que se indexa el .rx3stem). recortada, informe_recortada = decodifica_cancion("cancion-a.s16le", entera=False) desfase: int | None = None try: entera, informe_entera = decodifica_cancion("cancion-b.s16le", entera=True) except MoleError: # Un ffmpeg tan viejo que no conoce el flag, no un fichero roto: # la misma decodificación acaba de funcionar sin él. entera = None else: desfase = _colchon_codificador(recortada, entera) if desfase is None: # Quedarse en la rejilla del separador es correcto cuando el audio # no declara relleno (WAV, FLAC); si lo declara y no se ha podido # medir, mejor eso que una corrección equivocada. alineado = False cruda_cancion, informe_medido = recortada, informe_recortada if entera is not None: entera.unlink() else: colchon = desfase cruda_cancion, informe_medido = entera, informe_entera recortada.unlink() frames_objetivo = cruda_cancion.stat().st_size // 4 pico = _pico_medido(informe_medido) # Sin medida fiable no se toca nada: una corrección equivocada es # peor que dejar la voz como vino. if pico_normalizado and pico and pico > pico_normalizado: ganancia = pico / pico_normalizado avisa("Convirtiendo y alineando la voz…") cruda_voz = mesa_de_trabajo / "voz.s16le" argumentos = ["-i", str(voz), "-map", "0:a:0", "-vn"] cadena = [] if ganancia == 1.0 else [f"volume={ganancia:.9f}:precision=float"] cadena.append(f"aresample={FRECUENCIA}") # Silencio exacto por el relleno que la mesa reproducirá antes de la # primera muestra que el separador llegó a ver. if colchon: cadena.append(f"adelay=delays={colchon}S:all=1") cadena.extend(["apad", f"atrim=end_sample={frames_objetivo}"]) argumentos.extend(["-ac", str(CANALES)]) cadena.append(MEDIDOR) argumentos.extend(["-af", ",".join(cadena)]) argumentos.extend(["-f", FORMATO_FFMPEG, str(cruda_voz)]) informe_voz = _ffmpeg_pasada(ffmpeg, argumentos, con_informe=True) # Solo se pierden las muestras en las que la voz sola supera el # máximo; sin la corrección estarían mal todas. recortadas = max( (int(valor) for valor in MUESTRAS_RECORTADAS.findall(informe_voz)), default=0 ) talla = cruda_voz.stat().st_size if not talla or talla % BYTES_POR_FRAME: raise MoleError("la voz decodifica vacía o rota; ¿seguro que es un audio?") frames = talla // BYTES_POR_FRAME if frames != frames_objetivo: raise MoleError( "la voz no ha quedado del mismo largo que la canción " f"({frames} frames frente a {frames_objetivo})" ) avisa(f"Escribiendo {salida.name}…") cabecera = CABECERA.pack( MAGIA, FRECUENCIA, CANALES, ID_FORMATO, CABECERA.size, frames, b"\0" * 32 ) salida.parent.mkdir(parents=True, exist_ok=True) try: with salida.open("wb") as destino, cruda_voz.open("rb") as audio: destino.write(cabecera) while trozo := audio.read(1024 * 1024): destino.write(trozo) except BaseException: # Un fichero a medias en el USB solo confunde: fuera. try: salida.unlink() except OSError: pass raise return Resultado( salida=salida, frames=frames, segundos=frames / FRECUENCIA, bytes_audio=talla, ganancia=ganancia, recortadas=recortadas, colchon=colchon, alineado=alineado, ) # --------------------------------------------------------------------------- # La parte que habla contigo. # --------------------------------------------------------------------------- DESCRIPCION = """\ Convierte una voz ya separada (UVR, MDX, el programa que uses) en el fichero .rx3stem que entiende el XDJ-RX3 con el mod de MOLE. Tú le das dos cosas: la voz separada y la canción original TAL CUAL está en el USB. El script alinea la voz con la canción (sin eso el instrumental no cancela), calcula el nombre de fichero exacto que espera la mesa y lo deja en la carpeta RX3_STEMS del USB. Nada que renombrar ni que mover a mano. """ EJEMPLOS = """\ Ejemplos: python mole-stems.py voz.wav "E:\\Contents\\01 - Kasablanca - Nothing But You (Extended Mix).mp3" La voz que sacó tu separador y la canción tal cual está en el USB. El script detecta el USB (E:), calcula el nombre que espera la mesa y deja el fichero en E:\\RX3_STEMS\\. python mole-stems.py voz.flac "E:\\Contents\\cancion.mp3" --pico-normalizado 0.9 Igual, pero deshaciendo antes la normalización de volumen del separador (la opción «normalization» de UVR y MDX, que suele valer 0.9). Hace falta para que el instrumental cancele del todo. La forma cómoda es MOLE Studio, que separa, nombra y coloca todo por ti: https://get.molemusic.es """ class _Analizador(argparse.ArgumentParser): def error(self, message): # los errores de sintaxis, también en llano self.print_usage(sys.stderr) raise SystemExit(f"Error: {message}\nAyuda completa: python mole-stems.py --help") def _analizador() -> argparse.ArgumentParser: analizador = _Analizador( prog="mole-stems.py", usage='python mole-stems.py voz cancion [opciones]', description=DESCRIPCION, epilog=EJEMPLOS, formatter_class=argparse.RawDescriptionHelpFormatter, add_help=False, ) analizador._positionals.title = "Qué darle" analizador._optionals.title = "Opciones" analizador.add_argument( "voz", nargs="?", type=pathlib.Path, help="la voz ya separada (WAV, FLAC, MP3…)", ) analizador.add_argument( "cancion", nargs="?", type=pathlib.Path, metavar="cancion", help="la canción original, tal cual está en el USB", ) analizador.add_argument( "--salida", type=pathlib.Path, metavar="RUTA", help="guardar el fichero exactamente ahí (si RUTA es una carpeta, " "dentro de ella con el nombre calculado)", ) analizador.add_argument( "--usb", type=pathlib.Path, metavar="RUTA", help="raíz del USB (p. ej. E:\\), por si no se detecta sola", ) analizador.add_argument( "--pico-normalizado", type=float, metavar="PICO", dest="pico_normalizado", help="pico al que tu separador normalizó la canción antes de separar " "(la opción «normalization» de UVR/MDX, normalmente 0.9). Ponlo " "solo si tu separador normaliza: aquí se deshace ese cambio de " "volumen para que el instrumental cancele del todo", ) analizador.add_argument( "--ffmpeg", default="ffmpeg", metavar="RUTA", help="dónde está ffmpeg, si no está en el PATH", ) analizador.add_argument( "-h", "--help", action="help", help="muestra esta ayuda y sale", ) return analizador def _destino(args, cancion: pathlib.Path, nombre: str) -> tuple: """Dónde escribir, y la explicación de por qué ahí.""" if args.salida: salida = args.salida if salida.is_dir(): return salida / nombre, f"Destino elegido por ti: {salida}" return salida, None if args.usb: if not args.usb.is_dir(): raise MoleError(f"la ruta de --usb no existe o no es una carpeta: {args.usb}") return args.usb / "RX3_STEMS" / nombre, f"USB indicado por ti: {args.usb}" raiz = _raiz_usb(cancion) if raiz is not None: return raiz / "RX3_STEMS" / nombre, f"USB detectado: {raiz}" return cancion.parent / nombre, ( "No veo la canción dentro de un USB, así que dejaré el fichero junto " "a ella.\nLuego cópialo a la carpeta RX3_STEMS de la raíz del USB " "(o repite con --usb E:\\)." ) def _duracion(segundos: float) -> str: return f"{int(segundos // 60)}:{int(segundos % 60):02d}" def main(argv=None) -> int: if sys.version_info < (3, 9): print( "Este script necesita Python 3.9 o más nuevo " f"(el tuyo es {sys.version_info[0]}.{sys.version_info[1]}). " "Descarga la última versión en https://www.python.org/downloads/", file=sys.stderr, ) return 2 for salida_texto in (sys.stdout, sys.stderr): try: salida_texto.reconfigure(errors="replace") except Exception: pass args = _analizador().parse_args(argv) if args.voz is None or args.cancion is None: print( "Faltan datos. El uso mínimo es:\n\n" ' python mole-stems.py voz.wav "E:\\Contents\\cancion.mp3"\n\n' "(primero la voz separada, después la canción original del USB)\n" "Ayuda completa: python mole-stems.py --help", file=sys.stderr, ) return 2 try: voz = args.voz if not voz.is_file(): raise MoleError(f"no encuentro la voz separada: {voz}") cancion = args.cancion if not cancion.is_file(): raise MoleError(f"no encuentro la canción original: {cancion}") cancion = cancion.resolve() ffmpeg = _localizar_ffmpeg(args.ffmpeg) nombre = nombre_rx3stem(cancion) salida, nota = _destino(args, cancion, nombre) print("MOLE · stems para el XDJ-RX3") print(f" Voz: {voz}") print(f" Canción: {cancion}") if nota: print(f" {nota}") print(f" Escribiré: {salida}") print() resultado = escribir_rx3stem( voz, salida, cancion, ffmpeg=ffmpeg, pico_normalizado=args.pico_normalizado, avisa=lambda texto: print(f"» {texto}", flush=True), ) print() megas = resultado.bytes_audio / (1024 * 1024) print(f"Listo: {resultado.salida}") print(f" {_duracion(resultado.segundos)} min · 44.1 kHz estéreo · {megas:.1f} MB") if resultado.ganancia != 1.0: print(f" Volumen igualado con la canción original (×{resultado.ganancia:.3f}).") if resultado.recortadas: print( f" Aviso: {resultado.recortadas} muestras pasaban del máximo y se han " "recortado (no debería notarse)." ) if not resultado.alineado: print( " Nota: no he podido medir el pequeño desfase que meten los MP3/AAC.\n" " Si al quitar la voz en la mesa se oye la voz de fondo, repite el\n" " proceso separando desde el WAV o FLAC original." ) print("Expulsa el USB con seguridad y llévalo a la mesa: la voz ya se podrá quitar.") return 0 except MoleError as fallo: print(f"Error: {fallo}", file=sys.stderr) return 1 except KeyboardInterrupt: print("\nCancelado. No se ha dejado nada a medias en el USB.", file=sys.stderr) return 130 except OSError as fallo: print(f"Error al leer o escribir: {fallo}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())