tortoise-tts/webui.py

import os
import argparse
import time
import json
import base64
import re
import urllib.request

import torch
import torchaudio
import music_tag
import gradio as gr
import gradio.utils

from datetime import datetime

import tortoise.api

from tortoise.api import TextToSpeech
from tortoise.utils.audio import load_audio, load_voice, load_voices, get_voice_dir
from tortoise.utils.text import split_and_recombine_text
from tortoise.utils.device import get_device_name

voicefixer = None

def generate(
    text,
    delimiter,
    emotion,
    prompt,
    voice,
    mic_audio,
    voice_latents_chunks,
    seed,
    candidates,
    num_autoregressive_samples,
    diffusion_iterations,
    temperature,
    diffusion_sampler,
    breathing_room,
    cvvp_weight,
    top_p,
    diffusion_temperature,
    length_penalty,
    repetition_penalty,
    cond_free_k,
    experimental_checkboxes,
    progress=None
):
    global args
    global tts

    try:
        tts
    except NameError:
        raise gr.Error("TTS is still initializing...")

    if voice != "microphone":
        voices = [voice]
    else:
        voices = []

    if voice == "microphone":
        if mic_audio is None:
            raise gr.Error("Please provide audio from mic when choosing `microphone` as a voice input")
        mic = load_audio(mic_audio, tts.input_sample_rate)
        voice_samples, conditioning_latents = [mic], None
    elif voice == "random":
        voice_samples, conditioning_latents = None, tts.get_random_conditioning_latents()
    else:
        progress(0, desc="Loading voice...")
        voice_samples, conditioning_latents = load_voice(voice)

    if voice_samples is not None:
        sample_voice = torch.cat(voice_samples, dim=-1).squeeze().cpu()

        conditioning_latents = tts.get_conditioning_latents(voice_samples, return_mels=not args.latents_lean_and_mean, progress=progress, slices=voice_latents_chunks, force_cpu=args.force_cpu_for_conditioning_latents)
        if len(conditioning_latents) == 4:
            conditioning_latents = (conditioning_latents[0], conditioning_latents[1], conditioning_latents[2], None)
            
        if voice != "microphone":
            torch.save(conditioning_latents, f'{get_voice_dir()}/{voice}/cond_latents.pth')
        voice_samples = None
    else:
        if conditioning_latents is not None:
            sample_voice, _ = load_voice(voice, load_latents=False)
            sample_voice = torch.cat(sample_voice, dim=-1).squeeze().cpu()
        else:
            sample_voice = None

    if seed == 0:
        seed = None

    if conditioning_latents is not None and len(conditioning_latents) == 2 and cvvp_weight > 0:
        print("Requesting weighing against CVVP weight, but voice latents are missing some extra data. Please regenerate your voice latents.")
        cvvp_weight = 0


    settings = {
        'temperature': float(temperature),

        'top_p': float(top_p),
        'diffusion_temperature': float(diffusion_temperature),
        'length_penalty': float(length_penalty),
        'repetition_penalty': float(repetition_penalty),
        'cond_free_k': float(cond_free_k),

        'num_autoregressive_samples': num_autoregressive_samples,
        'sample_batch_size': args.sample_batch_size,
        'diffusion_iterations': diffusion_iterations,

        'voice_samples': voice_samples,
        'conditioning_latents': conditioning_latents,
        'use_deterministic_seed': seed,
        'return_deterministic_state': True,
        'k': candidates,
        'diffusion_sampler': diffusion_sampler,
        'breathing_room': breathing_room,
        'progress': progress,
        'half_p': "Half Precision" in experimental_checkboxes,
        'cond_free': "Conditioning-Free" in experimental_checkboxes,
        'cvvp_amount': cvvp_weight,
    }

    if delimiter == "\\n":
        delimiter = "\n"

    if delimiter != "" and delimiter in text:
        texts = text.split(delimiter)
    else:
        texts = split_and_recombine_text(text)
 
    full_start_time = time.time()
 
    outdir = f"./results/{voice}/"
    os.makedirs(outdir, exist_ok=True)

    audio_cache = {}

    resample = None
    # not a ternary in the event for some reason I want to rely on librosa's upsampling interpolator rather than torchaudio's, for some reason
    if tts.output_sample_rate != args.output_sample_rate:
        resampler = torchaudio.transforms.Resample(
            tts.output_sample_rate,
            args.output_sample_rate,
            lowpass_filter_width=16,
            rolloff=0.85,
            resampling_method="kaiser_window",
            beta=8.555504641634386,
        )

    volume_adjust = torchaudio.transforms.Vol(gain=args.output_volume, gain_type="amplitude") if args.output_volume != 1 else None

    idx = 0
    idx_cache = {}
    for i, file in enumerate(os.listdir(outdir)):
        filename = os.path.basename(file)
        if filename[-5:] == ".json":
            match = re.findall(rf"^{voice}_(\d+)(?:.+?)\.json$", filename)
        elif filename[-4:] == ".wav":
            match = re.findall(rf"^{voice}_(\d+)(?:.+?)\.wav$", filename)
        else:
            continue
        if match is None or len(match) == 0:
            idx = idx + 1 # safety
            continue
        match = match[0]
        key = match[0]
        idx_cache[key] = True

    idx = idx + len(idx_cache)

    # I know there's something to pad I don't care
    pad = ""
    if idx < 10000:
        pad = f"{pad}0"
    if idx < 1000:
        pad = f"{pad}0"
    if idx < 100:
        pad = f"{pad}0"
    if idx < 10:
        pad = f"{pad}0"
    idx = f"{pad}{idx}"

    def get_name(line=0, candidate=0, combined=False):
        name = f"{idx}"
        if combined:
            name = f"{name}_combined"
        elif len(texts) > 1:
            name = f"{name}_{line}"
        if candidates > 1:
            name = f"{name}_{candidate}"
        return name

    for line, cut_text in enumerate(texts):
        if emotion == "Custom":
            if prompt.strip() != "":
                cut_text = f"[{prompt},] {cut_text}"
        else:
            cut_text = f"[I am really {emotion.lower()},] {cut_text}"

        progress.msg_prefix = f'[{str(line+1)}/{str(len(texts))}]'
        print(f"{progress.msg_prefix} Generating line: {cut_text}")

        start_time = time.time()
        gen, additionals = tts.tts(cut_text, **settings )
        seed = additionals[0]
        run_time = time.time()-start_time
        print(f"Generating line took {run_time} seconds")
 
        if not isinstance(gen, list):
            gen = [gen]

        for j, g in enumerate(gen):
            audio = g.squeeze(0).cpu()
            name = get_name(line=line, candidate=j)
            audio_cache[name] = {
                'audio': audio,
                'text': cut_text,
                'time': run_time
            }
            # save here in case some error happens mid-batch
            torchaudio.save(f'{outdir}/{voice}_{name}.wav', audio, tts.output_sample_rate)

    for k in audio_cache:
        audio = audio_cache[k]['audio']

        if resampler is not None:
            audio = resampler(audio)
        if volume_adjust is not None:
            audio = volume_adjust(audio)

        audio_cache[k]['audio'] = audio
        torchaudio.save(f'{outdir}/{voice}_{k}.wav', audio, args.output_sample_rate)
 
    output_voices = []
    for candidate in range(candidates):
        if len(texts) > 1:
            audio_clips = []
            for line in range(len(texts)):
                name = get_name(line=line, candidate=candidate)
                audio = audio_cache[name]['audio']
                audio_clips.append(audio)
            
            name = get_name(candidate=candidate, combined=True)
            audio = torch.cat(audio_clips, dim=-1)
            torchaudio.save(f'{outdir}/{voice}_{name}.wav', audio, args.output_sample_rate)

            audio = audio.squeeze(0).cpu()
            audio_cache[name] = {
                'audio': audio,
                'text': text,
                'time': time.time()-full_start_time,
                'output': True
            }
        else:
            name = get_name(candidate=candidate)
            audio_cache[name]['output'] = True

    info = {
        'text': text,
        'delimiter': '\\n' if delimiter == "\n" else delimiter,
        'emotion': emotion,
        'prompt': prompt,
        'voice': voice,
        'seed': seed,
        'candidates': candidates,
        'num_autoregressive_samples': num_autoregressive_samples,
        'diffusion_iterations': diffusion_iterations,
        'temperature': temperature,
        'diffusion_sampler': diffusion_sampler,
        'breathing_room': breathing_room,
        'cvvp_weight': cvvp_weight,
        'top_p': top_p,
        'diffusion_temperature': diffusion_temperature,
        'length_penalty': length_penalty,
        'repetition_penalty': repetition_penalty,
        'cond_free_k': cond_free_k,
        'experimentals': experimental_checkboxes,
        'time': time.time()-full_start_time,
    }

    # kludgy yucky codesmells
    for name in audio_cache:
        if 'output' not in audio_cache[name]:
            continue

        output_voices.append(f'{outdir}/{voice}_{name}.wav')
        with open(f'{outdir}/{voice}_{name}.json', 'w', encoding="utf-8") as f:
            f.write(json.dumps(info, indent='\t') )

    if args.voice_fixer and voicefixer:
        fixed_output_voices = []
        for path in progress.tqdm(output_voices, desc="Running voicefix..."):
            fixed = path.replace(".wav", "_fixed.wav")
            voicefixer.restore(
                input=path,
                output=fixed,
                cuda=get_device_name() == "cuda" and args.voice_fixer_use_cuda,
                #mode=mode,
            )
            fixed_output_voices.append(fixed)
        output_voices = fixed_output_voices

    if voice is not None and conditioning_latents is not None:
        with open(f'{get_voice_dir()}/{voice}/cond_latents.pth', 'rb') as f:
            info['latents'] = base64.b64encode(f.read()).decode("ascii")

    if args.embed_output_metadata:
        for name in progress.tqdm(audio_cache, desc="Embedding metadata..."):
            info['text'] = audio_cache[name]['text']
            info['time'] = audio_cache[name]['time']

            metadata = music_tag.load_file(f"{outdir}/{voice}_{name}.wav")
            metadata['lyrics'] = json.dumps(info) 
            metadata.save()
 
    if sample_voice is not None:
        sample_voice = (tts.input_sample_rate, sample_voice.numpy())

    print(info['time'])
    print(output_voices)
    print(f"Generation took {info['time']} seconds, saved to '{output_voices[0]}'\n")

    info['seed'] = settings['use_deterministic_seed']
    if 'latents' in info:
        del info['latents']

    with open(f'./config/generate.json', 'w', encoding="utf-8") as f:
        f.write(json.dumps(info, indent='\t') )

    stats = [
        [ seed, "{:.3f}".format(info['time']) ]
    ]

    return (
        sample_voice,
        output_voices,
        stats,
    )

