This commit is contained in:
yumoqing 2024-10-22 22:59:29 +08:00
parent e80c8a3321
commit ffaa06c8fc
2 changed files with 100 additions and 166 deletions

View File

@ -1,6 +1,7 @@
{
"zmq_url" : "tcp://127.0.0.1:10003",
"sample_rate":16000,
"vocab_file":"",
"remove_silence":false,
"modelname":"F5-TTS",
"device":"cuda:0",

223
f5tts.py
View File

@ -21,8 +21,10 @@ from model.utils_infer import (
import os
import json
from time import time
from appPublic.dictObject import DictObject
from appPublic.zmq_reqrep import ZmqReplier
from appPublic.folderUtils import temp_file
from appPublic.jsonConfig import getConfig
n_mel_channels = 100
@ -34,110 +36,24 @@ ode_method = "euler"
sway_sampling_coef = -1.0
speed = 1.0
def main_process(ref_audio, ref_text, text_gen, model_obj, remove_silence):
main_voice = {"ref_audio": ref_audio, "ref_text": ref_text}
if "voices" not in config:
voices = {"main": main_voice}
else:
voices = config["voices"]
voices["main"] = main_voice
for voice in voices:
voices[voice]["ref_audio"], voices[voice]["ref_text"] = preprocess_ref_audio_text(
voices[voice]["ref_audio"], voices[voice]["ref_text"]
)
print("Voice:", voice)
print("Ref_audio:", voices[voice]["ref_audio"])
print("Ref_text:", voices[voice]["ref_text"])
generated_audio_segments = []
reg1 = r"(?=\[\w+\])"
chunks = re.split(reg1, text_gen)
reg2 = r"\[(\w+)\]"
for text in chunks:
match = re.match(reg2, text)
if match:
voice = match[1]
else:
print("No voice tag found, using main.")
voice = "main"
if voice not in voices:
print(f"Voice {voice} not found, using main.")
voice = "main"
text = re.sub(reg2, "", text)
gen_text = text.strip()
ref_audio = voices[voice]["ref_audio"]
ref_text = voices[voice]["ref_text"]
print(f"Voice: {voice}")
audio, final_sample_rate, spectragram = infer_process(ref_audio, ref_text, gen_text, model_obj)
generated_audio_segments.append(audio)
if generated_audio_segments:
final_wave = np.concatenate(generated_audio_segments)
with open(wave_path, "wb") as f:
sf.write(f.name, final_wave, final_sample_rate)
# Remove silence
if remove_silence:
remove_silence_for_generated_wav(f.name)
print(f.name)
class F5TTS:
def __init__(self):
config = getConfig()
self.sample_rate = config.sample_rate
self.remove_silence = config.remove_silence
self.modelname = config.modelname
self.ref_audio_fn = config.ref_audio_fn
self.load_vocoder_from_local = config.is_local or True
self.zmq_url = config.zmq_url
self.ref_text = config.ref_text
self.device = config.device
self.cross_fade_duration = config.cross_fade_duration
self.gen_ref_audio()
self.gen_ref_text()
self.replier = ZmqReplier(self.zmq_url, self.generate)
self.vocos = load_vocoder(is_local=is_local, local_path="../checkpoints/charactr/vocos-mel-24khz")
self.F5TTS_model_cfg = dict(
dim=1024, depth=22, heads=16, ff_mult=2, text_dim=512, conv_layers=4
)
self.E2TTS_model_cfg = dict(dim=1024, depth=24, heads=16, ff_mult=4)
self.model= self.load_model(self.modelname)
def gen_ref_audio(self):
"""
gen ref_audio
"""
audio, sr = torchaudio.load(self.ref_audio_fn)
if audio.shape[0] > 1:
audio = torch.mean(audio, dim=0, keepdim=True)
rms = torch.sqrt(torch.mean(torch.square(audio)))
if rms < target_rms:
audio = audio * target_rms / rms
if sr != self.sample_rate:
resampler = torchaudio.transforms.Resample(sr, self.sample_rate)
audio = resampler(audio)
self.ref_audio = audio
self.config = getConfig()
self.zmq_url = self.config.zmq_url
self.replier = ZmqReplier(self.config.zmq_url, self.generate)
# self.vocos = load_vocoder(is_local=True, local_path="../checkpoints/charactr/vocos-mel-24khz")
self.load_model()
self.setup_voice()
def run(self):
print(f'running {self.zmq_url}')
self.replier._run()
print('ended ...')
def gen_ref_text(self):
"""
"""
# Add the functionality to ensure it ends with ". "
ref_text = self.ref_text
if not ref_text.endswith(". ") and not ref_text.endswith(""):
if ref_text.endswith("."):
ref_text += " "
else:
ref_text += ". "
self.ref_text = ref_text
def load_model(self):
# load models
if self.modelname == "F5-TTS":
ckpt_file = ''
if self.config.modelname == "F5-TTS":
model_cls = DiT
model_cfg = dict(dim=1024, depth=22, heads=16, ff_mult=2, text_dim=512, conv_layers=4)
if ckpt_file == "":
@ -146,7 +62,7 @@ class F5TTS:
ckpt_step = 1200000
ckpt_file = str(cached_path(f"hf://SWivid/{repo_name}/{exp_name}/model_{ckpt_step}.safetensors"))
elif self.modelname == "E2-TTS":
elif self.config.modelname == "E2-TTS":
model_cls = UNetT
model_cfg = dict(dim=1024, depth=24, heads=16, ff_mult=4)
if ckpt_file == "":
@ -155,53 +71,8 @@ class F5TTS:
ckpt_step = 1200000
ckpt_file = str(cached_path(f"hf://SWivid/{repo_name}/{exp_name}/model_{ckpt_step}.safetensors"))
self.model = load_model(model_cls, model_cfg, ckpt_file, vocab_file)
def split_text(self, text):
max_chars = int(len(self.ref_text.encode('utf-8')) / (self.ref_audio.shape[-1] / self.sample_rate) * (25 - self.ref_audio.shape[-1] / self.sample_rate))
gen_text_batches = chunk_text(gen_text, max_chars=max_chars)
print('ref_text', ref_text)
def cross_fade_wave(self, waves):
final_wave = generated_waves[0]
for i in range(1, len(generated_waves)):
prev_wave = final_wave
next_wave = generated_waves[i]
# Calculate cross-fade samples, ensuring it does not exceed wave lengths
cross_fade_samples = int(self.cross_fade_duration * self.sample_rate)
cross_fade_samples = min(cross_fade_samples, len(prev_wave), len(next_wave))
if cross_fade_samples <= 0:
# No overlap possible, concatenate
final_wave = np.concatenate([prev_wave, next_wave])
continue
# Overlapping parts
prev_overlap = prev_wave[-cross_fade_samples:]
next_overlap = next_wave[:cross_fade_samples]
# Fade out and fade in
fade_out = np.linspace(1, 0, cross_fade_samples)
fade_in = np.linspace(0, 1, cross_fade_samples)
# Cross-faded overlap
cross_faded_overlap = prev_overlap * fade_out + next_overlap * fade_in
# Combine
new_wave = np.concatenate([
prev_wave[:-cross_fade_samples],
cross_faded_overlap,
next_wave[cross_fade_samples:]
])
final_wave = new_wave
return final_wave
def write_wave(wave):
fn = temp_file(suffix='.wav')
sf.write(fn, wave, self.sample_rate)
return fn
self.model = load_model(model_cls, model_cfg, ckpt_file,
self.config.vocab_file).to(self.config.device)
def generate(self, d):
msg= d.decode('utf-8')
@ -233,13 +104,75 @@ class F5TTS:
}
return json.dumps(d)
def setup_voice(self):
main_voice = {"ref_audio": self.config.ref_audio_fn,
"ref_text": self.config.ref_text}
if "voices" not in self.config:
voices = {"main": main_voice}
else:
voices = self.config["voices"]
voices["main"] = main_voice
for voice in voices:
voices[voice]["ref_audio"], voices[voice]["ref_text"] = preprocess_ref_audio_text(
voices[voice]["ref_audio"], voices[voice]["ref_text"]
)
print("Voice:", voice)
print("Ref_audio:", voices[voice]["ref_audio"])
print("Ref_text:", voices[voice]["ref_text"])
self.voices = voices
def inference(self, prompt):
text_gen = prompt
remove_silence = False
generated_audio_segments = []
reg1 = r"(?=\[\w+\])"
chunks = re.split(reg1, text_gen)
reg2 = r"\[(\w+)\]"
for text in chunks:
match = re.match(reg2, text)
if match:
voice = match[1]
else:
print("No voice tag found, using main.")
voice = "main"
if voice not in self.voices:
print(f"Voice {voice} not found, using main.")
voice = "main"
text = re.sub(reg2, "", text)
gen_text = text.strip()
ref_audio = self.voices[voice]["ref_audio"]
ref_text = self.voices[voice]["ref_text"]
print(f"Voice: {voice}, {self.model}")
audio, final_sample_rate, spectragram = \
infer_process(ref_audio, ref_text, gen_text, self.model)
generated_audio_segments.append(audio)
if generated_audio_segments:
final_wave = np.concatenate(generated_audio_segments)
fn = temp_file(suffix='.wav')
with open(fn, "wb") as f:
sf.write(f.name, final_wave, final_sample_rate)
# Remove silence
if remove_silence:
remove_silence_for_generated_wav(f.name)
return fn
if __name__ == '__main__':
workdir = os.getcwd()
config = getConfig(workdir)
# workdir = os.getcwd()
# config = getConfig(workdir)
tts = F5TTS()
print('here')
tts.run()
# tts.run()
while True:
print('prompt:')
p = input()
if p != '':
t1 = time()
f = tts.inference(p)
t2 = time()
print(f'{f}, cost {t2-t1} seconds')