kromskii2/alltalk_tts
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
037d774f2cd549a5071345e89f70192ca6f4d051
alltalk_tts/tts_server.py
T
erew123 f3ea73dec9 Add files via upload
2024-06-06 22:55:57 +01:00

1677 lines
95 KiB
Python
Raw Blame History

import os
import sys
import json
import time
import librosa
import logging
import importlib
import subprocess
from pathlib import Path
logging.disable(logging.WARNING)
#####################
# Webserver Imports #
#####################
from contextlib import asynccontextmanager
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from fastapi.middleware.cors import CORSMiddleware
from fastapi import FastAPI, Form, Request, Response, Depends, HTTPException, Query
from fastapi.responses import JSONResponse, HTMLResponse, RedirectResponse, FileResponse, StreamingResponse
##########################
# Generation API Imports #
##########################
import re
import html
import uuid
import hashlib
import numpy as np
import soundfile as sf
from typing import Union, Dict, List
from pydantic import BaseModel, ValidationError, Field, field_validator
########################################################################################
# START-UP # Silence RVC warning about torch.nn.utils.weight_norm even though not used #
########################################################################################
import warnings
warnings.filterwarnings("ignore", category=UserWarning, module="torch.nn.utils.weight_norm")
warnings.filterwarnings("ignore", category=UserWarning, module="torch.nn.functional", lineno=5476)
####################
# Setup local path #
####################
this_dir = Path(__file__).parent.resolve() # Set this_dir as the current alltalk_tts folder
####################################################
# Load params and api_defailts from confignew.json #
####################################################
def load_config():
global branding, output_directory, debug_transcode, debug_tts, debug_tts_variables, engines_available, engine_loaded, debug_rvc, debug_concat, debug_openai
# Define the path to the confignew.json file & load in api_defaults & params
configfile_path = this_dir / "confignew.json"
with open(configfile_path, "r") as configfile:
configfile_data = json.load(configfile)
api_defaults = {
"api_text_filtering": configfile_data["api_def"]["api_text_filtering"],
"api_narrator_enabled": configfile_data["api_def"]["api_narrator_enabled"],
"api_text_not_inside": configfile_data["api_def"]["api_text_not_inside"],
"api_language": configfile_data["api_def"]["api_language"],
"api_output_file_name": configfile_data["api_def"]["api_output_file_name"],
"api_output_file_timestamp": configfile_data["api_def"]["api_output_file_timestamp"],
"api_autoplay": configfile_data["api_def"]["api_autoplay"],
"api_autoplay_volume": configfile_data["api_def"]["api_autoplay_volume"],
"api_port_number": configfile_data["api_def"]["api_port_number"],
"api_allowed_filter": configfile_data["api_def"]["api_allowed_filter"],
"api_length_stripping": configfile_data["api_def"]["api_length_stripping"],
"api_max_characters": configfile_data["api_def"]["api_max_characters"],
"api_use_legacy_api": configfile_data["api_def"]["api_use_legacy_api"],
"api_legacy_ip_address": configfile_data["api_def"]["api_legacy_ip_address"],
}
rvc_settings = {
"rvc_enabled": configfile_data["rvc_settings"]["rvc_enabled"],
"rvc_char_model_file": configfile_data["rvc_settings"]["rvc_char_model_file"],
"rvc_narr_model_file": configfile_data["rvc_settings"]["rvc_narr_model_file"],
"split_audio": configfile_data["rvc_settings"]["split_audio"],
"autotune": configfile_data["rvc_settings"]["autotune"],
"pitch": configfile_data["rvc_settings"]["pitch"],
"filter_radius": configfile_data["rvc_settings"]["filter_radius"],
"index_rate": configfile_data["rvc_settings"]["index_rate"],
"rms_mix_rate": configfile_data["rvc_settings"]["rms_mix_rate"],
"protect": configfile_data["rvc_settings"]["protect"],
"hop_length": configfile_data["rvc_settings"]["hop_length"],
"f0method": configfile_data["rvc_settings"]["f0method"],
"embedder_model": configfile_data["rvc_settings"]["embedder_model"],
"training_data_size": configfile_data["rvc_settings"]["training_data_size"],
}
branding = configfile_data.get("branding", "")
debug_transcode = configfile_data.get("debugging").get("debug_transcode")
debug_tts = configfile_data.get("debugging").get("debug_tts")
debug_tts_variables = configfile_data.get("debugging").get("debug_tts_variables")
debug_rvc = configfile_data.get("debugging").get("debug_rvc")
debug_concat = configfile_data.get("debugging").get("debug_concat")
debug_openai = configfile_data.get("debugging").get("debug_openai")
output_directory = this_dir / configfile_data.get("output_folder", "")
output_directory.mkdir(parents=True, exist_ok=True)
tts_engines_file = os.path.join(this_dir, "system", "tts_engines", "tts_engines.json")
with open(tts_engines_file, "r") as f:
tts_engines_data = json.load(f)
engines_available = [engine["name"] for engine in tts_engines_data["engines_available"]] # List of the available TTS engines from tts_engines.json
engine_loaded = tts_engines_data["engine_loaded"] # The currently set TTS engine from tts_engines.json
selected_model = tts_engines_data["selected_model"]
# Conditionally import infer_pipeline if rvc_enabled is true
if rvc_settings["rvc_enabled"]:
from system.tts_engines.rvc.infer.infer import infer_pipeline
else:
infer_pipeline = None
return configfile_data, api_defaults, rvc_settings, infer_pipeline
# Call to load in the params and api_defaults
params, api_defaults, rvc_settings, infer_pipeline = load_config()
####################
# Check for FFMPEG #
####################
def check_ffmpeg(this_dir):
if sys.platform == "win32":
# Check for ffmpeg.exe in the specified directory on Windows
ffmpeg_path = os.path.join(this_dir, "system", "win_ffmpeg", "ffmpeg.exe")
if os.path.exists(ffmpeg_path):
return True
else:
# Check for FFmpeg on Linux and macOS using the command line
try:
subprocess.run(['ffmpeg', '-version'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
return True
except (subprocess.CalledProcessError, FileNotFoundError):
pass
return False
# Check if FFmpeg is installed
ffmpeg_installed = check_ffmpeg(this_dir)
if not ffmpeg_installed:
print(f"[{branding}ENG] \033[92mTranscoding :\033[94m ffmpeg not found\033[0m")
print(f"[{branding}ENG] FFmpeg is not installed. Transcoding will be disabled.")
print(f"[{branding}ENG] Please install FFmpeg on your system.")
if sys.platform == "win32":
print(f"[{branding}ENG] \033[92mTranscoding :\033[94m ffmpeg not found\033[0m")
print(f"[{branding}ENG] Installation instructions for Windows:")
print(f"[{branding}ENG] Copy the 'ffmpeg.exe' file to '{os.path.join(this_dir, 'system', 'win_ffmpeg')}'")
else:
print(f"[{branding}ENG] \033[92mTranscoding :\033[94m ffmpeg not found\033[0m")
print(f"[{branding}ENG] Installation instructions:")
print(f"[{branding}ENG] Linux (Debian-based systems): Run 'sudo apt-get install ffmpeg' in the terminal.")
print(f"[{branding}ENG] macOS: Run 'brew install ffmpeg' in the terminal (requires Homebrew).")
# You can choose to exit the script or continue without FFmpeg
# exit(1)
if ffmpeg_installed:
from ffmpeg.asyncio import FFmpeg
print(f"[{branding}ENG] \033[92mTranscoding :\033[93m ffmpeg found\033[0m")
################################
# Check for portaudio on Linux #
################################
try:
import sounddevice as sd
sounddevice_installed=True
except OSError:
print(f"[{branding}Startup] \033[91mInfo\033[0m PortAudio library not found. If you wish to play TTS in standalone mode through the API suite")
print(f"[{branding}Startup] \033[91mInfo\033[0m please install PortAudio. This will not affect any other features or use of Alltalk.")
print(f"[{branding}Startup] \033[91mInfo\033[0m If you don't know what the API suite is, then this message is nothing to worry about.")
sounddevice_installed=False
if sys.platform.startswith('linux'):
print(f"[{branding}Startup] \033[91mInfo\033[0m On Linux, you can use the following command to install PortAudio:")
print(f"[{branding}Startup] \033[91mInfo\033[0m sudo apt-get install portaudio19-dev")
#######################################################################
# Attempt to import the ModelLoader class for the selected TTS engine #
#######################################################################
if engine_loaded in engines_available:
loader_module = importlib.import_module(f"system.tts_engines.{engine_loaded}.model_engine")
tts_class = getattr(loader_module, "tts_class")
# Setup model_engine as the way to call the functions within the Class.
model_engine = tts_class()
else:
raise ValueError(f"Invalid TTS engine: {engine_loaded}")
##########################################
# Run setup function in the model_engine #
##########################################
@asynccontextmanager
async def startup_shutdown(no_actual_value_it_demanded_something_be_here):
try:
await model_engine.setup()
except FileNotFoundError as e:
print(f"Error during setup: {e}. Continuing without the TTS model.")
yield
# Shutdown logic
###############################
# Setup FastAPI with Lifespan #
###############################
app = FastAPI(lifespan=startup_shutdown)
# Allow all origins, and set other CORS options
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Set this to the specific origins you want to allow
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
##############################
# API Endpoint - /api/reload #
##############################
model_currently_changing = False
@app.route("/api/reload", methods=["POST"])
async def apifunction_reload(request: Request):
global model_currently_changing
if model_currently_changing == True:
return Response(content=json.dumps({"status": "model-is currently changing"}), media_type="application/json")
model_currently_changing == True
requested_model = request.query_params.get("tts_method")
if requested_model not in model_engine.available_models:
model_currently_changing == False
return {"status": "error", "message": "Invalid TTS method specified"}
# Call the handle_tts_method_change method from the imported class
success = await model_engine.handle_tts_method_change(requested_model)
if success:
model_engine.current_model_loaded = requested_model
print(f"[{branding}ENG]")
print(f"[{branding}ENG] {branding} Server Ready.")
model_currently_changing == False
return Response(content=json.dumps({"status": "model-success"}), media_type="application/json")
else:
model_currently_changing == False
return Response(content=json.dumps({"status": "model-failure"}), media_type="application/json")
####################################
# API Endpoint - /api/enginereload #
####################################
import asyncio
uvicorn_server = None
def restart_self():
"""Restart the current script."""
#print(f"[{branding}ENG] Restarting subprocess...")
os.execv(sys.executable, ['python'] + sys.argv)
async def handle_restart():
global uvicorn_server
if uvicorn_server:
print(f"[{branding}ENG] Stopping uvicorn server...")
uvicorn_server.should_exit = True
uvicorn_server.force_exit = True
# Check if the server has stopped
while not uvicorn_server.is_stopped:
await asyncio.sleep(0.1) # Check every 100ms if server is stopped
restart_self()
@app.post("/api/enginereload")
async def apifunction_reload(request: Request):
requested_engine = request.query_params.get("engine")
if requested_engine not in engines_available:
return {"status": "error", "message": "Invalid TTS engine specified"}
print(f"[{branding}ENG]")
print(f"[{branding}ENG] \033[94mChanging model loaded. Please wait.\033[00m")
print(f"[{branding}ENG]")
# Update the tts_engines.json file
tts_engines_file = os.path.join(this_dir, "system", "tts_engines", "tts_engines.json")
with open(tts_engines_file, "r") as f:
tts_engines_data = json.load(f)
for engine in tts_engines_data["engines_available"]:
if engine["name"] == requested_engine:
tts_engines_data["engine_loaded"] = requested_engine
tts_engines_data["selected_model"] = engine["selected_model"]
break
with open(tts_engines_file, "w") as f:
json.dump(tts_engines_data, f)
#print(f"[{branding}ENG] Reloading the engine...")
# Start the restart process in the background
asyncio.create_task(handle_restart())
# Never goes past the above asyncio as the script is killed at that point
return Response(content=json.dumps({"status": "engine-success"}), media_type="application/json")
#######################################
# API Endpoint - /api/stop-generation #
#######################################
# When this endpoint it called it will set tts_stop_generation in the model_engine to True, which can be used to interrupt the current TTS generation.
@app.put("/api/stop-generation")
async def apifunction_stop_generation():
if model_engine.tts_generating_lock and not model_engine.tts_stop_generation:
model_engine.tts_stop_generation = True
return {"message": "Cancelling current TTS generation"}
#############################
# API Endpoint - /api/audio #
#############################
# Gives web access to the output files as downloads
@app.get("/audio/{filename}")
async def apifunction_get_audio(filename: str):
audio_path = this_dir / output_directory / filename
return FileResponse(audio_path)
##################################
# API Endpoint - /api/audiocache #
##################################
@app.get("/audiocache/{filename}")
async def apifunction_get_audio(filename: str):
audio_path = Path({output_directory}) / filename
if not audio_path.is_file():
raise HTTPException(status_code=404, detail="File not found")
response = FileResponse(path=audio_path, media_type='audio/wav', filename=filename)
# Set caching headers
response.headers["Cache-Control"] = "public, max-age=604800" # Cache for one week
response.headers["ETag"] = str(audio_path.stat().st_mtime) # Use the file's last modified time as a simple ETag
return response
##############################
# API Endpoint - /api/voices #
##############################
@app.get("/api/voices")
async def apifunction_get_voices():
if not model_engine.multivoice_capable:
return {"status": "error", "message": f"The currently loaded TTS engine '{model_engine.engine_loaded}' does not support multiple voices."}
available_voices = model_engine.voices_file_list()
return {"status": "success", "voices": available_voices}
#################################
# API Endpoint - /api/rvcvoices #
#################################
@app.get("/api/rvcvoices")
async def apifunction_get_voices():
global params, api_defaults, rvc_settings, infer_pipeline
params, api_defaults, rvc_settings, infer_pipeline = load_config()
if not rvc_settings["rvc_enabled"]:
return {"status": "success", "rvcvoices": ["Disabled"]}
# Define the directory path
directory = os.path.join(this_dir, "models", "rvc_voices")
# Check if the directory exists
if not os.path.exists(directory):
return {"status": "success", "rvcvoices": ["Disabled"]}
# List all .pth files recursively in the directory
pth_files = []
for root, dirs, files in os.walk(directory):
for file in files:
if file.endswith(".pth"):
pth_files.append(os.path.relpath(os.path.join(root, file), start=directory))
# If no .pth files found, return "Disabled"
if not pth_files:
return {"status": "success", "rvcvoices": ["Disabled"]}
# Sort the list alphabetically
pth_files.sort()
# Add "Disabled" at the beginning of the list
pth_files.insert(0, "Disabled")
return {"status": "success", "rvcvoices": pth_files}
#####################################
# API Endpoint - /api/reload_config #
#####################################
@app.get("/api/reload_config")
async def apifunction_reload_config():
global params, api_defaults, rvc_settings, infer_pipeline
model_engine.available_models = model_engine.scan_models_folder()
params, api_defaults, rvc_settings, infer_pipeline = load_config()
return Response("Config file reloaded successfully")
#############################
# API Endpoint - /api/ready #
#############################
@app.get("/api/ready")
async def apifunction_ready():
if model_engine.setup_has_run:
return Response("Ready")
else:
return Response("Unloaded")
#######################################
# API Endpoint - /api/currentsettings #
#######################################
@app.get('/api/currentsettings')
def apifunction_get_current_settings():
settings = {
"engines_available": engines_available, # Lists the currently available TTS engines
"current_engine_loaded": model_engine.engine_loaded, # Lists the currently loaded TTS engine
"models_available": [{"name": name} for name in model_engine.available_models.keys()], # Lists the currently available models that can be loaded on this TTS engine.
"current_model_loaded": model_engine.current_model_loaded, # Lists the currently loaded model
"manufacturer_name": model_engine.manufacturer_name, # The name of the person/company/group who made the current TTS engine
"audio_format": model_engine.audio_format, # The default audio format the the currently loaded TTS engine generates in
"deepspeed_capable": model_engine.deepspeed_capable, # Is the TTS engine capable of supporting DeepSpeed
"deepspeed_available": model_engine.deepspeed_available, # Was DeepSpeed detected as being available for use
"deepspeed_enabled": model_engine.deepspeed_enabled, # Is DeepSpeed current enabled or disabled
"generationspeed_capable": model_engine.generationspeed_capable, # Does the TTS engine support changing speed of generated TTS
"generationspeed_set": model_engine.generationspeed_set, # What is the currently set TTS speed
"lowvram_capable": model_engine.lowvram_capable, # Does the TTS engine support moving the model between VRAM and System Ram
"lowvram_enabled": model_engine.lowvram_enabled, # What is the currently enabled/disabled state of low vram
"pitch_capable": model_engine.pitch_capable, # Does the TTS engine support pitch on tts generation
"pitch_set": model_engine.pitch_set, # What is the currently set value for pitch
"repetitionpenalty_capable": model_engine.repetitionpenalty_capable, # Does the TTS engine support differnt repitition penalties
"repetitionpenalty_set": model_engine.repetitionpenalty_set, # What is the currently set repetition penalty
"streaming_capable": model_engine.streaming_capable, # Does the model support streaming TTS generation
"temperature_capable": model_engine.temperature_capable, # Does the TTS engine support model temperature settings
"temperature_set": model_engine.temperature_set, # What is the currently set temperature
"ttsengines_installed": model_engine.engine_installed, #
"languages_capable": model_engine.languages_capable, # Does the TTS engine support multi languages
"multivoice_capable": model_engine.multivoice_capable, # Does the TTS engine support multi voice models
"multimodel_capable": model_engine.multimodel_capable, # Does the engine support loading many different models
}
return settings
######################################
# API Endpoint - /api/lowvramsetting #
######################################
@app.post("/api/lowvramsetting")
async def apifunction_low_vram(request: Request, new_low_vram_value: bool):
if not model_engine.lowvram_capable:
return {"status": "error", "message": f"The currently loaded TTS engine '{model_engine.engine_loaded}' does not support lowvram."}
try:
if new_low_vram_value is None:
raise ValueError("Missing 'low_vram' parameter")
if model_engine.lowvram_enabled == new_low_vram_value:
return Response(content=json.dumps({"status": "success", "message": f"[{branding}Model] LowVRAM is already {'enabled' if new_low_vram_value else 'disabled'}.",}))
model_engine.lowvram_enabled = new_low_vram_value
await model_engine.unload_model()
if model_engine.cuda_is_available:
if model_engine.lowvram_enabled:
model_engine.device = "cpu"
print(f"[{branding}Engine] \033[94mLowVRAM Enabled.\033[0m Model will move between \033[93mVRAM(cuda) <> System RAM(cpu)\033[0m")
else:
model_engine.device = "cuda"
print(f"[{branding}Engine] \033[94mLowVRAM Disabled.\033[0m Model will stay in \033[93mVRAM(cuda)\033[0m")
await model_engine.setup()
else:
# Handle the case where CUDA is not available
print(f"[{branding}Engine] \033[91mError:\033[0m Nvidia CUDA is not available on this system. Unable to use LowVRAM mode.")
model_engine.lowvram_enabled = False
return Response(content=json.dumps({"status": "lowvram-success"}))
except Exception as e:
return Response(content=json.dumps({"status": "error", "message": str(e)}))
#################################
# API Endpoint - /api/deepspeed #
#################################
@app.post("/api/deepspeed")
async def deepspeed(request: Request, new_deepspeed_value: bool):
if not model_engine.deepspeed_capable or not model_engine.deepspeed_available:
return {"status": "error", "message": f"The currently loaded TTS engine '{model_engine.engine_loaded}' does not support DeepSpeed or DeepSpeed is not available on this system."}
try:
if new_deepspeed_value is None:
raise ValueError("Missing 'deepspeed' parameter")
if model_engine.deepspeed_enabled == new_deepspeed_value:
return Response(
content=json.dumps({"status": "success", "message": f"DeepSpeed is already {'enabled' if new_deepspeed_value else 'disabled'}.",}))
model_engine.deepspeed_enabled = new_deepspeed_value
await model_engine.handle_deepspeed_change(new_deepspeed_value)
return Response(content=json.dumps({"status": "deepspeed-success"}))
except Exception as e:
return Response(content=json.dumps({"status": "error", "message": str(e)}))
#################################
# API Endpoint - /api/voice2rvc #
#################################
@app.post("/api/voice2rvc")
async def voice2rvc(input_tts_path: str = Form(...), output_rvc_path: str = Form(...), pth_name: str = Form(...)):
try:
# Handle "Disabled" case
if pth_name.lower() in ["disabled", "disable"]:
print(f"[{branding}TTS] \033[94mVoice2RVC Convert: No voice was specified or the name was Disabled\033[0m")
return {"status": "error", "message": "No voice was specified or the name was Disabled"}
input_tts_path = Path(input_tts_path)
output_rvc_path = Path(output_rvc_path)
# Check if input file exists
if not input_tts_path.is_file():
raise HTTPException(status_code=400, detail=f"Input file {input_tts_path} does not exist.")
# Define the model path based on the selected model name
pth_path = this_dir / "models" / "rvc_voices" / pth_name
# Check if model file exists
if not pth_path.is_file():
raise HTTPException(status_code=400, detail=f"Model file {pth_path} does not exist.")
# Run the RVC conversion
result_path = run_voice2rvc(input_tts_path, output_rvc_path, pth_path)
if result_path:
return {"status": "success", "output_path": str(result_path)}
else:
raise HTTPException(status_code=500, detail="RVC conversion failed.")
except Exception as e:
print(f"Error during Voice2RVC conversion: {e}")
raise HTTPException(status_code=500, detail=str(e))
# Define the run_rvc function
def run_voice2rvc(input_tts_path, output_rvc_path, pth_path):
print(f"[{branding}GEN] \033[94mVoice2RVC Convert: Started\033[0m")
generate_start_time = time.time()
f0up_key = rvc_settings["pitch"]
filter_radius = rvc_settings["filter_radius"]
index_rate = rvc_settings["index_rate"]
rms_mix_rate = rvc_settings["rms_mix_rate"]
protect = rvc_settings["protect"]
hop_length = rvc_settings["hop_length"]
f0method = rvc_settings["f0method"]
split_audio = rvc_settings["split_audio"]
f0autotune = rvc_settings["autotune"]
embedder_model = rvc_settings["embedder_model"]
training_data_size = rvc_settings["training_data_size"]
# Convert path variables to strings
input_tts_path = str(input_tts_path)
pth_path = str(pth_path)
# Check if the model file exists
if not os.path.isfile(pth_path):
print(f"Model file {pth_path} does not exist. Exiting.")
return
# Get the directory of the model file
model_dir = os.path.dirname(pth_path)
# Find all .index files in the model directory
index_files = [file for file in os.listdir(model_dir) if file.endswith(".index")]
if len(index_files) == 1:
index_path = str(os.path.join(model_dir, index_files[0]))
else:
index_path = ""
# Call the infer_pipeline function
infer_pipeline(f0up_key, filter_radius, index_rate, rms_mix_rate, protect, hop_length, f0method,
input_tts_path, output_rvc_path, pth_path, index_path, split_audio, f0autotune, embedder_model, training_data_size, debug_rvc)
generate_end_time = time.time()
generate_elapsed_time = generate_end_time - generate_start_time
print(f"[{branding}GEN] \033[94mVoice2RVC Convert: \033[91m{generate_elapsed_time:.2f} seconds.\033[0m")
return output_rvc_path
#################################
# API Endpoint - /api/voice2rvc #
#################################
@app.post("/api/voice2rvc")
async def voice2rvc(input_tts_path: str = Form(...), output_rvc_path: str = Form(...), pth_name: str = Form(...)):
try:
input_tts_path = Path(input_tts_path)
output_rvc_path = Path(output_rvc_path)
# Check if input file exists
if not input_tts_path.is_file():
raise HTTPException(status_code=400, detail=f"Input file {input_tts_path} does not exist.")
# Define the model path based on the selected model name
pth_path = this_dir / "models" / "rvc_voices" / pth_name
# Check if model file exists
if not pth_path.is_file():
raise HTTPException(status_code=400, detail=f"Model file {pth_path} does not exist.")
# Run the RVC conversion
result_path = run_voice2rvc(input_tts_path, output_rvc_path, pth_path)
if result_path:
return {"status": "success", "output_path": str(result_path)}
else:
raise HTTPException(status_code=500, detail="RVC conversion failed.")
except Exception as e:
print(f"Error during Voice2RVC conversion: {e}")
raise HTTPException(status_code=500, detail=str(e))
# Define the run_rvc function
def run_voice2rvc(input_tts_path, output_rvc_path, pth_path):
print(f"[{branding}GEN] \033[94mVoice2RVC Convert: Started\033[0m")
generate_start_time = time.time()
f0up_key = rvc_settings["pitch"]
filter_radius = rvc_settings["filter_radius"]
index_rate = rvc_settings["index_rate"]
rms_mix_rate = rvc_settings["rms_mix_rate"]
protect = rvc_settings["protect"]
hop_length = rvc_settings["hop_length"]
f0method = rvc_settings["f0method"]
split_audio = rvc_settings["split_audio"]
f0autotune = rvc_settings["autotune"]
embedder_model = rvc_settings["embedder_model"]
training_data_size = rvc_settings["training_data_size"]
# Convert path variables to strings
input_tts_path = str(input_tts_path)
pth_path = str(pth_path)
# Check if the model file exists
if not os.path.isfile(pth_path):
print(f"Model file {pth_path} does not exist. Exiting.")
return
# Get the directory of the model file
model_dir = os.path.dirname(pth_path)
# Find all .index files in the model directory
index_files = [file for file in os.listdir(model_dir) if file.endswith(".index")]
if len(index_files) == 1:
index_path = str(os.path.join(model_dir, index_files[0]))
else:
index_path = ""
# Call the infer_pipeline function
infer_pipeline(f0up_key, filter_radius, index_rate, rms_mix_rate, protect, hop_length, f0method,
input_tts_path, output_rvc_path, pth_path, index_path, split_audio, f0autotune, embedder_model, training_data_size, debug_rvc)
generate_end_time = time.time()
generate_elapsed_time = generate_end_time - generate_start_time
print(f"[{branding}GEN] \033[94mVoice2RVC Convert: \033[91m{generate_elapsed_time:.2f} seconds.\033[0m")
return output_rvc_path
##################################
# Transcode between file formats #
##################################
async def transcode_audio(input_file, output_format):
print(f"[{branding}Debug] *************************************************",) if debug_transcode else None
print(f"[{branding}Debug] transcode_audio function called (debug_transcode)",) if debug_transcode else None
print(f"[{branding}Debug] *************************************************",) if debug_transcode else None
print(f"[{branding}Debug] Input file : {input_file}") if debug_transcode else None
print(f"[{branding}Debug] Output format : {output_format}") if debug_transcode else None
if output_format == "Disabled":
print(f"[{branding}Debug] Transcode format is set to Disabled so skipping transcode.") if debug_transcode else None
return input_file
if not ffmpeg_installed:
print(f"[{branding}TTS] FFmpeg is not installed. Format conversion is not possible.")
raise Exception("FFmpeg is not installed. Format conversion is not possible.")
# Get the file extension of the input file
input_extension = os.path.splitext(input_file)[1][1:].lower()
print(f"[{branding}Debug] Input file extension: {input_extension}") if debug_transcode else None
# Check if the input extension matches the requested output format
if input_extension == output_format.lower():
print(f"[{branding}Debug] Input file is already in the requested format: {output_format}") if debug_transcode else None
return input_file
output_file = os.path.splitext(input_file)[0] + f".{output_format}"
print(f"[{branding}Debug] Output file: {output_file}") if debug_transcode else None
ffmpeg_path = "ffmpeg" # Default path for Linux and macOS
if sys.platform == "win32":
ffmpeg_path = os.path.join(this_dir, "system", "win_ffmpeg", "ffmpeg.exe")
ffmpeg = (FFmpeg(ffmpeg_path).option("y").input(input_file).output(output_file))
print(f"[{branding}Debug] Transcoding to {output_format}") if debug_transcode else None
if output_format == "opus":
print(f"[{branding}Debug] Configuring Opus options") if debug_transcode else None
ffmpeg.output(output_file, {"codec:a": "libopus", "b:a": "128k", "vbr": "on", "compression_level": 10, "frame_duration": 60, "application": "voip"})
elif output_format == "aac":
print(f"[{branding}Debug] Configuring AAC options") if debug_transcode else None
ffmpeg.output(output_file, {"codec:a": "aac", "b:a": "192k"})
elif output_format == "flac":
print(f"[{branding}Debug] Configuring FLAC options") if debug_transcode else None
ffmpeg.output(output_file, {"codec:a": "flac", "compression_level": 8})
elif output_format == "wav":
print(f"[{branding}Debug] Configuring WAV options") if debug_transcode else None
ffmpeg.output(output_file, {"codec:a": "pcm_s16le"})
elif output_format == "mp3":
print(f"[{branding}Debug] Configuring MP3 options") if debug_transcode else None
ffmpeg.output(output_file, {"codec:a": "libmp3lame", "b:a": "192k"})
else:
print(f"[{branding}TTS] Unsupported output format: {output_format}")
raise ValueError(f"Unsupported output format: {output_format}")
try:
print(f"[{branding}Debug] Starting transcoding process") if debug_transcode else None
await ffmpeg.execute()
print(f"[{branding}Debug] Transcoding completed successfully") if debug_transcode else None
except Exception as e:
print(f"[{branding}TTS] Error occurred during transcoding: {str(e)}")
raise
print(f"[{branding}Debug] Deleting original input file") if debug_transcode else None
os.remove(input_file)
print(f"[{branding}Debug] Transcoding process completed") if debug_transcode else None
print(f"[{branding}Debug] Transcoded file: {output_file}") if debug_transcode else None
return output_file
##############################
# Central Transcode function #
##############################
async def transcode_audio_if_necessary(output_file, model_audio_format, output_audio_format):
if debug_transcode:
print(f"[{branding}Debug] model_engine.audio_format is:", model_audio_format)
print(f"[{branding}Debug] audio format is:", output_audio_format)
print(f"[{branding}Debug] Entering the transcode condition")
try:
if debug_transcode:
print(f"[{branding}Debug] Calling transcode_audio function")
output_file = await transcode_audio(output_file, output_audio_format)
if debug_transcode:
print(f"[{branding}Debug] Transcode completed successfully")
except Exception as e:
print(f"[{branding}Debug] Error occurred during transcoding:", str(e))
raise
if debug_transcode:
print(f"[{branding}Debug] Transcode condition completed")
# Update the output file path and URLs based on the transcoded file
if debug_tts:
print(f"[{branding}Debug] Updating output file paths and URLs")
if params["api_def"]["api_use_legacy_api"]:
output_file_url = f'http://{api_defaults["api_legacy_ip_address"]}:{api_defaults["api_port_number"]}/audio/{os.path.basename(output_file)}'
output_cache_url = f'http://{api_defaults["api_legacy_ip_address"]}:{api_defaults["api_port_number"]}/audiocache/{os.path.basename(output_file)}'
else:
output_file_url = f'/audio/{os.path.basename(output_file)}'
output_cache_url = f'/audiocache/{os.path.basename(output_file)}'
if debug_tts:
print(f"[{branding}Debug] Output file paths and URLs updated")
return output_file, output_file_url, output_cache_url
##############################
#### Streaming Generation ####
##############################
@app.get("/api/tts-generate-streaming", response_class=StreamingResponse)
async def apifunction_generate_streaming(text: str, voice: str, language: str, output_file: str):
if model_engine.streaming_capable == False:
print(f"[{branding}GEN] Streaming is unavailable for this TTS Engine")
try:
output_file_path = f'{this_dir / output_directory / output_file}.{model_engine.audio_format}'
stream = await generate_audio(text, voice, language, model_engine.temperature_set, model_engine.repetitionpenalty_set, 1.0, 1.0, output_file_path, streaming=True)
return StreamingResponse(stream, media_type="audio/wav")
except Exception as e:
print(f"An error occurred: {e}")
return JSONResponse(content={"error": "An error occurred"}, status_code=500)
@app.post("/api/tts-generate-streaming", response_class=JSONResponse)
async def tts_generate_streaming(request: Request, text: str = Form(...), voice: str = Form(...), language: str = Form(...), output_file: str = Form(...)):
if model_engine.streaming_capable == False:
print(f"[{branding}GEN] Streaming is unavailable for this TTS Engine")
try:
output_file_path = f'{this_dir / output_directory / output_file}.{model_engine.audio_format}'
await generate_audio(text, voice, language, model_engine.temperature_set, model_engine.repetitionpenalty_set, "1.0", "1.0", output_file_path, streaming=False)
return JSONResponse(content={"output_file_path": str(output_file)}, status_code=200)
except Exception as e:
print(f"An error occurred: {e}")
return JSONResponse(content={"error": "An error occurred"}, status_code=500)
###################################
# Central generate_audio function #
###################################
async def generate_audio(text, voice, language, temperature, repetition_penalty, speed, pitch, output_file, streaming=False):
# Get the async generator from the internal function
response = model_engine.generate_tts(text, voice, language, temperature, repetition_penalty, speed, pitch, output_file, streaming)
# If streaming, then return the generator as-is, otherwise just exhaust it and return
if streaming:
return response
async for _ in response:
pass
###########################
#### PREVIEW VOICE API ####
###########################
@app.post("/api/previewvoice/", response_class=JSONResponse)
async def apifunction_preview_voice(request: Request, voice: str = Form(...)):
try:
# Hardcoded settings
language = "en"
output_file_name = "api_preview_voice"
# Clean the voice filename for inclusion in the text
clean_voice_filename = re.sub(r'\.wav$', '', voice.replace(' ', '_'))
clean_voice_filename = re.sub(r'[^a-zA-Z0-9]', ' ', clean_voice_filename)
# Generate the audio
text = f"Hello, this is a preview of voice {clean_voice_filename}."
# Generate the audio
output_file_path = this_dir / output_directory / f'{output_file_name}.{model_engine.audio_format}'
await generate_audio(text, voice, language, model_engine.temperature_set, model_engine.repetitionpenalty_set, 1.0, 0, output_file_path, streaming=False)
# Generate the URL
output_file_url = f'/audio/{output_file_name}.{model_engine.audio_format}'
# Return the response with both local file path and URL
return JSONResponse(
content={
"status": "generate-success",
"output_file_path": str(output_file_path),
"output_file_url": str(output_file_url),
},
status_code=200,
)
except Exception as e:
print(f"An error occurred: {e}")
return JSONResponse(content={"error": "An error occurred"}, status_code=500)
##################################################################
# API Endpoint - OpenAI Speech API compatable endpoint Validator #
##################################################################
class OpenAIInput(BaseModel):
model: str = Field(..., description="The TTS model to use. Currently ignored.")
input: str = Field(..., max_length=4096, description="The text to generate audio for.")
voice: str = Field(..., description="The voice to use when generating the audio.")
response_format: str = Field(default="wav", description="The format of the audio. Currently only 'wav' is supported.")
speed: float = Field(default=1.0, ge=0.25, le=4.0, description="The speed of the generated audio.")
@field_validator('voice')
def validate_voice(cls, value):
supported_voices = ["alloy", "echo", "fable", "nova", "onyx", "shimmer"]
if value not in supported_voices:
raise ValueError(f'Voice must be one of {supported_voices}')
return value
class OpenAIGenerator:
@staticmethod
def validate_input(json_data: dict) -> Union[None, str]:
try:
OpenAIInput(**json_data)
return None
except ValidationError as e:
errors = []
for err in e.errors():
field = err['loc'][0]
message = err['msg']
description = OpenAIInput.model_fields[field].field_info.description
errors.append(f"Error in field '{field}': {message}. Description: {description}")
return ', '.join(errors)
#########################################################################
# API Endpoint - OpenAI Speech API compatable endpoint /v1/audio/speech #
#########################################################################
@app.post("/v1/audio/speech", response_class=JSONResponse)
async def openai_tts_generate(request: Request):
try:
json_data = await request.json()
print(f"[{branding}Debug] Received JSON data: {json_data}") if debug_openai else None
validation_error = OpenAIGenerator.validate_input(json_data)
if validation_error:
print(f"Validation error: {validation_error}") if debug_openai else None
return JSONResponse(content={"error": validation_error}, status_code=400)
model = json_data["model"] # Currently ignored
input_text = json_data["input"]
voice = json_data["voice"]
response_format = json_data["response_format"].lower()
speed = json_data["speed"]
print(f"[{branding}Debug] Input text: {input_text}") if debug_openai else None
print(f"[{branding}Debug] Voice: {voice}") if debug_openai else None
print(f"[{branding}Debug] Speed: {speed}") if debug_openai else None
cleaned_string = html.unescape(standard_filtering(input_text))
# Map the OpenAI voice to the corresponding internal voice
voice_mapping = {
"alloy": model_engine.openai_alloy,
"echo": model_engine.openai_echo,
"fable": model_engine.openai_fable,
"nova": model_engine.openai_nova,
"onyx": model_engine.openai_onyx,
"shimmer": model_engine.openai_shimmer
}
mapped_voice = voice_mapping.get(voice)
if not mapped_voice:
print(f"Unsupported voice: {voice}")
return JSONResponse(content={"error": "Unsupported voice"}, status_code=400)
print(f"[{branding}Debug]Mapped voice: {mapped_voice}") if debug_openai else None
# Generate the audio
# Generate a unique filename
unique_id = uuid.uuid4()
timestamp = int(time.time())
output_file_path = f'{this_dir / output_directory / f"openai_output_{unique_id}_{timestamp}.{model_engine.audio_format}"}'
await generate_audio(cleaned_string, mapped_voice, "en", model_engine.temperature_set,
model_engine.repetitionpenalty_set, speed, model_engine.pitch_set,
output_file_path, streaming=False)
print(f"[{branding}Debug] Audio generated at: {output_file_path}") if debug_openai else None
if rvc_settings["rvc_enabled"]:
if rvc_settings["rvc_char_model_file"].lower() in ["disabled", "disable"]:
print(f"[{branding}Debug] Pass rvccharacter_voice_gen") if debug_openai else None
pass # Skip RVC processing for character part
else:
print(f"[{branding}Debug] send to rvc") if debug_openai else None
pth_path = this_dir / "models" / "rvc_voices" / rvc_settings["rvc_char_model_file"]
run_rvc(output_file_path, pth_path, infer_pipeline)
# Transcode the audio to the requested format
transcoded_file_path = await transcode_for_openai(output_file_path, response_format)
print(f"[{branding}Debug] Audio transcoded to: {transcoded_file_path}") if debug_openai else None
# Return the audio file
return FileResponse(transcoded_file_path, media_type=f"audio/{response_format}", filename=f"output.{response_format}")
except Exception as e:
print(f"[{branding}Debug] An error occurred: {str(e)}") if debug_openai else None
return JSONResponse(content={"error": "An error occurred"}, status_code=500)
###########################################################################
# API Endpoint - OpenAI Speech API compatable endpoint Transcode Function #
###########################################################################
async def transcode_for_openai(input_file, output_format):
print(f"[{branding}Debug] ************************************") if debug_openai else None
print(f"[{branding}Debug] transcode_for_openai function called") if debug_openai else None
print(f"[{branding}Debug] ************************************") if debug_openai else None
print(f"[{branding}Debug] Input file : {input_file}") if debug_openai else None
print(f"[{branding}Debug] Output format : {output_format}") if debug_openai else None
if not ffmpeg_installed:
print(f"[{branding}TTS] FFmpeg is not installed. Format conversion is not possible.")
raise Exception("FFmpeg is not installed. Format conversion is not possible.")
# Get the file extension of the input file
input_extension = os.path.splitext(input_file)[1][1:].lower()
print(f"[{branding}Debug] Input file extension: {input_extension}") if debug_openai else None
# Check if the input extension matches the requested output format
if input_extension == output_format.lower():
print(f"[{branding}Debug] Input file is already in the requested format: {output_format}") if debug_openai else None
return input_file
output_file = os.path.splitext(input_file)[0] + f".{output_format}"
print(f"[{branding}Debug] Output file: {output_file}") if debug_openai else None
ffmpeg_path = "ffmpeg" # Default path for Linux and macOS
if sys.platform == "win32":
ffmpeg_path = os.path.join(this_dir, "system", "win_ffmpeg", "ffmpeg.exe")
ffmpeg = (FFmpeg(ffmpeg_path).option("y").input(input_file).output(output_file))
print(f"[{branding}Debug] Transcoding to {output_format}") if debug_openai else None
if output_format == "opus":
print(f"[{branding}Debug] Configuring Opus options") if debug_openai else None
ffmpeg.output(output_file, {"codec:a": "libopus", "b:a": "128k", "vbr": "on", "compression_level": 10, "frame_duration": 60, "application": "voip"})
elif output_format == "aac":
print(f"[{branding}Debug] Configuring AAC options") if debug_openai else None
ffmpeg.output(output_file, {"codec:a": "aac", "b:a": "192k"})
elif output_format == "flac":
print(f"[{branding}Debug] Configuring FLAC options") if debug_openai else None
ffmpeg.output(output_file, {"codec:a": "flac", "compression_level": 8})
elif output_format == "wav":
print(f"[{branding}Debug] Configuring WAV options") if debug_openai else None
ffmpeg.output(output_file, {"codec:a": "pcm_s16le"})
elif output_format == "mp3":
print(f"[{branding}Debug] Configuring MP3 options") if debug_openai else None
ffmpeg.output(output_file, {"codec:a": "libmp3lame", "b:a": "192k"})
elif output_format in ["ogg", "m4a"]:
if output_format == "ogg":
print(f"[{branding}Debug] Configuring OGG options") if debug_openai else None
ffmpeg.output(output_file, {"codec:a": "libvorbis"})
elif output_format == "m4a":
print(f"[{branding}Debug] Configuring M4A options") if debug_openai else None
ffmpeg.output(output_file, {"codec:a": "aac", "b:a": "192k"})
else:
print(f"[{branding}TTS] Unsupported output format: {output_format}")
raise ValueError(f"Unsupported output format: {output_format}")
try:
print(f"[{branding}Debug] Starting transcoding process") if debug_openai else None
await ffmpeg.execute()
print(f"[{branding}Debug] Transcoding completed successfully") if debug_openai else None
except Exception as e:
print(f"[{branding}TTS] Error occurred during transcoding: {str(e)}")
raise
print(f"[{branding}Debug] Transcoding process completed") if debug_openai else None
print(f"[{branding}Debug] Transcoded file: {output_file}") if debug_openai else None
return output_file
#######################
# Play at the console #
#######################
def play_audio(file_path, volume):
print(f"[{branding}GEN] \033[94m File path is: {file_path}") if debug_tts else None
print(f"[{branding}GEN] \033[94m Volume is: {volume}") if debug_tts else None
normalized_file_path = os.path.normpath(file_path)
print(f"[{branding}GEN] \033[94m Normalized file path is: {normalized_file_path}") if debug_tts else None
directory = os.path.dirname(normalized_file_path)
if os.path.isdir(directory):
print(f"[{branding}GEN] \033[94m Directory contents: {os.listdir(directory)}") if debug_tts else None
else:
print(f"[{branding}GEN] \033[94mError: Directory does not exist: {directory}") if debug_tts else None
if not os.path.isfile(normalized_file_path):
print(f"[{branding}GEN] \033[94mError: File does not exist: {normalized_file_path}\033[0m") if debug_tts else None
return
# Check for AAC format
if normalized_file_path.lower().endswith('.aac'):
print(f"[{branding}GEN] \033[94mPlay Audio : \033[0mAAC format files cannot be played at the console. Please choose another format")
return
try:
print(f"[{branding}GEN] \033[94mPlay Audio : \033[0mPlaying audio at console")
data, fs = sf.read(normalized_file_path)
sd.play(volume * data, fs)
sd.wait()
except Exception as e:
print(f"[{branding}GEN] \033[94mError playing audio file: {e}\033[0m")
class Request(BaseModel):
# Define the structure of the 'Request' class if needed
pass
class TTSGenerator:
@staticmethod
def validate_json_input(json_data: Union[Dict, str]) -> Union[None, str]:
try:
if isinstance(json_data, str):
json_data = json.loads(json_data)
JSONInput(**json_data)
return None # JSON is valid
except ValidationError as e:
return str(e)
#################################################################
# /api/tts-generate Generation API Endpoint Narration Filtering #
#################################################################
def process_text(text):
# Normalize HTML encoded quotes
text = html.unescape(text)
# Replace ellipsis with a single dot
text = re.sub(r'\.{3,}', '.', text)
# Pattern to identify combined narrator and character speech
combined_pattern = r'(\*[^*"]+\*|"[^"*]+")'
# List to hold parts of speech along with their type
ordered_parts = []
# Track the start of the next segment
start = 0
# Find all matches
for match in re.finditer(combined_pattern, text):
# Add the text before the match, if any, as ambiguous
if start < match.start():
ambiguous_text = text[start:match.start()].strip()
if ambiguous_text:
ordered_parts.append(('ambiguous', ambiguous_text))
# Add the matched part as either narrator or character
matched_text = match.group(0)
if matched_text.startswith('*') and matched_text.endswith('*'):
ordered_parts.append(('narrator', matched_text.strip('*').strip()))
elif matched_text.startswith('"') and matched_text.endswith('"'):
ordered_parts.append(('character', matched_text.strip('"').strip()))
else:
# In case of mixed or improperly formatted parts
if '*' in matched_text:
ordered_parts.append(('narrator', matched_text.strip('*').strip('"')))
else:
ordered_parts.append(('character', matched_text.strip('"').strip('*')))
# Update the start of the next segment
start = match.end()
# Add any remaining text after the last match as ambiguous
if start < len(text):
ambiguous_text = text[start:].strip()
if ambiguous_text:
ordered_parts.append(('ambiguous', ambiguous_text))
return ordered_parts
def standard_filtering(text_input):
text_output = (text_input
.replace("***", "")
.replace("**", "")
.replace("*", "")
.replace("\n\n", "\n")
.replace("&#x27;", "'")
)
return text_output
#################################################################
# /api/tts-generate Generation API Endpoint Narration Combining #
#################################################################
def combine(output_file_timestamp, output_file_name, audio_files, target_sample_rate=44100):
audio = np.array([])
try:
for audio_file in audio_files:
normalized_audio_file = os.path.normpath(audio_file)
print(f"[{branding}Debug] Processing file: {normalized_audio_file}") if debug_concat else None
# Check if the file exists
if not os.path.isfile(normalized_audio_file):
print(f"[{branding}Debug] Error: File does not exist: {normalized_audio_file}") if debug_concat else None
return None, None
# Read the audio file
audio_data, current_sample_rate = sf.read(normalized_audio_file)
print(f"[{branding}Debug] Read file: {normalized_audio_file}, Sample rate: {current_sample_rate}, Data shape: {audio_data.shape}") if debug_concat else None
# Resample if necessary
if current_sample_rate != target_sample_rate:
print(f"[{branding}Debug] Resampling file from {current_sample_rate} to {target_sample_rate} Hz") if debug_concat else None
audio_data = librosa.resample(audio_data, orig_sr=current_sample_rate, target_sr=target_sample_rate)
# Concatenate audio data
if audio.size == 0:
audio = audio_data
else:
audio = np.concatenate((audio, audio_data))
if output_file_timestamp:
timestamp = int(time.time())
output_file_path = os.path.join(this_dir / "outputs" / f'{output_file_name}_{timestamp}_combined.wav')
sf.write(output_file_path, audio, target_sample_rate)
# Legacy API or New API return
if params["api_def"]["api_use_legacy_api"]:
output_file_url = f"http://{api_defaults['api_legacy_ip_address']}:{params['api_def']['api_port_number']}/audio/{output_file_name}_{timestamp}_combined.wav"
output_cache_url = f"http://{api_defaults['api_legacy_ip_address']}:{params['api_def']['api_port_number']}/audiocache/{output_file_name}_{timestamp}_combined.wav"
else:
output_file_url = f'/audio/{output_file_name}_{timestamp}_combined.wav'
output_cache_url = f'/audiocache/{output_file_name}_{timestamp}_combined.wav'
else:
output_file_path = os.path.join(this_dir / "outputs" / f'{output_file_name}_combined.wav')
sf.write(output_file_path, audio, target_sample_rate)
# Legacy API or New API return
if params["api_def"]["api_use_legacy_api"]:
output_file_url = f"http://{api_defaults['api_legacy_ip_address']}:{api_defaults['api_port_number']}/audio/{output_file_name}_combined.wav"
output_cache_url = f"http://{api_defaults['api_legacy_ip_address']}:{api_defaults['api_port_number']}/audiocache/{output_file_name}_combined.wav"
else:
output_file_url = f'/audio/{output_file_name}_combined.wav'
output_cache_url = f'/audiocache/{output_file_name}_combined.wav'
print(f"[{branding}Debug] Output file changed to:", output_file_path) if debug_concat else None
print(f"[{branding}Debug] Output file changed to:", output_file_url) if debug_concat else None
print(f"[{branding}Debug] Output file changed to:", output_cache_url) if debug_concat else None
except Exception as e:
print(f"Error occurred: {e}")
return output_file_path, output_file_url, output_cache_url
##################################
# Central RVC Generation request #
##################################
def run_rvc(input_tts_path, pth_path, infer_pipeline):
generate_start_time = time.time()
f0up_key = rvc_settings["pitch"]
filter_radius = rvc_settings["filter_radius"]
index_rate = rvc_settings["index_rate"]
rms_mix_rate = rvc_settings["rms_mix_rate"]
protect = rvc_settings["protect"]
hop_length = rvc_settings["hop_length"]
f0method = rvc_settings["f0method"]
split_audio = rvc_settings["split_audio"]
f0autotune = rvc_settings["autotune"]
embedder_model = rvc_settings["embedder_model"]
training_data_size = rvc_settings["training_data_size"]
# Convert path variables to strings
input_tts_path = str(input_tts_path)
pth_path = str(pth_path)
# Check if the model file exists
if not os.path.isfile(pth_path):
print(f"Model file {pth_path} does not exist. Exiting.")
return
# Get the directory of the model file
model_dir = os.path.dirname(pth_path)
# Get the filename of pth_path
pth_filename = os.path.basename(pth_path)
# Find all .index files in the model directory
index_files = [file for file in os.listdir(model_dir) if file.endswith(".index")]
if len(index_files) == 1:
index_path = str(os.path.join(model_dir, index_files[0]))
# Get the filename of index_path
index_filename = os.path.basename(index_path)
index_filename_print = index_filename
index_size_print = training_data_size
elif len(index_files) > 1:
print(f"[{branding}GEN] \033[94mRVC Convert :\033[0m Multiple RVC index files found in the models folder where \033[93m{pth_filename}\033[0m is")
print(f"[{branding}GEN] \033[94mRVC Convert :\033[0m located. Unable to determine which index to use. Continuing without an index file.")
index_path = ""
index_filename = None
index_filename_print = "None used"
index_size_print = "N/A"
else:
index_path = ""
index_filename = None
index_filename_print = "None used"
index_size_print = "N/A"
# Set the output path to be the same as the input path
output_rvc_path = input_tts_path
if debug_rvc:
debug_lines = [
f"[{branding}Debug] ************************************",
f"[{branding}Debug] run_rvc function called (debug_rvc)",
f"[{branding}Debug] ***********************************",
f"[{branding}Debug] f0up_key : {f0up_key}",
f"[{branding}Debug] filter_radius : {filter_radius}",
f"[{branding}Debug] index_rate : {index_rate}",
f"[{branding}Debug] rms_mix_rate : {rms_mix_rate}",
f"[{branding}Debug] protect : {protect}",
f"[{branding}Debug] hop_length : {hop_length}",
f"[{branding}Debug] f0method : {f0method}",
f"[{branding}Debug] input_tts_path : {input_tts_path}",
f"[{branding}Debug] output_rvc_path : {output_rvc_path}",
f"[{branding}Debug] pth_path : {pth_path}",
f"[{branding}Debug] index_path : {index_path}",
f"[{branding}Debug] split_audio : {split_audio}",
f"[{branding}Debug] f0autotune : {f0autotune}",
f"[{branding}Debug] embedder_model : {embedder_model}",
f"[{branding}Debug] training_data_size: {training_data_size}"
]
debug_info = "\n".join(debug_lines)
print(debug_info)
# Call the infer_pipeline function
infer_pipeline(f0up_key, filter_radius, index_rate, rms_mix_rate, protect, hop_length, f0method,
input_tts_path, output_rvc_path, pth_path, index_path, split_audio, f0autotune, embedder_model, training_data_size, debug_rvc)
generate_end_time = time.time()
generate_elapsed_time = generate_end_time - generate_start_time
print(f"[{branding}GEN] \033[94mRVC Convert : \033[94mModel: \033[93m{pth_filename} \033[94mIndex: \033[93m{index_filename_print}\033[0m")
print(f"[{branding}GEN] \033[94mRVC Convert : \033[93m{generate_elapsed_time:.2f} seconds. \033[94mMethod: \033[93m{f0method} \033[94mIndex size used \033[93m{index_size_print}\033[0m")
return
####################################################################
# /api/tts-generate Generation API Endpoint Narration RVC handling #
####################################################################
def process_rvc_narrator(part_type, voice_gen, default_model_file, output_file, infer_pipeline):
if voice_gen.lower() in ["disabled", "disable"]:
return
else:
voice_gen_path = this_dir / "models" / "rvc_voices" / voice_gen
pth_path = voice_gen_path if voice_gen else default_model_file
run_rvc(output_file, pth_path, infer_pipeline)
#############################################################
# /api/tts-generate Generation API Endpoint JSON validation #
#############################################################
class JSONInput(BaseModel):
text_input: str = Field(..., max_length=int(api_defaults["api_max_characters"]), description=f"text_input needs to be {api_defaults['api_max_characters']} characters or less.")
text_filtering: str = Field(..., pattern="^(none|standard|html)$", description="text_filtering needs to be 'none', 'standard' or 'html'.")
character_voice_gen: str = Field(..., pattern=r'^[\(\)a-zA-Z0-9\_\-./\s]+$', description="character_voice_gen needs to be the name of a valid voice for the loaded TTS engine.")
rvccharacter_voice_gen: str = Field(..., description="rvccharacter_voice_gen needs to be the name of a valid pth file in the 'folder\\file.pth' format or the word 'Disabled'.")
narrator_enabled: str = Field(..., pattern="^(true|false|silent)$", description="narrator_enabled needs to be 'true', 'false' or 'silent'.")
narrator_voice_gen: str = Field(..., pattern=r'^[\(\)a-zA-Z0-9\_\-./\s]+$', description="character_voice_gen needs to be the name of a valid voice for the loaded TTS engine.")
rvcnarrator_voice_gen: str = Field(..., description="rvcnarrator_voice_gen needs to be the name of a valid pth file in the 'folder\\file.pth' format or the word 'Disabled'.")
text_not_inside: str = Field(..., pattern="^(character|narrator|silent)$", description="text_not_inside needs to be 'character', 'narrator' or 'silent'.")
language: str = Field(..., pattern="^(ar|zh-cn|cs|nl|en|fr|de|hu|hi|it|ja|ko|pl|pt|ru|es|tr)$", description="language needs to be one of the following: ar, zh-cn, cs, nl, en, fr, de, hu, hi, it, ja, ko, pl, pt, ru, es, tr.")
output_file_name: str = Field(..., pattern="^[a-zA-Z0-9_]+$", description="output_file_name needs to be the name without any special characters or file extension, e.g., 'filename'.")
output_file_timestamp: bool = Field(..., description="output_file_timestamp needs to be true or false.")
autoplay: bool = Field(..., description="autoplay needs to be a true or false value.")
autoplay_volume: float = Field(..., ge=0.1, le=1.0, description="autoplay_volume needs to be from 0.1 to 1.0.")
speed: float = Field(..., ge=0.25, le=2.0, description="speed needs to be between 0.25 and 2.0.")
pitch: float = Field(..., ge=-10, le=10, description="pitch needs to be between -10 and 10.")
temperature: float = Field(..., ge=0.1, le=1.0, description="temperature needs to be between 0.1 and 1.0.")
repetition_penalty: float = Field(..., ge=1.0, le=20.0, description="repetition_penalty needs to be between 1.0 and 20.0.")
@classmethod
def validate_autoplay_volume(cls, value):
if not (0.1 <= value <= 1.0):
raise ValueError("Autoplay volume must be between 0.1 and 1.0")
return value
def validate_rvccharacter_voice_gen(cls, v):
if v.lower() == "disabled":
return v
pattern = re.compile(r'^.*\.(pth)$')
if not pattern.match(v):
raise ValueError("rvccharacter_voice_gen needs to be the name of a valid pth file in the 'folder\\file.pth' format or 'Disabled'.")
return v
def validate_rvccharacter_voice_gen(cls, v):
if v.lower() == "disabled":
return v
pattern = re.compile(r'^.*\.(pth)$')
if not pattern.match(v):
raise ValueError("rvcnarrator_voice_gen needs to be the name of a valid pth file in the 'folder\\file.pth' format or 'Disabled'.")
return v
def validate_json_input(json_input_data):
try:
JSONInput(**json_input_data)
except ValidationError as e:
error_messages = []
for error in e.errors():
field = error["loc"][0]
description = JSONInput.__fields__[field].description
error_messages.append(f"{field}: {description}")
error_message = "\n".join(error_messages)
print(f"[{branding}API] \033[91mError with API request:\033[0m", error_message)
return error_message
return None
###############################################################
# /api/tts-generate Generation API Endpoint pull fresh values #
###############################################################
def get_api_text_filtering():
return api_defaults["api_text_filtering"]
def get_api_narrator_enabled():
return api_defaults["api_narrator_enabled"]
def get_api_text_not_inside():
return api_defaults["api_text_not_inside"]
def get_api_language():
return api_defaults["api_language"]
def get_api_output_file_name():
return api_defaults["api_output_file_name"]
def get_api_output_file_timestamp():
return api_defaults["api_output_file_timestamp"]
def get_api_autoplay():
return api_defaults["api_autoplay"]
def get_api_autoplay_volume():
return api_defaults["api_autoplay_volume"]
def get_params_speed():
return model_engine.generationspeed_set
def get_params_temperature():
return model_engine.temperature_set
def get_params_repetition():
return model_engine.repetitionpenalty_set
def get_params_pitch():
return model_engine.pitch_set
def get_character_voice_gen():
return model_engine.def_character_voice
def get_rvccharacter_voice_gen():
return rvc_settings["rvc_char_model_file"]
def get_narrator_voice_gen():
return model_engine.def_narrator_voice
def get_rvcnarrator_voice_gen():
return rvc_settings["rvc_narr_model_file"]
#############################################
# /api/tts-generate Generation API Endpoint #
#############################################
@app.post("/api/tts-generate", response_class=JSONResponse)
async def apifunction_generate_tts_standard(
text_input: str = Form(...),
text_filtering: str = Form(None),
character_voice_gen: str = Form(None),
rvccharacter_voice_gen: str = Form(None),
narrator_enabled: str = Form(None),
narrator_voice_gen: str = Form(None),
rvcnarrator_voice_gen: str = Form(None),
text_not_inside: str = Form(None),
language: str = Form(None),
output_file_name: str = Form(None),
output_file_timestamp: bool = Form(None),
autoplay: bool = Form(None),
autoplay_volume: float = Form(None),
streaming: bool = Form(False),
speed: float = Form(None),
temperature: float = Form(None),
repetition_penalty: float = Form(None),
pitch: float = Form(None),
_text_filtering: str = Depends(get_api_text_filtering),
_character_voice_gen: str = Depends(get_character_voice_gen),
_rvccharacter_voice_gen: str = Depends(get_rvccharacter_voice_gen),
_narrator_enabled: str = Depends(get_api_narrator_enabled),
_narrator_voice_gen: str = Depends(get_narrator_voice_gen),
_rvcnarrator_voice_gen: str = Depends(get_rvcnarrator_voice_gen),
_text_not_inside: str = Depends(get_api_text_not_inside),
_language: str = Depends(get_api_language),
_output_file_name: str = Depends(get_api_output_file_name),
_output_file_timestamp: bool = Depends(get_api_output_file_timestamp),
_autoplay: bool = Depends(get_api_autoplay),
_autoplay_volume: float = Depends(get_api_autoplay_volume),
_speed: float = Depends(get_params_speed),
_temperature: float = Depends(get_params_temperature),
_repetition_penalty: float = Depends(get_params_repetition),
_pitch: float = Depends(get_params_pitch),
):
if debug_tts or debug_tts_variables:
debug_lines = [
f"[{branding}Debug] *******************************************************",
f"[{branding}Debug] /api/tts_generate function called (debug_tts_variables)",
f"[{branding}Debug] **** PRE validation checks ****",
f"[{branding}Debug] *******************************************************",
f"[{branding}Debug] Defaults will be pulled from Settings if not specified in API Request!",
f"[{branding}Debug] max_characters : {api_defaults['api_max_characters']}",
f"[{branding}Debug] length_stripping : {api_defaults['api_length_stripping']}",
f"[{branding}Debug] legacy_api : {api_defaults['api_use_legacy_api']}",
f"[{branding}Debug] legacy_api IPadd : {api_defaults['api_legacy_ip_address']}",
f"[{branding}Debug] allowed_filter : {api_defaults['api_allowed_filter']}",
f"[{branding}Debug] text_filtering : {text_filtering}",
f"[{branding}Debug] character_voice_gen : {character_voice_gen}",
f"[{branding}Debug] rvccharacter_voice_gen : {rvccharacter_voice_gen}",
f"[{branding}Debug] narrator_enabled : {narrator_enabled}",
f"[{branding}Debug] narrator_voice_gen : {narrator_voice_gen}",
f"[{branding}Debug] rvcnarrator_voice_gen : {rvcnarrator_voice_gen}",
f"[{branding}Debug] text_not_inside : {text_not_inside}",
f"[{branding}Debug] language : {language}",
f"[{branding}Debug] output_file_name : {output_file_name}",
f"[{branding}Debug] output_file_timestamp : {output_file_timestamp}",
f"[{branding}Debug] autoplay : {autoplay}",
f"[{branding}Debug] autoplay_volume : {autoplay_volume}",
f"[{branding}Debug] streaming : {streaming}",
f"[{branding}Debug] speed : {speed}",
f"[{branding}Debug] temperature : {temperature}",
f"[{branding}Debug] repetition_penalty : {repetition_penalty}",
f"[{branding}Debug] pitch : {pitch}"
]
debug_info = "\n".join(debug_lines)
print(debug_info)
text_filtering = text_filtering or _text_filtering
character_voice_gen = character_voice_gen or _character_voice_gen
rvccharacter_voice_gen = rvccharacter_voice_gen or _rvccharacter_voice_gen
narrator_enabled = narrator_enabled or _narrator_enabled # NEED TO COME BACK AND LOOK AT THIS!!!
narrator_voice_gen = narrator_voice_gen or _narrator_voice_gen
rvcnarrator_voice_gen = rvcnarrator_voice_gen or _rvcnarrator_voice_gen
text_not_inside = text_not_inside or _text_not_inside
language = language or _language
output_file_name = output_file_name or _output_file_name
output_file_timestamp = output_file_timestamp if output_file_timestamp is not None else _output_file_timestamp
autoplay = autoplay if autoplay is not None else _autoplay
autoplay_volume = autoplay_volume or _autoplay_volume
speed = speed or _speed
temperature = temperature or _temperature
repetition_penalty = repetition_penalty or _repetition_penalty
pitch = pitch or _pitch
try:
json_input_data = {
"text_input": text_input,
"text_filtering": text_filtering,
"character_voice_gen": character_voice_gen,
"rvccharacter_voice_gen": rvccharacter_voice_gen,
"narrator_enabled": narrator_enabled,
"narrator_voice_gen": narrator_voice_gen,
"rvcnarrator_voice_gen": rvcnarrator_voice_gen,
"text_not_inside": text_not_inside,
"language": language,
"output_file_name": output_file_name,
"output_file_timestamp": output_file_timestamp,
"autoplay": autoplay,
"autoplay_volume": autoplay_volume,
"streaming": streaming,
"speed": speed,
"temperature": temperature,
"repetition_penalty": repetition_penalty,
"pitch": pitch,
}
JSONresult = validate_json_input(json_input_data)
if JSONresult is None:
pass
else:
return JSONResponse(content={"error": JSONresult}, status_code=400)
if debug_tts or debug_tts_variables:
debug_lines = [
f"[{branding}Debug] *******************************************************",
f"[{branding}Debug] /api/tts_generate function called (debug_tts_variables)",
f"[{branding}Debug] **** POST validation checks ****",
f"[{branding}Debug] *******************************************************",
f"[{branding}Debug] Defaults will be pulled from Settings if not specified in API Request!",
f"[{branding}Debug] max_characters : {api_defaults['api_max_characters']}",
f"[{branding}Debug] length_stripping : {api_defaults['api_length_stripping']}",
f"[{branding}Debug] legacy_api : {api_defaults['api_use_legacy_api']}",
f"[{branding}Debug] legacy_api IPadd : {api_defaults['api_legacy_ip_address']}",
f"[{branding}Debug] allowed_filter : {api_defaults['api_allowed_filter']}",
f"[{branding}Debug] text_filtering : {text_filtering}",
f"[{branding}Debug] character_voice_gen : {character_voice_gen}",
f"[{branding}Debug] rvccharacter_voice_gen : {rvccharacter_voice_gen}",
f"[{branding}Debug] narrator_enabled : {narrator_enabled}",
f"[{branding}Debug] narrator_voice_gen : {narrator_voice_gen}",
f"[{branding}Debug] rvcnarrator_voice_gen : {rvcnarrator_voice_gen}",
f"[{branding}Debug] text_not_inside : {text_not_inside}",
f"[{branding}Debug] language : {language}",
f"[{branding}Debug] output_file_name : {output_file_name}",
f"[{branding}Debug] output_file_timestamp : {output_file_timestamp}",
f"[{branding}Debug] autoplay : {autoplay}",
f"[{branding}Debug] autoplay_volume : {autoplay_volume}",
f"[{branding}Debug] streaming : {streaming}",
f"[{branding}Debug] speed : {speed}",
f"[{branding}Debug] temperature : {temperature}",
f"[{branding}Debug] repetition_penalty : {repetition_penalty}",
f"[{branding}Debug] pitch : {pitch}"
]
debug_info = "\n".join(debug_lines)
print(debug_info)
if narrator_enabled.lower() == "silent" and text_not_inside.lower() == "silent":
print(f"[{branding}GEN] \033[92mBoth Narrator & Text-not-inside are set to \033[91msilent\033[92m. If you get no TTS, this is why.\033[0m")
if narrator_enabled.lower() in ["true", "silent"]:
if model_engine.lowvram_enabled and model_engine.device == "cpu":
await model_engine.handle_lowvram_change()
model_engine.tts_narrator_generatingtts = True
print(f"[{branding}Debug] Moved into Narrator") if debug_tts else None
print(f"[{branding}Debug] original text : {text_input}") if debug_tts else None
processed_parts = process_text(text_input)
audio_files_all_paragraphs = []
for part_type, part in processed_parts:
# Skip parts that are too short
if len(part.strip()) <= int(params["api_def"]["api_length_stripping"]):
continue
# Determine the voice to use based on the part type
if part_type == 'narrator':
if narrator_enabled.lower() == "silent":
print(f"[{branding}GEN] \033[95mNarrator Silent:\033[0m", part) # Green
continue # Skip generating audio for narrator parts if set to "silent"
voice_to_use = narrator_voice_gen
print(f"[{branding}GEN] \033[92mNarrator:\033[0m", part) # Purple
elif part_type == 'character':
voice_to_use = character_voice_gen
print(f"[{branding}GEN] \033[36mCharacter:\033[0m", part) # Yellow
else:
# Handle ambiguous parts based on user preference
if text_not_inside == "silent":
print(f"[{branding}GEN] \033[95mText-not-inside Silent:\033[0m", part) # Purple
continue # Skip generating audio for ambiguous parts if set to "silent"
voice_to_use = character_voice_gen if text_not_inside == "character" else narrator_voice_gen
voice_description = "\033[36mCharacter (Text-not-inside)\033[0m" if text_not_inside == "character" else "\033[92mNarrator (Text-not-inside)\033[0m"
print(f"[{branding}GEN] {voice_description}:", part)
# Replace multiple exclamation marks, question marks, or other punctuation with a single instance
cleaned_part = re.sub(r'([!?.])\1+', r'\1', part)
# Further clean to remove any other unwanted characters
cleaned_part = re.sub(rf'{api_defaults["api_allowed_filter"]}', '', cleaned_part)
print(f"[{branding}Debug] text after api_allowed: {text_input}") if debug_tts else None
# Remove all newline characters (single or multiple)
cleaned_part = re.sub(r'\n+', ' ', cleaned_part)
output_file = this_dir / output_directory / f'{output_file_name}_{uuid.uuid4()}_{int(time.time())}.{model_engine.audio_format}'
output_file_str = output_file.as_posix()
response = await generate_audio(cleaned_part, voice_to_use, language,temperature, repetition_penalty, speed, pitch, output_file_str, streaming)
audio_path = output_file_str
if rvc_settings["rvc_enabled"]:
if part_type == 'character':
process_rvc_narrator(part_type, rvccharacter_voice_gen, rvc_settings["rvc_char_model_file"], output_file, infer_pipeline)
elif part_type == 'narrator':
process_rvc_narrator(part_type, rvcnarrator_voice_gen, rvc_settings["rvc_narr_model_file"], output_file, infer_pipeline)
else:
if text_not_inside == 'character':
process_rvc_narrator('character', rvccharacter_voice_gen, rvc_settings["rvc_char_model_file"], output_file, infer_pipeline)
elif text_not_inside == 'narrator':
process_rvc_narrator('narrator', rvcnarrator_voice_gen, rvc_settings["rvc_narr_model_file"], output_file, infer_pipeline)
print(f"[{branding}Debug] Appending audio path to list") if debug_tts else None
audio_files_all_paragraphs.append(audio_path)
# Combine audio files across paragraphs
model_engine.tts_narrator_generatingtts = False
print(f"[{branding}Debug] Narrator sending to combine") if debug_tts else None
output_file_path, output_file_url, output_cache_url = combine(output_file_timestamp, output_file_name, audio_files_all_paragraphs)
# Transcode audio if necessary
model_audio_format = str(model_engine.audio_format).lower()
output_audio_format = str(params["transcode_audio_format"]).lower()
if output_audio_format == "disabled":
pass
else:
if model_audio_format != output_audio_format and not model_engine.tts_narrator_generatingtts:
print(f"[{branding}Debug] Pre Transcode call - Output file: {output_file_path}") if debug_tts else None
output_file_path, output_file_url, output_cache_url = await transcode_audio_if_necessary(output_file_path, model_audio_format, output_audio_format)
print(f"[{branding}Debug] Post Transcode call - Output file: {output_file_path}") if debug_tts else None
if sounddevice_installed == False:
autoplay = False
if autoplay:
play_audio(output_file_path, autoplay_volume)
# Move model back to cpu system ram if needed.
if model_engine.lowvram_enabled and model_engine.device == "cuda" and model_engine.lowvram_capable:
await model_engine.handle_lowvram_change()
return JSONResponse(content={"status": "generate-success", "output_file_path": str(output_file_path), "output_file_url": str(output_file_url), "output_cache_url": str(output_cache_url)}, status_code=200)
else:
print(f"[{branding}Debug] Moved into Standard generation") if debug_tts else None
print(f"[{branding}Debug] original text : {text_input}") if debug_tts else None
if output_file_timestamp:
timestamp = int(time.time())
# Generate a standard UUID
original_uuid = uuid.uuid4()
# Hash the UUID using SHA-256
hash_object = hashlib.sha256(str(original_uuid).encode())
hashed_uuid = hash_object.hexdigest()
# Truncate to the desired length, for example, 16 characters
short_uuid = hashed_uuid[:5]
output_file_path = this_dir / output_directory / f'{output_file_name}_{timestamp}{short_uuid}.{model_engine.audio_format}'
#Legacy API or New API return
if params["api_def"]["api_use_legacy_api"]:
output_file_url = f'http://{api_defaults["api_legacy_ip_address"]}:{api_defaults["api_port_number"]}/audio/{output_file_name}_{timestamp}{short_uuid}.{model_engine.audio_format}'
output_cache_url = f'http://{api_defaults["api_legacy_ip_address"]}:{api_defaults["api_port_number"]}/audiocache/{output_file_name}_{timestamp}{short_uuid}.{model_engine.audio_format}'
else:
output_file_url = f'/audio/{output_file_name}_{timestamp}{short_uuid}.{model_engine.audio_format}'
output_cache_url = f'/audiocache/{output_file_name}_{timestamp}{short_uuid}.{model_engine.audio_format}'
else:
output_file_path = this_dir / output_directory / f"{output_file_name}.{model_engine.audio_format}"
# Legacy API or New API return
if params["api_def"]["api_use_legacy_api"]:
output_file_url = f'http://{api_defaults["api_legacy_ip_address"]}:{api_defaults["api_port_number"]}/audio/{output_file_name}.{model_engine.audio_format}'
output_cache_url = f'http://{api_defaults["api_legacy_ip_address"]}:{api_defaults["api_port_number"]}/audiocache/{output_file_name}.{model_engine.audio_format}'
else:
output_file_url = f'/audio/{output_file_name}.{model_engine.audio_format}'
output_cache_url = f'/audiocache/{output_file_name}.{model_engine.audio_format}'
if text_filtering == "html":
cleaned_string = html.unescape(standard_filtering(text_input))
cleaned_string = re.sub(r'([!?.])\1+', r'\1', cleaned_string)
# Further clean to remove any other unwanted characters
cleaned_string = re.sub(rf'{api_defaults["api_allowed_filter"]}', '', cleaned_string)
print(f"[{branding}Debug] HTML Filtering - text after api_allowed: {cleaned_string}") if debug_tts else None
# Remove all newline characters (single or multiple)
cleaned_string = re.sub(r'\n+', ' ', cleaned_string)
elif text_filtering == "standard":
cleaned_string = re.sub(r'([!?.])\1+', r'\1', text_input)
# Further clean to remove any other unwanted characters
cleaned_string = re.sub(rf'{api_defaults["api_allowed_filter"]}', '', cleaned_string)
print(f"[{branding}Debug] Standard Filtering - text after api_allowed: {cleaned_string}") if debug_tts else None
# Remove all newline characters (single or multiple)
cleaned_string = re.sub(r'\n+', ' ', cleaned_string)
else:
print(f"[{branding}Debug] No filtering text : {text_input}") if debug_tts else None
cleaned_string = text_input
print(f"[{branding}GEN]", cleaned_string)
print(f"[{branding}Debug] Sending request to generate_audio in tts_server.py") if debug_tts else None
try:
if streaming:
response = await generate_audio(cleaned_string, character_voice_gen, language, temperature, repetition_penalty, speed, pitch, output_file_path, streaming)
return StreamingResponse(response, media_type="audio/wav")
else:
response = await generate_audio(cleaned_string, character_voice_gen, language, temperature, repetition_penalty, speed, pitch, output_file_path, streaming)
if rvc_settings["rvc_enabled"]:
if rvccharacter_voice_gen.lower() in ["disabled", "disable"]:
print(f"[{branding}Debug] Pass rvccharacter_voice_gen") if debug_tts else None
pass # Skip RVC processing for character part
else:
print(f"[{branding}Debug] send to rvc") if debug_tts else None
rvccharacter_voice_gen = this_dir / "models" / "rvc_voices" / rvccharacter_voice_gen
pth_path = rvccharacter_voice_gen if rvccharacter_voice_gen else rvc_settings["rvc_char_model_file"]
run_rvc(output_file_path, pth_path, infer_pipeline)
# Transcode audio if necessary
model_audio_format = str(model_engine.audio_format).lower()
output_audio_format = str(params["transcode_audio_format"]).lower()
print(f"[{branding}Debug] At Model audio format") if debug_tts else None
if output_audio_format == "disabled":
pass
else:
if model_audio_format != output_audio_format and not model_engine.tts_narrator_generatingtts:
print(f"[{branding}Debug] Pre Transcode call - Output file: {output_file_path}") if debug_tts else None
output_file_path, output_file_url, output_cache_url = await transcode_audio_if_necessary(output_file_path, model_audio_format, output_audio_format)
print(f"[{branding}Debug] Post Transcode call - Output file: {output_file_path}") if debug_tts else None
if sounddevice_installed == False:
autoplay = False
if autoplay:
play_audio(output_file_path, autoplay_volume)
return JSONResponse(content={"status": "generate-success", "output_file_path": str(output_file_path), "output_file_url": str(output_file_url), "output_cache_url": str(output_cache_url)}, status_code=200)
except Exception as e:
return JSONResponse(content={"status": "generate-failure", "error": "An error occurred"}, status_code=500)
except Exception as e:
return JSONResponse(content={"status": "generate-failure", "error": "An error occurred"}, status_code=500)
#############################
#### Word Add-in Sharing ####
#############################
# Mount the static files from the 'word_addin' directory
app.mount("/api/word_addin", StaticFiles(directory=os.path.join(this_dir / 'system' / 'word_addin')), name="word_addin")
#############################################
#### TTS Generator Comparision Endpoints ####
#############################################
import aiofiles
class TTSItem(BaseModel):
id: int
fileUrl: str
text: str
characterVoice: str
language: str
class TTSData(BaseModel):
ttsList: List[TTSItem]
@app.post("/api/save-tts-data")
async def apifunction_save_tts_data(tts_data: List[TTSItem]):
# Convert the list of Pydantic models to a list of dictionaries
tts_data_list = [item.dict() for item in tts_data]
# Serialize the list of dictionaries to a JSON string
tts_data_json = json.dumps(tts_data_list, indent=4)
async with aiofiles.open(this_dir / output_directory / "ttsList.json", 'w') as f:
await f.write(tts_data_json)
return {"message": "Data saved successfully"}
########################################
# Trigger TTS Gen Text/Speech Analysis #
########################################
@app.get("/api/trigger-analysis")
async def apifunction_trigger_analysis(threshold: int = Query(default=98)):
venv_path = sys.prefix
env = os.environ.copy()
env["PATH"] = os.path.join(venv_path, "bin") + ":" + env["PATH"]
ttslist_path = this_dir / output_directory / "ttsList.json"
wavfile_path = this_dir / output_directory
subprocess.run(["python", "tts_diff.py", f"--threshold={threshold}", f"--ttslistpath={ttslist_path}", f"--wavfilespath={wavfile_path}"], cwd=this_dir / "system" / "tts_diff", env=env)
# Read the analysis summary
try:
with open(this_dir / output_directory / "analysis_summary.json", "r") as summary_file:
summary_data = json.load(summary_file)
except FileNotFoundError:
summary_data = {"error": "Analysis summary file not found."}
return {"message": "Analysis Completed", "summary": summary_data}
###########################################
# TTS Generator SRT Subtitiles generation #
###########################################
@app.get("/api/srt-generation")
async def apifunction_srt_generation():
venv_path = sys.prefix
env = os.environ.copy()
env["PATH"] = os.path.join(venv_path, "bin") + ":" + env["PATH"]
ttslist_path = this_dir / output_directory / "ttsList.json"
wavfile_path = this_dir / output_directory
subprocess.run(["python", "tts_srt.py", f"--ttslistpath={ttslist_path}", f"--wavfilespath={wavfile_path}"], cwd=this_dir / "system" / "tts_srt", env=env)
srt_file_path = this_dir / output_directory / "subtitles.srt"
if not srt_file_path.exists():
raise HTTPException(status_code=404, detail="Subtitle file not found.")
return FileResponse(path=srt_file_path, filename="subtitles.srt", media_type='application/octet-stream')
#################################
# Static Mount for file serving #
#################################
app.mount("/static", StaticFiles(directory=str(this_dir / "system")), name="static")
########################################
# Legacy JSON update settings function #
########################################
# Setup logging
logging.basicConfig(level=logging.DEBUG)
def get_json_data():
with open(this_dir / "confignew.json", "r") as json_file:
data = json.load(json_file)
return data
@app.get("/settings")
async def get_settings(request: Request):
data = get_json_data()
return templates.TemplateResponse("admin.html", {"request": request, "data": data})
@app.get("/settings-json")
async def get_settings_json():
return get_json_data()
@app.post("/update-settings")
async def update_settings(
request: Request,
delete_output_wavs: str = Form(...),
gradio_interface: str = Form(...),
gradio_port_number: int = Form(...),
api_port_number: int = Form(...),
):
data = get_json_data()
# Update the settings based on the form values
data["delete_output_wavs"] = delete_output_wavs
data["gradio_interface"] = gradio_interface.lower() == 'true'
data["gradio_port_number"] = gradio_port_number
data["api_def"]["api_port_number"] = api_port_number
# Save the updated settings back to the JSON file
with open(this_dir / "confignew.json", "w") as json_file:
json.dump(data, json_file, indent=4)
# Redirect to the settings page to display the updated settings
return templates.TemplateResponse("admin.html", {"request": request, "data": data})
# Create an instance of Jinja2Templates for rendering HTML templates
templates = Jinja2Templates(directory=this_dir / "system")
# Get the admin interface template
template = templates.get_template("admin.html")
# Render the template with the dynamic values
rendered_html = template.render(params=params)
###################################################
#### Webserver Startup & Initial model Loading ####
###################################################
@app.get("/")
async def read_root():
return HTMLResponse(content=rendered_html, status_code=200)
# Start Uvicorn Webserver
port_parameter = int(params["api_def"]["api_port_number"])
if __name__ == "__main__":
import uvicorn
# Start Uvicorn Webserver
uvicorn_server = uvicorn.run(app, host="0.0.0.0", port=port_parameter, log_level="debug")
Reference in new issue View Git Blame Copy Permalink