tts-voice-creator-clone-and-design-2/guided_installer.py

#!/usr/bin/env python3
from __future__ import annotations

import os
import platform
import queue
import shutil
import subprocess
import threading
import webbrowser
from pathlib import Path
import tkinter as tk
from tkinter import filedialog, messagebox, ttk


ROOT = Path(__file__).resolve().parent
GENERATED_DIR = ROOT / "installer" / "generated"
COMPOSE_PATH = GENERATED_DIR / "docker-compose.guided.yml"
DOCKER_DOCS = "https://docs.docker.com/engine/install/"
OPEN_WEBUI_DOCS = "https://docs.openwebui.com/features/audio/"


MODEL_EXPLANATIONS = {
    "clone": (
        "Voice Clone",
        "Uses recorded reference voices from the Voice Library. This is the everyday server for exported voices and Open WebUI.",
        "faster-qwen3-tts",
        "8020",
        "/models/Qwen3-TTS",
    ),
    "design": (
        "Voice Design",
        "Generates a voice from a written description. Great for prototyping characters without a recording.",
        "faster-qwen3-tts-voicedesign",
        "8021",
        "/models/Qwen3-TTS-VoiceDesign",
    ),
    "custom": (
        "Custom Voice",
        "Uses Qwen3 CustomVoice speaker IDs and optional instructions. Useful when you already maintain named model speakers.",
        "faster-qwen3-tts-customvoice",
        "8022",
        "/models/Qwen3-TTS-CustomVoice",
    ),
}


