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
|
"""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):
super().__init__(chunks_dir, quiet=quiet)
# 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")[:200]
except Exception:
pass
raise RuntimeError(f"Faster TTS server returned HTTP {exc.code}: {detail}") 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
|