def compute_latents(voice, voice_latents_chunks, progress=gr.Progress(track_tqdm=True)):
    global tts
    try:
        tts
    except NameError:
        raise gr.Error("TTS is still initializing...")

    voice_samples, conditioning_latents = load_voice(voice, load_latents=False)

    if voice_samples is None:
        return

    conditioning_latents = tts.get_conditioning_latents(voice_samples, return_mels=not args.latents_lean_and_mean, progress=progress, slices=voice_latents_chunks, force_cpu=args.force_cpu_for_conditioning_latents)

    if len(conditioning_latents) == 4:
        conditioning_latents = (conditioning_latents[0], conditioning_latents[1], conditioning_latents[2], None)
            
    torch.save(conditioning_latents, f'{get_voice_dir()}/{voice}/cond_latents.pth')

    return voice

def update_presets(value):
    PRESETS = {
        'Ultra Fast': {'num_autoregressive_samples': 16, 'diffusion_iterations': 30, 'cond_free': False},
        'Fast': {'num_autoregressive_samples': 96, 'diffusion_iterations': 80},
        'Standard': {'num_autoregressive_samples': 256, 'diffusion_iterations': 200},
        'High Quality': {'num_autoregressive_samples': 256, 'diffusion_iterations': 400},
    }
    
    if value in PRESETS:
        preset = PRESETS[value]
        return (gr.update(value=preset['num_autoregressive_samples']), gr.update(value=preset['diffusion_iterations']))
    else:
        return (gr.update(), gr.update())