class GuidedInstaller(tk.Tk):
    def __init__(self) -> None:
        super().__init__()
        self.title("TTS Voice Creator Guided Installer")
        self.geometry("1120x780")
        self.minsize(920, 680)
        self.log_queue: queue.Queue[str] = queue.Queue()
        self._build_vars()
        self._build_ui()
        self.after(120, self._drain_log)
        self.check_environment()

    def _build_vars(self) -> None:
        home = Path.home()
        self.voice_dir = tk.StringVar(value=str(home / "TTS_Voices"))
        self.model_dir = tk.StringVar(value=str(home / "TTS_Models"))
        self.creator_port = tk.StringVar(value="7860")
        self.network_name = tk.StringVar(value="tts_net")
        self.tts_image = tk.StringVar(value="faster-qwen3-tts-dgx-spark:v4")
        self.install_portainer = tk.BooleanVar(value=False)
        self.model_enabled = {key: tk.BooleanVar(value=(key == "clone")) for key in MODEL_EXPLANATIONS}
        self.model_ports = {key: tk.StringVar(value=meta[3]) for key, meta in MODEL_EXPLANATIONS.items()}
        self.model_paths = {key: tk.StringVar(value=str(Path(self.model_dir.get()) / Path(meta[4]).name)) for key, meta in MODEL_EXPLANATIONS.items()}
        self.extra_tts_name = tk.StringVar(value="")
        self.extra_tts_url = tk.StringVar(value="")

    def _build_ui(self) -> None:
        self.columnconfigure(0, weight=0)
        self.columnconfigure(1, weight=1)
        self.rowconfigure(0, weight=1)

        left = ttk.Frame(self, padding=14)
        left.grid(row=0, column=0, sticky="ns")
        right = ttk.Frame(self, padding=14)
        right.grid(row=0, column=1, sticky="nsew")
        right.columnconfigure(0, weight=1)
        right.rowconfigure(1, weight=1)

        ttk.Label(left, text="Install Checks", font=("", 15, "bold")).pack(anchor="w")
        self.status = tk.Text(left, width=38, height=15, wrap="word")
        self.status.pack(fill="x", pady=(8, 12))
        ttk.Button(left, text="Re-check this computer", command=self.check_environment).pack(fill="x", pady=3)
        ttk.Button(left, text="Open Docker install guide", command=lambda: webbrowser.open(DOCKER_DOCS)).pack(fill="x", pady=3)
        ttk.Button(left, text="Install Docker on Linux", command=self.install_docker_linux).pack(fill="x", pady=3)
        ttk.Button(left, text="Install Portainer", command=self.install_portainer_now).pack(fill="x", pady=3)
        ttk.Button(left, text="Generate compose file", command=self.generate_compose).pack(fill="x", pady=(14, 3))
        ttk.Button(left, text="Install / update selected stack", command=self.install_stack).pack(fill="x", pady=3)
        ttk.Button(left, text="Open Open WebUI audio guide", command=lambda: webbrowser.open(OPEN_WEBUI_DOCS)).pack(fill="x", pady=(14, 3))

        nb = ttk.Notebook(right)
        nb.grid(row=0, column=0, sticky="nsew")
        right.rowconfigure(0, weight=1)

        self._setup_tab(nb)
        self._models_tab(nb)
        self._openwebui_tab(nb)
        self._log_tab(nb)

    def _setup_tab(self, nb: ttk.Notebook) -> None:
        tab = ttk.Frame(nb, padding=12)
        tab.columnconfigure(1, weight=1)
        nb.add(tab, text="1. Basics")
        rows = [
            ("Voice library folder", self.voice_dir, True),
            ("Model cache / LLM folder", self.model_dir, True),
            ("Creator app port", self.creator_port, False),
            ("Docker network", self.network_name, False),
            ("Qwen TTS Docker image", self.tts_image, False),
        ]
        for row, (label, var, browse) in enumerate(rows):
            ttk.Label(tab, text=label).grid(row=row, column=0, sticky="w", pady=6)
            ttk.Entry(tab, textvariable=var).grid(row=row, column=1, sticky="ew", pady=6, padx=8)
            if browse:
                ttk.Button(tab, text="Browse", command=lambda v=var: self.pick_dir(v)).grid(row=row, column=2, pady=6)
        ttk.Checkbutton(tab, text="Also install Portainer web UI", variable=self.install_portainer).grid(row=len(rows), column=1, sticky="w", pady=8)
        text = (
            "What these pieces do:\n\n"
            "Docker runs the app and TTS servers in containers.\n"
            "Portainer is optional. It gives non-command-line users a web dashboard for containers.\n"
            "The voice folder stores cloned voices and metadata.\n"
            "The model folder stores large Qwen model files, ideally on a fast disk with enough space."
        )
        ttk.Label(tab, text=text, wraplength=680, justify="left").grid(row=len(rows)+1, column=0, columnspan=3, sticky="ew", pady=16)

    def _models_tab(self, nb: ttk.Notebook) -> None:
        tab = ttk.Frame(nb, padding=12)
        tab.columnconfigure(2, weight=1)
        nb.add(tab, text="2. Qwen Models")
        ttk.Label(tab, text="Install", font=("", 10, "bold")).grid(row=0, column=0, sticky="w")
        ttk.Label(tab, text="Model", font=("", 10, "bold")).grid(row=0, column=1, sticky="w")
        ttk.Label(tab, text="Model location", font=("", 10, "bold")).grid(row=0, column=2, sticky="w")
        ttk.Label(tab, text="Port", font=("", 10, "bold")).grid(row=0, column=3, sticky="w")
        for i, (key, meta) in enumerate(MODEL_EXPLANATIONS.items(), start=1):
            title, desc, _container, _port, _path = meta
            ttk.Checkbutton(tab, variable=self.model_enabled[key]).grid(row=i, column=0, sticky="nw", pady=8)
            ttk.Label(tab, text=f"{title}\n{desc}", wraplength=260, justify="left").grid(row=i, column=1, sticky="w", pady=8, padx=8)
            ttk.Entry(tab, textvariable=self.model_paths[key]).grid(row=i, column=2, sticky="ew", pady=8, padx=8)
            ttk.Entry(tab, textvariable=self.model_ports[key], width=8).grid(row=i, column=3, sticky="w", pady=8)

        ttk.Label(tab, text="Optional external TTS endpoint", font=("", 11, "bold")).grid(row=5, column=0, columnspan=4, sticky="w", pady=(22, 6))
        ttk.Label(tab, text="Name").grid(row=6, column=0, sticky="w")
        ttk.Entry(tab, textvariable=self.extra_tts_name).grid(row=6, column=1, sticky="ew", padx=8)
        ttk.Label(tab, text="Base URL").grid(row=6, column=2, sticky="e")
        ttk.Entry(tab, textvariable=self.extra_tts_url).grid(row=6, column=3, sticky="ew")
        ttk.Label(tab, text="Use this if you already have another OpenAI-compatible TTS server. The creator can route to it later by setting the TTS API URL in Settings.", wraplength=760).grid(row=7, column=0, columnspan=4, sticky="w", pady=10)

    def _openwebui_tab(self, nb: ttk.Notebook) -> None:
        tab = ttk.Frame(nb, padding=12)
        tab.columnconfigure(0, weight=1)
        nb.add(tab, text="3. Open WebUI")
        self.openwebui_text = tk.Text(tab, height=24, wrap="word")
        self.openwebui_text.grid(row=0, column=0, sticky="nsew")
        ttk.Button(tab, text="Refresh instructions", command=self.refresh_openwebui_text).grid(row=1, column=0, sticky="e", pady=8)
        self.refresh_openwebui_text()

    def _log_tab(self, nb: ttk.Notebook) -> None:
        tab = ttk.Frame(nb, padding=12)
        tab.rowconfigure(0, weight=1)
        tab.columnconfigure(0, weight=1)
        nb.add(tab, text="Log")
        self.log = tk.Text(tab, wrap="word")
        self.log.grid(row=0, column=0, sticky="nsew")

    def pick_dir(self, var: tk.StringVar) -> None:
        chosen = filedialog.askdirectory(initialdir=var.get() or str(Path.home()))
        if chosen:
            var.set(chosen)

    def log_line(self, msg: str) -> None:
        self.log_queue.put(msg.rstrip() + "\n")

    def _drain_log(self) -> None:
        while True:
            try:
                msg = self.log_queue.get_nowait()
            except queue.Empty:
                break
            self.log.insert("end", msg)
            self.log.see("end")
        self.after(120, self._drain_log)

    def run_threaded(self, title: str, commands: list[list[str]]) -> None:
        def worker():
            self.log_line(f"\n== {title} ==")
            for cmd in commands:
                self.log_line("$ " + " ".join(cmd))
                try:
                    proc = subprocess.run(cmd, cwd=ROOT, text=True, capture_output=True)
                    if proc.stdout:
                        self.log_line(proc.stdout)
                    if proc.stderr:
                        self.log_line(proc.stderr)
                    if proc.returncode:
                        self.log_line(f"Command exited with {proc.returncode}")
                        break
                except Exception as exc:
                    self.log_line(f"Failed: {exc}")
                    break
        threading.Thread(target=worker, daemon=True).start()

    def command_ok(self, cmd: list[str]) -> bool:
        try:
            return subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=6).returncode == 0
        except Exception:
            return False

    def docker_names(self) -> set[str]:
        try:
            out = subprocess.check_output(["docker", "ps", "-a", "--format", "{{.Names}}"], text=True, timeout=8)
            return {line.strip() for line in out.splitlines() if line.strip()}
        except Exception:
            return set()

    def check_environment(self) -> None:
        docker_bin = shutil.which("docker")
        compose_ok = self.command_ok(["docker", "compose", "version"]) if docker_bin else False
        daemon_ok = self.command_ok(["docker", "info"]) if docker_bin else False
        names = self.docker_names() if daemon_ok else set()
        lines = [
            f"System: {platform.system()} {platform.release()}",
            f"Docker command: {'found' if docker_bin else 'missing'}",
            f"Docker daemon: {'running' if daemon_ok else 'not reachable'}",
            f"Docker Compose plugin: {'found' if compose_ok else 'missing'}",
            f"Portainer: {'installed' if any('portainer' in n for n in names) else 'not found'}",
            f"Creator app: {'installed' if 'tts-voice-creator-clone-and-design' in names or 'tts-voice-creator' in names or 'voice-clone-factory' in names else 'not found'}",
            f"Qwen clone TTS: {'installed' if 'faster-qwen3-tts' in names else 'not found'}",
            f"Qwen Voice Design: {'installed' if 'faster-qwen3-tts-voicedesign' in names else 'not found'}",
            f"Qwen Custom Voice: {'installed' if 'faster-qwen3-tts-customvoice' in names else 'not found'}",
        ]
        self.status.delete("1.0", "end")
        self.status.insert("end", "\n".join(lines))

    def install_portainer_now(self) -> None:
        if not messagebox.askyesno("Install Portainer", "Install or update Portainer CE with Docker?"):
            return
        self.run_threaded("Install Portainer", [
            ["docker", "volume", "create", "portainer_data"],
            [
                "docker", "run", "-d", "--name", "portainer", "--restart=always",
                "-p", "9000:9000", "-p", "9443:9443",
                "-v", "/var/run/docker.sock:/var/run/docker.sock",
                "-v", "portainer_data:/data",
                "portainer/portainer-ce:latest",
            ],
        ])

    def install_docker_linux(self) -> None:
        if platform.system().lower() != "linux":
            messagebox.showinfo("Docker installer", "This one-click installer is only for Linux. Use the Docker install guide for this operating system.")
            return
        if not messagebox.askyesno(
            "Install Docker",
            "Run Docker's official Linux convenience installer?\n\nThis may ask for sudo privileges and needs internet access.",
        ):
            return
        self.run_threaded("Install Docker", [
            ["curl", "-fsSL", "https://get.docker.com", "-o", "/tmp/get-docker.sh"],
            ["sh", "/tmp/get-docker.sh"],
            ["docker", "version"],
        ])

    def generate_compose(self) -> None:
        GENERATED_DIR.mkdir(parents=True, exist_ok=True)
        content = self.compose_text()
        COMPOSE_PATH.write_text(content)
        self.log_line(f"Generated {COMPOSE_PATH}")
        messagebox.showinfo("Compose generated", f"Generated:\n{COMPOSE_PATH}")

    def install_stack(self) -> None:
        self.generate_compose()
        if self.install_portainer.get():
            self.install_portainer_now()
        if not messagebox.askyesno("Install stack", "Run docker compose up -d for the generated stack?"):
            return
        self.run_threaded("Install selected stack", [
            ["docker", "network", "create", self.network_name.get()],
            ["docker", "compose", "-f", str(COMPOSE_PATH), "up", "-d", "--build"],
        ])

    def compose_text(self) -> str:
        network = self.network_name.get().strip() or "tts_net"
        voice_dir = self.voice_dir.get().strip() or "./voices"
        model_dir = self.model_dir.get().strip() or "./models"
        image = self.tts_image.get().strip() or "faster-qwen3-tts-dgx-spark:v4"
        creator_port = self.creator_port.get().strip() or "7860"
        services = [
            "services:",
            "  tts-voice-creator-clone-and-design:",
            "    build:",
            "      context: ../..",
            "      dockerfile: Dockerfile",
            "    image: tts-voice-creator-clone-and-design:latest",
            "    container_name: tts-voice-creator-clone-and-design",
            "    restart: unless-stopped",
            "    ports:",
            f"      - \"{creator_port}:7860\"",
            "    volumes:",
            f"      - {voice_dir}:/voices:rw",
            "      - tts-voice-creator-clone-and-design:/home/app/.config/tts-voice-creator",
            "    environment:",
            "      - PYTHONUNBUFFERED=1",
            "      - VOICES_DIR=/voices",
            "      - OUTPUT_DIR=/voices/active_voices",
            "      - TTS_CONTAINER_NAME=faster-qwen3-tts",
            "    extra_hosts:",
            "      - \"host.docker.internal:host-gateway\"",
            "    networks:",
            f"      - {network}",
            "",
        ]
        for key, meta in MODEL_EXPLANATIONS.items():
            if not self.model_enabled[key].get():
                continue
            _title, _desc, container, _default_port, _default_model = meta
            host_port = self.model_ports[key].get().strip() or _default_port
            model_path = self.model_paths[key].get().strip() or str(Path(model_dir) / Path(_default_model).name)
            services.extend(self.tts_service_yaml(key, container, image, host_port, model_path, network))

        services.extend([
            "volumes:",
            "  tts-voice-creator-clone-and-design:",
            "",
            "networks:",
            f"  {network}:",
            "    external: true",
            "",
        ])
        return "\n".join(services)

    def tts_service_yaml(self, key: str, container: str, image: str, host_port: str, model_path: str, network: str) -> list[str]:
        lines = [
            f"  {container}:",
            f"    image: {image}",
            f"    container_name: {container}",
            "    restart: unless-stopped",
            "    ports:",
            f"      - \"{host_port}:8000\"",
            "    volumes:",
            f"      - {model_path}:/models/{Path(model_path).name}:ro",
            f"      - {ROOT / 'support' / 'faster-qwen3-tts-config'}:/config:ro",
            "    environment:",
            "      - PYTHONUNBUFFERED=1",
            "    networks:",
            f"      - {network}",
        ]
        if key == "design":
            lines.extend([
                "    command: python /config/run_voicedesign_server.py --host 0.0.0.0 --port 8000 --model /models/" + Path(model_path).name,
            ])
        elif key == "custom":
            lines.extend([
                "    command: python /config/run_customvoice_server.py --host 0.0.0.0 --port 8000 --model /models/" + Path(model_path).name,
            ])
        lines.append("")
        return lines

    def refresh_openwebui_text(self) -> None:
        creator = f"http://YOUR_HOST:{self.creator_port.get() or '7860'}/v1"
        text = f"""Open WebUI setup

Without routing:
1. Open Open WebUI Admin Settings -> Audio.
2. Choose OpenAI-compatible TTS.
3. Base URL: http://YOUR_HOST:{self.model_ports['clone'].get() or '8020'}/v1
4. API key: dummy
5. Model: qwen3-tts or tts-1
6. Voice: one real active voice, such as EN_F_Anna.

With routing:
1. Open Open WebUI Admin Settings -> Audio.
2. Choose OpenAI-compatible TTS.
3. Base URL: {creator}
4. API key: dummy
5. Model: tts-1
6. Voice: default
7. In TTS Voice Creator -> Routing, map:
   Open WebUI + default + EN -> EN_F_YourEnglishVoice
   Open WebUI + default + DE -> DE_M_YourGermanVoice

What to tell users:
- Voice Clone is the normal production path for recorded voices.
- Voice Design is for creating voices from descriptions.
- Custom Voice is for named Qwen speakers/custom voice model workflows.
- Portainer is optional. Use it when you want a browser dashboard for containers.
"""
        self.openwebui_text.delete("1.0", "end")
        self.openwebui_text.insert("end", text)


if __name__ == "__main__":
    GuidedInstaller().mainloop()