xVASynth-TTS / app.py
Pendrokar's picture
catch model raised except
a65b632
raw
history blame
4.32 kB
import os
import sys
import time
import requests
from subprocess import Popen, PIPE
import threading
# from huggingface_hub import hf_hub_download
import gradio as gr
def run_xvaserver():
try:
# start the process without waiting for a response
print('Running xVAServer subprocess...\n')
xvaserver = Popen(['python', 'server.py'], stdout=PIPE, stderr=PIPE, universal_newlines=True)
except:
print('Could not run xVASynth.')
sys.exit(0)
# Wait for a moment to ensure the server starts up
time.sleep(10)
# Check if the server is running
if xvaserver.poll() is not None:
print("Web server failed to start.")
sys.exit(0)
# contact local xVASynth server; ~2 second timeout
print('Attempting to connect to xVASynth...')
try:
response = requests.get('http://0.0.0.0:8008')
response.raise_for_status() # If the response contains an HTTP error status code, raise an exception
except requests.exceptions.RequestException as err:
print('Failed to connect!')
return
print('xVAServer running on port 8008')
# Read and print stdout and stderr of the subprocess
while True:
output = xvaserver.stdout.readline()
if output == '' and xvaserver.poll() is not None:
break
if output:
print(output.strip())
error = xvaserver.stderr.readline()
if error == '' and xvaserver.poll() is not None:
break
if error:
print(error.strip(), file=sys.stderr)
# Wait for the process to exit
xvaserver.wait()
def load_model():
model_name = "Pendrokar/TorchMoji"
# model_path = hf_hub_download(repo_id=model_name, filename="ccby_nvidia_hifi_6670_M.pt")
# model_json_path = hf_hub_download(repo_id=model_name, filename="ccby_nvidia_hifi_6670_M.json")
model_path = '/tmp/hfcache/models--Pendrokar--xvapitch_nvidia_6670/snapshots/2e138a7c459fb1cb1182dd7bc66813f5325d30fd/ccby_nvidia_hifi_6670_M.pt'
model_json_path = '/tmp/hfcache/models--Pendrokar--xvapitch_nvidia_6670/snapshots/2e138a7c459fb1cb1182dd7bc66813f5325d30fd/ccby_nvidia_hifi_6670_M.json'
try:
os.symlink(model_path, os.path.join('./models/ccby/', os.path.basename(model_path)))
os.symlink(model_json_path, os.path.join('./models/ccby/', os.path.basename(model_json_path)))
except:
print('Failed creating symlinks, they probably already exist')
model_type = 'xVAPitch'
language = 'en'
data = {
'outputs': None,
'version': '3.0',
'model': 'ccby/ccby_nvidia_hifi_6670_M',
'modelType': model_type,
'base_lang': language,
'pluginsContext': '{}',
}
try:
response = requests.post('http://0.0.0.0:8008/loadModel', json=data)
response.raise_for_status() # If the response contains an HTTP error status code, raise an exception
except requests.exceptions.RequestException as err:
print('Failed to load voice model!')
return
def predict(input, pacing):
model_type = 'xVAPitch'
line = 'Test'
pace = pacing if pacing else 1.0
save_path = 'test.wav'
language = 'en'
base_speaker_emb = []
use_sr = 0
use_cleanup = 0
data = {
'pluginsContext': '{}',
'modelType': model_type,
'sequence': line,
'pace': pace,
'outfile': save_path,
'vocoder': 'n/a',
'base_lang': language,
'base_emb': base_speaker_emb,
'useSR': use_sr,
'useCleanup': use_cleanup,
}
try:
response = requests.post('http://0.0.0.0:8008/synthesize', json=data)
response.raise_for_status() # If the response contains an HTTP error status code, raise an exception
except requests.exceptions.RequestException as err:
print('Failed to synthesize!')
print('server.log contents:')
with open('server.log', 'r') as f:
print(f.read())
return 22100, os.open(save_path, "rb")
input_textbox = gr.Textbox(
label="Input Text",
lines=1,
autofocus=True
)
slider = gr.Slider(0.0, 2.0, value=1.0, step=0.1, label="Pacing")
gradio_app = gr.Interface(
predict,
[
input_textbox,
slider
],
outputs= "audio",
title="xVASynth (WIP)",
)
if __name__ == "__main__":
# Run the web server in a separate thread
web_server_thread = threading.Thread(target=run_xvaserver)
print('Starting xVAServer thread')
web_server_thread.start()
# load default voice model
load_model()
# predicted = predict('test', 1.0)
# print(predicted)
print('running Gradio interface')
gradio_app.launch()
# Wait for the web server thread to finish (shouldn't be reached in normal execution)
web_server_thread.join()