1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
"""Client for the faster-qwen3-tts OpenAI-compatible server."""
import json
import logging
import urllib.error
import urllib.request
import wave
from pathlib import Path
from typing import List, Optional
from .. import config
from ..chunking import split_into_chunks
from .base import BaseTTSClient, ConversionCancelled
logger = logging.getLogger(__name__)
# The 12Hz codec the faster server synthesizes with outputs 24 kHz audio.
SAMPLE_RATE = 24000
class FasterTTSClient(BaseTTSClient):
"""Generates audio chunks through a faster-qwen3-tts server.
Talks to the OpenAI-compatible server shipped in the faster-qwen3-tts
repository (examples/openai_server.py). The reference voice (ref audio,
ref text) and language are configured on the server itself via
--ref-audio/--ref-text or a --voices JSON file; this client only sends
text. Unlike the Qwen demo, the server performs one generation per
request, so long chunks are sub-chunked client-side.
"""
def __init__(self, chunks_dir: Path,
voice: Optional[str] = None, api_url: Optional[str] = None,
quiet: bool = False, cancel=None):
super().__init__(chunks_dir, quiet=quiet, cancel=cancel)
# The voice is per-run (--voice / the Generate form's Voice pick);
# there is no configured default.
self.voice = (voice or "").strip()
if not self.voice:
raise RuntimeError(
"The faster backend requires a voice: pass --voice NAME "
"naming a key in the server's voices.json (see README).")
self.api_url = (api_url or config.FASTER_API_URL).rstrip("/")
self._check_health()
def _check_health(self) -> None:
"""Verify the server is reachable and its model is loaded."""
url = f"{self.api_url}/health"
try:
with urllib.request.urlopen(url, timeout=10) as response:
payload = json.loads(response.read().decode("utf-8"))
except Exception as exc:
raise RuntimeError(
f"Faster TTS server not reachable at {url}: {exc}. "
"Start the faster-qwen3-tts OpenAI-compatible server first "
"(see the 'Faster backend' section of the README)."
) from exc
if not payload.get("model_loaded"):
raise RuntimeError(
"The faster TTS server is running but its model is not loaded yet; "
"wait for model download and startup to finish, then retry."
)
self._report(f"[OK] Connected to faster TTS API at {self.api_url} (voice '{self.voice}')")
self._report(f"[INFO] The server silently falls back to its first configured voice if "
f"'{self.voice}' is not defined in its voice config (see README).")
# ------------------------------------------------------------------
# HTTP requests
# ------------------------------------------------------------------
def _request_pcm(self, text: str) -> bytes:
"""POST one sub-chunk and return raw 16-bit mono PCM bytes."""
url = f"{self.api_url}/v1/audio/speech"
payload = json.dumps({
"model": "tts-1",
"input": text,
"voice": self.voice,
"response_format": "pcm",
}).encode("utf-8")
request = urllib.request.Request(
url, data=payload, headers={"Content-Type": "application/json"}, method="POST")
try:
with urllib.request.urlopen(request, timeout=config.API_TIMEOUT) as response:
pcm = response.read()
except urllib.error.HTTPError as exc:
detail = ""
try:
detail = exc.read().decode("utf-8", errors="replace")
except Exception:
pass
raise RuntimeError(
f"Faster TTS server returned HTTP {exc.code}: {detail[:200]}"
) from exc
except urllib.error.URLError as exc:
raise RuntimeError(f"Faster TTS request failed: {exc.reason}") from exc
if not pcm:
raise RuntimeError("Faster TTS server returned empty audio")
return pcm
# ------------------------------------------------------------------
# Chunk generation
# ------------------------------------------------------------------
def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]:
"""Generate one audio chunk; returns its path in the chunks folder."""
try:
sub_chunks = split_into_chunks(text, max_words=config.CHUNK_SIZE)
if not sub_chunks:
raise RuntimeError("No text to synthesize")
pcm_parts: List[bytes] = []
with self._chunk_heartbeat(chunk_num):
for sub_text in sub_chunks:
pcm_parts.append(self._request_pcm(sub_text))
output_path = self._chunk_path(chunk_num, ".wav")
with wave.open(str(output_path), "wb") as wav_file:
wav_file.setnchannels(1)
wav_file.setsampwidth(2)
wav_file.setframerate(SAMPLE_RATE)
wav_file.writeframes(b"".join(pcm_parts))
logger.debug("Chunk %d generated (%d sub-chunks)", chunk_num, len(sub_chunks))
return str(output_path)
except ConversionCancelled:
raise
except Exception as exc:
logger.error("Faster chunk processing failed for chunk %d: %s", chunk_num, exc)
return None
|
