Files
Athena-Deck/separator_worker.py
T

32 lines
1.8 KiB
Python

"""One native separation process; exact pinned package, no arbitrary model code."""
import json,sys
from pathlib import Path
from audio_separator.separator import Separator
import torch
model_dir,model,input_file,output_dir=sys.argv[1:5]
target=sys.argv[5] if len(sys.argv)>5 else ''
from separator_runtime import verify_package
if not verify_package(model_dir,model):raise RuntimeError('Modellpaket-Prüfsumme stimmt nicht.')
if not torch.cuda.is_available():raise RuntimeError('CUDA nicht verfügbar; kein stiller CPU-Fallback.')
separator=Separator(model_file_dir=model_dir,output_dir=output_dir,output_format='WAV',use_soundfile=True,demucs_params={'segment_size':7,'shifts':2,'overlap':.25},mdxc_params={'segment_size':256,'override_model_segment_size':False,'batch_size':1,'overlap':2,'pitch_shift':0})
separator.load_model(model)
files=separator.separate(input_file)
root=Path(output_dir).resolve();names=[]
for name in files:
path=(root/name).resolve()
if path.parent!=root or not path.is_file() or path.suffix.lower()!='.wav':raise RuntimeError('Ungültige Ausgabe.')
names.append(path.name)
if not names:raise RuntimeError('Keine Spuren erzeugt.')
if target:
import subprocess
chosen=next((n for n in names if target in Path(n).stem.lower()),None)
if not chosen:raise RuntimeError('Zielspur fehlt.')
others=[root/n for n in names if n!=chosen]
if not others:raise RuntimeError('Restspuren fehlen.')
args=['ffmpeg','-hide_banner','-loglevel','error','-y']
for f in others:args+=['-i',str(f)]
args+=['-filter_complex',''.join(f'[{i}:a]' for i in range(len(others)))+f'amix=inputs={len(others)}:normalize=0:dropout_transition=0[rest]','-map','[rest]','-c:a','pcm_s16le',str(root/'rest.wav')]
subprocess.run(args,check=True,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,timeout=180)
names=[chosen,'rest.wav']
(root/'results.json').write_text(json.dumps(names))