| import os |
| import sys |
| import importlib.util |
| import torch |
| import torchaudio |
| import numpy as np |
| from huggingface_hub import snapshot_download, hf_hub_download |
| import subprocess |
| import uuid |
| import soundfile as sf |
| import spaces |
| import librosa |
| import shutil |
| import re |
| import gradio as gr |
|
|
| |
|
|
| def install_espeak(): |
| try: |
| result = subprocess.run(["which", "espeak-ng"], capture_output=True, text=True) |
| if result.returncode != 0: |
| print("Installing espeak-ng...") |
| subprocess.run(["apt-get", "update"], check=True) |
| subprocess.run(["apt-get", "install", "-y", "espeak-ng", "espeak-ng-data"], check=True) |
| except Exception as e: |
| print(f"Error installing espeak-ng: {e}") |
|
|
| install_espeak() |
|
|
| def patch_langsegment_init(): |
| try: |
| spec = importlib.util.find_spec("LangSegment") |
| if spec is None or spec.origin is None: return |
| init_path = os.path.join(os.path.dirname(spec.origin), '__init__.py') |
| with open(init_path, 'r') as f: lines = f.readlines() |
| modified = False |
| new_lines =[] |
| target_line_prefix = "from .LangSegment import" |
| for line in lines: |
| if line.strip().startswith(target_line_prefix) and ('setLangfilters' in line or 'getLangfilters' in line): |
| mod_line = line.replace(',setLangfilters', '').replace(',getLangfilters', '') |
| mod_line = mod_line.replace('setLangfilters,', '').replace('getLangfilters,', '').rstrip(',') |
| new_lines.append(mod_line + '\n') |
| modified = True |
| else: |
| new_lines.append(line) |
| if modified: |
| with open(init_path, 'w') as f: f.writelines(new_lines) |
| try: |
| import LangSegment |
| importlib.reload(LangSegment) |
| except: pass |
| except: pass |
|
|
| patch_langsegment_init() |
|
|
| if not os.path.exists("Amphion"): |
| print("Cloning Amphion repository...") |
| subprocess.run(["git", "clone", "https://github.com/open-mmlab/Amphion.git"]) |
|
|
| amphion_path = os.path.abspath("Amphion") |
| if amphion_path not in sys.path: |
| sys.path.append(amphion_path) |
|
|
| |
| def patch_amphion_llama_config(): |
| try: |
| for root, dirs, files in os.walk(amphion_path): |
| for file in files: |
| if file.endswith(".py"): |
| file_path = os.path.join(root, file) |
| try: |
| with open(file_path, "r", encoding="utf-8") as f: |
| content = f.read() |
| |
| new_content = re.sub( |
| r'LlamaConfig\(\s*0\s*,\s*256\s*,\s*1024\s*,\s*1\s*,\s*1\s*\)', |
| r'LlamaConfig(vocab_size=0, hidden_size=256, intermediate_size=1024, num_hidden_layers=1, num_attention_heads=1)', |
| content |
| ) |
| |
| if new_content != content: |
| with open(file_path, "w", encoding="utf-8") as f: |
| f.write(new_content) |
| print(f"Patched {file} successfully.") |
| except Exception: |
| pass |
| except Exception as e: |
| print(f"Failed to patch Amphion: {e}") |
|
|
| patch_amphion_llama_config() |
|
|
| |
| import transformers |
| from transformers.models.llama.configuration_llama import LlamaConfig |
|
|
| _original_llama_init = LlamaConfig.__init__ |
| def _patched_llama_init(self, *args, **kwargs): |
| kwargs.setdefault('rope_theta', 10000.0) |
| kwargs.setdefault('attention_bias', False) |
| kwargs.setdefault('rope_scaling', None) |
| kwargs.setdefault('max_position_embeddings', 4096) |
| |
| _original_llama_init(self, *args, **kwargs) |
| |
| if not hasattr(self, 'rope_theta'): |
| self.rope_theta = 10000.0 |
| if not hasattr(self, 'attention_bias'): |
| self.attention_bias = False |
|
|
| LlamaConfig.__init__ = _patched_llama_init |
|
|
| os.makedirs("wav", exist_ok=True) |
| os.makedirs("ckpts/Vevo", exist_ok=True) |
|
|
| try: |
| from models.vc.vevo.vevo_utils import VevoInferencePipeline |
| except ImportError: |
| sys.path.append(os.path.join(amphion_path)) |
| from models.vc.vevo.vevo_utils import VevoInferencePipeline |
|
|
| |
|
|
| def save_audio_pcm16(waveform, output_path, sample_rate=24000): |
| try: |
| if isinstance(waveform, torch.Tensor): |
| waveform = waveform.detach().cpu() |
| if waveform.dim() == 2 and waveform.shape[0] == 1: |
| waveform = waveform.squeeze(0) |
| waveform = waveform.numpy() |
| sf.write(output_path, waveform, sample_rate, subtype='PCM_16') |
| except Exception as e: |
| print(f"Save error: {e}") |
|
|
| downloaded_resources = { |
| "configs": False, |
| "tokenizer_vq8192": False, |
| "fmt_Vq8192ToMels": False, |
| "vocoder": False |
| } |
|
|
| def setup_configs(): |
| if downloaded_resources["configs"]: return |
| target_config_path = "models/vc/vevo/config" |
| os.makedirs(target_config_path, exist_ok=True) |
| |
| config_files =["Vq8192ToMels.json", "Vocoder.json", "hubert_large_l18_mean_std.npz"] |
| source_config_path = os.path.join(amphion_path, "models/vc/vevo/config") |
|
|
| for file in config_files: |
| target_file_path = f"{target_config_path}/{file}" |
| source_file_path = os.path.join(source_config_path, file) |
| |
| if not os.path.exists(target_file_path): |
| if os.path.exists(source_file_path): |
| shutil.copy(source_file_path, target_file_path) |
| else: |
| try: |
| file_data = hf_hub_download(repo_id="amphion/Vevo", filename=f"config/{file}", repo_type="model") |
| if os.path.exists(target_file_path): os.remove(target_file_path) |
| subprocess.run(["cp", file_data, target_file_path]) |
| except Exception as e: |
| print(f"Error downloading config {file}: {e}") |
| |
| downloaded_resources["configs"] = True |
|
|
| setup_configs() |
| device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu") |
| inference_pipelines = {} |
|
|
| downloaded_content_style_tokenizer_path = None |
| downloaded_fmt_path = None |
| downloaded_vocoder_path = None |
|
|
| def preload_all_resources(): |
| setup_configs() |
| global downloaded_content_style_tokenizer_path, downloaded_fmt_path, downloaded_vocoder_path |
| if not downloaded_resources["tokenizer_vq8192"]: |
| downloaded_content_style_tokenizer_path = snapshot_download(repo_id="amphion/Vevo", repo_type="model", cache_dir="./ckpts/Vevo", allow_patterns=["tokenizer/vq8192/"]) |
| downloaded_resources["tokenizer_vq8192"] = True |
| if not downloaded_resources["fmt_Vq8192ToMels"]: |
| downloaded_fmt_path = snapshot_download(repo_id="amphion/Vevo", repo_type="model", cache_dir="./ckpts/Vevo", allow_patterns=["acoustic_modeling/Vq8192ToMels/"]) |
| downloaded_resources["fmt_Vq8192ToMels"] = True |
| if not downloaded_resources["vocoder"]: |
| downloaded_vocoder_path = snapshot_download(repo_id="amphion/Vevo", repo_type="model", cache_dir="./ckpts/Vevo", allow_patterns=["acoustic_modeling/Vocoder/*"]) |
| downloaded_resources["vocoder"] = True |
|
|
| preload_all_resources() |
|
|
| def get_pipeline(): |
| if "timbre" in inference_pipelines: return inference_pipelines["timbre"] |
| tokenizer_path = os.path.join(downloaded_content_style_tokenizer_path, "tokenizer/vq8192") |
| fmt_ckpt = os.path.join(downloaded_fmt_path, "acoustic_modeling/Vq8192ToMels") |
| vocoder_ckpt = os.path.join(downloaded_vocoder_path, "acoustic_modeling/Vocoder") |
| pipeline = VevoInferencePipeline( |
| content_style_tokenizer_ckpt_path=tokenizer_path, |
| fmt_cfg_path="./models/vc/vevo/config/Vq8192ToMels.json", |
| fmt_ckpt_path=fmt_ckpt, |
| vocoder_cfg_path="./models/vc/vevo/config/Vocoder.json", |
| vocoder_ckpt_path=vocoder_ckpt, |
| device=device, |
| ) |
| inference_pipelines["timbre"] = pipeline |
| return pipeline |
|
|
|
|
| |
| def smart_split_audio_vevo(y, sr, min_segment=4): |
| total_samples = len(y) |
| current_sample = 0 |
| chunks = [] |
| |
| |
| if total_samples <= 8 * sr: |
| return [y] |
| |
| while current_sample < total_samples: |
| |
| if total_samples - current_sample <= min_segment * sr: |
| chunks.append(y[current_sample:]) |
| break |
| |
| |
| steps = [8, 12, 15, 20] |
| silence_threshold = 0.015 |
| |
| best_split_point = None |
| best_rms_val = float('inf') |
| |
| for max_sec in steps: |
| start_search = current_sample + int(min_segment * sr) |
| end_search = current_sample + int(max_sec * sr) |
| |
| if start_search >= total_samples: |
| best_split_point = total_samples |
| break |
| |
| if end_search > total_samples: |
| end_search = total_samples |
| |
| search_region = y[start_search:end_search] |
| if len(search_region) == 0: |
| best_split_point = total_samples |
| break |
| |
| |
| rms = librosa.feature.rms(y=search_region, frame_length=2048, hop_length=512)[0] |
| if len(rms) == 0: |
| continue |
| |
| min_rms_idx = np.argmin(rms) |
| min_rms_val = rms[min_rms_idx] |
| candidate_point = start_search + (min_rms_idx * 512) |
| |
| |
| if min_rms_val < best_rms_val: |
| best_rms_val = min_rms_val |
| best_split_point = candidate_point |
| |
| |
| if min_rms_val < silence_threshold: |
| break |
| |
| if best_split_point is None: |
| best_split_point = min(current_sample + int(8 * sr), total_samples) |
| |
| chunk_audio = y[current_sample:best_split_point] |
| chunks.append(chunk_audio) |
| current_sample = best_split_point |
| |
| |
| if len(chunks) > 1 and len(chunks[-1]) < int(1.5 * sr): |
| last_chunk = chunks.pop() |
| combined_len = len(chunks[-1]) + len(last_chunk) |
| |
| if combined_len <= int(20 * sr): |
| chunks[-1] = np.concatenate([chunks[-1], last_chunk]) |
| else: |
| chunks.append(last_chunk) |
| |
| return chunks |
|
|
|
|
| |
| |
| |
|
|
| @spaces.GPU(duration=120) |
| def predict_voice_conversion(source_audio_path, ref_audio_path): |
| if source_audio_path is None or ref_audio_path is None: |
| raise gr.Error("لطفاً هر دو فایل صوتی (اصلی و مرجع) را وارد کنید.") |
|
|
| task_id = str(uuid.uuid4()) |
| out_path = f"wav/{task_id}_out.wav" |
| SR = 24000 |
|
|
| try: |
| |
| content_data, _ = librosa.load(source_audio_path, sr=SR) |
| |
| |
| ref_data, _ = librosa.load(ref_audio_path, sr=SR) |
| ref_tensor = torch.FloatTensor(ref_data).unsqueeze(0) |
| ref_tensor = ref_tensor / (torch.max(torch.abs(ref_tensor)) + 1e-6) * 0.95 |
| |
| temp_r = f"wav/{task_id}_temp_r.wav" |
| save_audio_pcm16(ref_tensor, temp_r, SR) |
|
|
| |
| chunks = smart_split_audio_vevo(content_data, SR, min_segment=4) |
|
|
| generated_chunks = [] |
| pipeline = get_pipeline() |
|
|
| |
| for idx, chunk in enumerate(chunks): |
| chunk_tensor = torch.FloatTensor(chunk).unsqueeze(0) |
| chunk_tensor = chunk_tensor / (torch.max(torch.abs(chunk_tensor)) + 1e-6) * 0.95 |
| |
| temp_c_chunk = f"wav/{task_id}_temp_c_{idx}.wav" |
| save_audio_pcm16(chunk_tensor, temp_c_chunk, SR) |
|
|
| gen = pipeline.inference_fm( |
| src_wav_path=temp_c_chunk, |
| timbre_ref_wav_path=temp_r, |
| flow_matching_steps=32, |
| ) |
|
|
| if torch.isnan(gen).any(): |
| gen = torch.nan_to_num(gen, nan=0.0) |
| gen_np = gen.detach().cpu().squeeze().numpy() |
| |
| generated_chunks.append(gen_np) |
|
|
| if os.path.exists(temp_c_chunk): |
| os.remove(temp_c_chunk) |
|
|
| |
| final_gen = np.concatenate(generated_chunks, axis=0) |
|
|
| |
| save_audio_pcm16(final_gen, out_path, SR) |
|
|
| if os.path.exists(temp_r): |
| os.remove(temp_r) |
|
|
| return out_path |
|
|
| except Exception as e: |
| raise gr.Error(f"خطایی در حین پردازش رخ داد: {str(e)}") |
|
|
| |
| with gr.Blocks() as demo: |
| gr.Markdown("# سامانه تبدیل صدا (Vevo Voice Conversion) - نسخه Zero-GPU PRO") |
| |
| with gr.Row(): |
| with gr.Column(): |
| source_input = gr.Audio(type="filepath", label="فایل صوتی اصلی (Source Audio)") |
| ref_input = gr.Audio(type="filepath", label="فایل صوتی مرجع/هدف (Reference Timbre)") |
| submit_btn = gr.Button("شروع فرآیند تبدیل صدا", variant="primary") |
| |
| with gr.Column(): |
| audio_output = gr.Audio(label="خروجی صدای شبیهسازی شده", type="filepath") |
|
|
| submit_btn.click( |
| fn=predict_voice_conversion, |
| inputs=[source_input, ref_input], |
| outputs=audio_output |
| ) |
|
|
| if __name__ == "__main__": |
| demo.queue(max_size=20).launch(server_name="0.0.0.0", server_port=7860, theme=gr.themes.Soft()) |
