import os
import sys
import torch
import gradio as gr
from pydub import AudioSegment
import mimetypes
sys.path.append('./Amphion')
import Amphion.models.vc.vevo.vevo_utils as vevo_utils
from huggingface_hub import snapshot_download
def load_model():
print("Loading model...")
device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")
print(f"Using device: {device}")
cache_dir = "./ckpts/Vevo"
os.makedirs(cache_dir, exist_ok=True)
# Content Tokenizer
local_dir = snapshot_download(
repo_id="amphion/Vevo",
repo_type="model",
cache_dir=cache_dir,
allow_patterns=["tokenizer/vq32/*"],
)
content_tokenizer_ckpt_path = os.path.join(
local_dir, "tokenizer/vq32/hubert_large_l18_c32.pkl"
)
# Content-Style Tokenizer
local_dir = snapshot_download(
repo_id="amphion/Vevo",
repo_type="model",
cache_dir=cache_dir,
allow_patterns=["tokenizer/vq8192/*"],
)
content_style_tokenizer_ckpt_path = os.path.join(local_dir, "tokenizer/vq8192")
# Autoregressive Transformer
local_dir = snapshot_download(
repo_id="amphion/Vevo",
repo_type="model",
cache_dir=cache_dir,
allow_patterns=["contentstyle_modeling/Vq32ToVq8192/*"],
)
ar_cfg_path = "./config/Vq32ToVq8192.json"
ar_ckpt_path = os.path.join(local_dir, "contentstyle_modeling/Vq32ToVq8192")
# Flow Matching Transformer
local_dir = snapshot_download(
repo_id="amphion/Vevo",
repo_type="model",
cache_dir=cache_dir,
allow_patterns=["acoustic_modeling/Vq8192ToMels/*"],
)
fmt_cfg_path = "./config/Vq8192ToMels.json"
fmt_ckpt_path = os.path.join(local_dir, "acoustic_modeling/Vq8192ToMels")
# Vocoder
local_dir = snapshot_download(
repo_id="amphion/Vevo",
repo_type="model",
cache_dir=cache_dir,
allow_patterns=["acoustic_modeling/Vocoder/*"],
)
vocoder_cfg_path = "./Amphion/models/vc/vevo/config/Vocoder.json"
vocoder_ckpt_path = os.path.join(local_dir, "acoustic_modeling/Vocoder")
print("Initializing pipeline...")
pipeline = vevo_utils.VevoInferencePipeline(
content_tokenizer_ckpt_path=content_tokenizer_ckpt_path,
content_style_tokenizer_ckpt_path=content_style_tokenizer_ckpt_path,
ar_cfg_path=ar_cfg_path,
ar_ckpt_path=ar_ckpt_path,
fmt_cfg_path=fmt_cfg_path,
fmt_ckpt_path=fmt_ckpt_path,
vocoder_cfg_path=vocoder_cfg_path,
vocoder_ckpt_path=vocoder_ckpt_path,
device=device
)
print("Model loaded successfully!")
return pipeline
def convert_to_wav(audio_path):
if audio_path is None:
return None
mime, _ = mimetypes.guess_type(audio_path)
if mime == 'audio/wav' or mime == 'audio/x-wav':
return audio_path
elif mime == 'audio/mpeg':
seg = AudioSegment.from_mp3(audio_path)
wav_path = audio_path.rsplit('.', 1)[0] + '.wav'
seg.export(wav_path, format="wav")
return wav_path
else:
raise ValueError(f"Unsupported audio format: {mime}")
def process_audio(mode, content_audio, ref_style_audio, ref_timbre_audio,
src_text, ref_text, src_language, ref_language, steps,
progress=gr.Progress()):
try:
output_dir = "outputs"
os.makedirs(output_dir, exist_ok=True)
output_path = os.path.join(output_dir, "output.wav")
# Convert uploaded audio files to WAV if needed
if content_audio:
content_path = convert_to_wav(content_audio)
else:
content_path = None
if ref_style_audio:
ref_style_path = convert_to_wav(ref_style_audio)
else:
ref_style_path = None
if ref_timbre_audio:
ref_timbre_path = convert_to_wav(ref_timbre_audio)
else:
ref_timbre_path = None
progress(0.2, "Processing audio...")
# Run inference based on mode
if mode == 'voice':
if not all([content_path, ref_style_path, ref_timbre_path]):
raise gr.Error("Voice mode requires all audio inputs")
gen_audio = inference_pipeline.inference_ar_and_fm(
src_wav_path=content_path,
src_text=None,
style_ref_wav_path=ref_style_path,
timbre_ref_wav_path=ref_timbre_path,
flow_matching_steps=steps
)
elif mode == 'timbre':
if not all([content_path, ref_timbre_path]):
raise gr.Error("Timbre mode requires source and timbre reference audio")
gen_audio = inference_pipeline.inference_fm(
src_wav_path=content_path,
timbre_ref_wav_path=ref_timbre_path,
flow_matching_steps=steps
)
elif mode == 'tts':
if not all([ref_style_path, ref_timbre_path]) or not src_text:
raise gr.Error("TTS mode requires style audio, timbre audio, and source text")
gen_audio = inference_pipeline.inference_ar_and_fm(
src_wav_path=None,
src_text=src_text,
style_ref_wav_path=ref_style_path,
timbre_ref_wav_path=ref_timbre_path,
style_ref_wav_text=ref_text if ref_text else None,
src_text_language=src_language,
style_ref_wav_text_language=ref_language
)
progress(0.8, "Saving generated audio...")
# Save and return the generated audio
vevo_utils.save_audio(gen_audio, target_sample_rate=48000, output_path=output_path)
return output_path
except Exception as e:
raise gr.Error(str(e))
# Initialize the model
inference_pipeline = load_model()
# Create the Gradio interface
with gr.Blocks(title="Vevo Voice Conversion") as demo:
gr.Markdown("# Vevo Voice Conversion")
with gr.Row():
mode = gr.Radio(
choices=["voice", "timbre", "tts"],
value="timbre",
label="Inference Mode",
interactive=True
)
with gr.Tabs():
with gr.TabItem("Audio Inputs"):
content_audio = gr.Audio(
label="Source Audio",
type="filepath",
interactive=True
)
ref_style_audio = gr.Audio(
label="Reference Style Audio",
type="filepath",
interactive=True
)
ref_timbre_audio = gr.Audio(
label="Reference Timbre Audio",
type="filepath",
interactive=True
)
with gr.TabItem("Text Inputs (TTS Mode)"):
src_text = gr.Textbox(
label="Source Text",
placeholder="Enter text for TTS mode",
interactive=True
)
ref_text = gr.Textbox(
label="Reference Style Text (Optional)",
placeholder="Enter reference text",
interactive=True
)
with gr.Row():
src_language = gr.Dropdown(
choices=["en", "zh"],
value="en",
label="Source Language",
interactive=True
)
ref_language = gr.Dropdown(
choices=["en", "zh"],
value="en",
label="Reference Language",
interactive=True
)
with gr.Row():
steps = gr.Slider(
minimum=1,
maximum=64,
value=32,
step=1,
label="Flow Matching Steps"
)
with gr.Row():
with gr.Column():
submit_btn = gr.Button("Generate")
error_box = gr.Textbox(label="Status", interactive=False)
output_audio = gr.Audio(label="Generated Audio")
def process_with_error_handling(*args):
try:
result = process_audio(*args)
error_box.update(value="Success!")
return [result, "Success!"]
except Exception as e:
error_msg = str(e)
return [None, error_msg]
submit_btn.click(
fn=process_with_error_handling,
inputs=[
mode,
content_audio,
ref_style_audio,
ref_timbre_audio,
src_text,
ref_text,
src_language,
ref_language,
steps
],
outputs=[output_audio, error_box]
)
# Example usage text
gr.Markdown("""
## Quick Start Guide
1. Select your mode:
- **Voice**: Full voice conversion (style + timbre)
- **Timbre**: Only voice timbre conversion
- **TTS**: Text-to-speech with voice cloning
2. For Voice/Timbre modes:
- Upload source audio (what you want to convert)
- Upload reference audio(s)
3. For TTS mode:
- Enter your text
- Select language
- Upload reference audio(s)
4. Adjust steps slider:
- Higher values = better quality but slower
- Lower values = faster but lower quality
5. Click Generate and wait for processing
""")
if __name__ == "__main__":
demo.queue().launch()