def read_generate_settings(file, read_latents=True, read_json=True):
    j = None
    latents = None

    if file is not None:
        if hasattr(file, 'name'):
            file = file.name

        if file[-4:] == ".wav":
            metadata = music_tag.load_file(file)
            if 'lyrics' in metadata:
                j = json.loads(str(metadata['lyrics']))
        elif file[-5:] == ".json":
            with open(file, 'r') as f:
                j = json.load(f)

    if j is None:
        gr.Error("No metadata found in audio file to read")
    else:
        if 'latents' in j:
            if read_latents:
                latents = base64.b64decode(j['latents'])
            del j['latents']
        

        if "time" in j:
            j["time"] = "{:.3f}".format(j["time"])

    return (
        j,
        latents,
    )

def import_voice(file, saveAs = None):
    j, latents = read_generate_settings(file, read_latents=True)
    
    if j is not None and saveAs is None:
        saveAs = j['voice']
    if saveAs is None or saveAs == "":
        raise gr.Error("Specify a voice name")

    outdir = f'{get_voice_dir()}/{saveAs}/'
    os.makedirs(outdir, exist_ok=True)
    if latents:
        with open(f'{outdir}/cond_latents.pth', 'wb') as f:
            f.write(latents)
        latents = f'{outdir}/cond_latents.pth'
        print(f"Imported latents to {latents}")
    else:
        filename = file.name
        if filename[-4:] != ".wav":
            raise gr.Error("Please convert to a WAV first")

        path = f"{outdir}/{os.path.basename(filename)}"
        waveform, sampling_rate = torchaudio.load(filename)

        if args.voice_fixer:
            # resample to best bandwidth since voicefixer will do it anyways through librosa
            if sampling_rate != 44100:
                print(f"Resampling imported voice sample: {path}")
                resampler = torchaudio.transforms.Resample(
                    sampling_rate,
                    44100,
                    lowpass_filter_width=16,
                    rolloff=0.85,
                    resampling_method="kaiser_window",
                    beta=8.555504641634386,
                )
                waveform = resampler(waveform)
                sampling_rate = 44100

            torchaudio.save(path, waveform, sampling_rate)

            print(f"Running 'voicefixer' on voice sample: {path}")
            voicefixer.restore(
                input = path,
                output = path,
                cuda=get_device_name() == "cuda" and args.voice_fixer_use_cuda,
                #mode=mode,
            )
        else:
            torchaudio.save(path, waveform, sampling_rate)


        print(f"Imported voice to {path}")


