Spaces:
Running
Running
| # ================================================================================================== | |
| # DEEPFAKE AUDIO - toolbox/__init__.py (Interactive Logic Controller) | |
| # ================================================================================================== | |
| # | |
| # 📝 DESCRIPTION | |
| # This module implements the main controller for the SV2TTS toolbox. It | |
| # orchestrates the interaction between the User Interface (PyQt5), the neural | |
| # encoder, synthesizer, and vocoder modules. It manages dataset browsing, | |
| # real-time voice recording, and the cloning workflow. | |
| # | |
| # 👤 AUTHORS | |
| # - Amey Thakur (https://github.com/Amey-Thakur) | |
| # - Mega Satish (https://github.com/msatmod) | |
| # | |
| # 🤝🏻 CREDITS | |
| # Original Real-Time Voice Cloning methodology by CorentinJ | |
| # Repository: https://github.com/CorentinJ/Real-Time-Voice-Cloning | |
| # | |
| # 🔗 PROJECT LINKS | |
| # Repository: https://github.com/Amey-Thakur/DEEPFAKE-AUDIO | |
| # Video Demo: https://youtu.be/i3wnBcbHDbs | |
| # Research: https://github.com/Amey-Thakur/DEEPFAKE-AUDIO/blob/main/DEEPFAKE-AUDIO.ipynb | |
| # | |
| # 📜 LICENSE | |
| # Released under the MIT License | |
| # Release Date: 2021-02-06 | |
| # ================================================================================================== | |
| import sys | |
| import traceback | |
| from pathlib import Path | |
| from time import perf_counter as timer | |
| import numpy as np | |
| import torch | |
| from encoder import inference as encoder | |
| from synthesizer.inference import Synthesizer | |
| from toolbox.ui import UI | |
| from toolbox.utterance import Utterance | |
| from vocoder import inference as vocoder | |
| # Dataset Catalog: Predefined paths for recognized speech datasets | |
| recognized_datasets = [ | |
| "LibriSpeech/dev-clean", | |
| "LibriSpeech/dev-other", | |
| "LibriSpeech/test-clean", | |
| "LibriSpeech/test-other", | |
| "LibriSpeech/train-clean-100", | |
| "LibriSpeech/train-clean-360", | |
| "LibriSpeech/train-other-500", | |
| "LibriTTS/dev-clean", | |
| "LibriTTS/dev-other", | |
| "LibriTTS/test-clean", | |
| "LibriTTS/test-other", | |
| "LibriTTS/train-clean-100", | |
| "LibriTTS/train-clean-360", | |
| "LibriTTS/train-other-500", | |
| "LJSpeech-1.1", | |
| "VoxCeleb1/wav", | |
| "VoxCeleb1/test_wav", | |
| "VoxCeleb2/dev/aac", | |
| "VoxCeleb2/test/aac", | |
| "VCTK-Corpus/wav48", | |
| ] | |
| # Memory Limit: Maximum cached waveforms to prevent OOM errors | |
| MAX_WAVS = 15 | |
| class Toolbox: | |
| """ | |
| Main Orchestrator: | |
| Binds neural inference backends to the graphical interface. | |
| Handles application state, hardware interfacing, and error logging. | |
| """ | |
| def __init__(self, datasets_root: Path, models_dir: Path, seed: int=None): | |
| sys.excepthook = self.excepthook | |
| self.datasets_root = datasets_root | |
| self.utterances = set() | |
| self.current_generated = (None, None, None, None) # speaker_name, spec, breaks, wav | |
| self.synthesizer = None # type: Synthesizer | |
| self.current_wav = None | |
| self.waves_list = [] | |
| self.waves_count = 0 | |
| self.waves_namelist = [] | |
| # Feature Detection: Optional WebRTC Voice Activity Detection (VAD) | |
| try: | |
| import webrtcvad | |
| self.trim_silences = True | |
| except: | |
| self.trim_silences = False | |
| self.ui = UI() | |
| self.reset_ui(models_dir, seed) | |
| self.setup_events() | |
| self.ui.start() | |
| def excepthook(self, exc_type, exc_value, exc_tb): | |
| """Diagnostic Supervisor: Intercepts and logs application crashes to the UI.""" | |
| traceback.print_exception(exc_type, exc_value, exc_tb) | |
| self.ui.log("Exception: %s" % exc_value) | |
| def setup_events(self): | |
| """Event Binder: Connects UI signals to toolbox control logic.""" | |
| # Dataset, speaker and utterance selection | |
| self.ui.browser_load_button.clicked.connect(lambda: self.load_from_browser()) | |
| random_func = lambda level: lambda: self.ui.populate_browser(self.datasets_root, | |
| recognized_datasets, | |
| level) | |
| self.ui.random_dataset_button.clicked.connect(random_func(0)) | |
| self.ui.random_speaker_button.clicked.connect(random_func(1)) | |
| self.ui.random_utterance_button.clicked.connect(random_func(2)) | |
| self.ui.dataset_box.currentIndexChanged.connect(random_func(1)) | |
| self.ui.speaker_box.currentIndexChanged.connect(random_func(2)) | |
| # Model selection and initialization | |
| self.ui.encoder_box.currentIndexChanged.connect(self.init_encoder) | |
| self.ui.synthesizer_box.currentIndexChanged.connect(lambda: setattr(self, 'synthesizer', None)) | |
| self.ui.vocoder_box.currentIndexChanged.connect(self.init_vocoder) | |
| # File exploration and playback controls | |
| self.ui.browser_browse_button.clicked.connect(lambda: self.load_from_browser(self.ui.browse_file())) | |
| self.ui.utterance_history.currentIndexChanged.connect(lambda: self.ui.draw_utterance(self.ui.selected_utterance, "current")) | |
| self.ui.play_button.clicked.connect(lambda: self.ui.play(self.ui.selected_utterance.wav, Synthesizer.sample_rate)) | |
| self.ui.stop_button.clicked.connect(self.ui.stop) | |
| self.ui.record_button.clicked.connect(self.record) | |
| # Audio and Persistence | |
| self.ui.setup_audio_devices(Synthesizer.sample_rate) | |
| self.ui.replay_wav_button.clicked.connect(self.replay_last_wav) | |
| self.ui.export_wav_button.clicked.connect(self.export_current_wave) | |
| self.ui.waves_cb.currentIndexChanged.connect(self.set_current_wav) | |
| # Synthesis and Parameterization | |
| self.ui.generate_button.clicked.connect(lambda: self.synthesize() or self.vocode()) | |
| self.ui.synthesize_button.clicked.connect(self.synthesize) | |
| self.ui.vocode_button.clicked.connect(self.vocode) | |
| self.ui.random_seed_checkbox.clicked.connect(self.update_seed_textbox) | |
| self.ui.clear_button.clicked.connect(self.clear_utterances) | |
| def set_current_wav(self, index): | |
| """Archives the currently active waveform.""" | |
| self.current_wav = self.waves_list[index] | |
| def export_current_wave(self): | |
| """Persistence Bridge: Saves the currently active waveform to the filesystem.""" | |
| self.ui.save_audio_file(self.current_wav, Synthesizer.sample_rate) | |
| def replay_last_wav(self): | |
| """Audio Feedback: Replays the generated cloning result.""" | |
| self.ui.play(self.current_wav, Synthesizer.sample_rate) | |
| def reset_ui(self, models_dir: Path, seed: int=None): | |
| """Structural Sync: Refreshes models and dataset browser state.""" | |
| self.ui.populate_browser(self.datasets_root, recognized_datasets, 0, True) | |
| self.ui.populate_models(models_dir) | |
| self.ui.populate_gen_options(seed, self.trim_silences) | |
| def load_from_browser(self, fpath=None): | |
| """Asset Loader: Imports audio files from the dataset browser or local paths.""" | |
| if fpath is None: | |
| fpath = Path(self.datasets_root, | |
| self.ui.current_dataset_name, | |
| self.ui.current_speaker_name, | |
| self.ui.current_utterance_name) | |
| name = str(fpath.relative_to(self.datasets_root)) | |
| speaker_name = self.ui.current_dataset_name + '_' + self.ui.current_speaker_name | |
| if self.ui.auto_next_checkbox.isChecked(): | |
| self.ui.browser_select_next() | |
| elif fpath == "": return | |
| else: | |
| name = fpath.name | |
| speaker_name = fpath.parent.name | |
| wav = Synthesizer.load_preprocess_wav(fpath) | |
| self.ui.log("Loaded %s" % name) | |
| self.add_real_utterance(wav, name, speaker_name) | |
| def record(self): | |
| """Input Bridge: Captures 5 seconds of live audio from the microphone.""" | |
| wav = self.ui.record_one(encoder.sampling_rate, 5) | |
| if wav is None: return | |
| self.ui.play(wav, encoder.sampling_rate) | |
| speaker_name = "user01" | |
| name = speaker_name + "_rec_%05d" % np.random.randint(100000) | |
| self.add_real_utterance(wav, name, speaker_name) | |
| def add_real_utterance(self, wav, name, speaker_name): | |
| """Analytic Registration: Computes embedding and spectrogram for a real sample.""" | |
| spec = Synthesizer.make_spectrogram(wav) | |
| self.ui.draw_spec(spec, "current") | |
| if not encoder.is_loaded(): self.init_encoder() | |
| encoder_wav = encoder.preprocess_wav(wav) | |
| embed, partial_embeds, _ = encoder.embed_utterance(encoder_wav, return_partials=True) | |
| utterance = Utterance(name, speaker_name, wav, spec, embed, partial_embeds, False) | |
| self.utterances.add(utterance) | |
| self.ui.register_utterance(utterance) | |
| self.ui.draw_embed(embed, name, "current") | |
| self.ui.draw_umap_projections(self.utterances) | |
| def clear_utterances(self): | |
| """Memory Cleanse: Resets the projection view and session history.""" | |
| self.utterances.clear() | |
| self.ui.draw_umap_projections(self.utterances) | |
| def synthesize(self): | |
| """Neural Synthesis Phase: Generates Mel-Spectrogram frames from text.""" | |
| self.ui.log("Generating the mel spectrogram...") | |
| self.ui.set_loading(1) | |
| seed = int(self.ui.seed_textbox.text()) if self.ui.random_seed_checkbox.isChecked() else None | |
| if seed is not None: | |
| self.ui.populate_gen_options(seed, self.trim_silences) | |
| torch.manual_seed(seed) | |
| if self.synthesizer is None or seed is not None: | |
| self.init_synthesizer() | |
| texts = self.ui.text_prompt.toPlainText().split("\n") | |
| embed = self.ui.selected_utterance.embed | |
| specs = self.synthesizer.synthesize_spectrograms(texts, [embed] * len(texts)) | |
| breaks = [spec.shape[1] for spec in specs] | |
| spec = np.concatenate(specs, axis=1) | |
| self.ui.draw_spec(spec, "generated") | |
| self.current_generated = (self.ui.selected_utterance.speaker_name, spec, breaks, None) | |
| self.ui.set_loading(0) | |
| def vocode(self): | |
| """Neural Vocoding Phase: Transforms Mel-Spectrograms into audible waveforms.""" | |
| speaker_name, spec, breaks, _ = self.current_generated | |
| assert spec is not None | |
| seed = int(self.ui.seed_textbox.text()) if self.ui.random_seed_checkbox.isChecked() else None | |
| if seed is not None: | |
| self.ui.populate_gen_options(seed, self.trim_silences) | |
| torch.manual_seed(seed) | |
| if not vocoder.is_loaded() or seed is not None: | |
| self.init_vocoder() | |
| def vocoder_progress(i, seq_len, b_size, gen_rate): | |
| real_time_factor = (gen_rate / Synthesizer.sample_rate) * 1000 | |
| line = "Waveform generation: %d/%d (batch size: %d, rate: %.1fkHz - %.2fx real time)" \ | |
| % (i * b_size, seq_len * b_size, b_size, gen_rate, real_time_factor) | |
| self.ui.log(line, "overwrite") | |
| self.ui.set_loading(i, seq_len) | |
| if self.ui.current_vocoder_fpath is not None: | |
| wav = vocoder.infer_waveform(spec, progress_callback=vocoder_progress) | |
| else: | |
| self.ui.log("Waveform generation with Griffin-Lim... ") | |
| wav = Synthesizer.griffin_lim(spec) | |
| self.ui.set_loading(0) | |
| self.ui.log(" Done!", "append") | |
| # Concat with breath breaks and post-process | |
| b_ends = np.cumsum(np.array(breaks) * Synthesizer.hparams.hop_size) | |
| b_starts = np.concatenate(([0], b_ends[:-1])) | |
| wavs = [wav[start:end] for start, end, in zip(b_starts, b_ends)] | |
| breath_breaks = [np.zeros(int(0.15 * Synthesizer.sample_rate))] * len(breaks) | |
| wav = np.concatenate([i for w, b in zip(wavs, breath_breaks) for i in (w, b)]) | |
| if self.ui.trim_silences_checkbox.isChecked(): | |
| wav = encoder.preprocess_wav(wav) | |
| wav = wav / np.abs(wav).max() * 0.97 | |
| self.ui.play(wav, Synthesizer.sample_rate) | |
| # Update Session History | |
| wav_name = str(self.waves_count + 1) | |
| self.waves_count += 1 | |
| if self.waves_count > MAX_WAVS: | |
| self.waves_list.pop() | |
| self.waves_namelist.pop() | |
| self.waves_list.insert(0, wav) | |
| self.waves_namelist.insert(0, wav_name) | |
| self.ui.waves_cb.disconnect() | |
| self.ui.waves_cb_model.setStringList(self.waves_namelist) | |
| self.ui.waves_cb.setCurrentIndex(0) | |
| self.ui.waves_cb.currentIndexChanged.connect(self.set_current_wav) | |
| self.set_current_wav(0) | |
| self.ui.replay_wav_button.setDisabled(False) | |
| self.ui.export_wav_button.setDisabled(False) | |
| # Clone Verification: Compute embedding for the generated voice | |
| if not encoder.is_loaded(): self.init_encoder() | |
| encoder_wav = encoder.preprocess_wav(wav) | |
| embed, partial_embeds, _ = encoder.embed_utterance(encoder_wav, return_partials=True) | |
| name = speaker_name + "_gen_%05d" % np.random.randint(100000) | |
| utterance = Utterance(name, speaker_name, wav, spec, embed, partial_embeds, True) | |
| self.utterances.add(utterance) | |
| self.ui.draw_embed(embed, name, "generated") | |
| self.ui.draw_umap_projections(self.utterances) | |
| def init_encoder(self): | |
| """Neural Wake-up: Loads the speaker encoder model into memory.""" | |
| model_fpath = self.ui.current_encoder_fpath | |
| self.ui.log("Loading the encoder %s... " % model_fpath) | |
| self.ui.set_loading(1) | |
| start = timer() | |
| encoder.load_model(model_fpath) | |
| self.ui.log("Done (%dms)." % int(1000 * (timer() - start)), "append") | |
| self.ui.set_loading(0) | |
| def init_synthesizer(self): | |
| """Neural Wake-up: Loads the Tacotron-2 synthesizer model.""" | |
| model_fpath = self.ui.current_synthesizer_fpath | |
| self.ui.log("Loading the synthesizer %s... " % model_fpath) | |
| self.ui.set_loading(1) | |
| start = timer() | |
| self.synthesizer = Synthesizer(model_fpath) | |
| self.ui.log("Done (%dms)." % int(1000 * (timer() - start)), "append") | |
| self.ui.set_loading(0) | |
| def init_vocoder(self): | |
| """Neural Wake-up: Loads the WaveRNN vocoder model.""" | |
| model_fpath = self.ui.current_vocoder_fpath | |
| if model_fpath is None: return | |
| self.ui.log("Loading the vocoder %s... " % model_fpath) | |
| self.ui.set_loading(1) | |
| start = timer() | |
| vocoder.load_model(model_fpath) | |
| self.ui.log("Done (%dms)." % int(1000 * (timer() - start)), "append") | |
| self.ui.set_loading(0) | |
| def update_seed_textbox(self): | |
| """UI Logic: Synchronizes seed control state.""" | |
| self.ui.update_seed_textbox() | |