mARTin-B78
003e4f9f46
The emotion quick-pickers added in 1.20.5 guarded with `typeof REH_EMOTIONS === 'undefined'`, but REH_EMOTIONS is a const declared later in the bundle's single shared scope — `typeof` on a const in its temporal dead zone throws instead of returning "undefined", which aborted top-level initialization for every module bundled after tts-preview.js. The pickers now read window.REH_EMOTIONS on a deferred macrotask. Also: emotion instructions are now always built in English (spoken text and the native-accent clause stay in the book's language), which controlled A/B testing showed produces a far cleaner prosodic gradient from Qwen3-TTS; and Fish-Speech now receives temperature/top_p/repetition_penalty, which it was the only backend never to have forwarded. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
805 lines
35 KiB
Python
805 lines
35 KiB
Python
"""TTS request helpers: config, audio request, streaming, voice design, NVIDIA, benchmark."""
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import struct
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
from pydub import AudioSegment
|
|
|
|
from core.config import (
|
|
_clean_preview_backend,
|
|
_preview_backend_base_url,
|
|
_tts_extra_params,
|
|
_apply_tts_extra_params,
|
|
_post_tts_with_fallback,
|
|
)
|
|
from core.constants import (
|
|
_VOICES_DIR_DEFAULT,
|
|
_VOICE_DESIGN_DEFAULT,
|
|
_VOICE_DESIGN_MODEL,
|
|
_MAX_TTS_OUTPUT_SECONDS,
|
|
)
|
|
from core.audio import _to_wav_24k, _duration as _dur
|
|
from core.voice import _find_voice_audio, _read_reference_text, _load_meta
|
|
|
|
|
|
def _apply_voice_pinned_seed(payload: dict, voice: str, settings: dict) -> dict:
|
|
"""Fall back to a voice's own saved seed (set via the Seed Finder tool) when
|
|
the request didn't already pin one explicitly.
|
|
|
|
Confirmed live: a voice's pinned seed was saved to its meta.json but never
|
|
read back anywhere outside the Seed Finder's own one-off benchmarking
|
|
codepath — every normal generation (Try It Out, Rehearsal, audiobook
|
|
export) left the seed unset, so all that per-voice seed-hunting work had
|
|
zero effect on real output. `"seed" not in payload` at this point means no
|
|
caller-level override (e.g. Seed Finder itself testing a candidate) came
|
|
through _apply_tts_extra_params, so this only ever fills a gap, never
|
|
clobbers an explicit choice.
|
|
"""
|
|
if "seed" in payload:
|
|
return payload
|
|
try:
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
wav = _find_voice_audio(voice, scan_dir)
|
|
if wav is not None:
|
|
seed = _load_meta(wav).get("seed")
|
|
if seed is not None:
|
|
payload["seed"] = int(seed)
|
|
except Exception:
|
|
pass
|
|
return payload
|
|
|
|
|
|
# ── Language helpers ──────────────────────────────────────────────────────────
|
|
|
|
_NVIDIA_LANGUAGE_CODES = {
|
|
"EN": "en-US", "DE": "de-DE", "ES": "es-ES", "FR": "fr-FR",
|
|
"IT": "it-IT", "PT": "pt-PT", "NL": "nl-NL", "PL": "pl-PL",
|
|
}
|
|
|
|
# Maps 2-letter prefix from voice ID (e.g. "DE_M_…") to full language name
|
|
# used by the faster-qwen3-tts clone server's `language` field.
|
|
_VOICE_LANG_NAMES = {
|
|
"EN": "English", "DE": "German", "ES": "Spanish", "FR": "French",
|
|
"IT": "Italian", "PT": "Portuguese", "NL": "Dutch", "PL": "Polish",
|
|
"ZH": "Chinese", "JA": "Japanese", "KO": "Korean", "RU": "Russian",
|
|
"AR": "Arabic", "HI": "Hindi", "TR": "Turkish", "SV": "Swedish",
|
|
}
|
|
|
|
|
|
def _voice_language_name(voice: str) -> str:
|
|
"""Return the full language name inferred from the voice ID prefix (e.g. DE_M_ → German)."""
|
|
prefix = str(voice or "").split("_", 1)[0].upper()
|
|
return _VOICE_LANG_NAMES.get(prefix, "")
|
|
|
|
|
|
def _nvidia_clone_language_code(voice: str, language: str = "") -> str:
|
|
raw = str(language or "").strip()
|
|
if raw and raw.lower() not in {"auto", "*"}:
|
|
return raw
|
|
prefix = str(voice or "").split("_", 1)[0].upper()
|
|
return _NVIDIA_LANGUAGE_CODES.get(prefix, "en-US")
|
|
|
|
|
|
def _form_value(value) -> str:
|
|
if isinstance(value, bool):
|
|
return "true" if value else "false"
|
|
return str(value)
|
|
|
|
|
|
# ── WAV helpers ───────────────────────────────────────────────────────────────
|
|
|
|
def _wav_data_offset(data: bytes) -> int | None:
|
|
if len(data) < 12:
|
|
return None
|
|
if data[:4] != b"RIFF" or data[8:12] != b"WAVE":
|
|
return 0
|
|
pos = 12
|
|
while pos + 8 <= len(data):
|
|
chunk_sz = struct.unpack_from("<I", data, pos + 4)[0]
|
|
if data[pos:pos + 4] == b"data":
|
|
return pos + 8
|
|
pos += 8 + chunk_sz + (chunk_sz % 2)
|
|
return None
|
|
|
|
|
|
def _audio_duration_from_bytes(audio: bytes, media_type: str) -> float | None:
|
|
try:
|
|
source_format = "wav" if audio[:4] == b"RIFF" or "wav" in media_type.lower() else None
|
|
return len(AudioSegment.from_file(io.BytesIO(audio), format=source_format)) / 1000.0
|
|
except Exception:
|
|
offset = _wav_data_offset(audio)
|
|
if offset is not None and offset > 0 and len(audio) > offset:
|
|
return (len(audio) - offset) / (24000 * 2)
|
|
return None
|
|
|
|
|
|
# ── TTS request config & audio ────────────────────────────────────────────────
|
|
|
|
def _tts_request_config(
|
|
text: str,
|
|
voice: str,
|
|
settings: dict,
|
|
response_format: str = "wav",
|
|
instruct: str = "",
|
|
url_override: str = "",
|
|
api_key_override: str | None = None,
|
|
backend_override: str = "",
|
|
extra_backend: str = "",
|
|
) -> tuple[str, dict, dict]:
|
|
from core.constants import _TTS_DEFAULT
|
|
from core.validation import _validate_http_url
|
|
|
|
tts_url = _validate_http_url(url_override or settings.get("tts_url", _TTS_DEFAULT), allow_private=True).rstrip("/")
|
|
backend = backend_override or settings.get("tts_backend", "openai")
|
|
tts_key = (api_key_override if api_key_override is not None else settings.get("tts_api_key", "")).strip()
|
|
tts_hdrs = {"Authorization": f"Bearer {tts_key}"} if tts_key else {}
|
|
|
|
if backend == "localai":
|
|
endpoint, payload = f"{tts_url}/tts", {"input": text, "model": voice, "response_format": response_format}
|
|
elif backend == "pocket":
|
|
endpoint, payload = f"{tts_url}/v1/audio/speech", {"input": text, "voice": voice, "response_format": response_format}
|
|
else:
|
|
endpoint, payload = f"{tts_url}/v1/audio/speech", {"model": "tts-1", "input": text, "voice": voice, "response_format": response_format}
|
|
if instruct.strip():
|
|
payload["instruct"] = instruct.strip()
|
|
# Include language derived from voice ID prefix so the TTS server knows what
|
|
# language to synthesise (e.g. DE_M_… → German). Only set when the backend
|
|
# is the local faster-qwen3-tts clone server — skip for OpenAI / cloud APIs.
|
|
_eff_backend = extra_backend or backend_override or "voice_clone"
|
|
if _eff_backend in ("voice_clone", "streaming", "customvoice"):
|
|
lang = _voice_language_name(voice)
|
|
if lang:
|
|
payload["language"] = lang
|
|
_apply_tts_extra_params(payload, settings, _eff_backend)
|
|
if _eff_backend in ("voice_clone", "streaming", "customvoice"):
|
|
_apply_voice_pinned_seed(payload, voice, settings)
|
|
return endpoint, payload, tts_hdrs
|
|
|
|
|
|
def _tts_request_audio(
|
|
text: str,
|
|
voice: str,
|
|
settings: dict,
|
|
instruct: str = "",
|
|
url_override: str = "",
|
|
api_key_override: str | None = None,
|
|
backend_override: str = "",
|
|
extra_backend: str = "",
|
|
) -> tuple[bytes, str]:
|
|
endpoint, payload, tts_hdrs = _tts_request_config(
|
|
text, voice, settings, "wav", instruct, url_override, api_key_override, backend_override, extra_backend
|
|
)
|
|
resp = _post_tts_with_fallback(endpoint, payload, tts_hdrs, timeout=120)
|
|
resp.raise_for_status()
|
|
audio = resp.content
|
|
if not audio:
|
|
raise RuntimeError("backend returned empty audio")
|
|
media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav"
|
|
return audio, media_type
|
|
|
|
|
|
# ── TTS streaming helpers ─────────────────────────────────────────────────────
|
|
|
|
def _tts_stream_request_config(text: str, voice: str, settings: dict, instruct: str = "") -> tuple[str, dict, dict]:
|
|
from core.constants import _TTS_STREAM_DEFAULT
|
|
from core.validation import _validate_http_url
|
|
|
|
stream_url = settings.get("tts_stream_url") or settings.get("tts_url") or _TTS_STREAM_DEFAULT
|
|
tts_url = _validate_http_url(stream_url, allow_private=True).rstrip("/")
|
|
tts_key = settings.get("tts_api_key", "").strip()
|
|
tts_hdrs = {"Authorization": f"Bearer {tts_key}"} if tts_key else {}
|
|
payload = {"model": "tts-1", "input": text, "voice": voice, "response_format": "wav"}
|
|
if instruct.strip():
|
|
payload["instruct"] = instruct.strip()
|
|
_apply_tts_extra_params(payload, settings, "streaming")
|
|
return f"{tts_url}/v1/audio/speech", payload, tts_hdrs
|
|
|
|
|
|
def _open_tts_stream_response(text: str, voice: str, settings: dict, instruct: str = "") -> requests.Response:
|
|
endpoint, payload, tts_hdrs = _tts_stream_request_config(text, voice, settings, instruct)
|
|
resp = _post_tts_with_fallback(endpoint, payload, tts_hdrs, stream=True, timeout=(10, 900))
|
|
try:
|
|
resp.raise_for_status()
|
|
except Exception as exc:
|
|
detail = ""
|
|
try:
|
|
detail = resp.text[:500]
|
|
except Exception:
|
|
pass
|
|
resp.close()
|
|
raise RuntimeError(f"streaming backend error: {exc}{(': ' + detail) if detail else ''}") from exc
|
|
return resp
|
|
|
|
|
|
def _read_tts_stream_response(resp: requests.Response) -> tuple[bytes, str]:
|
|
try:
|
|
media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav"
|
|
return b"".join(chunk for chunk in resp.iter_content(chunk_size=64 * 1024) if chunk), media_type
|
|
finally:
|
|
resp.close()
|
|
|
|
|
|
def _iter_tts_stream_response(resp: requests.Response):
|
|
try:
|
|
for chunk in resp.iter_content(chunk_size=64 * 1024):
|
|
if chunk:
|
|
yield chunk
|
|
finally:
|
|
resp.close()
|
|
|
|
|
|
# ── Voice design request helpers ──────────────────────────────────────────────
|
|
|
|
def _apply_voice_design_gender(instruct: str, gender: str) -> str:
|
|
gender = (gender or "").strip().upper()
|
|
if gender == "F":
|
|
prefix = (
|
|
"MANDATORY SPEAKER IDENTITY: Female speaker / woman.\n"
|
|
"gender: Female.\n"
|
|
"Use a clearly feminine vocal timbre, light-to-medium resonance, and soprano or mezzo-soprano pitch range.\n"
|
|
"Avoid male baritone, bass, chest-heavy, or masculine vocal qualities."
|
|
)
|
|
elif gender == "M":
|
|
prefix = (
|
|
"MANDATORY SPEAKER IDENTITY: Male speaker / man.\n"
|
|
"gender: Male.\n"
|
|
"Use a clearly masculine vocal timbre, medium-to-deep resonance, and tenor, baritone, or bass pitch range.\n"
|
|
"Avoid feminine soprano or mezzo-soprano vocal qualities."
|
|
)
|
|
else:
|
|
return instruct
|
|
return f"{prefix}\n\n{instruct.strip()}"
|
|
|
|
|
|
def _voice_design_request_audio(
|
|
instruct: str,
|
|
text: str,
|
|
language: str,
|
|
settings: dict,
|
|
gender: str = "",
|
|
) -> tuple[bytes, str]:
|
|
from core.validation import _validate_http_url
|
|
vd_url = _validate_http_url(settings.get("voice_design_url") or _VOICE_DESIGN_DEFAULT, allow_private=True).rstrip("/")
|
|
vd_key = (settings.get("voice_design_api_key") or settings.get("tts_api_key", "")).strip()
|
|
vd_hdrs = {"Authorization": f"Bearer {vd_key}"} if vd_key else {}
|
|
payload = {
|
|
"model": _VOICE_DESIGN_MODEL,
|
|
"input": text,
|
|
"instruct": _apply_voice_design_gender(instruct, gender),
|
|
"language": language,
|
|
"response_format": "wav",
|
|
}
|
|
_apply_tts_extra_params(payload, settings, "voice_design")
|
|
resp = _post_tts_with_fallback(f"{vd_url}/v1/audio/speech", payload, vd_hdrs, timeout=180)
|
|
resp.raise_for_status()
|
|
audio = resp.content
|
|
if not audio:
|
|
raise RuntimeError("backend returned empty audio")
|
|
media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav"
|
|
return audio, media_type
|
|
|
|
|
|
def _voice_design_voice_request_audio(
|
|
voice: str,
|
|
text: str,
|
|
settings: dict,
|
|
instruct: str = "",
|
|
language: str = "Auto",
|
|
) -> tuple[bytes, str]:
|
|
from core.validation import _validate_http_url
|
|
vd_url = _validate_http_url(settings.get("voice_design_url") or _VOICE_DESIGN_DEFAULT, allow_private=True).rstrip("/")
|
|
vd_key = (settings.get("voice_design_api_key") or settings.get("tts_api_key", "")).strip()
|
|
vd_hdrs = {"Authorization": f"Bearer {vd_key}"} if vd_key else {}
|
|
payload = {
|
|
"model": _VOICE_DESIGN_MODEL,
|
|
"input": text,
|
|
"voice": voice,
|
|
"response_format": "wav",
|
|
}
|
|
if instruct.strip():
|
|
payload["instruct"] = instruct.strip()
|
|
if language and language != "Auto":
|
|
payload["language"] = language
|
|
# "voice_design_playback", not "voice_design" — this is reading an
|
|
# ALREADY-saved voice back for a line of dialogue, not creating a new
|
|
# one, so it needs voice_clone-grade stability, not design-time
|
|
# randomness. See the config-side comment for the full story.
|
|
_apply_tts_extra_params(payload, settings, "voice_design_playback")
|
|
_apply_voice_pinned_seed(payload, voice, settings)
|
|
resp = _post_tts_with_fallback(f"{vd_url}/v1/audio/speech", payload, vd_hdrs, timeout=180)
|
|
resp.raise_for_status()
|
|
audio = resp.content
|
|
if not audio:
|
|
raise RuntimeError("backend returned empty audio")
|
|
media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav"
|
|
return audio, media_type
|
|
|
|
|
|
# ── NVIDIA clone request ──────────────────────────────────────────────────────
|
|
|
|
def _nvidia_clone_request_audio(
|
|
text: str,
|
|
voice: str,
|
|
settings: dict,
|
|
mode: str = "zeroshot",
|
|
reference_transcript: str = "",
|
|
language: str = "",
|
|
) -> tuple[bytes, str]:
|
|
from core.validation import _validate_http_url
|
|
|
|
mode = "flow" if str(mode).lower().endswith("flow") else "zeroshot"
|
|
backend = "nvidia_flow" if mode == "flow" else "nvidia_zeroshot"
|
|
base_url = _validate_http_url(_preview_backend_base_url(settings, backend), allow_private=True).rstrip("/")
|
|
key = (settings.get("voice_design_api_key") or settings.get("tts_api_key", "")).strip()
|
|
headers = {"Authorization": f"Bearer {key}"} if key else {}
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
audio_path = _find_voice_audio(voice, scan_dir)
|
|
if audio_path is None:
|
|
raise RuntimeError(f"reference voice not found in library: {voice}")
|
|
prompt_wav = _to_wav_24k(audio_path)
|
|
has_ref, saved_transcript = _read_reference_text(audio_path)
|
|
prompt_transcript = str(reference_transcript or saved_transcript or "").strip()
|
|
if mode == "flow" and not prompt_transcript:
|
|
raise RuntimeError("NVIDIA Magpie Flow requires the selected voice to have an exact saved reference transcript")
|
|
data = {
|
|
"language": _nvidia_clone_language_code(voice, language),
|
|
"text": text,
|
|
}
|
|
if mode == "flow":
|
|
data["audio_prompt_transcript"] = prompt_transcript
|
|
params = _tts_extra_params(settings, backend)
|
|
for k, v in params.items():
|
|
if k in {"audio_prompt", "audio_prompt_transcript", "text", "language"}:
|
|
continue
|
|
data[k] = _form_value(v)
|
|
endpoint = f"{base_url}/v1/audio/synthesize"
|
|
with prompt_wav.open("rb") as f:
|
|
files = {"audio_prompt": ("prompt.wav", f, "audio/wav")}
|
|
resp = requests.post(endpoint, data=data, files=files, headers=headers, timeout=180)
|
|
if resp.status_code in {400, 404, 415, 422} and params:
|
|
try:
|
|
resp.close()
|
|
except Exception:
|
|
pass
|
|
fallback = {k: v for k, v in data.items() if k in {"language", "text", "audio_prompt_transcript"}}
|
|
with prompt_wav.open("rb") as f:
|
|
files = {"audio_prompt": ("prompt.wav", f, "audio/wav")}
|
|
resp = requests.post(endpoint, data=fallback, files=files, headers=headers, timeout=180)
|
|
resp.raise_for_status()
|
|
audio = resp.content
|
|
if not audio:
|
|
raise RuntimeError("NVIDIA clone backend returned empty audio")
|
|
media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav"
|
|
return audio, media_type
|
|
|
|
|
|
# ── Fish-Speech request (clone from saved WAV + inline emotion markers) ──────
|
|
|
|
# Compact German→English fallback for the emotion word Qwen3-TTS's native-language
|
|
# instruct sentence carries (e.g. "Sprich in einem bedrohlich Tonfall.") — Fish-Speech's
|
|
# docs require English tags "regardless of the spoken language". This is only a
|
|
# server-side safety net for callers that never went through the client's own
|
|
# _rehInlineTone/_rehEmotionEnglishTag translation (static/js/rehearser.js) and only
|
|
# send `instruct`; keep it in sync with that JS table if it grows.
|
|
_FISHSPEECH_EMOTION_DE_EN = {
|
|
"wütend": "angry", "zornig": "angry", "traurig": "sad", "ängstlich": "scared",
|
|
"furchtsam": "fearful", "fröhlich": "happy", "glücklich": "happy",
|
|
"flüsternd": "whispering", "aufgeregt": "excited", "überrascht": "surprised",
|
|
"verzweifelt": "desperate", "resigniert": "resigned", "entschlossen": "determined",
|
|
"selbstbewusst": "confident", "schüchtern": "shy", "ironisch": "sarcastic",
|
|
"sarkastisch": "sarcastic", "verächtlich": "contemptuous", "ernst": "serious",
|
|
"streng": "stern", "befehlend": "commanding", "sanft": "gentle", "zärtlich": "tender",
|
|
"kalt": "cold", "gelangweilt": "bored", "geheimnisvoll": "mysterious",
|
|
"bedrohlich": "threatening", "dramatisch": "dramatic", "ruhig": "calm",
|
|
"schockiert": "shocked", "verwirrt": "confused", "weinend": "tearful",
|
|
"trauernd": "grieving", "schroff": "curt", "freundlich": "friendly",
|
|
"spielerisch": "playful", "romantisch": "romantic", "erleichtert": "relieved",
|
|
"neugierig": "curious", "müde": "weary", "bemerkend": "remarking",
|
|
"flehend": "pleading", "warnend": "warning", "trotzig": "defiant",
|
|
"erschrocken": "startled",
|
|
}
|
|
|
|
|
|
def _fishspeech_emotion_prefix(instruct: str) -> str:
|
|
"""Turn the per-line style instruction into a Fish-Speech inline emotion marker.
|
|
|
|
Fish-Speech (S2-Pro) requires SQUARE brackets for a tag to be treated as a silent
|
|
control instruction — round parentheses get read aloud as literal text instead
|
|
(confirmed live: "(excited)" was spoken as "Hexited") — and the tag word itself
|
|
must be English regardless of the instruct sentence's own language (per Fish
|
|
Audio's docs). Matches both the EN template ("Speak in a X manner.") and the DE
|
|
template ("Sprich in einem X Tonfall.") from _BUILD_INSTRUCT_TEMPLATES.
|
|
"""
|
|
import re
|
|
s = (instruct or "").strip()
|
|
if not s:
|
|
return ""
|
|
m = re.search(r"speak(?:ing)?\s+in\s+(?:a|an)\s+([a-z\- ]+?)\s+manner", s, re.I)
|
|
if m:
|
|
# Matched the EN template — the word is already English, use as-is.
|
|
return f"[{m.group(1).strip().lower()}] "
|
|
m = re.search(r"sprich\s+in\s+einem\s+([a-zäöüß\- ]+?)\s+tonfall", s, re.I)
|
|
if m:
|
|
# Matched the DE template — the word is German and MUST be translated;
|
|
# it will almost always be pure a-z letters too, so there is no reliable
|
|
# way to tell "already English" apart from "German" by charset alone here.
|
|
word = m.group(1).strip().lower()
|
|
tag = _FISHSPEECH_EMOTION_DE_EN.get(word, "")
|
|
return f"[{tag}] " if tag else ""
|
|
# Free-typed style instruction (Read Aloud / Try a Voice let you type anything,
|
|
# not just the two fixed templates above) — no reliable way to tell English apart
|
|
# from another language here, so fall back to the original behavior: use it
|
|
# verbatim if short enough to plausibly be a tag. Confirmed as the pre-existing,
|
|
# working behavior for manually-typed English instructions on those two pages;
|
|
# only the automated Rehearser/Studio pipeline's two known templates are handled
|
|
# more precisely above.
|
|
return f"[{s}] " if len(s) <= 40 else ""
|
|
|
|
|
|
def _fishspeech_request_audio(
|
|
text: str,
|
|
voice: str,
|
|
settings: dict,
|
|
instruct: str = "",
|
|
language: str = "",
|
|
) -> tuple[bytes, str]:
|
|
"""Synthesize via Fish-Speech: clone the voice's saved reference WAV (consistent
|
|
speaker identity) while applying the line's tone as an inline emotion marker."""
|
|
import base64
|
|
import hashlib
|
|
from core.validation import _validate_http_url
|
|
base_url = _validate_http_url(_preview_backend_base_url(settings, "fishspeech"), allow_private=True).rstrip("/")
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
audio_path = _find_voice_audio(voice, scan_dir)
|
|
if audio_path is None:
|
|
raise RuntimeError(f"reference voice not found in library: {voice}")
|
|
wav = _to_wav_24k(audio_path)
|
|
_has_ref, ref_text = _read_reference_text(audio_path)
|
|
audio_b64 = base64.b64encode(wav.read_bytes()).decode("ascii")
|
|
# Stable per-voice seed → reduces run-to-run drift on top of the reference clone.
|
|
seed = int(hashlib.md5(voice.encode("utf-8")).hexdigest()[:8], 16)
|
|
|
|
# The Rehearser/Studio pipeline already embeds an English [tag] straight into
|
|
# `text` client-side (_rehInlineTone in rehearser.js) — only fall back to deriving
|
|
# one from `instruct` here for callers that don't (e.g. a direct API call that
|
|
# skips the client helper), to avoid double-tagging the same line.
|
|
already_tagged = text.lstrip().startswith("[")
|
|
payload = {
|
|
"text": text if already_tagged else _fishspeech_emotion_prefix(instruct) + text,
|
|
"format": "wav",
|
|
"references": [{"audio": audio_b64, "text": ref_text or ""}],
|
|
"seed": seed,
|
|
"use_memory_cache": "on",
|
|
"chunk_length": 200,
|
|
"normalize": True,
|
|
}
|
|
# Fish-Speech's own request schema (ServeTTSRequest) exposes temperature/top_p/
|
|
# repetition_penalty, but this call never forwarded them — every line synthesized
|
|
# at the server's fixed default (temperature 0.8), regardless of Settings. Confirmed
|
|
# live as the likely cause of emotion tags barely differentiating from each other
|
|
# (a controlled same-line test showed happy/angry/excited/scared/shouting all
|
|
# collapsing into nearly identical pitch/loudness — the model wasn't being given
|
|
# room to vary). Every other backend already routes through this same helper; Fish
|
|
# was the one exception.
|
|
_apply_tts_extra_params(payload, settings, "fishspeech")
|
|
resp = _post_tts_with_fallback(f"{base_url}/v1/tts", payload, {}, timeout=180)
|
|
resp.raise_for_status()
|
|
audio = resp.content
|
|
if not audio or len(audio) < 256:
|
|
raise RuntimeError("Fish-Speech returned empty audio")
|
|
media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav"
|
|
return audio, media_type
|
|
|
|
|
|
# ── VibeVoice request ─────────────────────────────────────────────────────────
|
|
|
|
def _vibevoice_request_audio(text: str, settings: dict) -> tuple[bytes, str]:
|
|
url = _preview_backend_base_url(settings, "vibevoice").rstrip("/")
|
|
resp = requests.post(f"{url}/tts", json={"text": text}, timeout=60)
|
|
resp.raise_for_status()
|
|
audio = resp.content
|
|
if not audio:
|
|
raise RuntimeError("VibeVoice backend returned empty audio")
|
|
media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav"
|
|
return audio, media_type
|
|
|
|
|
|
# ── Preview request dispatcher ────────────────────────────────────────────────
|
|
|
|
def _preview_request_audio(
|
|
text: str,
|
|
voice: str,
|
|
settings: dict,
|
|
instruct: str = "",
|
|
backend: str = "voice_clone",
|
|
) -> tuple[bytes, str]:
|
|
backend = _clean_preview_backend(backend)
|
|
if backend == "streaming":
|
|
resp = _open_tts_stream_response(text, voice, settings, instruct)
|
|
audio, media_type = _read_tts_stream_response(resp)
|
|
if not audio:
|
|
raise RuntimeError("backend returned empty audio")
|
|
return audio, media_type
|
|
if backend == "customvoice":
|
|
return _tts_request_audio(
|
|
text, voice, settings, instruct,
|
|
_preview_backend_base_url(settings, "customvoice"),
|
|
settings.get("tts_api_key", ""),
|
|
"openai",
|
|
"customvoice",
|
|
)
|
|
if backend == "voice_design":
|
|
# Every OTHER backend passes an explicit language derived from the
|
|
# voice's own id prefix (see _tts_request_config above) — this one
|
|
# left it at the default "Auto" for every single line, letting the
|
|
# engine guess from scratch each time instead of being told what it
|
|
# already knows. Short dialogue lines (a few words) are exactly the
|
|
# case where language auto-detection is least reliable, and a wrong
|
|
# guess here plausibly contributes to the accent/pronunciation
|
|
# inconsistency reported for designed voices.
|
|
lang = _voice_language_name(voice) or "Auto"
|
|
return _voice_design_voice_request_audio(voice, text, settings, instruct, lang)
|
|
if backend == "nvidia_magpie":
|
|
return _tts_request_audio(
|
|
text, voice, settings, instruct,
|
|
_preview_backend_base_url(settings, "nvidia_magpie"),
|
|
settings.get("tts_api_key", ""),
|
|
"nvidia_magpie",
|
|
"nvidia_magpie",
|
|
)
|
|
if backend == "nvidia_zeroshot":
|
|
return _nvidia_clone_request_audio(text, voice, settings, "zeroshot")
|
|
if backend == "nvidia_flow":
|
|
return _nvidia_clone_request_audio(text, voice, settings, "flow")
|
|
if backend == "kokoro":
|
|
return _tts_request_audio(
|
|
text, voice, settings, instruct,
|
|
url_override=_preview_backend_base_url(settings, "kokoro"),
|
|
api_key_override=settings.get("tts_api_key", ""),
|
|
backend_override="openai",
|
|
extra_backend="kokoro",
|
|
)
|
|
if backend == "vibevoice":
|
|
return _vibevoice_request_audio(text, settings)
|
|
if backend == "fishspeech":
|
|
return _fishspeech_request_audio(text, voice, settings, instruct)
|
|
if backend == "xtts":
|
|
return _tts_request_audio(
|
|
text, voice, settings, instruct,
|
|
url_override=_preview_backend_base_url(settings, "xtts"),
|
|
api_key_override=settings.get("tts_api_key", ""),
|
|
backend_override="openai",
|
|
extra_backend="xtts",
|
|
)
|
|
return _tts_request_audio(text, voice, settings, instruct)
|
|
|
|
|
|
# ── Response format helpers ───────────────────────────────────────────────────
|
|
|
|
def _requested_response_format(data: dict) -> str:
|
|
requested = str(data.get("response_format") or data.get("format") or "wav").strip().lower()
|
|
if requested in {"mp3", "mpeg"}:
|
|
return "mp3"
|
|
if requested in {"wav", "pcm"}:
|
|
return "wav"
|
|
return "wav"
|
|
|
|
|
|
def _audio_ext_media(response_format: str) -> tuple[str, str]:
|
|
if response_format == "mp3":
|
|
return "mp3", "audio/mpeg"
|
|
return "wav", "audio/wav"
|
|
|
|
|
|
def _prepare_proxy_audio(audio: bytes, media_type: str, response_format: str) -> tuple[bytes, str, str, float | None, bool]:
|
|
"""Rewrite backend audio so clients receive playable headers and requested format."""
|
|
try:
|
|
source_format = "wav" if audio[:4] == b"RIFF" or "wav" in media_type.lower() else None
|
|
segment = AudioSegment.from_file(io.BytesIO(audio), format=source_format)
|
|
except Exception:
|
|
ext, wanted_media = _audio_ext_media(response_format)
|
|
return audio, media_type or wanted_media, ext, None, False
|
|
|
|
duration = len(segment) / 1000.0
|
|
clipped = False
|
|
if _MAX_TTS_OUTPUT_SECONDS > 0 and duration > _MAX_TTS_OUTPUT_SECONDS:
|
|
segment = segment[:int(_MAX_TTS_OUTPUT_SECONDS * 1000)]
|
|
duration = len(segment) / 1000.0
|
|
clipped = True
|
|
ext, wanted_media = _audio_ext_media(response_format)
|
|
out = io.BytesIO()
|
|
export_format = "mp3" if response_format == "mp3" else "wav"
|
|
segment.export(out, format=export_format)
|
|
return out.getvalue(), wanted_media, ext, duration, clipped
|
|
|
|
|
|
# ── Route sound helpers ───────────────────────────────────────────────────────
|
|
|
|
def _route_sound_path(settings: dict, value: str) -> Path | None:
|
|
from core.validation import _safe_child_path
|
|
from core.voice import _AUDIO_EXTS
|
|
|
|
value = str(value or "").strip()
|
|
if not value:
|
|
return None
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
candidate = Path(value)
|
|
if not candidate.is_absolute():
|
|
candidate = scan_dir / candidate
|
|
path = _safe_child_path(scan_dir, candidate)
|
|
if not path.exists() or not path.is_file() or path.suffix.lower() not in _AUDIO_EXTS:
|
|
raise RuntimeError(f"Route sound not found or unsupported: {value}")
|
|
return path
|
|
|
|
|
|
def _sound_segment(path: Path) -> AudioSegment:
|
|
return AudioSegment.from_file(str(path)).set_channels(1).set_sample_width(2).set_frame_rate(24000)
|
|
|
|
|
|
def _apply_route_sounds(audio: bytes, media_type: str, route: dict | None, settings: dict) -> tuple[bytes, str, list[str]]:
|
|
if not route:
|
|
return audio, media_type, []
|
|
before = _route_sound_path(settings, str(route.get("before_sound", "")))
|
|
after = _route_sound_path(settings, str(route.get("after_sound", "")))
|
|
try:
|
|
speed = float(route.get("speed", 1.0) or 1.0)
|
|
except (TypeError, ValueError):
|
|
speed = 1.0
|
|
has_speed = abs(speed - 1.0) > 0.01
|
|
if not before and not after and not has_speed:
|
|
return audio, media_type, []
|
|
|
|
source_format = "wav" if audio[:4] == b"RIFF" or "wav" in media_type.lower() else None
|
|
speech = AudioSegment.from_file(io.BytesIO(audio), format=source_format)
|
|
speech = speech.set_channels(1).set_sample_width(2).set_frame_rate(24000)
|
|
applied = []
|
|
if has_speed:
|
|
# Only the spoken voice is stretched — a before/after chime or
|
|
# page-turn sound stays at its own natural speed, added after.
|
|
from core.audio import _change_tempo
|
|
buf = io.BytesIO()
|
|
speech.export(buf, format="wav")
|
|
speech = AudioSegment.from_file(io.BytesIO(_change_tempo(buf.getvalue(), speed)), format="wav")
|
|
applied.append(f"speed:{speed}x")
|
|
combined = AudioSegment.empty()
|
|
if before:
|
|
combined += _sound_segment(before)
|
|
applied.append(f"before:{before.name}")
|
|
combined += speech
|
|
if after:
|
|
combined += _sound_segment(after)
|
|
applied.append(f"after:{after.name}")
|
|
|
|
out = io.BytesIO()
|
|
combined.export(out, format="wav")
|
|
return out.getvalue(), "audio/wav", applied
|
|
|
|
|
|
# ── Voice design dialogue helpers ─────────────────────────────────────────────
|
|
|
|
def _parse_voice_design_dialogue(instruct: str, script: str) -> tuple[dict[str, str], list[tuple[str, str]]] | None:
|
|
import re
|
|
speakers: dict[str, str] = {}
|
|
for raw in instruct.splitlines():
|
|
line = raw.strip()
|
|
if not line:
|
|
continue
|
|
match = re.match(r'^"?([^":]+)"?\s*:\s*"?(.+?)"?$', line)
|
|
if match:
|
|
speakers[match.group(1).strip()] = match.group(2).strip()
|
|
|
|
turns: list[tuple[str, str]] = []
|
|
for raw in script.splitlines():
|
|
line = raw.strip()
|
|
if not line:
|
|
continue
|
|
match = re.match(r"^([^:]{1,40}):\s*(.+)$", line)
|
|
if match:
|
|
speaker = match.group(1).strip()
|
|
text = match.group(2).strip()
|
|
if speaker in speakers and text:
|
|
turns.append((speaker, text))
|
|
|
|
if len(speakers) < 2 or len(turns) < 2:
|
|
return None
|
|
if len({speaker for speaker, _text in turns}) < 2:
|
|
return None
|
|
return speakers, turns
|
|
|
|
|
|
def _infer_voice_design_gender(description: str) -> str:
|
|
import re
|
|
text = f" {description.lower()} "
|
|
if re.search(r"\b(female|woman|girl|feminine|soprano|mezzo-soprano|mezzo)\b", text):
|
|
return "F"
|
|
if re.search(r"\b(male|man|boy|masculine|tenor|baritone|bass)\b", text):
|
|
return "M"
|
|
return ""
|
|
|
|
|
|
def _voice_design_dialogue_request_audio(
|
|
speakers: dict[str, str],
|
|
turns: list[tuple[str, str]],
|
|
language: str,
|
|
settings: dict,
|
|
) -> tuple[bytes, str]:
|
|
from core.audio import _audio_segment_from_wav, _wav_bytes_from_segment
|
|
combined = AudioSegment.silent(duration=120, frame_rate=24000).set_channels(1).set_sample_width(2)
|
|
pause = AudioSegment.silent(duration=180, frame_rate=24000).set_channels(1).set_sample_width(2)
|
|
|
|
for speaker, text in turns:
|
|
description = speakers[speaker]
|
|
turn_instruct = f'Speaker "{speaker}".\n{description}'
|
|
audio, _media_type = _voice_design_request_audio(
|
|
turn_instruct,
|
|
text,
|
|
language,
|
|
settings,
|
|
gender=_infer_voice_design_gender(description),
|
|
)
|
|
combined += _audio_segment_from_wav(audio) + pause
|
|
|
|
return _wav_bytes_from_segment(combined), "audio/wav"
|
|
|
|
|
|
# ── Benchmark request ─────────────────────────────────────────────────────────
|
|
|
|
def _tts_benchmark_request(text: str, voice: str, settings: dict, label: str, is_designed: bool = False) -> dict:
|
|
start = time.perf_counter()
|
|
first_audio_at = None
|
|
raw = bytearray()
|
|
media_type = "audio/wav"
|
|
|
|
if is_designed:
|
|
# A designed voice has no reference WAV to clone from — it can only
|
|
# ever be synthesized through the voice_design engine (same dispatch
|
|
# as /api/tts-preview's backend=='voice_design' branch), never the
|
|
# generic voice_clone-style request this function otherwise builds.
|
|
# Previously every voice benchmarked through the one fixed tts_url
|
|
# regardless of origin, so a designed voice's benchmark only ever
|
|
# "worked" by coincidence when that unrelated clone engine happened
|
|
# to also be reachable — confirmed live: with it down, EVERY voice
|
|
# in an all-designed batch failed the benchmark even though the
|
|
# voice_design engine itself was reachable the whole time.
|
|
audio_bytes, media_type = _voice_design_voice_request_audio(voice, text, settings)
|
|
raw.extend(audio_bytes)
|
|
first_audio_at = time.perf_counter()
|
|
else:
|
|
endpoint, payload, tts_hdrs = _tts_request_config(text, voice, settings, "wav")
|
|
with _post_tts_with_fallback(endpoint, payload, tts_hdrs, stream=True, timeout=180) as resp:
|
|
resp.raise_for_status()
|
|
media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav"
|
|
for chunk in resp.iter_content(chunk_size=512):
|
|
if not chunk:
|
|
continue
|
|
raw.extend(chunk)
|
|
if first_audio_at is None:
|
|
if "wav" in media_type.lower():
|
|
offset = _wav_data_offset(bytes(raw))
|
|
if offset is not None and len(raw) > offset:
|
|
first_audio_at = time.perf_counter()
|
|
else:
|
|
first_audio_at = time.perf_counter()
|
|
|
|
total = time.perf_counter() - start
|
|
if not raw:
|
|
raise RuntimeError("backend returned empty audio")
|
|
audio_sec = _audio_duration_from_bytes(bytes(raw), media_type)
|
|
rtf = total / audio_sec if audio_sec and audio_sec > 0 else None
|
|
speed = audio_sec / total if audio_sec and total > 0 else None
|
|
return {
|
|
"ok": True,
|
|
"label": label,
|
|
"text": text,
|
|
"ttfa_ms": round(((first_audio_at or time.perf_counter()) - start) * 1000, 1),
|
|
"total_sec": round(total, 3),
|
|
"audio_sec": round(audio_sec, 3) if audio_sec is not None else None,
|
|
"rtf": round(rtf, 3) if rtf is not None else None,
|
|
"speed": round(speed, 3) if speed is not None else None,
|
|
"bytes": len(raw),
|
|
}
|