def import_generate_settings(file="./config/generate.json"):
    settings, _ = read_generate_settings(file, read_latents=False)
    
    if settings is None:
        return None

    return (
        None if 'text' not in settings else settings['text'],
        None if 'delimiter' not in settings else settings['delimiter'],
        None if 'emotion' not in settings else settings['emotion'],
        None if 'prompt' not in settings else settings['prompt'],
        None if 'voice' not in settings else settings['voice'],
        None,
        None,
        None if 'seed' not in settings else settings['seed'],
        None if 'candidates' not in settings else settings['candidates'],
        None if 'num_autoregressive_samples' not in settings else settings['num_autoregressive_samples'],
        None if 'diffusion_iterations' not in settings else settings['diffusion_iterations'],
        0.8 if 'temperature' not in settings else settings['temperature'],
        "DDIM" if 'diffusion_sampler' not in settings else settings['diffusion_sampler'],
        8   if 'breathing_room' not in settings else settings['breathing_room'],
        0.0 if 'cvvp_weight' not in settings else settings['cvvp_weight'],
        0.8 if 'top_p' not in settings else settings['top_p'],
        1.0 if 'diffusion_temperature' not in settings else settings['diffusion_temperature'],
        1.0 if 'length_penalty' not in settings else settings['length_penalty'],
        2.0 if 'repetition_penalty' not in settings else settings['repetition_penalty'],
        2.0 if 'cond_free_k' not in settings else settings['cond_free_k'],
        None if 'experimentals' not in settings else settings['experimentals'],
    )

def curl(url):
    try:
        req = urllib.request.Request(url, headers={'User-Agent': 'Python'})
        conn = urllib.request.urlopen(req)
        data = conn.read()
        data = data.decode()
        data = json.loads(data)
        conn.close()
        return data
    except Exception as e:
        print(e)
        return None

def check_for_updates():
    if not os.path.isfile('./.git/FETCH_HEAD'):
        print("Cannot check for updates: not from a git repo")
        return False

    with open(f'./.git/FETCH_HEAD', 'r', encoding="utf-8") as f:
        head = f.read()
    
    match = re.findall(r"^([a-f0-9]+).+?https:\/\/(.+?)\/(.+?)\/(.+?)\n", head)
    if match is None or len(match) == 0:
        print("Cannot check for updates: cannot parse FETCH_HEAD")
        return False

    match = match[0]

    local = match[0]
    host = match[1]
    owner = match[2]
    repo = match[3]

    res = curl(f"https://{host}/api/v1/repos/{owner}/{repo}/branches/") #this only works for gitea instances

    if res is None or len(res) == 0:
        print("Cannot check for updates: cannot fetch from remote")
        return False

    remote = res[0]["commit"]["id"]

    if remote != local:
        print(f"New version found: {local[:8]} => {remote[:8]}")
        return True

    return False

def reload_tts():
    global tts
    del tts
    tts = setup_tortoise(restart=True)

def cancel_generate():
    tortoise.api.STOP_SIGNAL = True

def get_voice_list(dir=get_voice_dir()):
    return sorted([d for d in os.listdir(dir) if os.path.isdir(os.path.join(dir, d)) and len(os.listdir(os.path.join(dir, d))) > 0 ]) + ["microphone", "random"]

def update_voices():
    return gr.Dropdown.update(choices=get_voice_list())

def export_exec_settings( listen, share, check_for_updates, models_from_local_only, low_vram, embed_output_metadata, latents_lean_and_mean, voice_fixer, voice_fixer_use_cuda, force_cpu_for_conditioning_latents, sample_batch_size, concurrency_count, output_sample_rate, output_volume ):
    args.listen = listen
    args.share = share
    args.check_for_updates = check_for_updates
    args.models_from_local_only = models_from_local_only
    args.low_vram = low_vram
    args.force_cpu_for_conditioning_latents = force_cpu_for_conditioning_latents
    args.sample_batch_size = sample_batch_size
    args.embed_output_metadata = embed_output_metadata
    args.latents_lean_and_mean = latents_lean_and_mean
    args.voice_fixer = voice_fixer
    args.voice_fixer_use_cuda = voice_fixer_use_cuda
    args.concurrency_count = concurrency_count
    args.output_sample_rate = output_sample_rate
    args.output_volume = output_volume

    settings = {
        'listen': None if args.listen else args.listen,
        'share': args.share,
        'low-vram':args.low_vram,
        'check-for-updates':args.check_for_updates,
        'models-from-local-only':args.models_from_local_only,
        'force-cpu-for-conditioning-latents': args.force_cpu_for_conditioning_latents,
        'sample-batch-size': args.sample_batch_size,
        'embed-output-metadata': args.embed_output_metadata,
        'latents-lean-and-mean': args.latents_lean_and_mean,
        'voice-fixer': args.voice_fixer,
        'voice-fixer-use-cuda': args.voice_fixer_use_cuda,
        'concurrency-count': args.concurrency_count,
        'output-sample-rate': args.output_sample_rate,
        'output-volume': args.output_volume,
    }

    with open(f'./config/exec.json', 'w', encoding="utf-8") as f:
        f.write(json.dumps(settings, indent='\t') )

