App.py

from kokoro import KModel, KPipeline
import gradio as gr
import os
import random
import torch
import logging
import soundfile as sf

# Optional: import Resemblyzer for voice cloning (install via pip install resemblyzer)
try:
    from resemblyzer import VoiceEncoder, preprocess_wav
    encoder = VoiceEncoder()
except ImportError:
    encoder = None

# Configuration
VOICE_DIR = r"D:\New folder (2)\model\voices"
OUTPUT_DIR = r"D:\New folder (2)\output_audio"
TEXT = "Hello, this is a test of the Kokoro TTS system."

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Device setup
CUDA_AVAILABLE = torch.cuda.is_available()
device = "cuda" if CUDA_AVAILABLE else "cpu"
logger.info(f"Using hardware: {device}")

# Load models for CPU and GPU (if available)
models = {gpu: KModel("hexgrad/Kokoro-82M").to("cuda" if gpu else "cpu").eval() for gpu in [False] + ([True] if CUDA_AVAILABLE else [])}

# Define pipelines for American ('a') and British ('b') English
pipelines = {
    'a': KPipeline(model=models[False], lang_code='a', device='cpu'),  # American English
    'b': KPipeline(model=models[False], lang_code='b', device='cpu')   # British English
}

# Set custom pronunciations for "kokoro" in both American and British modes
try:
    pipelines["a"].g2p.lexicon.golds["kokoro"] = "kˈOkəɹO"
    pipelines["b"].g2p.lexicon.golds["kokoro"] = "kˈQkəɹQ"
except AttributeError as e:
    logger.warning(f"Could not set custom pronunciations: {e}")

def forward_gpu(text, voice_path, speed):
    # Use the GPU model directly without spaces.GPU decorator
    pipeline = pipelines[voice_path[0]]
    # Ensure the pipeline uses the GPU model
    pipeline.model = models[True]  # Switch to GPU model
    generator = pipeline(text, voice=voice_path, speed=speed)
    for _, _, audio in generator:
        return audio
    return None

def generate_first(text, voice="af_bella.pt", speed=1, use_gpu=CUDA_AVAILABLE, clone_voice_file=None):
    voice_path = os.path.join(VOICE_DIR, voice)
    if not os.path.exists(voice_path):
        raise FileNotFoundError(f"Voice file not found: {voice_path}")

    pipeline = pipelines[voice[0]]

    # If a clone file is provided and the encoder is available, try to clone the voice
    if clone_voice_file is not None and encoder is not None:
        try:
            # clone_voice_file is a file path (string) in Gradio with type="filepath"
            wav = preprocess_wav(clone_voice_file)
            cloned_voice = torch.tensor(encoder.embed_utterance(wav), device=device).unsqueeze(0)
            temp_voice_path = os.path.join(VOICE_DIR, "cloned_voice.pt")
            torch.save(cloned_voice, temp_voice_path)
            voice_path = temp_voice_path
        except Exception as e:
            logger.error(f"Error cloning voice: {e}")
            voice_path = os.path.join(VOICE_DIR, voice)

    use_gpu = use_gpu and CUDA_AVAILABLE
    try:
        if use_gpu:
            audio = forward_gpu(text, voice_path, speed)
        else:
            pipeline.model = models[False]  # Ensure CPU model is used
            generator = pipeline(text, voice=voice_path, speed=speed)
            for _, ps, audio in generator:
                return (24000, audio.numpy()), ps
    except gr.exceptions.Error as e:
        if use_gpu:
            gr.Warning(str(e))
            gr.Info("Retrying with CPU. To avoid this error, change Hardware to CPU.")
            pipeline.model = models[False]  # Switch to CPU model
            generator = pipeline(text, voice=voice_path, speed=speed)
            for _, ps, audio in generator:
                return (24000, audio.numpy()), ps
        else:
            raise gr.Error(e)
    return None, ""

def predict(text, voice="af_bella.pt", speed=1):
    return generate_first(text, voice, speed, use_gpu=False)[0]

def tokenize_first(text, voice="af_bella.pt"):
    voice_path = os.path.join(VOICE_DIR, voice)
    if not os.path.exists(voice_path):
        raise FileNotFoundError(f"Voice file not found: {voice_path}")

    pipeline = pipelines[voice[0]]
    generator = pipeline(text, voice=voice_path)
    for _, ps, _ in generator:
        return ps
    return ""

def generate_all(text, voice="af_bella.pt", speed=1, use_gpu=CUDA_AVAILABLE):
    voice_path = os.path.join(VOICE_DIR, voice)
    if not os.path.exists(voice_path):
        raise FileNotFoundError(f"Voice file not found: {voice_path}")

    pipeline = pipelines[voice[0]]
    use_gpu = use_gpu and CUDA_AVAILABLE
    first = True
    if use_gpu:
        pipeline.model = models[True]  # Switch to GPU model
    else:
        pipeline.model = models[False]  # Switch to CPU model
    generator = pipeline(text, voice=voice_path, speed=speed)
    for _, _, audio in generator:
        yield 24000, audio.numpy()
        if first:
            first = False
            yield 24000, torch.zeros(1).numpy()

# Load random quotes and sample texts
try:
    with open("en.txt", "r") as r:
        random_quotes = [line.strip() for line in r]
except FileNotFoundError:
    random_quotes = ["Hello, this is a test of the Kokoro TTS system."]

def get_random_quote():
    return random.choice(random_quotes)

def get_gatsby():
    try:
        with open("gatsby5k.md", "r") as r:
            return r.read().strip()
    except FileNotFoundError:
        return "The Great Gatsby text not found."

