Create coqui.py

coqui.py

@@ -0,0 +1,389 @@
+import sys
+import io, os, stat
+import subprocess
+import random
+from zipfile import ZipFile
+import uuid
+import time
+import torch
+import torchaudio
+import numpy as np
+
+
+#update gradio to faster streaming
+#download for mecab
+os.system('python -m unidic download')
+
+# By using XTTS you agree to CPML license https://coqui.ai/cpml
+os.environ["COQUI_TOS_AGREED"] = "1"
+
+# langid is used to detect language for longer text
+# Most users expect text to be their own language, there is checkbox to disable it
+import langid
+import base64
+import csv
+from io import StringIO
+import datetime
+import re
+
+from scipy.io.wavfile import write
+from pydub import AudioSegment
+
+from TTS.api import TTS
+from TTS.tts.configs.xtts_config import XttsConfig
+from TTS.tts.models.xtts import Xtts
+from TTS.utils.generic_utils import get_user_data_dir
+
+HF_TOKEN = os.environ.get("HF_TOKEN")
+
+from huggingface_hub import HfApi
+# will use api to restart space on a unrecoverable error
+api = HfApi(token=HF_TOKEN)
+repo_id = "coqui/xtts"
+
+# This will trigger downloading model
+print("Downloading if not downloaded Coqui XTTS V2")
+from TTS.utils.manage import ModelManager
+
+model_name = "tts_models/multilingual/multi-dataset/xtts_v2"
+ModelManager().download_model(model_name)
+model_path = os.path.join(get_user_data_dir("tts"), model_name.replace("/", "--"))
+print("XTTS downloaded")
+
+config = XttsConfig()
+config.load_json(os.path.join(model_path, "config.json"))
+
+model = Xtts.init_from_config(config)
+model.load_checkpoint(
+ config,
+ checkpoint_path=os.path.join(model_path, "model.pth"),
+ vocab_path=os.path.join(model_path, "vocab.json"),
+ eval=True,
+ use_deepspeed=True,
+)
+model.cuda()
+
+# This is for debugging purposes only
+DEVICE_ASSERT_DETECTED = 0
+DEVICE_ASSERT_PROMPT = None
+DEVICE_ASSERT_LANG = None
+
+supported_languages = config.languages
+def numpy_to_mp3(audio_array, sampling_rate):
+ # Normalize audio_array if it's floating-point
+ if np.issubdtype(audio_array.dtype, np.floating):
+ max_val = np.max(np.abs(audio_array))
+ audio_array = (audio_array / max_val) * 32767 # Normalize to 16-bit range
+ audio_array = audio_array.astype(np.int16)
+
+ # Create an audio segment from the numpy array
+ audio_segment = AudioSegment(
+ audio_array.tobytes(),
+ frame_rate=sampling_rate,
+ sample_width=audio_array.dtype.itemsize,
+ channels=1
+ )
+
+ # Export the audio segment to MP3 bytes - use a high bitrate to maximise quality
+ mp3_io = io.BytesIO()
+ audio_segment.export(mp3_io, format="mp3", bitrate="320k")
+
+ # Get the MP3 bytes
+ mp3_bytes = mp3_io.getvalue()
+ mp3_io.close()
+
+ return mp3_bytes
+
+def predict(
+ prompt,
+ language,
+ audio_file_pth,
+ mic_file_path,
+ use_mic,
+ voice_cleanup,
+ no_lang_auto_detect,
+ agree,
+):
+ if agree == True:
+ if language not in supported_languages:
+ gr.Warning(
+ f"Language you put {language} in is not in is not in our Supported Languages, please choose from dropdown"
+ )
+
+ return (
+ None,
+ )
+
+ language_predicted = langid.classify(prompt)[
+ 0
+ ].strip() # strip need as there is space at end!
+
+ # tts expects chinese as zh-cn
+ if language_predicted == "zh":
+ # we use zh-cn
+ language_predicted = "zh-cn"
+
+ print(f"Detected language:{language_predicted}, Chosen language:{language}")
+
+ # After text character length 15 trigger language detection
+ if len(prompt) > 15:
+ # allow any language for short text as some may be common
+ # If user unchecks language autodetection it will not trigger
+ # You may remove this completely for own use
+ if language_predicted != language and not no_lang_auto_detect:
+ # Please duplicate and remove this check if you really want this
+ # Or auto-detector fails to identify language (which it can on pretty short text or mixed text)
+ gr.Warning(
+ f"It looks like your text isn’t the language you chose , if you’re sure the text is the same language you chose, please check disable language auto-detection checkbox"
+ )
+
+ return (
+ None,
+ )
+
+ if use_mic == True:
+ if mic_file_path is not None:
+ speaker_wav = mic_file_path
+ else:
+ gr.Warning(
+ "Please record your voice with Microphone, or uncheck Use Microphone to use reference audios"
+ )
+ return (
+ None,
+ )
+
+ else:
+ speaker_wav = audio_file_pth
+
+ # Filtering for microphone input, as it has BG noise, maybe silence in beginning and end
+ # This is fast filtering not perfect
+
+ # Apply all on demand
+ lowpassfilter = denoise = trim = loudness = True
+
+ if lowpassfilter:
+ lowpass_highpass = "lowpass=8000,highpass=75,"
+ else:
+ lowpass_highpass = ""
+
+ if trim:
+ # better to remove silence in beginning and end for microphone
+ trim_silence = "areverse,silenceremove=start_periods=1:start_silence=0:start_threshold=0.02,areverse,silenceremove=start_periods=1:start_silence=0:start_threshold=0.02,"
+ else:
+ trim_silence = ""
+
+ if voice_cleanup:
+ try:
+ out_filename = (
+ speaker_wav + str(uuid.uuid4()) + ".wav"
+ ) # ffmpeg to know output format
+
+ # we will use newer ffmpeg as that has afftn denoise filter
+ shell_command = f"./ffmpeg -y -i {speaker_wav} -af {lowpass_highpass}{trim_silence} {out_filename}".split(
+ " "
+ )
+
+ command_result = subprocess.run(
+ [item for item in shell_command],
+ capture_output=False,
+ text=True,
+ check=True,
+ )
+ speaker_wav = out_filename
+ print("Filtered microphone input")
+ except subprocess.CalledProcessError:
+ # There was an error - command exited with non-zero code
+ print("Error: failed filtering, use original microphone input")
+ else:
+ speaker_wav = speaker_wav
+
+ if len(prompt) < 2:
+ gr.Warning("Please give a longer prompt text")
+ return (
+ None,
+ )
+ if len(prompt) > 1000:
+ gr.Warning(
+ "Text length limited to 200 characters for this demo, please try shorter text. You can clone this space and edit code for your own usage"
+ )
+ return (
+ None,
+ )
+ global DEVICE_ASSERT_DETECTED
+ if DEVICE_ASSERT_DETECTED:
+ global DEVICE_ASSERT_PROMPT
+ global DEVICE_ASSERT_LANG
+ # It will likely never come here as we restart space on first unrecoverable error now
+ print(
+ f"Unrecoverable exception caused by language:{DEVICE_ASSERT_LANG} prompt:{DEVICE_ASSERT_PROMPT}"
+ )
+
+ # HF Space specific.. This error is unrecoverable need to restart space
+ space = api.get_space_runtime(repo_id=repo_id)
+ if space.stage != "BUILDING":
+ api.restart_space(repo_id=repo_id)
+ else:
+ print("TRIED TO RESTART but space is building")
+
+ try:
+ metrics_text = ""
+ t_latent = time.time()
+
+ # note diffusion_conditioning not used on hifigan (default mode), it will be empty but need to pass it to model.inference
+ try:
+ (
+ gpt_cond_latent,
+ speaker_embedding,
+ ) = model.get_conditioning_latents(audio_path=speaker_wav, gpt_cond_len=30, gpt_cond_chunk_len=4, max_ref_length=60)
+ except Exception as e:
+ print("Speaker encoding error", str(e))
+ gr.Warning(
+ "It appears something wrong with reference, did you unmute your microphone?"
+ )
+ return (
+ None,
+ )
+
+ latent_calculation_time = time.time() - t_latent
+ # metrics_text=f"Embedding calculation time: {latent_calculation_time:.2f} seconds\n"
+
+ # temporary comma fix
+ prompt = re.sub("([^\x00-\x7F]|\w)(\.|\。|\?)", r"\1 \2\2", prompt)
+
+ wav_chunks = []
+ ## Direct mode
+ """
+ print("I: Generating new audio...")
+ t0 = time.time()
+ out = model.inference(
+ prompt,
+ language,
+ gpt_cond_latent,
+ speaker_embedding,
+ repetition_penalty=5.0,
+ temperature=0.75,
+ )
+ inference_time = time.time() - t0
+ print(f"I: Time to generate audio: {round(inference_time*1000)} milliseconds")
+ metrics_text+=f"Time to generate audio: {round(inference_time*1000)} milliseconds\n"
+ real_time_factor= (time.time() - t0) / out['wav'].shape[-1] * 24000
+ print(f"Real-time factor (RTF): {real_time_factor}")
+ metrics_text+=f"Real-time factor (RTF): {real_time_factor:.2f}\n"
+ torchaudio.save("output.wav", torch.tensor(out["wav"]).unsqueeze(0), 24000)
+ """
+ print("I: Generating new audio in streaming mode...")
+ t0 = time.time()
+ chunks = model.inference_stream(
+ prompt,
+ language,
+ gpt_cond_latent,
+ speaker_embedding,
+ repetition_penalty=7.0,
+ temperature=0.85,
+ )
+
+ first_chunk = True
+ for i, chunk in enumerate(chunks):
+ if first_chunk:
+ first_chunk_time = time.time() - t0
+ metrics_text += f"Latency to first audio chunk: {round(first_chunk_time*1000)} milliseconds\n"
+ first_chunk = False
+
+ # Convert chunk to numpy array and return it
+ chunk_np = chunk.cpu().numpy()
+ print('chunk',i)
+ yield (24000, chunk_np)
+ wav_chunks.append(chunk)
+ 
+ print(f"Received chunk {i} of audio length {chunk.shape[-1]}")
+ inference_time = time.time() - t0
+ print(
+ f"I: Time to generate audio: {round(inference_time*1000)} milliseconds"
+ )
+ # metrics_text += (
+ # f"Time to generate audio: {round(inference_time*1000)} milliseconds\n"
+ #)
+
+ except RuntimeError as e:
+ if "device-side assert" in str(e):
+ # cannot do anything on cuda device side error, need tor estart
+ print(
+ f"Exit due to: Unrecoverable exception caused by language:{language} prompt:{prompt}",
+ flush=True,
+ )
+ gr.Warning("Unhandled Exception encounter, please retry in a minute")
+ print("Cuda device-assert Runtime encountered need restart")
+ if not DEVICE_ASSERT_DETECTED:
+ DEVICE_ASSERT_DETECTED = 1
+ DEVICE_ASSERT_PROMPT = prompt
+ DEVICE_ASSERT_LANG = language
+
+ # just before restarting save what caused the issue so we can handle it in future
+ # Uploading Error data only happens for unrecovarable error
+ error_time = datetime.datetime.now().strftime("%d-%m-%Y-%H:%M:%S")
+ error_data = [
+ error_time,
+ prompt,
+ language,
+ audio_file_pth,
+ mic_file_path,
+ use_mic,
+ voice_cleanup,
+ no_lang_auto_detect,
+ agree,
+ ]
+ error_data = [str(e) if type(e) != str else e for e in error_data]
+ print(error_data)
+ print(speaker_wav)
+ write_io = StringIO()
+ csv.writer(write_io).writerows([error_data])
+ csv_upload = write_io.getvalue().encode()
+
+ filename = error_time + "_" + str(uuid.uuid4()) + ".csv"
+ print("Writing error csv")
+ error_api = HfApi()
+ error_api.upload_file(
+ path_or_fileobj=csv_upload,
+ path_in_repo=filename,
+ repo_id="coqui/xtts-flagged-dataset",
+ repo_type="dataset",
+ )
+
+ # speaker_wav
+ print("Writing error reference audio")
+ speaker_filename = (
+ error_time + "_reference_" + str(uuid.uuid4()) + ".wav"
+ )
+ error_api = HfApi()
+ error_api.upload_file(
+ path_or_fileobj=speaker_wav,
+ path_in_repo=speaker_filename,
+ repo_id="coqui/xtts-flagged-dataset",
+ repo_type="dataset",
+ )
+
+ # HF Space specific.. This error is unrecoverable need to restart space
+ space = api.get_space_runtime(repo_id=repo_id)
+ if space.stage != "BUILDING":
+ api.restart_space(repo_id=repo_id)
+ else:
+ print("TRIED TO RESTART but space is building")
+
+ else:
+ if "Failed to decode" in str(e):
+ print("Speaker encoding error", str(e))
+ gr.Warning(
+ "It appears something wrong with reference, did you unmute your microphone?"
+ )
+ else:
+ print("RuntimeError: non device-side assert error:", str(e))
+ gr.Warning("Something unexpected happened please retry again.")
+ return (
+ None,
+ )
+
+ else:
+ gr.Warning("Please accept the Terms & Condition!")
+ return (
+ None,
+ )