def setup_args():
    default_arguments = {
        'share': False,
        'listen': None,
        'check-for-updates': False,
        'models-from-local-only': False,
        'low-vram': False,
        'sample-batch-size': None,
        'embed-output-metadata': True,
        'latents-lean-and-mean': True,
        'voice-fixer': True,
        'voice-fixer-use-cuda': True,
        'force-cpu-for-conditioning-latents': False,
        'concurrency-count': 2,
        'output-sample-rate': 44100,
        'output-volume': 1,
    }

    if os.path.isfile('./config/exec.json'):
        with open(f'./config/exec.json', 'r', encoding="utf-8") as f:
            overrides = json.load(f)
            for k in overrides:
                default_arguments[k] = overrides[k]

    parser = argparse.ArgumentParser()
    parser.add_argument("--share", action='store_true', default=default_arguments['share'], help="Lets Gradio return a public URL to use anywhere")
    parser.add_argument("--listen", default=default_arguments['listen'], help="Path for Gradio to listen on")
    parser.add_argument("--check-for-updates", action='store_true', default=default_arguments['check-for-updates'], help="Checks for update on startup")
    parser.add_argument("--models-from-local-only", action='store_true', default=default_arguments['models-from-local-only'], help="Only loads models from disk, does not check for updates for models")
    parser.add_argument("--low-vram", action='store_true', default=default_arguments['low-vram'], help="Disables some optimizations that increases VRAM usage")
    parser.add_argument("--no-embed-output-metadata", action='store_false', default=not default_arguments['embed-output-metadata'], help="Disables embedding output metadata into resulting WAV files for easily fetching its settings used with the web UI (data is stored in the lyrics metadata tag)")
    parser.add_argument("--latents-lean-and-mean", action='store_true', default=default_arguments['latents-lean-and-mean'], help="Exports the bare essentials for latents.")
    parser.add_argument("--voice-fixer", action='store_true', default=default_arguments['voice-fixer'], help="Uses python module 'voicefixer' to improve audio quality, if available.")
    parser.add_argument("--voice-fixer-use-cuda", action='store_true', default=default_arguments['voice-fixer-use-cuda'], help="Hints to voicefixer to use CUDA, if available.")
    parser.add_argument("--force-cpu-for-conditioning-latents", default=default_arguments['force-cpu-for-conditioning-latents'], action='store_true', help="Forces computing conditional latents to be done on the CPU (if you constantyl OOM on low chunk counts)")
    parser.add_argument("--sample-batch-size", default=default_arguments['sample-batch-size'], type=int, help="Sets how many batches to use during the autoregressive samples pass")
    parser.add_argument("--concurrency-count", type=int, default=default_arguments['concurrency-count'], help="How many Gradio events to process at once")
    parser.add_argument("--output-sample-rate", type=int, default=default_arguments['output-sample-rate'], help="Sample rate to resample the output to (from 24KHz)")
    parser.add_argument("--output-volume", type=float, default=default_arguments['output-volume'], help="Adjusts volume of output")
    args = parser.parse_args()

    args.embed_output_metadata = not args.no_embed_output_metadata

    args.listen_host = None
    args.listen_port = None
    args.listen_path = None
    if args.listen:
        try:
            match = re.findall(r"^(?:(.+?):(\d+))?(\/.+?)?$", args.listen)[0]

            args.listen_host = match[0] if match[0] != "" else "127.0.0.1"
            args.listen_port = match[1] if match[1] != "" else None
            args.listen_path = match[2] if match[2] != "" else "/"
        except Exception as e:
            pass

    if args.listen_port is not None:
        args.listen_port = int(args.listen_port)
    
    return args

def setup_tortoise(restart=False):
    global args
    global tts
    global voicefixer

    if args.voice_fixer and not restart:
        try:
            from voicefixer import VoiceFixer
            print("Initializating voice-fixer")
            voicefixer = VoiceFixer()
            print("initialized voice-fixer")
        except Exception as e:
            pass

    print("Initializating TorToiSe...")
    tts = TextToSpeech(minor_optimizations=not args.low_vram)
    print("TorToiSe initialized, ready for generation.")
    return tts

