uxoxo/eb2ab
0
1import os2import numpy as np3import regex as re4import scipy.fftpack5import soundfile as sf6import subprocess7import shutil8 9from io import BytesIO10from pydub import AudioSegment, silence11from pydub.silence import detect_silence12 13from lib.conf import voice_formats, default_audio_proc_samplerate14from lib.models import TTS_ENGINES, models15 16# Try to import BackgroundDetector, but make it optional17try:18 from lib.classes.background_detector import BackgroundDetector19 BACKGROUND_DETECTOR_AVAILABLE = True20except (ImportError, AttributeError) as e:21 print(f"Warning: BackgroundDetector not available: {e}")22 print("Voice extraction will skip background detection step.")23 BACKGROUND_DETECTOR_AVAILABLE = False24 BackgroundDetector = None25 26class VoiceExtractor:27 28 def __init__(self, session, voice_file, voice_name):29 self.wav_file = None30 self.session = session31 self.voice_file = voice_file32 self.voice_name = voice_name33 self.voice_track = 'vocals.wav'34 self.samplerate = models[session['tts_engine']][session['fine_tuned']]['samplerate']35 self.output_dir = self.session['voice_dir']36 self.demucs_dir = os.path.join(self.output_dir, 'htdemucs', voice_name)37 self.silence_threshold = -6038 39 def _validate_format(self):40 file_extension = os.path.splitext(self.voice_file)[1].lower()41 if file_extension in voice_formats:42 msg = 'Input file valid'43 return True, msg44 error = f'Unsupported file format: {file_extension}. Supported formats are: {", ".join(voice_formats)}'45 return False, error46 47 def _convert2wav(self):48 try:49 self.wav_file = os.path.join(self.session['voice_dir'], f'{self.voice_name}.wav')50 ffmpeg_cmd = [51 shutil.which('ffmpeg'), '-hide_banner', '-nostats', '-i', self.voice_file,52 '-ac', '1',53 '-y', self.wav_file54 ]55 process = subprocess.Popen(56 ffmpeg_cmd,57 env={},58 stdout=subprocess.PIPE,59 stderr=subprocess.STDOUT,60 text=True,61 universal_newlines=True,62 encoding='utf-8'63 )64 for line in process.stdout:65 print(line, end='') # Print each line of stdout66 process.wait()67 if process.returncode != 0:68 error = f'_convert2wav(): process.returncode: {process.returncode}'69 elif not os.path.exists(self.wav_file) or os.path.getsize(self.wav_file) == 0:70 error = f'_convert2wav output error: {self.wav_file} was not created or is empty.' 71 else:72 msg = 'Conversion to .wav format for processing successful'73 return True, msg74 except subprocess.CalledProcessError as e:75 error = f'convert2wav fmpeg.Error: {e.stderr.decode()}'76 raise ValueError(error)77 except Exception as e:78 error = f'_convert2wav() error: {e}'79 raise ValueError(error)80 return False, error81 82 def _detect_background(self):83 try:84 if not BACKGROUND_DETECTOR_AVAILABLE:85 msg = 'Background detection unavailable (pyannote-audio issue). Assuming background present, will run demucs...'86 print(msg)87 # Assume background is present to be safe - will run demucs separation88 return True, True, msg89 90 msg = 'Detecting any background noise or music...'91 print(msg)92 detector = BackgroundDetector(wav_file=self.wav_file)93 status, report = detector.detect(vad_ratio_thresh=0.15)94 print(report)95 if status:96 msg = 'Background noise or music detected. Proceeding voice extraction...'97 else:98 msg = 'No background noise or music detected. Skipping separation...'99 return True, status, msg100 except Exception as e:101 error = f'_detect_background() error: {e}'102 raise ValueError(error)103 return False, False, error104 105 def _demucs_voice(self):106 try: 107 cmd = [108 "demucs",109 "--verbose",110 "--two-stems=vocals",111 "--out", self.output_dir,112 self.wav_file113 ]114 try:115 process = subprocess.run(cmd, check=True)116 self.voice_track = os.path.join(self.demucs_dir, self.voice_track)117 msg = 'Voice track isolation successful'118 return True, msg119 except subprocess.CalledProcessError as e:120 error = (121 f'_demucs_voice() subprocess CalledProcessError error: {e.returncode}\n\n'122 f'stdout: {e.output}\n\n'123 f'stderr: {e.stderr}'124 )125 raise ValueError(error)126 except FileNotFoundError:127 error = f'_demucs_voice() subprocess FileNotFoundError error: The "demucs" command was not found. Ensure it is installed and in PATH.'128 raise ValueError(error)129 except Exception as e: 130 error = f'_demucs_voice() subprocess Exception error: {str(e)}'131 raise ValueError(error)132 except Exception as e:133 error = f'_demucs_voice() error: {e}'134 raise ValueError(error)135 return False, error136 137 def _remove_silences(self, audio, silence_threshold, min_silence_len=200, keep_silence=300):138 final_audio = AudioSegment.silent(duration=0)139 chunks = silence.split_on_silence(140 audio,141 min_silence_len=min_silence_len,142 silence_thresh=silence_threshold,143 keep_silence=keep_silence144 )145 for chunk in chunks:146 final_audio += chunk147 final_audio.export(self.voice_track, format='wav')148 149 def _trim_and_clean(self,silence_threshold, min_silence_len=200, chunk_size=100):150 try:151 audio = AudioSegment.from_file(self.voice_track)152 total_duration = len(audio) # Total duration in milliseconds153 min_required_duration = 20000 if self.session['tts_engine'] == TTS_ENGINES['BARK'] else 12000154 msg = f"Removing long pauses..."155 print(msg)156 self._remove_silences(audio, silence_threshold)157 if total_duration <= min_required_duration:158 msg = f"Audio is only {total_duration/1000:.2f}s long; skipping audio trimming..."159 return True, msg160 else:161 if total_duration > (min_required_duration * 2):162 msg = f"Audio longer than the max allowed. Proceeding to audio trimming..." 163 print(msg)164 window = min_required_duration165 hop = max(1, window // 4)166 best_var = -float("inf")167 best_start = 0168 sr = audio.frame_rate169 for start in range(0, total_duration - window + 1, hop):170 chunk = audio[start : start + window]171 samples = np.array(chunk.get_array_of_samples()).astype(float)172 # 1) FFT + magnitude173 spectrum = np.abs(scipy.fftpack.fft(samples))174 # 2) turn into a probability distribution175 p = spectrum / (np.sum(spectrum) + 1e-10)176 # 3) spectral entropy177 entropy = -np.sum(p * np.log2(p + 1e-10))178 if entropy > best_var:179 best_var = entropy180 best_start = start181 best_end = best_start + window182 msg = (183 f"Selected most‐diverse‐spectrum window "184 f"{best_start/1000:.2f}s–{best_end/1000:.2f}s "185 f"(@ entropy {best_var:.2f} bits)"186 )187 print(msg)188 # 1) find all silent spans in the file189 silence_spans = detect_silence(190 audio,191 min_silence_len=min_silence_len,192 silence_thresh=silence_threshold193 )194 # silence_spans = [ [start_ms, end_ms], … ]195 # 2) snap best_start *backward* to the end of the last silence before it196 prev_ends = [end for (start, end) in silence_spans if end <= best_start]197 if prev_ends:198 new_start = max(prev_ends)199 else:200 new_start = 0201 # 3) snap best_end *forward* to the start of the first silence after it202 next_starts = [start for (start, end) in silence_spans if start >= best_end]203 if next_starts:204 new_end = min(next_starts)205 else:206 new_end = total_duration207 # 4) update your slice bounds208 best_start, best_end = new_start, new_end209 else:210 best_start = 0211 best_end = total_duration212 trimmed_audio = audio[best_start:best_end]213 trimmed_audio.export(self.voice_track, format='wav')214 msg = 'Audio trimmed and cleaned!'215 return True, msg216 except Exception as e:217 error = f'_trim_and_clean() error: {e}'218 raise ValueError(error)219 220 def _normalize_audio(self):221 error = ''222 try:223 proc_voice_file = os.path.join(self.session['voice_dir'], f'{self.voice_name}_proc.wav')224 final_voice_file = os.path.join(self.session['voice_dir'], f'{self.voice_name}.wav')225 ffmpeg_cmd = [shutil.which('ffmpeg'), '-hide_banner', '-nostats', '-i', self.voice_track]226 filter_complex = (227 'agate=threshold=-25dB:ratio=1.4:attack=10:release=250,'228 'afftdn=nf=-70,'229 'acompressor=threshold=-20dB:ratio=2:attack=80:release=200:makeup=1dB,'230 'loudnorm=I=-14:TP=-3:LRA=7:linear=true,'231 'equalizer=f=150:t=q:w=2:g=1,'232 'equalizer=f=250:t=q:w=2:g=-3,'233 'equalizer=f=3000:t=q:w=2:g=2,'234 'equalizer=f=5500:t=q:w=2:g=-4,'235 'equalizer=f=9000:t=q:w=2:g=-2,'236 'highpass=f=63[audio]'237 )238 ffmpeg_cmd += [239 '-filter_complex', filter_complex,240 '-map', '[audio]',241 '-ar', f'{default_audio_proc_samplerate}',242 '-y', proc_voice_file243 ]244 try:245 process = subprocess.Popen(246 ffmpeg_cmd,247 env={},248 stdout=subprocess.PIPE, 249 stderr=subprocess.PIPE,250 encoding='utf-8',251 errors='ignore'252 )253 for line in process.stdout:254 print(line, end='') # Print each line of stdout255 process.wait()256 if process.returncode != 0:257 error = f'_normalize_audio(): process.returncode: {process.returncode}'258 elif not os.path.exists(proc_voice_file) or os.path.getsize(proc_voice_file) == 0:259 error = f'_normalize_audio() error: {proc_voice_file} was not created or is empty.'260 else:261 os.replace(proc_voice_file, final_voice_file)262 shutil.rmtree(self.demucs_dir, ignore_errors=True)263 msg = 'Audio normalization successful!'264 return True, msg265 except subprocess.CalledProcessError as e:266 error = f'_normalize_audio() ffmpeg.Error: {e.stderr.decode()}'267 except FileNotFoundError as e:268 error = '_normalize_audio() FileNotFoundError: {e} Input file or FFmpeg PATH not found!'269 except Exception as e:270 error = f'_normalize_audio() error: {e}'271 return False, error272 273 def extract_voice(self):274 success = False275 msg = None276 try:277 success, msg = self._validate_format()278 print(msg)279 if success:280 success, msg = self._convert2wav()281 print(msg)282 if success:283 success, status, msg = self._detect_background()284 print(msg)285 if success:286 if status:287 success, msg = self._demucs_voice()288 print(msg)289 else:290 self.voice_track = self.wav_file291 if success:292 success, msg = self._trim_and_clean(self.silence_threshold)293 print(msg)294 if success:295 success, msg = self._normalize_audio()296 print(msg)297 except Exception as e:298 msg = f'extract_voice() error: {e}'299 raise ValueError(msg)300 shutil.rmtree(self.demucs_dir, ignore_errors=True)301 return success, msg