| r""" |
| NeuTTS Gradio Studio - single-file Windows-friendly UI |
| ====================================================== |
| |
| Start: |
| py -3.11 -m venv .venv |
| .\.venv\Scripts\Activate.ps1 |
| pip install --upgrade pip "setuptools<81" wheel |
| pip install -r requirements_neutts_clone_v3.txt |
| |
| Optional für GGUF-Modelle: |
| pip install llama-cpp-python --prefer-binary --extra-index-url https://abetlen.github.io/llama-cpp-python/whl/cpu |
| |
| Optional für ONNX-Decoder: |
| pip install "neutts[onnx]" |
| |
| Windows/eSpeak-ng, falls nötig: |
| setx PHONEMIZER_ESPEAK_LIBRARY "C:\Program Files\eSpeak NG\libespeak-ng.dll" |
| setx PHONEMIZER_ESPEAK_PATH "C:\Program Files\eSpeak NG" |
| |
| Ordnerstruktur neben dieser Datei: |
| voices/<Stimmenname>/reference.wav |
| voices/<Stimmenname>/reference.txt |
| outputs/*.wav |
| outputs/history.jsonl |
| """ |
|
|
| from __future__ import annotations |
|
|
| import asyncio |
| import json |
| import logging |
| import os |
| import re |
| import shutil |
| import subprocess |
| import sys |
| import time |
| import traceback |
| from datetime import datetime |
| from enum import Enum |
| from json import JSONDecodeError |
| from pathlib import Path |
| from typing import Any, Dict, List, Tuple |
|
|
| |
| os.environ.setdefault("GRADIO_ANALYTICS_ENABLED", "False") |
| os.environ.setdefault("HF_HUB_DISABLE_SYMLINKS_WARNING", "1") |
| |
| os.environ.setdefault("GRADIO_SSR_MODE", "False") |
|
|
|
|
| def apply_windows_asyncio_policy() -> None: |
| """Nutzt unter Windows den Selector-Loop, bevor Gradio/Uvicorn importiert werden.""" |
| if sys.platform.startswith("win") and hasattr(asyncio, "WindowsSelectorEventLoopPolicy"): |
| asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) |
|
|
|
|
| def install_windows_connection_reset_guard() -> None: |
| """ |
| Fängt nur den bekannten Windows-Proactor-Race-Condition-Fall ab. |
| |
| Hintergrund: Manche lokale Browser-/WebSocket-Verbindungen werden vom Client |
| geschlossen, während Python den Socket im Proactor-Transport ebenfalls schließen |
| will. Das ist bei Gradio/Uvicorn unter Windows nervig, aber nicht der TTS-Fehler. |
| Andere ConnectionResetError-Fälle werden nicht geschluckt. |
| """ |
| if not sys.platform.startswith("win"): |
| return |
| try: |
| import asyncio.proactor_events as proactor_events |
| except ImportError: |
| return |
|
|
| transport_cls = getattr(proactor_events, "_ProactorBasePipeTransport", None) |
| if transport_cls is None: |
| return |
|
|
| original = getattr(transport_cls, "_call_connection_lost", None) |
| if original is None or getattr(original, "_neutts_guarded", False): |
| return |
|
|
| def guarded_call_connection_lost(self, exc): |
| try: |
| return original(self, exc) |
| except ConnectionResetError as err: |
| if getattr(err, "winerror", None) == 10054: |
| return None |
| raise |
|
|
| guarded_call_connection_lost._neutts_guarded = True |
| transport_cls._call_connection_lost = guarded_call_connection_lost |
|
|
|
|
| def install_unix_invalid_fd_cleanup_guard() -> None: |
| """ |
| Linux/Hugging-Face-Guard gegen bekannten asyncio-Cleanup-Noise: |
| ValueError: Invalid file descriptor: -1 |
| |
| Wichtig: Es wird nicht pauschal alles geschluckt. Wir fangen gezielt den |
| Invalid-file-descriptor-Fall im Selector-Loop-Cleanup ab. Zusätzlich gibt |
| es dieselbe frühe Absicherung in sitecustomize.py, die vor app.py geladen wird. |
| """ |
| if sys.platform.startswith("win"): |
| return |
|
|
| try: |
| import asyncio.selector_events as selector_events |
|
|
| selector_loop_cls = getattr(selector_events, "BaseSelectorEventLoop", None) |
| original_remove_reader = getattr(selector_loop_cls, "_remove_reader", None) if selector_loop_cls else None |
|
|
| if original_remove_reader is not None and not getattr(original_remove_reader, "_neutts_invalid_fd_guarded", False): |
|
|
| def guarded_remove_reader(self, fd, *args, **kwargs): |
| try: |
| return original_remove_reader(self, fd, *args, **kwargs) |
| except ValueError as exc: |
| if "Invalid file descriptor" in str(exc): |
| return False |
| raise |
|
|
| guarded_remove_reader._neutts_invalid_fd_guarded = True |
| selector_loop_cls._remove_reader = guarded_remove_reader |
| except (ImportError, AttributeError, TypeError): |
| pass |
|
|
| original_del = getattr(asyncio.BaseEventLoop, "__del__", None) |
| if original_del is None or getattr(original_del, "_neutts_invalid_fd_guarded", False): |
| return |
|
|
| def guarded_event_loop_del(self): |
| try: |
| return original_del(self) |
| except ValueError as exc: |
| if "Invalid file descriptor" in str(exc): |
| return None |
| raise |
|
|
| guarded_event_loop_del._neutts_invalid_fd_guarded = True |
| asyncio.BaseEventLoop.__del__ = guarded_event_loop_del |
|
|
|
|
| def configure_quiet_runtime_logging() -> None: |
| """Reduziert nur bekannte Third-Party-Statuslogs, nicht unsere Fehlerausgaben.""" |
| logging.getLogger("torch").setLevel(logging.ERROR) |
| logging.getLogger("torch.utils._pytree").setLevel(logging.ERROR) |
| logging.getLogger("huggingface_hub.file_download").setLevel(logging.ERROR) |
| logging.getLogger("uvicorn.error").setLevel(logging.WARNING) |
| logging.getLogger("uvicorn.access").setLevel(logging.WARNING) |
|
|
|
|
| apply_windows_asyncio_policy() |
| install_windows_connection_reset_guard() |
| configure_quiet_runtime_logging() |
|
|
| import gradio as gr |
| import numpy as np |
| import pandas as pd |
| import soundfile as sf |
|
|
| try: |
| import librosa |
| except ImportError: |
| librosa = None |
|
|
| import audio_fx |
| import voice_persistence |
|
|
| APP_DIR = Path(__file__).resolve().parent |
| VOICES_DIR = APP_DIR / "voices" |
| OUTPUTS_DIR = APP_DIR / "outputs" |
| MEDIA_WORK_DIR = OUTPUTS_DIR / "prepared_media" |
| HISTORY_FILE = OUTPUTS_DIR / "history.jsonl" |
| SAMPLE_RATE = 24_000 |
| HISTORY_COLUMNS = ["Zeit", "Modell", "Stimme", "Zeichen", "Dauer", "Audio", "RTF", "Output"] |
| APP_HANDLED_ERRORS = (OSError, RuntimeError, ValueError, TypeError, ImportError, AttributeError) |
|
|
| AUDIO_EXTENSIONS = {".wav", ".mp3", ".m4a", ".flac", ".ogg", ".aac", ".wma"} |
| VIDEO_EXTENSIONS = {".mp4", ".avi", ".mov", ".mkv", ".webm", ".m4v", ".mpeg", ".mpg"} |
|
|
| VOICES_DIR.mkdir(parents=True, exist_ok=True) |
| OUTPUTS_DIR.mkdir(parents=True, exist_ok=True) |
| MEDIA_WORK_DIR.mkdir(parents=True, exist_ok=True) |
|
|
| |
| MODEL_PRESETS: Dict[str, str] = { |
| "Nano German · klein/schnell/deutsch": "neuphonic/neutts-nano-german", |
| "Nano German Q4 GGUF · deutsch/CPU/llama-cpp": "neuphonic/neutts-nano-german-q4-gguf", |
| "Nano Multilingual · klein/mehrsprachig": "neuphonic/neutts-nano", |
| "Nano Multilingual Q4 GGUF · CPU/llama-cpp": "neuphonic/neutts-nano-q4-gguf", |
| "Air · gross/bessere Qualität": "neuphonic/neutts-air", |
| "Air Q4 GGUF · gross/CPU/llama-cpp": "neuphonic/neutts-air-q4-gguf", |
| "Air Q8 GGUF · gross/besser/mehr RAM": "neuphonic/neutts-air-q8-gguf", |
| } |
|
|
| CODEC_PRESETS: Dict[str, str] = { |
| "NeuCodec · Standard": "neuphonic/neucodec", |
| "NeuCodec ONNX Decoder · optional/schneller": "neuphonic/neucodec-onnx-decoder", |
| } |
|
|
| WHISPER_MODEL_PRESETS: Dict[str, str] = { |
| "Tiny · schnell/ungenauer": "tiny", |
| "Base · guter Start/CPU": "base", |
| "Small · besser/langsamer": "small", |
| "Medium · deutlich besser/sehr langsam CPU": "medium", |
| } |
|
|
| |
| _MODEL_CACHE: Dict[Tuple[str, str, str, str], Any] = {} |
| _REF_CACHE: Dict[Tuple[str, int, str, str], Any] = {} |
| _WHISPER_CACHE: Dict[Tuple[str, str, str], Any] = {} |
| _REFERENCE_WAV_BY_NAME: Dict[str, str] = {} |
|
|
|
|
| def patch_torch_pytree_enum_register_constant() -> str: |
| """ |
| Verhindert die kommende PyTorch-Deprecation sauber an der Ursache. |
| |
| Einige Dependencies rufen torch.utils._pytree.register_constant() noch für Enum-Klassen auf. |
| PyTorch unterstützt Enum-Klassen inzwischen nativ. Deshalb wird der alte Call für |
| Enum-Subclasses zu einem No-op gemacht; alle anderen register_constant-Aufrufe |
| laufen unverändert weiter. |
| """ |
| try: |
| from torch.utils import _pytree |
| except ImportError as exc: |
| return f"⚠️ PyTorch pytree konnte nicht geladen werden: {exc}" |
|
|
| original = getattr(_pytree, "register_constant", None) |
| if original is None: |
| return "ℹ️ PyTorch register_constant nicht vorhanden; kein Patch nötig." |
| if getattr(original, "_neutts_enum_guarded", False): |
| return "✅ PyTorch pytree Enum-Guard war bereits aktiv." |
|
|
| def safe_register_constant(*args, **kwargs): |
| target = args[0] if args else kwargs.get("cls") |
| if isinstance(target, type) and issubclass(target, Enum): |
| return None |
| return original(*args, **kwargs) |
|
|
| safe_register_constant._neutts_enum_guarded = True |
| _pytree.register_constant = safe_register_constant |
| return "✅ PyTorch pytree Enum-Guard aktiv." |
|
|
|
|
| def patch_torch_weight_norm() -> str: |
| """ |
| Nutzt die neue PyTorch-Parametrization-API, bevor NeuTTS/NeuCodec geladen wird. |
| """ |
| try: |
| import torch |
| from torch.nn.utils import parametrizations |
|
|
| torch.nn.utils.weight_norm = parametrizations.weight_norm |
| return "✅ PyTorch weight_norm auf parametrizations.weight_norm gemappt." |
| except (ImportError, AttributeError, TypeError) as exc: |
| return f"⚠️ PyTorch weight_norm konnte nicht gepatcht werden: {exc}" |
|
|
|
|
| TORCH_PATCH_STATUS = "\n".join([ |
| patch_torch_pytree_enum_register_constant(), |
| patch_torch_weight_norm(), |
| ]) |
|
|
|
|
| def safe_name(value: str | None, fallback: str = "voice") -> str: |
| value = (value or "").strip() |
| value = re.sub(r"[^a-zA-Z0-9_.\- äöüÄÖÜß]+", "_", value) |
| value = value.strip(" ._-") |
| return value or fallback |
|
|
|
|
| def timestamp() -> str: |
| return datetime.now().strftime("%Y%m%d_%H%M%S") |
|
|
|
|
| def format_seconds(seconds: float) -> str: |
| if seconds < 60: |
| return f"{seconds:.2f} s" |
| minutes = int(seconds // 60) |
| rest = seconds - minutes * 60 |
| return f"{minutes} min {rest:.1f} s" |
|
|
|
|
| def read_text(path: Path) -> str: |
| return path.read_text(encoding="utf-8").strip() |
|
|
|
|
| def write_jsonl(row: dict[str, Any]) -> None: |
| with HISTORY_FILE.open("a", encoding="utf-8") as f: |
| f.write(json.dumps(row, ensure_ascii=False) + "\n") |
|
|
|
|
| def load_history(limit: int = 30) -> pd.DataFrame: |
| if not HISTORY_FILE.exists(): |
| return pd.DataFrame(columns=HISTORY_COLUMNS) |
|
|
| rows: List[List[str]] = [] |
| lines = HISTORY_FILE.read_text(encoding="utf-8").splitlines()[-limit:] |
|
|
| for line in reversed(lines): |
| try: |
| item = json.loads(line) |
| except JSONDecodeError: |
| continue |
|
|
| rows.append( |
| [ |
| str(item.get("time", "")), |
| str(item.get("model", "")), |
| str(item.get("voice", "")), |
| str(item.get("chars", "")), |
| str(item.get("elapsed", "")), |
| str(item.get("audio_duration", "")), |
| str(item.get("rtf", "")), |
| str(item.get("output", "")), |
| ] |
| ) |
|
|
| return pd.DataFrame(rows, columns=HISTORY_COLUMNS) |
|
|
|
|
| def build_voice_index() -> Dict[str, Tuple[Path, Path]]: |
| """Findet gespeicherte Stimmen und optionale Sample-Stimmen aus dem Repo-Ordner.""" |
| voices: Dict[str, Tuple[Path, Path]] = {} |
|
|
| sample_candidates = [ |
| ("Sample · Greta Deutsch", APP_DIR / "samples" / "greta.wav", APP_DIR / "samples" / "greta.txt"), |
| ("Sample · Jo Englisch", APP_DIR / "samples" / "jo.wav", APP_DIR / "samples" / "jo.txt"), |
| ("Sample · Dave Englisch", APP_DIR / "samples" / "dave.wav", APP_DIR / "samples" / "dave.txt"), |
| ("Sample · Mateo Spanisch", APP_DIR / "samples" / "mateo.wav", APP_DIR / "samples" / "mateo.txt"), |
| ("Sample · Juliette Französisch", APP_DIR / "samples" / "juliette.wav", APP_DIR / "samples" / "juliette.txt"), |
| ] |
| for label, wav_path, txt_path in sample_candidates: |
| if wav_path.exists() and txt_path.exists(): |
| voices[label] = (wav_path, txt_path) |
|
|
| for voice_dir in sorted(VOICES_DIR.iterdir() if VOICES_DIR.exists() else []): |
| if not voice_dir.is_dir(): |
| continue |
| wav_path = voice_dir / "reference.wav" |
| txt_path = voice_dir / "reference.txt" |
| if wav_path.exists() and txt_path.exists(): |
| voices[f"{voice_dir.name}"] = (wav_path, txt_path) |
|
|
| return voices |
|
|
|
|
| def voice_choices() -> List[str]: |
| return list(build_voice_index().keys()) |
|
|
|
|
| def refresh_voices(): |
| choices = voice_choices() |
| value = choices[0] if choices else None |
| msg = "✅ Stimmenliste aktualisiert." if choices else "⚠️ Noch keine Stimme gespeichert. Lege rechts eine Referenzstimme an." |
| return gr.update(choices=choices, value=value), msg |
|
|
|
|
| def normalize_or_copy_audio(src: str | Path, dst: Path, normalize: bool) -> None: |
| src = Path(src) |
| dst.parent.mkdir(parents=True, exist_ok=True) |
|
|
| if normalize: |
| if librosa is None: |
| raise RuntimeError("librosa ist nicht installiert. Installiere es mit: pip install librosa") |
| audio, _sr = librosa.load(str(src), sr=SAMPLE_RATE, mono=True) |
| peak = float(np.max(np.abs(audio))) if audio.size else 0.0 |
| if peak > 0: |
| audio = audio / max(peak, 1e-9) * 0.92 |
| sf.write(str(dst), audio, SAMPLE_RATE) |
| else: |
| |
| shutil.copyfile(src, dst) |
|
|
|
|
| def get_audio_duration(path: str | Path) -> float: |
| try: |
| info = sf.info(str(path)) |
| if info.samplerate > 0: |
| return float(info.frames) / float(info.samplerate) |
| except (OSError, RuntimeError, ValueError): |
| pass |
|
|
| if librosa is not None: |
| try: |
| return float(librosa.get_duration(path=str(path))) |
| except (OSError, RuntimeError, ValueError): |
| return 0.0 |
|
|
| return 0.0 |
|
|
|
|
| def resolve_ffmpeg_binary() -> str: |
| """Findet zuerst das projektlokale ffmpeg, danach ein global installiertes ffmpeg.""" |
| candidates = [] |
|
|
| if sys.platform.startswith("win"): |
| candidates.append(APP_DIR / "tools" / "ffmpeg" / "bin" / "ffmpeg.exe") |
| else: |
| candidates.append(APP_DIR / "tools" / "ffmpeg" / "bin" / "ffmpeg") |
|
|
| candidates.append(APP_DIR / "tools" / "ffmpeg" / "bin" / "ffmpeg.exe") |
| candidates.append(APP_DIR / "tools" / "ffmpeg" / "bin" / "ffmpeg") |
|
|
| for candidate in candidates: |
| if candidate.exists(): |
| return str(candidate) |
|
|
| system_ffmpeg = shutil.which("ffmpeg") |
| if system_ffmpeg: |
| return system_ffmpeg |
|
|
| raise RuntimeError( |
| "ffmpeg wurde nicht gefunden. Lege es lokal unter " |
| "`tools/ffmpeg/bin/ffmpeg.exe` ab oder installiere ffmpeg im PATH." |
| ) |
|
|
|
|
| def resolve_existing_media_path(path_value: str | Path | None) -> Path | None: |
| """ |
| Gradio verliert beim Audio-Editor manchmal den Ordner und gibt nur den |
| Dateinamen zurück, z. B. `20260602_203047_reference_audio.wav`. |
| Diese Funktion rekonstruiert den echten Pfad aus unseren Arbeitsordnern. |
| """ |
| if path_value is None: |
| return None |
|
|
| raw_path = Path(str(path_value)) |
|
|
| if raw_path.exists(): |
| return raw_path.resolve() |
|
|
| mapped_path = _REFERENCE_WAV_BY_NAME.get(raw_path.name) |
| if mapped_path and Path(mapped_path).exists(): |
| return Path(mapped_path).resolve() |
|
|
| search_dirs = [ |
| MEDIA_WORK_DIR, |
| OUTPUTS_DIR, |
| APP_DIR, |
| Path.cwd(), |
| ] |
|
|
| for base_dir in search_dirs: |
| candidate = base_dir / raw_path.name |
| if candidate.exists(): |
| return candidate.resolve() |
|
|
| |
| |
| if MEDIA_WORK_DIR.exists(): |
| matches = list(MEDIA_WORK_DIR.rglob(raw_path.name)) |
| if matches: |
| return matches[-1].resolve() |
|
|
| return raw_path |
|
|
|
|
| def uploaded_value_to_path(value: Any) -> Path | None: |
| """ |
| Macht Gradio-Dateiwerte robust: |
| - gr.File(type="filepath") kann als str kommen |
| - gr.Audio(editable=True) kann nach dem Schneiden nur einen Dateinamen liefern |
| - je nach Gradio-Version auch als Objekt/Dict mit path/name |
| - bei Listen wird der erste Eintrag genommen |
| """ |
| if value is None: |
| return None |
|
|
| if isinstance(value, (list, tuple)) and value: |
| |
| if not (len(value) == 2 and isinstance(value[0], int)): |
| return uploaded_value_to_path(value[0]) |
|
|
| if isinstance(value, dict): |
| for key in ("path", "name", "orig_name"): |
| candidate = value.get(key) |
| if candidate: |
| return resolve_existing_media_path(candidate) |
| return None |
|
|
| for attr in ("path", "name", "orig_name"): |
| candidate = getattr(value, attr, None) |
| if candidate: |
| return resolve_existing_media_path(candidate) |
|
|
| if isinstance(value, (str, Path)): |
| return resolve_existing_media_path(value) |
|
|
| return None |
|
|
|
|
| def is_video_file(path: str | Path | None) -> bool: |
| if not path: |
| return False |
| return Path(path).suffix.lower() in VIDEO_EXTENSIONS |
|
|
|
|
| def is_audio_file(path: str | Path | None) -> bool: |
| if not path: |
| return False |
| return Path(path).suffix.lower() in AUDIO_EXTENSIONS |
|
|
|
|
| def write_audio_tuple_as_reference_wav(audio_value: Any, reason: str = "edited_reference") -> Path: |
| """ |
| Falls Gradio nach dem Schneiden ein Audio-Tuple liefert, wird es sofort als mono/24kHz WAV materialisiert. |
| """ |
| if not isinstance(audio_value, tuple) or len(audio_value) != 2: |
| raise TypeError("Audio-Wert ist kein gültiges Gradio-Audio-Tuple.") |
|
|
| sr, audio = audio_value |
| arr = np.asarray(audio, dtype=np.float32) |
|
|
| if arr.ndim > 1: |
| arr = np.mean(arr, axis=1) |
|
|
| |
| if arr.size: |
| max_abs = float(np.max(np.abs(arr))) |
| if max_abs > 1.5: |
| arr = arr / 32768.0 |
|
|
| if int(sr) != SAMPLE_RATE: |
| if librosa is None: |
| raise RuntimeError("librosa fehlt zum Resampling der bearbeiteten Audiodatei.") |
| arr = librosa.resample(arr, orig_sr=int(sr), target_sr=SAMPLE_RATE) |
|
|
| peak = float(np.max(np.abs(arr))) if arr.size else 0.0 |
| if peak > 0: |
| arr = arr / max(peak, 1e-9) * 0.92 |
|
|
| out_path = MEDIA_WORK_DIR / f"{timestamp()}_{safe_name(reason, 'edited_reference')}.wav" |
| sf.write(str(out_path), arr, SAMPLE_RATE, subtype="PCM_16") |
| _REFERENCE_WAV_BY_NAME[out_path.name] = str(out_path.resolve()) |
| return out_path |
|
|
|
|
| def convert_media_to_reference_wav(src: Any, reason: str = "reference") -> Path: |
| """ |
| Extrahiert/konvertiert Audio oder Video sofort zu mono/24kHz WAV. |
| Wichtig: Kein Format wird nur anhand der Dateiendung verworfen, weil Gradio-Tempdateien |
| je nach Version/Browser manchmal andere Namen bekommen. ffmpeg darf selbst prüfen. |
| """ |
| |
| if isinstance(src, tuple) and len(src) == 2: |
| return write_audio_tuple_as_reference_wav(src, reason) |
|
|
| src_path = uploaded_value_to_path(src) |
| if src_path is None: |
| raise ValueError("Kein gültiger Medienpfad von Gradio erhalten.") |
| if not src_path.exists(): |
| raise FileNotFoundError(f"Datei nicht gefunden: {src_path}") |
|
|
| ffmpeg_bin = resolve_ffmpeg_binary() |
| out_path = MEDIA_WORK_DIR / f"{timestamp()}_{safe_name(reason, 'reference')}_{safe_name(src_path.stem, 'media')}.wav" |
|
|
| cmd = [ |
| ffmpeg_bin, |
| "-y", |
| "-hide_banner", |
| "-loglevel", |
| "error", |
| "-i", |
| str(src_path), |
| "-map", |
| "0:a:0", |
| "-vn", |
| "-ac", |
| "1", |
| "-ar", |
| str(SAMPLE_RATE), |
| "-sample_fmt", |
| "s16", |
| str(out_path), |
| ] |
|
|
| creationflags = 0 |
| if sys.platform.startswith("win"): |
| creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0) |
|
|
| result = subprocess.run( |
| cmd, |
| check=False, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE, |
| text=True, |
| encoding="utf-8", |
| errors="replace", |
| creationflags=creationflags, |
| ) |
|
|
| if result.returncode != 0: |
| tail = (result.stderr or result.stdout or "").strip()[-3000:] |
| raise RuntimeError( |
| "ffmpeg-Konvertierung fehlgeschlagen. " |
| "Prüfe, ob die Datei eine Audiospur enthält und ob ffmpeg.exe unter tools/ffmpeg/bin liegt.\n\n" |
| f"{tail}" |
| ) |
|
|
| if not out_path.exists() or out_path.stat().st_size == 0: |
| raise RuntimeError("ffmpeg hat keine gültige WAV-Datei erzeugt.") |
|
|
| _REFERENCE_WAV_BY_NAME[out_path.name] = str(out_path.resolve()) |
| return out_path |
|
|
|
|
| def prepare_reference_media(reference_media: Any): |
| """Ein Eingang für Audio oder Video: sofort zu mono/24kHz WAV vorbereiten und im Audio-Editor anzeigen.""" |
| try: |
| src_path = uploaded_value_to_path(reference_media) |
| if src_path is None: |
| return None, "⚠️ Noch keine Audio- oder Video-Datei ausgewählt." |
|
|
| wav_path = convert_media_to_reference_wav(reference_media, "reference") |
| duration = get_audio_duration(wav_path) |
|
|
| if is_video_file(src_path): |
| source_type = "Video" |
| elif is_audio_file(src_path): |
| source_type = "Audio" |
| else: |
| source_type = "Medium" |
|
|
| return ( |
| str(wav_path), |
| f"✅ {source_type} wurde automatisch zu mono/24kHz WAV vorbereitet.\n\n" |
| f"**Quelle:** `{src_path.name}`\n\n" |
| f"**Jetzt im Audiofeld schneiden/bearbeiten und danach transkribieren oder speichern.**\n\n" |
| f"**Dauer:** `{format_seconds(duration)}`", |
| ) |
| except APP_HANDLED_ERRORS as exc: |
| tb = traceback.format_exc() |
| return ( |
| None, |
| f"❌ Medienvorbereitung fehlgeschlagen:\n\n```text\n{exc}\n```\n\n" |
| f"<details><summary>Traceback</summary>\n\n```text\n{tb}\n```\n\n</details>", |
| ) |
|
|
|
|
| def pick_reference_media( |
| prepared_audio: Any, |
| reference_media: Any, |
| ) -> str | None: |
| """ |
| Nimmt die bearbeitete/vorbereitete WAV aus dem Audiofeld. |
| Wichtig: Gradio kann nach dem Audio-Editor nur den Dateinamen liefern. |
| Deshalb wird der Pfad zuerst über resolve_existing_media_path rekonstruiert. |
| """ |
| if isinstance(prepared_audio, tuple) and len(prepared_audio) == 2: |
| return str(write_audio_tuple_as_reference_wav(prepared_audio, "edited_reference")) |
|
|
| prepared_path = uploaded_value_to_path(prepared_audio) |
| if prepared_path is not None and prepared_path.exists(): |
| |
| |
| |
| if prepared_path.suffix.lower() == ".wav": |
| return str(prepared_path.resolve()) |
| return str(convert_media_to_reference_wav(prepared_path, "edited_reference")) |
|
|
| |
| converted, _status = prepare_reference_media(reference_media) |
| if converted: |
| converted_path = uploaded_value_to_path(converted) |
| if converted_path is not None and converted_path.exists(): |
| return str(converted_path.resolve()) |
| return converted |
|
|
| return None |
|
|
|
|
| def save_voice_style_metadata( |
| target_dir: Path, |
| emotion_label: str | None, |
| preserve_laughter: bool, |
| non_speech_notes: str | None, |
| ) -> None: |
| style_data = { |
| "emotion_label": (emotion_label or "").strip(), |
| "preserve_laughter": bool(preserve_laughter), |
| "non_speech_notes": (non_speech_notes or "").strip(), |
| "updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), |
| } |
| (target_dir / "style.json").write_text(json.dumps(style_data, ensure_ascii=False, indent=2), encoding="utf-8") |
|
|
|
|
| def apply_tts_event_markers(text: str, marker_mode: str | None) -> str: |
| """ |
| Ein pragmatischer Marker-Fallback. |
| NeuTTS versteht keine garantierten [lacht]-Tokens. Damit es trotzdem im TTS |
| wahrnehmbar wird, können Marker optional in sprechbare Lautäußerungen gewandelt werden. |
| """ |
| mode = (marker_mode or "Marker unverändert lassen").strip() |
| if mode == "Marker unverändert lassen": |
| return text |
|
|
| replacements = { |
| r"\[(lacht|lachen|laughs|laugh)\]": " haha ", |
| r"\[(lacht kurz|kurzes lachen)\]": " haha ", |
| r"\[(lacht herzlich|herzliches lachen|lacht laut)\]": " hahahaha ", |
| r"\[(kichert|giggles)\]": " hihi ", |
| r"\[(seufzt|sighs)\]": " hach ", |
| r"\[(atmet|breathes|atmen)\]": " ", |
| r"\((lacht|lachen|laughs|laugh)\)": " haha ", |
| r"\((kichert|giggles)\)": " hihi ", |
| r"\((seufzt|sighs)\)": " hach ", |
| } |
|
|
| converted = text |
| for pattern, repl in replacements.items(): |
| converted = re.sub(pattern, repl, converted, flags=re.IGNORECASE) |
|
|
| converted = re.sub(r"\s+", " ", converted).strip() |
| return converted |
|
|
|
|
| def save_voice( |
| voice_name: str | None, |
| prepared_audio: str | None, |
| reference_media: str | None, |
| ref_text: str | None, |
| emotion_label: str | None, |
| preserve_laughter: bool, |
| non_speech_notes: str | None, |
| ): |
| if not voice_name or not voice_name.strip(): |
| return gr.update(choices=voice_choices()), "❌ Bitte einen Namen für die Stimme eingeben.", None |
|
|
| source_audio = pick_reference_media(prepared_audio, reference_media) |
| if not source_audio: |
| return gr.update(choices=voice_choices()), "❌ Bitte eine Audio- oder Video-Datei hochladen.", None |
|
|
| if not ref_text or not ref_text.strip(): |
| return gr.update(choices=voice_choices()), "❌ Bitte den exakt gesprochenen Referenztext eingeben oder vorher transkribieren.", None |
|
|
| name = safe_name(voice_name, "voice") |
| target_dir = VOICES_DIR / name |
| target_dir.mkdir(parents=True, exist_ok=True) |
| wav_path = target_dir / "reference.wav" |
| txt_path = target_dir / "reference.txt" |
|
|
| try: |
| normalize_or_copy_audio(source_audio, wav_path, True) |
| txt_path.write_text(ref_text.strip(), encoding="utf-8") |
| save_voice_style_metadata(target_dir, emotion_label, preserve_laughter, non_speech_notes) |
|
|
| choices = voice_choices() |
| style_line = "" |
| if emotion_label or preserve_laughter or non_speech_notes: |
| style_line = "\n\nStil-Metadaten: `style.json` gespeichert." |
|
|
| |
| sync_note = voice_persistence.push_voice_to_dataset(VOICES_DIR, name) |
|
|
| return ( |
| gr.update(choices=choices, value=name), |
| f"✅ Stimme gespeichert: `{target_dir}`\n\nReferenz: `{wav_path.name}` + `{txt_path.name}`{style_line}{sync_note}", |
| str(wav_path), |
| ) |
| except APP_HANDLED_ERRORS as exc: |
| return gr.update(choices=voice_choices()), f"❌ Stimme konnte nicht gespeichert werden:\n\n```text\n{exc}\n```", None |
|
|
|
|
| def resolve_model(model_label: str | None, custom_model: str | None) -> str: |
| custom_model = (custom_model or "").strip() |
| if custom_model: |
| return custom_model |
| if model_label and model_label in MODEL_PRESETS: |
| return MODEL_PRESETS[model_label] |
| return MODEL_PRESETS["Nano German · klein/schnell/deutsch"] |
|
|
|
|
| def resolve_codec(codec_label: str | None) -> str: |
| if codec_label and codec_label in CODEC_PRESETS: |
| return CODEC_PRESETS[codec_label] |
| return CODEC_PRESETS["NeuCodec · Standard"] |
|
|
|
|
| def resolve_whisper_model(model_label: str | None) -> str: |
| if model_label and model_label in WHISPER_MODEL_PRESETS: |
| return WHISPER_MODEL_PRESETS[model_label] |
| return WHISPER_MODEL_PRESETS["Base · guter Start/CPU"] |
|
|
|
|
| def get_tts(backbone_repo: str, backbone_device: str, codec_repo: str, codec_device: str): |
| key = (backbone_repo, backbone_device, codec_repo, codec_device) |
| if key in _MODEL_CACHE: |
| return _MODEL_CACHE[key] |
|
|
| from neutts import NeuTTS |
|
|
| tts = NeuTTS( |
| backbone_repo=backbone_repo, |
| backbone_device=backbone_device, |
| codec_repo=codec_repo, |
| codec_device=codec_device, |
| ) |
| _MODEL_CACHE[key] = tts |
| return tts |
|
|
|
|
| def get_ref_codes(tts: Any, ref_audio_path: Path, codec_repo: str, codec_device: str): |
| stat = ref_audio_path.stat() |
| key = (str(ref_audio_path.resolve()), stat.st_mtime_ns, codec_repo, codec_device) |
| if key in _REF_CACHE: |
| return _REF_CACHE[key] |
| ref_codes = tts.encode_reference(str(ref_audio_path)) |
| _REF_CACHE[key] = ref_codes |
| return ref_codes |
|
|
|
|
| def get_whisper(model_size: str, device: str, compute_type: str): |
| key = (model_size, device, compute_type) |
| if key in _WHISPER_CACHE: |
| return _WHISPER_CACHE[key] |
|
|
| from faster_whisper import WhisperModel |
|
|
| model = WhisperModel(model_size, device=device, compute_type=compute_type) |
| _WHISPER_CACHE[key] = model |
| return model |
|
|
|
|
| def split_text(text: str | None, max_chars: int) -> List[str]: |
| text = " ".join((text or "").strip().split()) |
| if not text: |
| return [] |
| if len(text) <= max_chars: |
| return [text] |
|
|
| |
| sentences = re.split(r"(?<=[.!?…])\s+", text) |
| chunks: List[str] = [] |
| current = "" |
|
|
| for sentence in sentences: |
| sentence = sentence.strip() |
| if not sentence: |
| continue |
| if len(sentence) > max_chars: |
| |
| words = sentence.split() |
| for word in words: |
| if len(current) + len(word) + 1 > max_chars and current: |
| chunks.append(current.strip()) |
| current = word |
| else: |
| current = f"{current} {word}".strip() |
| continue |
| if len(current) + len(sentence) + 1 <= max_chars: |
| current = f"{current} {sentence}".strip() |
| else: |
| if current: |
| chunks.append(current.strip()) |
| current = sentence |
|
|
| if current: |
| chunks.append(current.strip()) |
| return chunks |
|
|
|
|
| def ensure_mono_float32(wav: Any) -> np.ndarray: |
| arr = np.asarray(wav, dtype=np.float32) |
| arr = np.squeeze(arr) |
| if arr.ndim > 1: |
| arr = np.mean(arr, axis=1).astype(np.float32) |
| return arr |
|
|
|
|
| def make_output_path(prefix: str | None, voice: str | None, model_repo: str) -> Path: |
| clean_prefix = safe_name(prefix or "neutts", "neutts") |
| clean_voice = safe_name(voice or "voice", "voice") |
| clean_model = safe_name(model_repo.split("/")[-1], "model") |
| return OUTPUTS_DIR / f"{timestamp()}_{clean_prefix}_{clean_voice}_{clean_model}.wav" |
|
|
|
|
| def transcribe_reference_audio( |
| prepared_audio: str | None, |
| reference_media: str | None, |
| whisper_model_label: str | None, |
| whisper_language: str | None, |
| whisper_device: str | None, |
| whisper_compute_type: str | None, |
| vad_filter_enabled: bool, |
| preserve_laughter_hint: bool, |
| emotion_label: str | None, |
| non_speech_notes: str | None, |
| ): |
| source_audio = pick_reference_media(prepared_audio, reference_media) |
| if not source_audio: |
| return "", "❌ Bitte zuerst eine Audio- oder Video-Datei hochladen." |
|
|
| model_size = resolve_whisper_model(whisper_model_label) |
| device = (whisper_device or "cpu").strip() or "cpu" |
| compute_type = (whisper_compute_type or "int8").strip() or "int8" |
| language = (whisper_language or "auto").strip() |
| language_arg = None if language == "auto" else language |
|
|
| started = time.perf_counter() |
| try: |
| model = get_whisper(model_size, device, compute_type) |
| segments, info = model.transcribe( |
| source_audio, |
| language=language_arg, |
| beam_size=5, |
| vad_filter=bool(vad_filter_enabled), |
| ) |
| text = " ".join(segment.text.strip() for segment in segments if segment.text.strip()).strip() |
| elapsed = time.perf_counter() - started |
| detected_lang = getattr(info, "language", "unbekannt") |
| probability = getattr(info, "language_probability", 0.0) |
|
|
| marker_notes = [] |
| if preserve_laughter_hint: |
| marker_notes.append("[lacht]") |
| if emotion_label and emotion_label.strip(): |
| marker_notes.append(f"[Stil: {emotion_label.strip()}]") |
| if non_speech_notes and non_speech_notes.strip(): |
| marker_notes.append(non_speech_notes.strip()) |
|
|
| |
| |
| if marker_notes: |
| text = f"{text}\n\n{' '.join(marker_notes)}".strip() |
|
|
| if not text: |
| return "", f"⚠️ Keine Sprache erkannt. Dauer: {format_seconds(elapsed)}" |
|
|
| laughter_msg = ( |
| "\n\n**Lach-/Emotionshinweis:** Marker wurden als editierbare Hinweise angefügt." |
| if marker_notes |
| else "\n\n**Lach-/Emotionshinweis:** Whisper erkennt Lachen nicht zuverlässig automatisch. Nutze bei Bedarf die Marker-Optionen." |
| ) |
|
|
| return ( |
| text, |
| f"✅ Transkription fertig.\n\n" |
| f"**Whisper-Modell:** `{model_size}`\n\n" |
| f"**Sprache:** `{detected_lang}` ({probability:.2%})\n\n" |
| f"**VAD-Filter:** `{'an' if vad_filter_enabled else 'aus'}`\n\n" |
| f"**Dauer:** `{format_seconds(elapsed)}`" |
| f"{laughter_msg}", |
| ) |
| except ImportError: |
| return "", "❌ faster-whisper fehlt. Installiere es mit: `pip install faster-whisper`" |
| except APP_HANDLED_ERRORS as exc: |
| tb = traceback.format_exc() |
| return "", f"❌ Transkription fehlgeschlagen:\n\n```text\n{exc}\n```\n\n<details><summary>Traceback</summary>\n\n```text\n{tb}\n```\n\n</details>" |
|
|
|
|
| def synthesize( |
| input_text: str | None, |
| model_label: str | None, |
| custom_model: str | None, |
| codec_label: str | None, |
| voice_label: str | None, |
| backbone_device: str | None, |
| codec_device: str | None, |
| split_enabled: bool, |
| max_chars: int | float, |
| pause_ms: int | float, |
| tail_padding_ms: int | float, |
| event_marker_mode: str | None, |
| output_prefix: str | None, |
| synthesis_speed: float | int | None = 1.0, |
| denoise_enabled: bool = False, |
| ): |
| if not input_text or not input_text.strip(): |
| return None, None, "❌ Bitte Text eingeben.", load_history() |
|
|
| voices = build_voice_index() |
| if not voices: |
| return None, None, "❌ Keine Stimme gefunden. Speichere zuerst eine Referenzstimme.", load_history() |
| if not voice_label or voice_label not in voices: |
| return None, None, "❌ Bitte eine vorhandene Stimme auswählen.", load_history() |
|
|
| model_repo = resolve_model(model_label, custom_model) |
| codec_repo = resolve_codec(codec_label) |
| backbone = (backbone_device or "cpu").strip() or "cpu" |
| codec = (codec_device or "cpu").strip() or "cpu" |
| ref_audio_path, ref_text_path = voices[voice_label] |
| synthesis_text = apply_tts_event_markers(input_text.strip(), event_marker_mode) |
|
|
| started = time.perf_counter() |
| try: |
| ref_text = read_text(ref_text_path) |
| tts = get_tts(model_repo, backbone, codec_repo, codec) |
| ref_codes = get_ref_codes(tts, ref_audio_path, codec_repo, codec) |
|
|
| if split_enabled: |
| chunks = split_text(synthesis_text, int(max_chars or 220)) |
| else: |
| chunks = [synthesis_text] |
|
|
| if not chunks: |
| return None, None, "❌ Nach dem Text-Splitting ist kein Text übrig geblieben.", load_history() |
|
|
| pause = np.zeros(int(SAMPLE_RATE * max(0, int(pause_ms or 0)) / 1000), dtype=np.float32) |
| parts: List[np.ndarray] = [] |
|
|
| for idx, chunk in enumerate(chunks, start=1): |
| wav = tts.infer(chunk, ref_codes, ref_text) |
| parts.append(ensure_mono_float32(wav)) |
| if idx < len(chunks) and len(pause) > 0: |
| parts.append(pause) |
|
|
| tail_padding = np.zeros( |
| int(SAMPLE_RATE * max(0, int(tail_padding_ms or 0)) / 1000), |
| dtype=np.float32, |
| ) |
| if len(tail_padding) > 0: |
| parts.append(tail_padding) |
|
|
| full_wav = np.concatenate(parts) if len(parts) > 1 else parts[0] |
|
|
| |
| full_wav = audio_fx.apply_speed(full_wav, synthesis_speed) |
| if denoise_enabled: |
| full_wav = audio_fx.apply_denoise(full_wav, SAMPLE_RATE) |
|
|
| out_path = make_output_path(output_prefix, voice_label, model_repo) |
| sf.write(str(out_path), full_wav, SAMPLE_RATE) |
|
|
| elapsed = time.perf_counter() - started |
| audio_duration = get_audio_duration(out_path) |
| rtf = elapsed / audio_duration if audio_duration > 0 else 0.0 |
|
|
| elapsed_text = format_seconds(elapsed) |
| audio_text = format_seconds(audio_duration) if audio_duration > 0 else "unbekannt" |
| rtf_text = f"{rtf:.2f}x" if rtf > 0 else "unbekannt" |
|
|
| write_jsonl( |
| { |
| "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), |
| "model": model_repo, |
| "codec": codec_repo, |
| "voice": voice_label, |
| "chars": len(synthesis_text), |
| "chunks": len(chunks), |
| "elapsed": elapsed_text, |
| "audio_duration": audio_text, |
| "rtf": rtf_text, |
| "output": str(out_path), |
| } |
| ) |
|
|
| status = ( |
| f"✅ Fertig.\n\n" |
| f"**Modell:** `{model_repo}`\n\n" |
| f"**Codec:** `{codec_repo}`\n\n" |
| f"**Stimme:** `{voice_label}`\n\n" |
| f"**Chunks:** `{len(chunks)}`\n\n" |
| f"**Marker-Modus:** `{event_marker_mode or 'Marker unverändert lassen'}`\n\n" |
| f"**Generierungsdauer:** `{elapsed_text}`\n\n" |
| f"**Audio-Länge:** `{audio_text}`\n\n" |
| f"**Realtime-Faktor:** `{rtf_text}`\n\n" |
| f"**Gespeichert:** `{out_path}`" |
| ) |
| return str(out_path), str(out_path), status, load_history() |
|
|
| except APP_HANDLED_ERRORS as exc: |
| tb = traceback.format_exc() |
| helpful = "" |
| if "llama" in tb.lower() or "gguf" in model_repo.lower(): |
| helpful = ( |
| "\n\n💡 GGUF-Hinweis: Installiere `llama-cpp-python` am besten als Wheel, nicht per OpenBLAS-Source-Build:\n\n" |
| "```powershell\n" |
| "pip install llama-cpp-python --prefer-binary --extra-index-url https://abetlen.github.io/llama-cpp-python/whl/cpu\n" |
| "```" |
| ) |
| if "espeak" in tb.lower() or "phonemizer" in tb.lower(): |
| helpful += ( |
| "\n\n💡 eSpeak-Hinweis: Prüfe `PHONEMIZER_ESPEAK_LIBRARY` und `PHONEMIZER_ESPEAK_PATH`." |
| ) |
| return ( |
| None, |
| None, |
| f"❌ Fehler bei der Synthese:\n\n```text\n{exc}\n```{helpful}\n\n<details><summary>Traceback</summary>\n\n```text\n{tb}\n```\n\n</details>", |
| load_history(), |
| ) |
|
|
|
|
| def clear_model_cache(): |
| _MODEL_CACHE.clear() |
| _REF_CACHE.clear() |
| _WHISPER_CACHE.clear() |
| return "🧹 Modell-, Referenz- und Whisper-Cache geleert." |
|
|
|
|
| CUSTOM_CSS = """ |
| :root { |
| --lara-orange: #ff8a2a; |
| --lara-bg: #070b18; |
| } |
| .gradio-container { |
| background: radial-gradient(circle at top left, rgba(255,138,42,0.13), transparent 28%), |
| linear-gradient(135deg, #070b18 0%, #10182b 55%, #080a12 100%) !important; |
| } |
| #hero-box { |
| border: 1px solid rgba(255,255,255,0.12); |
| border-radius: 22px; |
| padding: 22px; |
| background: rgba(255,255,255,0.06); |
| backdrop-filter: blur(14px); |
| box-shadow: 0 0 40px rgba(255,138,42,0.12); |
| } |
| #hero-title h1, #hero-title h2, #hero-title h3 { |
| color: #fff !important; |
| } |
| #hero-sub { |
| color: rgba(255,255,255,0.72) !important; |
| } |
| button.primary { |
| box-shadow: 0 0 24px rgba(255,138,42,0.28) !important; |
| } |
| """ |
|
|
|
|
| def build_app() -> gr.Blocks: |
| initial_voices = voice_choices() |
| initial_voice = initial_voices[0] if initial_voices else None |
|
|
| |
| |
| with gr.Blocks(title="NeuTTS Gradio Studio", analytics_enabled=False) as demo: |
| gr.Markdown( |
| """ |
| <div id="hero-box"> |
| <div id="hero-title"> |
| <h1>🎙️ NeuTTS Gradio Studio</h1> |
| </div> |
| <div id="hero-sub"> |
| Lokale TTS-Oberfläche für NeuTTS: Modelle wechseln, Stimmen speichern, Referenzen klonen und WAV-Dateien direkt auf die Festplatte schreiben. |
| </div> |
| </div> |
| """ |
| ) |
|
|
| with gr.Tab("🎧 Synthese"): |
| with gr.Row(): |
| with gr.Column(scale=7): |
| input_text = gr.Textbox( |
| label="Text", |
| placeholder="Schreib hier rein, was die Stimme sprechen soll...", |
| lines=8, |
| value="Hallo Sascha, das ist NeuTTS in einer kleinen lokalen Gradio-Oberfläche. Wenn das läuft, bekommt die App ein Käsebrot.", |
| ) |
| with gr.Row(): |
| model_dropdown = gr.Dropdown( |
| label="Modell", |
| choices=list(MODEL_PRESETS.keys()), |
| value="Nano German · klein/schnell/deutsch", |
| ) |
| codec_dropdown = gr.Dropdown( |
| label="Codec", |
| choices=list(CODEC_PRESETS.keys()), |
| value="NeuCodec · Standard", |
| ) |
| custom_model = gr.Textbox( |
| label="Custom Hugging-Face-Model-Repo-ID optional", |
| placeholder="z. B. neuphonic/neutts-nano-german-q4-gguf — leer lassen, wenn Dropdown genutzt wird", |
| ) |
| with gr.Row(): |
| voice_dropdown = gr.Dropdown( |
| label="Stimme", |
| choices=initial_voices, |
| value=initial_voice, |
| allow_custom_value=False, |
| ) |
| refresh_btn = gr.Button("🔄 Stimmen neu laden") |
| with gr.Accordion("⚙️ Optionen", open=True): |
| with gr.Row(): |
| backbone_device = gr.Dropdown( |
| label="Backbone Device", |
| choices=["cpu", "cuda", "gpu"], |
| value="cpu", |
| info="CPU für Windows sicher. Bei GGUF/GPU ggf. 'gpu' testen, wenn llama-cpp passend gebaut ist.", |
| ) |
| codec_device = gr.Dropdown( |
| label="Codec Device", |
| choices=["cpu", "cuda"], |
| value="cpu", |
| ) |
| with gr.Row(): |
| split_enabled = gr.Checkbox(label="Lange Texte automatisch splitten", value=True) |
| max_chars = gr.Slider(label="Max. Zeichen pro Chunk", minimum=80, maximum=600, value=260, step=20) |
| pause_ms = gr.Slider(label="Pause zwischen Chunks in ms", minimum=0, maximum=1200, value=220, step=20) |
| tail_padding_ms = gr.Slider(label="Finale Stille am Ende in ms", minimum=0, maximum=2000, value=500, step=50) |
| with gr.Row(): |
| synthesis_speed = gr.Slider( |
| label="Geschwindigkeit", |
| minimum=0.5, maximum=2.0, value=1.0, step=0.05, |
| info="Tempo der Ausgabe, pitch-erhaltend (time-stretch).", |
| ) |
| denoise_enabled = gr.Checkbox( |
| label="Rauschunterdrückung", |
| value=False, |
| info="Stationäres Denoise am erzeugten Audio.", |
| ) |
| event_marker_mode = gr.Dropdown( |
| label="Lach-/Emotionsmarker im TTS-Text", |
| choices=["Marker unverändert lassen", "Marker in sprechbare Laute umwandeln"], |
| value="Marker in sprechbare Laute umwandeln", |
| info="Beispiel: [lacht] wird zu 'haha'. NeuTTS unterstützt Lachen nicht garantiert als eigenes Token.", |
| ) |
| output_prefix = gr.Textbox(label="Dateiname-Prefix", value="neutts_test") |
|
|
| generate_btn = gr.Button("🔥 Sprache erzeugen", variant="primary") |
|
|
| with gr.Column(scale=5): |
| out_audio = gr.Audio(label="Ausgabe", type="filepath") |
| out_file = gr.File(label="Gespeicherte WAV-Datei") |
| status = gr.Markdown("Bereit. Kleine TTS-Maschine wartet auf Futter.") |
|
|
| with gr.Tab("🗣️ Stimmen verwalten"): |
| gr.Markdown( |
| """ |
| **Referenzregel:** Lade oben genau **ein** Medium hoch: Audio **oder** Video. Die App wandelt sofort lokal mit `tools/ffmpeg/bin/ffmpeg.exe` in mono/24kHz WAV um. |
| Danach schneidest du direkt im angezeigten Audiofeld den relevanten Bereich, z. B. 10–20 Sekunden saubere Sprache ohne zweite Stimme. |
| Lachen/Emotionen werden nicht zuverlässig automatisch von Whisper erkannt; dafür gibt es editierbare Marker und Stil-Metadaten. |
| """ |
| ) |
| with gr.Row(): |
| with gr.Column(scale=5): |
| new_voice_name = gr.Textbox(label="Name der Stimme", placeholder="z. B. Lara_Test_01") |
|
|
| reference_media = gr.File( |
| label="Referenzmedium hochladen (Audio oder Video)", |
| file_types=[ |
| ".wav", ".mp3", ".m4a", ".flac", ".ogg", ".aac", ".wma", |
| ".mp4", ".avi", ".mov", ".mkv", ".webm", ".m4v", ".mpeg", ".mpg", |
| ], |
| type="filepath", |
| ) |
| media_status = gr.Markdown("Noch kein Medium vorbereitet. Lade oben Audio oder Video hoch; es wird automatisch zu mono/24kHz WAV gewandelt.") |
| prepared_ref_audio = gr.Audio( |
| label="Bearbeitbare Referenz-WAV (hier schneiden)", |
| type="filepath", |
| interactive=True, |
| editable=True, |
| format="wav", |
| ) |
|
|
| with gr.Accordion("📝 Referenztext transkribieren", open=True): |
| with gr.Row(): |
| whisper_model_dropdown = gr.Dropdown( |
| label="Whisper-Modell", |
| choices=list(WHISPER_MODEL_PRESETS.keys()), |
| value="Base · guter Start/CPU", |
| ) |
| whisper_language = gr.Dropdown( |
| label="Sprache", |
| choices=["auto", "de", "en", "fr", "es"], |
| value="de", |
| ) |
| with gr.Row(): |
| whisper_device = gr.Dropdown( |
| label="Whisper Device", |
| choices=["cpu", "cuda"], |
| value="cpu", |
| ) |
| whisper_compute_type = gr.Dropdown( |
| label="Whisper Compute Type", |
| choices=["int8", "int8_float32", "float32", "float16"], |
| value="int8", |
| info="CPU: int8. CUDA: float16 ist oft schneller, wenn passend installiert.", |
| ) |
| with gr.Row(): |
| vad_filter_enabled = gr.Checkbox( |
| label="VAD-Filter nutzen", |
| value=True, |
| info="An = sauberer Sprachtext. Aus = Lachen/Atmer bleiben eher im Audio, Whisper erkennt sie aber nicht garantiert als Text.", |
| ) |
| preserve_laughter_hint = gr.Checkbox( |
| label="Lachen/Non-Speech als Marker berücksichtigen", |
| value=True, |
| info="Fügt editierbare Marker wie [lacht] hinzu, statt so zu tun, als hätte Whisper sie sicher erkannt.", |
| ) |
|
|
| emotion_label = gr.Dropdown( |
| label="Emotionaler Stil der Referenz", |
| choices=[ |
| "", |
| "neutral", |
| "fröhlich", |
| "lachend", |
| "ruhig", |
| "ernst", |
| "aufgeregt", |
| "zärtlich", |
| "energisch", |
| ], |
| value="", |
| allow_custom_value=True, |
| ) |
| non_speech_notes = gr.Textbox( |
| label="Manuelle Lach-/Atem-/Emotionsmarker", |
| placeholder="z. B. [lacht], [kichert], [seufzt], [atmet ein], fröhlich und locker", |
| lines=2, |
| ) |
| transcribe_btn = gr.Button("📝 Referenz transkribieren") |
|
|
| ref_text = gr.Textbox( |
| label="Exakter Referenztext", |
| lines=5, |
| placeholder="Das muss exakt das sein, was in der Referenz-WAV gesprochen wird. Marker kannst du bewusst ergänzen, wenn du sie beim Stiltest nutzen willst.", |
| ) |
| save_voice_btn = gr.Button("💾 Stimme speichern", variant="primary") |
| with gr.Column(scale=5): |
| voice_status = gr.Markdown("Noch nichts gespeichert.") |
| saved_ref_audio = gr.Audio(label="Gespeicherte Referenz", type="filepath") |
| gr.Markdown( |
| """ |
| **Hinweis zu Lachen & Emotion:** |
| NeuTTS kann Stil aus der Referenz übernehmen, aber ein echtes `[lacht]`-Token ist nicht garantiert. |
| Für Tests kannst du im Synthese-Text Marker wie `[lacht]` verwenden und im Synthese-Tab in sprechbare Laute umwandeln lassen. |
| """ |
| ) |
|
|
| with gr.Tab("📜 Verlauf"): |
| history_df = gr.Dataframe( |
| headers=HISTORY_COLUMNS, |
| value=load_history(), |
| label="Letzte Generierungen", |
| interactive=False, |
| wrap=True, |
| ) |
| with gr.Row(): |
| reload_history_btn = gr.Button("🔄 Verlauf neu laden") |
| clear_cache_btn = gr.Button("🧹 Modellcache leeren") |
| cache_status = gr.Markdown(TORCH_PATCH_STATUS + "\n\n" + globals().get("PERSISTENCE_STATUS", "")) |
|
|
| refresh_btn.click(refresh_voices, outputs=[voice_dropdown, status]) |
| reference_media.change( |
| prepare_reference_media, |
| inputs=[reference_media], |
| outputs=[prepared_ref_audio, media_status], |
| api_name="prepare_reference", |
| ) |
| transcribe_btn.click( |
| transcribe_reference_audio, |
| inputs=[ |
| prepared_ref_audio, |
| reference_media, |
| whisper_model_dropdown, |
| whisper_language, |
| whisper_device, |
| whisper_compute_type, |
| vad_filter_enabled, |
| preserve_laughter_hint, |
| emotion_label, |
| non_speech_notes, |
| ], |
| outputs=[ref_text, voice_status], |
| api_name="transcribe_reference", |
| ) |
| save_voice_btn.click( |
| save_voice, |
| inputs=[ |
| new_voice_name, |
| prepared_ref_audio, |
| reference_media, |
| ref_text, |
| emotion_label, |
| preserve_laughter_hint, |
| non_speech_notes, |
| ], |
| outputs=[voice_dropdown, voice_status, saved_ref_audio], |
| api_name="save_voice", |
| ) |
| generate_btn.click( |
| synthesize, |
| inputs=[ |
| input_text, |
| model_dropdown, |
| custom_model, |
| codec_dropdown, |
| voice_dropdown, |
| backbone_device, |
| codec_device, |
| split_enabled, |
| max_chars, |
| pause_ms, |
| tail_padding_ms, |
| event_marker_mode, |
| output_prefix, |
| synthesis_speed, |
| denoise_enabled, |
| ], |
| outputs=[out_audio, out_file, status, history_df], |
| api_name="synthesize", |
| ) |
| reload_history_btn.click(lambda: load_history(), outputs=[history_df]) |
| clear_cache_btn.click(clear_model_cache, outputs=[cache_status]) |
|
|
| return demo |
|
|
|
|
| |
| |
| PERSISTENCE_STATUS = voice_persistence.sync_voices_from_dataset(VOICES_DIR) |
| print(f"[neutts] {PERSISTENCE_STATUS}") |
|
|
| |
| |
| demo = build_app() |
|
|
|
|
| def launch_app(app: gr.Blocks = demo) -> None: |
| """Startet lokal freundlich, aber Hugging-Face-Space-kompatibel.""" |
| launch_kwargs = { |
| "show_error": True, |
| "theme": gr.themes.Soft(), |
| "css": CUSTOM_CSS, |
| |
| |
| "ssr_mode": False, |
| } |
|
|
| |
| |
| is_hf_space = bool(os.getenv("SPACE_ID") or os.getenv("SPACE_HOST")) |
| if not is_hf_space: |
| launch_kwargs.update( |
| { |
| "server_name": "127.0.0.1", |
| "server_port": 7860, |
| "inbrowser": True, |
| } |
| ) |
|
|
| app.queue(default_concurrency_limit=1, max_size=8).launch(**launch_kwargs) |
|
|
|
|
| if __name__ == "__main__": |
| launch_app() |
|
|
