|
|
import warnings |
|
|
warnings.filterwarnings("ignore", category=UserWarning) |
|
|
warnings.filterwarnings("ignore", category=FutureWarning) |
|
|
import logging |
|
|
logging.getLogger().setLevel(logging.ERROR) |
|
|
|
|
|
import argparse |
|
|
import os |
|
|
import sys |
|
|
import time |
|
|
|
|
|
TTS_ENGINE = None |
|
|
os.environ["CUDA_LAUNCH_BLOCKING"] = "1" |
|
|
|
|
|
def server_mode(args): |
|
|
while True: |
|
|
input = sys.stdin.readline().strip() |
|
|
input = input.split("voice") |
|
|
|
|
|
try: args.speed = float(input[0]) |
|
|
except: args.speed = 1 |
|
|
|
|
|
try: args.voice = int(input[1]) |
|
|
except: args.voice = 8 |
|
|
|
|
|
output_path = initiate(args) |
|
|
|
|
|
print(output_path) |
|
|
sys.stdout.flush() |
|
|
|
|
|
def current_env(): |
|
|
"""Detect current virtual environment.""" |
|
|
venv_path = os.environ.get("VIRTUAL_ENV") |
|
|
if venv_path: |
|
|
return os.path.basename(venv_path) |
|
|
raise ValueError("Please set env first") |
|
|
|
|
|
def initiate(args): |
|
|
model = args.get('model') if isinstance(args, dict) else getattr(args, 'model', None) |
|
|
if not model: |
|
|
if current_env() == "kokoro_env": |
|
|
from .engines.kokoro import KokoroTTSProcessor as TTSEngine |
|
|
elif current_env() == "kitten_env": |
|
|
from .engines.kitten import KittenTTSProcessor as TTSEngine |
|
|
else: |
|
|
from .engines.chatterbox import ChatterboxTTSProcessor as TTSEngine |
|
|
else: |
|
|
if model == "kokoro": |
|
|
from .engines.kokoro import KokoroTTSProcessor as TTSEngine |
|
|
elif model == "kitten": |
|
|
from .engines.kitten import KittenTTSProcessor as TTSEngine |
|
|
else: |
|
|
from .engines.chatterbox import ChatterboxTTSProcessor as TTSEngine |
|
|
|
|
|
global TTS_ENGINE |
|
|
if not TTS_ENGINE: |
|
|
TTS_ENGINE = TTSEngine(stream_audio=args.stream_text) |
|
|
|
|
|
try: |
|
|
import torch |
|
|
import gc |
|
|
torch.cuda.empty_cache() |
|
|
torch.cuda.synchronize() |
|
|
gc.collect() |
|
|
gc.collect() |
|
|
time.sleep(1) |
|
|
print("\n🧹 Cleared PyTorch CUDA cache") |
|
|
except: pass |
|
|
|
|
|
if args.stream_text: |
|
|
TTS_ENGINE.stream_real_time_text(args) |
|
|
text = TTS_ENGINE.read_content_file() |
|
|
for text_chunk in text.split(): |
|
|
TTS_ENGINE.feed_text_chunk(text_chunk) |
|
|
time.sleep(0.1) |
|
|
|
|
|
TTS_ENGINE.stop_all_streaming() |
|
|
else: |
|
|
TTS_ENGINE.save_audio(args) |
|
|
|
|
|
|
|
|
def main(): |
|
|
"""Main entry point.""" |
|
|
parser = argparse.ArgumentParser( |
|
|
description="Text-to-Speech processor" |
|
|
) |
|
|
parser.add_argument( |
|
|
"--server-mode", |
|
|
action="store_true", |
|
|
help="Run in server mode (read commands from stdin)" |
|
|
) |
|
|
parser.add_argument( |
|
|
"--speed", |
|
|
type=float, |
|
|
help=f"Speech speed" |
|
|
) |
|
|
parser.add_argument( |
|
|
"--voice", |
|
|
type=int, |
|
|
help=f"Voice index" |
|
|
) |
|
|
parser.add_argument( |
|
|
"--stream-text", |
|
|
action="store_true", |
|
|
help="Enable streaming text output" |
|
|
) |
|
|
parser.add_argument( |
|
|
"--model", |
|
|
help="model name" |
|
|
) |
|
|
|
|
|
|
|
|
args = parser.parse_args() |
|
|
|
|
|
if args.server_mode: |
|
|
server_mode(args) |
|
|
else: |
|
|
success = initiate(args) |
|
|
return 0 if success else 1 |
|
|
|
|
|
if __name__ == "__main__": |
|
|
main() |
|
|
|