def setup_gradio():
    global args
    
    if not args.share:
        def noop(function, return_value=None):
            def wrapped(*args, **kwargs):
                return return_value
            return wrapped
        gradio.utils.version_check = noop(gradio.utils.version_check)
        gradio.utils.initiated_analytics = noop(gradio.utils.initiated_analytics)
        gradio.utils.launch_analytics = noop(gradio.utils.launch_analytics)
        gradio.utils.integration_analytics = noop(gradio.utils.integration_analytics)
        gradio.utils.error_analytics = noop(gradio.utils.error_analytics)
        gradio.utils.log_feature_analytics = noop(gradio.utils.log_feature_analytics)
        #gradio.utils.get_local_ip_address = noop(gradio.utils.get_local_ip_address, 'localhost')

    if args.models_from_local_only:
        os.environ['TRANSFORMERS_OFFLINE']='1'

    with gr.Blocks() as webui:
        with gr.Tab("Generate"):
            with gr.Row():
                with gr.Column():
                    text = gr.Textbox(lines=4, label="Prompt")
            with gr.Row():
                with gr.Column():
                    delimiter = gr.Textbox(lines=1, label="Line Delimiter", placeholder="\\n")

                    emotion = gr.Radio(
                        ["Happy", "Sad", "Angry", "Disgusted", "Arrogant", "Custom"],
                        value="Custom",
                        label="Emotion",
                        type="value",
                        interactive=True
                    )
                    prompt = gr.Textbox(lines=1, label="Custom Emotion + Prompt (if selected)")
                    voice = gr.Dropdown(
                        get_voice_list(),
                        label="Voice",
                        type="value",
                    )
                    mic_audio = gr.Audio(
                        label="Microphone Source",
                        source="microphone",
                        type="filepath",
                    )
                    refresh_voices = gr.Button(value="Refresh Voice List")
                    refresh_voices.click(update_voices,
                        inputs=None,
                        outputs=voice
                    )
                    voice_latents_chunks = gr.Slider(label="Voice Chunks", minimum=1, maximum=64, value=1, step=1)
                    recompute_voice_latents = gr.Button(value="(Re)Compute Voice Latents")
                    recompute_voice_latents.click(compute_latents,
                        inputs=[
                            voice,
                            voice_latents_chunks,
                        ],
                        outputs=voice,
                    )
                    
                    prompt.change(fn=lambda value: gr.update(value="Custom"),
                        inputs=prompt,
                        outputs=emotion
                    )
                    mic_audio.change(fn=lambda value: gr.update(value="microphone"),
                        inputs=mic_audio,
                        outputs=voice
                    )
                with gr.Column():
                    candidates = gr.Slider(value=1, minimum=1, maximum=6, step=1, label="Candidates")
                    seed = gr.Number(value=0, precision=0, label="Seed")

                    preset = gr.Radio(
                        ["Ultra Fast", "Fast", "Standard", "High Quality"],
                        label="Preset",
                        type="value",
                    )
                    num_autoregressive_samples = gr.Slider(value=128, minimum=0, maximum=512, step=1, label="Samples")
                    diffusion_iterations = gr.Slider(value=128, minimum=0, maximum=512, step=1, label="Iterations")

                    temperature = gr.Slider(value=0.2, minimum=0, maximum=1, step=0.1, label="Temperature")
                    breathing_room = gr.Slider(value=8, minimum=1, maximum=32, step=1, label="Pause Size")
                    diffusion_sampler = gr.Radio(
                        ["P", "DDIM"], # + ["K_Euler_A", "DPM++2M"],
                        value="P",
                        label="Diffusion Samplers",
                        type="value",
                    )

                    preset.change(fn=update_presets,
                        inputs=preset,
                        outputs=[
                            num_autoregressive_samples,
                            diffusion_iterations,
                        ],
                    )

                    show_experimental_settings = gr.Checkbox(label="Show Experimental Settings")
                    reset_generation_settings_button = gr.Button(value="Reset to Default")
                with gr.Column(visible=False) as col:
                    experimental_column = col

                    experimental_checkboxes = gr.CheckboxGroup(["Half Precision", "Conditioning-Free"], value=["Conditioning-Free"], label="Experimental Flags")
                    cvvp_weight = gr.Slider(value=0, minimum=0, maximum=1, label="CVVP Weight")
                    top_p = gr.Slider(value=0.8, minimum=0, maximum=1, label="Top P")
                    diffusion_temperature = gr.Slider(value=1.0, minimum=0, maximum=1, label="Diffusion Temperature")
                    length_penalty = gr.Slider(value=1.0, minimum=0, maximum=8, label="Length Penalty")
                    repetition_penalty = gr.Slider(value=2.0, minimum=0, maximum=8, label="Repetition Penalty")
                    cond_free_k = gr.Slider(value=2.0, minimum=0, maximum=4, label="Conditioning-Free K")

                    show_experimental_settings.change(
                        fn=lambda x: gr.update(visible=x),
                        inputs=show_experimental_settings,
                        outputs=experimental_column
                    )
                with gr.Column():
                    submit = gr.Button(value="Generate")
                    stop = gr.Button(value="Stop")

                    generation_results = gr.Dataframe(label="Results", headers=["Seed", "Time"], visible=False)
                    source_sample = gr.Audio(label="Source Sample", visible=False)
                    output_audio = gr.Audio(label="Output")
                    candidates_list = gr.Dropdown(label="Candidates", type="value", visible=False)
                    output_pick = gr.Button(value="Select Candidate", visible=False)
                    
        with gr.Tab("History"):
            with gr.Row():
                with gr.Column():
                    headers = {
                        "Name": "",
                        "Samples": "num_autoregressive_samples",
                        "Iterations": "diffusion_iterations",
                        "Temp.": "temperature",
                        "Sampler": "diffusion_sampler",
                        "CVVP": "cvvp_weight",
                        "Top P": "top_p",
                        "Diff. Temp.": "diffusion_temperature",
                        "Len Pen": "length_penalty",
                        "Rep Pen": "repetition_penalty",
                        "Cond-Free K": "cond_free_k",
                        "Time": "time",
                    }
                    history_info = gr.Dataframe(label="Results", headers=list(headers.keys()))
            with gr.Row():
                with gr.Column():
                    history_voices = gr.Dropdown(
                        get_voice_list("./results/"),
                        label="Voice",
                        type="value",
                    )

                    history_view_results_button = gr.Button(value="View Files")
                with gr.Column():
                    history_results_list = gr.Dropdown(label="Results",type="value", interactive=True)
                    history_view_result_button = gr.Button(value="View File")
                with gr.Column():
                    history_audio = gr.Audio()
                    history_copy_settings_button = gr.Button(value="Copy Settings")
                
                def history_view_results( voice ):
                    results = []
                    files = []
                    outdir = f"./results/{voice}/"
                    for i, file in enumerate(sorted(os.listdir(outdir))):
                        if file[-4:] != ".wav":
                            continue

                        metadata, _ = read_generate_settings(f"{outdir}/{file}", read_latents=False)
                        if metadata is None:
                            continue
                            
                        values = []
                        for k in headers:
                            v = file
                            if k != "Name":
                                v = metadata[headers[k]]
                            values.append(v)


                        files.append(file)
                        results.append(values)

                    return (
                        results,
                        gr.Dropdown.update(choices=sorted(files))
                    )

                history_view_results_button.click(
                    fn=history_view_results,
                    inputs=history_voices,
                    outputs=[
                        history_info,
                        history_results_list,
                    ]
                )
                history_view_result_button.click(
                    fn=lambda voice, file: f"./results/{voice}/{file}",
                    inputs=[
                        history_voices,
                        history_results_list,
                    ],
                    outputs=history_audio
                )
        with gr.Tab("Utilities"):
            with gr.Row():
                with gr.Column():
                    audio_in = gr.File(type="file", label="Audio Input", file_types=["audio"])
                    copy_button = gr.Button(value="Copy Settings")
                    import_voice_name = gr.Textbox(label="Voice Name")
                    import_voice_button = gr.Button(value="Import Voice")
                with gr.Column():
                    metadata_out = gr.JSON(label="Audio Metadata")
                    latents_out = gr.File(type="binary", label="Voice Latents")

                    def read_generate_settings_proxy(file, saveAs='.temp'):
                        j, latents = read_generate_settings(file)

                        if latents:
                            outdir = f'{get_voice_dir()}/{saveAs}/'
                            os.makedirs(outdir, exist_ok=True)
                            with open(f'{outdir}/cond_latents.pth', 'wb') as f:
                                f.write(latents)
                            
                            latents = f'{outdir}/cond_latents.pth'

                        return (
                            j,
                            gr.update(value=latents, visible=latents is not None),
                            None if j is None else j['voice']
                        )

                    audio_in.upload(
                        fn=read_generate_settings_proxy,
                        inputs=audio_in,
                        outputs=[
                            metadata_out,
                            latents_out,
                            import_voice_name
                        ]
                    )

                import_voice_button.click(
                    fn=import_voice,
                    inputs=[
                        audio_in,
                        import_voice_name,
                    ]
                )
        with gr.Tab("Settings"):
            with gr.Row():
                exec_inputs = []
                with gr.Column():
                    exec_inputs = exec_inputs + [
                        gr.Textbox(label="Listen", value=args.listen, placeholder="127.0.0.1:7860/"),
                        gr.Checkbox(label="Public Share Gradio", value=args.share),
                        gr.Checkbox(label="Check For Updates", value=args.check_for_updates),
                        gr.Checkbox(label="Only Load Models Locally", value=args.models_from_local_only),
                        gr.Checkbox(label="Low VRAM", value=args.low_vram),
                        gr.Checkbox(label="Embed Output Metadata", value=args.embed_output_metadata),
                        gr.Checkbox(label="Slimmer Computed Latents", value=args.latents_lean_and_mean),
                        gr.Checkbox(label="Voice Fixer", value=args.voice_fixer),
                        gr.Checkbox(label="Use CUDA for Voice Fixer", value=args.voice_fixer_use_cuda),
                        gr.Checkbox(label="Force CPU for Conditioning Latents", value=args.force_cpu_for_conditioning_latents),
                    ]
                    gr.Button(value="Check for Updates").click(check_for_updates)
                    gr.Button(value="Reload TTS").click(reload_tts)
                with gr.Column():
                    exec_inputs = exec_inputs + [
                        gr.Number(label="Sample Batch Size", precision=0, value=args.sample_batch_size),
                        gr.Number(label="Concurrency Count", precision=0, value=args.concurrency_count),
                        gr.Number(label="Ouptut Sample Rate", precision=0, value=args.output_sample_rate),
                        gr.Slider(label="Ouptut Volume", minimum=0, maximum=2, value=args.output_volume),
                    ]

                for i in exec_inputs:
                    i.change(
                        fn=export_exec_settings,
                        inputs=exec_inputs
                    )

        input_settings = [
            text,
            delimiter,
            emotion,
            prompt,
            voice,
            mic_audio,
            voice_latents_chunks,
            seed,
            candidates,
            num_autoregressive_samples,
            diffusion_iterations,
            temperature,
            diffusion_sampler,
            breathing_room,
            cvvp_weight,
            top_p,
            diffusion_temperature,
            length_penalty,
            repetition_penalty,
            cond_free_k,
            experimental_checkboxes,
        ]

        # YUCK
        def run_generation(
            text,
            delimiter,
            emotion,
            prompt,
            voice,
            mic_audio,
            voice_latents_chunks,
            seed,
            candidates,
            num_autoregressive_samples,
            diffusion_iterations,
            temperature,
            diffusion_sampler,
            breathing_room,
            cvvp_weight,
            top_p,
            diffusion_temperature,
            length_penalty,
            repetition_penalty,
            cond_free_k,
            experimental_checkboxes,
            progress=gr.Progress(track_tqdm=True)
        ):
            try:
                sample, outputs, stats = generate(
                    text,
                    delimiter,
                    emotion,
                    prompt,
                    voice,
                    mic_audio,
                    voice_latents_chunks,
                    seed,
                    candidates,
                    num_autoregressive_samples,
                    diffusion_iterations,
                    temperature,
                    diffusion_sampler,
                    breathing_room,
                    cvvp_weight,
                    top_p,
                    diffusion_temperature,
                    length_penalty,
                    repetition_penalty,
                    cond_free_k,
                    experimental_checkboxes,
                    progress
                )
            except Exception as e:
                message = str(e)
                if message == "Kill signal detected":
                    reload_tts()

                raise gr.Error(message)
            

            return (
                outputs[0],
                gr.update(value=sample, visible=sample is not None),
                gr.update(choices=outputs, value=outputs[0], visible=len(outputs) > 1, interactive=True),
                gr.update(visible=len(outputs) > 1),
                gr.update(value=stats, visible=True),
            )

        output_pick.click(
            lambda x: x,
            inputs=candidates_list,
            outputs=output_audio,
        )

        submit.click(
            lambda: (gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False)),
            outputs=[source_sample, candidates_list, output_pick, generation_results],
        )

        submit_event = submit.click(run_generation,
            inputs=input_settings,
            outputs=[output_audio, source_sample, candidates_list, output_pick, generation_results],
        )


        copy_button.click(import_generate_settings,
            inputs=audio_in, # JSON elements cannot be used as inputs
            outputs=input_settings
        )

        def reset_generation_settings():
            with open(f'./config/generate.json', 'w', encoding="utf-8") as f:
                f.write(json.dumps({}, indent='\t') )
            return import_generate_settings()

        reset_generation_settings_button.click(
            fn=reset_generation_settings,
            inputs=None,
            outputs=input_settings
        )

        def history_copy_settings( voice, file ):
            settings = import_generate_settings( f"./results/{voice}/{file}" )
            return settings

        history_copy_settings_button.click(history_copy_settings,
            inputs=[
                history_voices,
                history_results_list,
            ],
            outputs=input_settings
        )

        if os.path.isfile('./config/generate.json'):
            webui.load(import_generate_settings, inputs=None, outputs=input_settings)
        
        if args.check_for_updates:
            webui.load(check_for_updates)

        stop.click(fn=cancel_generate, inputs=None, outputs=None, cancels=[submit_event])


    webui.queue(concurrency_count=args.concurrency_count)

    return webui