"""Bounded WAV tests with coordinated GPU ownership and disposable input.""" import io,json,os,re,signal,subprocess,threading,time,uuid,wave from pathlib import Path from image_test import probe from separator_runtime import MODELS def validate_profiles(settings): if not isinstance(settings,dict) or set(settings)-set(MODELS):raise ValueError('Unbekanntes Audio-Trennprofil.') for value in settings.values(): if not isinstance(value,dict) or set(value)!={'enabled','device'} or type(value['enabled']) is not bool or not isinstance(value['device'],str) or value['device']!='auto' and not re.fullmatch(r'GPU-[a-fA-F0-9-]{16,64}',value['device']):raise ValueError('Ungültiges Audio-Trennprofil.') return settings class SeparatorTests: def __init__(self,root,runtime,scheduler): self.root=Path(root);self.runtime=runtime;self.scheduler=scheduler;self.lock=threading.RLock();self.job=None;self.process=None;self.cancel=threading.Event();self.thread=None def profiles(self): path=self.root.parent/'separator-profiles.json' settings=validate_profiles(json.loads(path.read_text())) if path.exists() else {} return {name:dict(settings.get(name,dict(enabled=True,device='auto'))) for name in MODELS} def configure(self,data): if not isinstance(data,dict) or set(data)!={'model','enabled','device'} or data['model'] not in MODELS:raise ValueError('Modell, Aktivierung und GPU erforderlich.') validate_profiles({data['model']:{k:data[k] for k in ('enabled','device')}}) if data['device']!='auto' and not any(g['uuid']==data['device'] for g in probe()):raise ValueError('Gewählte GPU ist nicht vorhanden.') with self.lock: if self.job and self.job['state']=='running':raise ValueError('Zuerst den laufenden Trennauftrag beenden.') settings=self.profiles();settings[data['model']]={k:data[k] for k in ('enabled','device')} from auth import atomic_write atomic_write(self.root.parent/'separator-profiles.json',settings) return self.status() def status(self): try:devices=[dict(uuid=g['uuid'],name=g['name']) for g in probe()];hardware_error=None except (OSError,ValueError,subprocess.SubprocessError):devices=[];hardware_error='GPU-Erkennung derzeit nicht verfügbar.' with self.lock:return dict(job=dict(self.job) if self.job else None,profiles=self.profiles(),devices=devices,hardware_error=hardware_error) def select_gpu(self,model): settings=self.profiles()[model] if not settings['enabled']:raise ValueError('Dieses Audio-Trennprofil ist deaktiviert. Unter Profile aktivieren.') available=[g for g in probe() if not g['processes'] and g['free_mib']>4000] if settings['device']!='auto': gpu=next((g for g in available if g['uuid']==settings['device']),None) if not gpu:raise ValueError('Die gewählte GPU ist nicht ausreichend frei. GPU-Auswahl im Profil prüfen; kein stiller Wechsel.') return gpu return next((g for term in ('RTX 5080','RTX 3060','') for g in available if term in g['name']),None) def start(self,model,audio,target=None,full=False): if isinstance(model,str) and model in MODELS and not self.profiles()[model]['enabled']:raise ValueError('Dieses Audio-Trennprofil ist deaktiviert. Unter Profile aktivieren.') if full and self.scheduler.gpu_mode!='separator':raise ValueError('Bitte unter Steuerung Audio-Trennung starten.') targets={'vocals':'model_bs_roformer_ep_317_sdr_12.9755.ckpt','drums':'htdemucs_ft.yaml','bass':'htdemucs_ft.yaml','guitar':'htdemucs_6s.yaml','piano':'htdemucs_6s.yaml','other':'htdemucs_6s.yaml'} if target is not None and targets.get(target)!=model:raise ValueError('Zielspur und Modellpaket passen nicht zusammen.') if not isinstance(model,str) or model not in MODELS or not self.runtime.status()['installed'] or not self.runtime.model_ready(model):raise ValueError('Laufzeit und Modellpaket zuerst unter Audio Separator einrichten.') if not isinstance(audio,bytes) or not 44<=len(audio)<=(256 if full else 8)*1024**2:raise ValueError('WAV überschreitet die erlaubte Upload-Größe.') try: with wave.open(io.BytesIO(audio)) as wav: if wav.getnchannels() not in (1,2) or wav.getsampwidth()!=2 or not 8000<=wav.getframerate()<=48000 or not 0deadline:raise ValueError('Trennauftrag überschreitet das Zeitlimit.') with self.lock:self.job['elapsed_seconds']=round(time.time()-self.job['started_at']) diagnostics.close();error=(directory/'worker.tmp').read_bytes()[-65536:].decode(errors='replace') if self.process.returncode:raise ValueError('GPU-Speicher reicht nicht (OOM).' if 'out of memory' in error.lower() else 'Audio Separator hat den Test abgebrochen. Modellpaket und Laufzeit prüfen.') names=json.loads((directory/'results.json').read_text()) if not names or any(Path(n).name!=n or not n.endswith('.wav') or not (directory/n).is_file() for n in names):raise ValueError('Ungültige Ergebnisdateien.') with self.lock:self.job['files']=[dict(index=i,name=n) for i,n in enumerate(names)] (directory/'complete').touch();state='complete';phase='Spuren fertig · Modell entladen' except InterruptedError:state='cancelled';phase='Trenntest abgebrochen' except Exception as exc:phase=str(exc) if isinstance(exc,ValueError) else 'Trenntest fehlgeschlagen.' finally: process=self.process if process and process.poll() is None: try:os.killpg(process.pid,signal.SIGTERM);process.wait(5) except subprocess.TimeoutExpired:os.killpg(process.pid,signal.SIGKILL);process.wait() except ProcessLookupError:pass if 'diagnostics' in locals():diagnostics.close() (directory/'input.wav').unlink(missing_ok=True);(directory/'worker.tmp').unlink(missing_ok=True) with self.lock:self.process=None;self.job.update(state=state,phase=phase,finished_at=time.time()) if release:release() def stop(self): self.cancel.set() if self.thread:self.thread.join(7) return {'cancellation_requested':True} def audio(self,ident,index): if not isinstance(ident,str) or len(ident)!=32 or any(c not in '0123456789abcdef' for c in ident):raise ValueError('Ungültige Auftrag-ID.') directory=self.root/ident if not (directory/'complete').is_file():raise ValueError('Auftrag nicht fertig.') names=json.loads((directory/'results.json').read_text()) if type(index) is not int or not 0<=index512*1024**2:raise ValueError('Spur zu groß.') return path.read_bytes()