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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
"""Shared TTS client plumbing: cancellation, retries, chunk bookkeeping."""
import contextlib
import logging
import random
import threading
import time
from pathlib import Path
from typing import Optional
from .. import config
logger = logging.getLogger(__name__)
class ConversionCancelled(Exception):
"""Raised when the run's cancel event is set (between requests)."""
class NonRetryableTTSError(RuntimeError):
"""A deterministic server-side request/config error; retrying cannot help.
Raised by clients for failures that will reproduce identically on every
attempt (missing server-side reference transcripts, unknown model ids,
missing model contracts, ...). process_chunk_with_retry skips its
remaining attempts and back-off sleeps for these and re-raises, so the
converter aborts with the actionable message instead of burning the
retry budget on a request that can never succeed. Subclasses RuntimeError
so except-handlers written for the plain HTTP-error case keep working.
"""
# How a run supplies its voice: a built-in CustomVoice speaker, by cloning
# a reference audio clip (the faster and audiocpp backends always clone
# server-side; only the Qwen client branches on this at request time), or
# designed from an instruction (Qwen's VoiceDesign model).
VOICE_MODE_CUSTOM = "custom_voice"
VOICE_MODE_CLONE = "voice_clone"
VOICE_MODE_DESIGN = "voice_design"
VOICE_MODES = (VOICE_MODE_CUSTOM, VOICE_MODE_CLONE, VOICE_MODE_DESIGN)
def resolve_request_seed() -> int:
"""Resolve the seed sent with every request.
Returns config.SEED as-is, or (with CONSTANT_SEED and SEED < 0) one
random value drawn per run, meant to be reused for every request so
the voice stays consistent across chunk boundaries. Without
CONSTANT_SEED, -1 is returned so the server re-samples the voice on
every generation.
"""
seed = config.SEED
if config.CONSTANT_SEED and seed < 0:
seed = random.randrange(2 ** 31)
return seed
class BaseTTSClient:
"""Shared chunk retry logic, heartbeat, and chunk file bookkeeping.
CHUNKS_DIR is the scratch folder the generated chunk files are written
to — provided by the converter that owns the run's folders, never a
module global, so concurrent runs (and tests) cannot step on each other.
"""
# Class-level defaults so a partially-constructed instance behaves like
# a plain console run (tests build clients via __new__).
cancel = None
quiet = False
def __init__(self, chunks_dir: Path, quiet: bool = False,
cancel: Optional[threading.Event] = None):
self.chunks_dir = Path(chunks_dir)
# Quiet silences console prints (the run view owns the screen).
self.quiet = bool(quiet)
# Set by the converter when the run is cancellable (the TUI run
# view): a threading.Event that, once set, aborts the run between
# requests (and interrupts retry back-off sleeps). Assigned here —
# before the subclass's connect logic runs — so a cancel pressed
# while the client is still connecting is not lost.
self.cancel = cancel
def _report(self, message: str) -> None:
"""Print a console line unless quiet (the run view owns the screen)."""
if not self.quiet:
print(message)
def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]:
"""Generate one audio chunk; returns its path in the chunks folder."""
raise NotImplementedError
def _cancel_requested(self) -> bool:
"""True when the run's cancel event has been set (if any)."""
return isinstance(self.cancel, threading.Event) \
and self.cancel.is_set()
def _check_cancelled(self) -> None:
"""Raise ConversionCancelled when the cancel event is set."""
if self._cancel_requested():
raise ConversionCancelled("Cancelled by user")
def _sleep(self, seconds: float) -> None:
"""Sleep SECONDS, cut short (raising) when the cancel event sets."""
if isinstance(self.cancel, threading.Event):
if self.cancel.wait(seconds):
raise ConversionCancelled("Cancelled by user")
else:
time.sleep(seconds)
def _chunk_path(self, chunk_num: int, suffix: str) -> Path:
"""Resolve the target path for a chunk, removing stale files first.
Any stale chunk file for this index is removed so a retry or extension
change can never leave two files matching chunk_NNNN.*.
"""
for stale in self.chunks_dir.glob(f"chunk_{chunk_num:04d}.*"):
try:
stale.unlink()
except OSError as exc:
logger.debug("Could not remove stale chunk file %s: %s", stale, exc)
return self.chunks_dir / f"chunk_{chunk_num:04d}{suffix}"
def process_chunk_with_retry(self, chunk_num: int, text: str) -> Optional[Path]:
"""Process a chunk with retry logic.
Returns the generated chunk file's path, or None when all attempts
failed. Raises ConversionCancelled when the run was cancelled.
NonRetryableTTSError is logged once and re-raised without consuming
the remaining attempts (the identical request can never succeed).
"""
for attempt in range(config.MAX_RETRIES):
self._check_cancelled()
try:
result = self.generate_chunk(text, chunk_num)
if result and Path(result).exists():
return Path(result)
logger.warning("Chunk %d attempt %d failed", chunk_num, attempt + 1)
except ConversionCancelled:
raise
except NonRetryableTTSError as exc:
logger.error("Chunk %d error (not retried): %s", chunk_num, exc)
raise
except Exception as exc:
logger.warning("Chunk %d attempt %d error: %s", chunk_num, attempt + 1, exc)
if attempt < config.MAX_RETRIES - 1:
sleep_time = 5 + (2 ** attempt)
logger.info("Waiting %ds before retry...", sleep_time)
self._sleep(sleep_time)
logger.error("Chunk %d failed after %d attempts", chunk_num, config.MAX_RETRIES)
return None
@contextlib.contextmanager
def _chunk_heartbeat(self, chunk_num: int):
"""Log a periodic "still working" record while a request generates."""
stop = threading.Event()
subject = f"Chunk {chunk_num}"
def _beat():
start = time.time()
while not stop.wait(config.HEARTBEAT_INTERVAL_SECONDS):
elapsed = time.time() - start
if self.quiet:
logger.info("%s still generating — %dm %ds elapsed",
subject, int(elapsed // 60), int(elapsed % 60))
else:
print(f"[...] {subject} still generating — "
f"{int(elapsed // 60)}m {int(elapsed % 60)}s elapsed",
flush=True)
thread = threading.Thread(target=_beat, daemon=True)
thread.start()
try:
yield
finally:
stop.set()
thread.join()
|
