Spaces:
Runtime error
Runtime error
# Prediction interface for Cog ⚙️ | |
# https://cog.run/python | |
from cog import BasePredictor, Input, Path | |
import os | |
import re | |
import torch | |
import torchaudio | |
import numpy as np | |
import tempfile | |
from einops import rearrange | |
from ema_pytorch import EMA | |
from vocos import Vocos | |
from pydub import AudioSegment | |
from model import CFM, UNetT, DiT, MMDiT | |
from cached_path import cached_path | |
from model.utils import ( | |
get_tokenizer, | |
convert_char_to_pinyin, | |
save_spectrogram, | |
) | |
from transformers import pipeline | |
import librosa | |
device = "cuda" if torch.cuda.is_available() else "mps" if torch.backends.mps.is_available() else "cpu" | |
target_sample_rate = 24000 | |
n_mel_channels = 100 | |
hop_length = 256 | |
target_rms = 0.1 | |
nfe_step = 32 # 16, 32 | |
cfg_strength = 2.0 | |
ode_method = 'euler' | |
sway_sampling_coef = -1.0 | |
speed = 1.0 | |
# fix_duration = 27 # None or float (duration in seconds) | |
fix_duration = None | |
class Predictor(BasePredictor): | |
def load_model(exp_name, model_cls, model_cfg, ckpt_step): | |
checkpoint = torch.load(str(cached_path(f"hf://SWivid/F5-TTS/{exp_name}/model_{ckpt_step}.pt")), map_location=device) | |
vocab_char_map, vocab_size = get_tokenizer("Emilia_ZH_EN", "pinyin") | |
model = CFM( | |
transformer=model_cls( | |
**model_cfg, | |
text_num_embeds=vocab_size, | |
mel_dim=n_mel_channels | |
), | |
mel_spec_kwargs=dict( | |
target_sample_rate=target_sample_rate, | |
n_mel_channels=n_mel_channels, | |
hop_length=hop_length, | |
), | |
odeint_kwargs=dict( | |
method=ode_method, | |
), | |
vocab_char_map=vocab_char_map, | |
).to(device) | |
ema_model = EMA(model, include_online_model=False).to(device) | |
ema_model.load_state_dict(checkpoint['ema_model_state_dict']) | |
ema_model.copy_params_from_ema_to_model() | |
return ema_model, model | |
def setup(self) -> None: | |
"""Load the model into memory to make running multiple predictions efficient""" | |
# self.model = torch.load("./weights.pth") | |
print("Loading Whisper model...") | |
self.pipe = pipeline( | |
"automatic-speech-recognition", | |
model="openai/whisper-large-v3-turbo", | |
torch_dtype=torch.float16, | |
device=device, | |
) | |
print("Loading F5-TTS model...") | |
F5TTS_model_cfg = dict(dim=1024, depth=22, heads=16, ff_mult=2, text_dim=512, conv_layers=4) | |
self.F5TTS_ema_model, self.F5TTS_base_model = self.load_model("F5TTS_Base", DiT, F5TTS_model_cfg, 1200000) | |
def predict( | |
self, | |
gen_text: str = Input(description="Text to generate"), | |
ref_audio_orig: Path = Input(description="Reference audio"), | |
remove_silence: bool = Input(description="Remove silences", default=True), | |
) -> Path: | |
"""Run a single prediction on the model""" | |
model_choice = "F5-TTS" | |
print(gen_text) | |
if len(gen_text) > 200: | |
raise gr.Error("Please keep your text under 200 chars.") | |
gr.Info("Converting audio...") | |
with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as f: | |
aseg = AudioSegment.from_file(ref_audio_orig) | |
audio_duration = len(aseg) | |
if audio_duration > 15000: | |
gr.Warning("Audio is over 15s, clipping to only first 15s.") | |
aseg = aseg[:15000] | |
aseg.export(f.name, format="wav") | |
ref_audio = f.name | |
ema_model = self.F5TTS_ema_model | |
base_model = self.F5TTS_base_model | |
if not ref_text.strip(): | |
gr.Info("No reference text provided, transcribing reference audio...") | |
ref_text = outputs = self.pipe( | |
ref_audio, | |
chunk_length_s=30, | |
batch_size=128, | |
generate_kwargs={"task": "transcribe"}, | |
return_timestamps=False, | |
)['text'].strip() | |
gr.Info("Finished transcription") | |
else: | |
gr.Info("Using custom reference text...") | |
audio, sr = torchaudio.load(ref_audio) | |
rms = torch.sqrt(torch.mean(torch.square(audio))) | |
if rms < target_rms: | |
audio = audio * target_rms / rms | |
if sr != target_sample_rate: | |
resampler = torchaudio.transforms.Resample(sr, target_sample_rate) | |
audio = resampler(audio) | |
audio = audio.to(device) | |
# Prepare the text | |
text_list = [ref_text + gen_text] | |
final_text_list = convert_char_to_pinyin(text_list) | |
# Calculate duration | |
ref_audio_len = audio.shape[-1] // hop_length | |
# if fix_duration is not None: | |
# duration = int(fix_duration * target_sample_rate / hop_length) | |
# else: | |
zh_pause_punc = r"。,、;:?!" | |
ref_text_len = len(ref_text) + len(re.findall(zh_pause_punc, ref_text)) | |
gen_text_len = len(gen_text) + len(re.findall(zh_pause_punc, gen_text)) | |
duration = ref_audio_len + int(ref_audio_len / ref_text_len * gen_text_len / speed) | |
# inference | |
gr.Info(f"Generating audio using F5-TTS") | |
with torch.inference_mode(): | |
generated, _ = base_model.sample( | |
cond=audio, | |
text=final_text_list, | |
duration=duration, | |
steps=nfe_step, | |
cfg_strength=cfg_strength, | |
sway_sampling_coef=sway_sampling_coef, | |
) | |
generated = generated[:, ref_audio_len:, :] | |
generated_mel_spec = rearrange(generated, '1 n d -> 1 d n') | |
gr.Info("Running vocoder") | |
vocos = Vocos.from_pretrained("charactr/vocos-mel-24khz") | |
generated_wave = vocos.decode(generated_mel_spec.cpu()) | |
if rms < target_rms: | |
generated_wave = generated_wave * rms / target_rms | |
# wav -> numpy | |
generated_wave = generated_wave.squeeze().cpu().numpy() | |
if remove_silence: | |
gr.Info("Removing audio silences... This may take a moment") | |
non_silent_intervals = librosa.effects.split(generated_wave, top_db=30) | |
non_silent_wave = np.array([]) | |
for interval in non_silent_intervals: | |
start, end = interval | |
non_silent_wave = np.concatenate([non_silent_wave, generated_wave[start:end]]) | |
generated_wave = non_silent_wave | |
# spectogram | |
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp_wav: | |
wav_path = tmp_wav.name | |
torchaudio.save(wav_path, torch.tensor(generated_wave), target_sample_rate) | |
return wav_path |