def get_frankenstein():
    try:
        with open("frankenstein5k.md", "r") as r:
            return r.read().strip()
    except FileNotFoundError:
        return "Frankenstein text not found."

# Dynamically load all .pt voice files from VOICE_DIR
def load_voice_choices():
    voice_files = [f for f in os.listdir(VOICE_DIR) if f.endswith('.pt')]
    choices = {}
    for voice_file in voice_files:
        # Determine the voice type based on the prefix
        prefix = voice_file[:2]
        if prefix == 'af':
            label = f"🇺🇸 🚺 {voice_file[3:-3].capitalize()}"
        elif prefix == 'am':
            label = f"🇺🇸 🚹 {voice_file[3:-3].capitalize()}"
        elif prefix == 'bf':
            label = f"🇬🇧 🚺 {voice_file[3:-3].capitalize()}"
        elif prefix == 'bm':
            label = f"🇬🇧 🚹 {voice_file[3:-3].capitalize()}"
        else:
            label = f"Unknown {voice_file[:-3]}"
        choices[label] = voice_file
    return choices

CHOICES = load_voice_choices()

# Log available voices
for label, voice_path in CHOICES.items():
    full_path = os.path.join(VOICE_DIR, voice_path)
    if not os.path.exists(full_path):
        logger.warning(f"Voice file not found: {full_path}")
    else:
        logger.info(f"Loaded voice: {label} ({voice_path})")

# If no voices are found, add a default fallback
if not CHOICES:
    logger.warning("No voice files found in VOICE_DIR. Adding a placeholder.")
    CHOICES = {"🇺🇸 🚺 Bella 🔥": "af_bella.pt"}

TOKEN_NOTE = '''
💡 Customize pronunciation with Markdown link syntax and /slashes/ like [Kokoro](/kˈOkəɹO/)

💬 To adjust intonation, try punctuation ;:,.!?—…"()“” or stress ˈ and ˌ

⬇️ Lower stress [1 level](-1) or [2 levels](-2)

⬆️ Raise stress 1 level [or](+2) 2 levels (only works on less stressed, usually short words)
'''

with gr.Blocks() as generate_tab:
    out_audio = gr.Audio(label="Output Audio", interactive=False, streaming=False, autoplay=True)
    generate_btn = gr.Button("Generate", variant="primary")
    with gr.Accordion("Output Tokens", open=True):
        out_ps = gr.Textbox(interactive=False, show_label=False,
                            info="Tokens used to generate the audio, up to 510 context length.")
        tokenize_btn = gr.Button("Tokenize", variant="secondary")
        gr.Markdown(TOKEN_NOTE)
        predict_btn = gr.Button("Predict", variant="secondary", visible=False)

with gr.Blocks() as stream_tab:
    out_stream = gr.Audio(label="Output Audio Stream", interactive=False, streaming=True, autoplay=True)
    with gr.Row():
        stream_btn = gr.Button("Stream", variant="primary")
        stop_btn = gr.Button("Stop", variant="stop")
    with gr.Accordion("Note", open=True):
        gr.Markdown("⚠️ There is an unknown Gradio bug that might yield no audio the first time you click Stream.")
        gr.DuplicateButton()

API_OPEN = True
with gr.Blocks() as app:
    with gr.Row():
        with gr.Column():
            text = gr.Textbox(label="Input Text", info="Arbitrarily many characters supported")
            with gr.Row():
                voice = gr.Dropdown(list(CHOICES.items()), value="af_bella.pt" if "af_bella.pt" in CHOICES.values() else list(CHOICES.values())[0], label="Voice",
                                    info="Quality and availability vary by language")
                use_gpu = gr.Dropdown(
                    [("GPU 🚀", True), ("CPU 🐌", False)],
                    value=CUDA_AVAILABLE,
                    label="Hardware",
                    info="GPU is usually faster, but may require CUDA support",
                    interactive=CUDA_AVAILABLE
                )
            speed = gr.Slider(minimum=0.5, maximum=2, value=1, step=0.1, label="Speed")
            clone_voice_file = gr.File(label="Clone Voice Sample (Optional)", file_count="single", type="filepath")
            random_btn = gr.Button("🎲 Random Quote 💬", variant="secondary")
            with gr.Row():
                gatsby_btn = gr.Button("🥂 Gatsby 📕", variant="secondary")
                frankenstein_btn = gr.Button("💀 Frankenstein 📗", variant="secondary")
        with gr.Column():
            gr.TabbedInterface([generate_tab, stream_tab], ["Generate", "Stream"])
    random_btn.click(fn=get_random_quote, inputs=[], outputs=[text])
    gatsby_btn.click(fn=get_gatsby, inputs=[], outputs=[text])
    frankenstein_btn.click(fn=get_frankenstein, inputs=[], outputs=[text])
    generate_btn.click(fn=generate_first, inputs=[text, voice, speed, use_gpu, clone_voice_file],
                       outputs=[out_audio, out_ps])
    tokenize_btn.click(fn=tokenize_first, inputs=[text, voice], outputs=[out_ps])
    stream_event = stream_btn.click(fn=generate_all, inputs=[text, voice, speed, use_gpu], outputs=[out_stream])
    stop_btn.click(fn=None, cancels=[stream_event])
    predict_btn.click(fn=predict, inputs=[text, voice, speed], outputs=[out_audio])

if __name__ == "__main__":
    app.queue(api_open=API_OPEN).launch(
        server_name="127.0.0.1",
        server_port=40001,
        show_api=API_OPEN,
        inbrowser=True
    )