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
|
"""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)."""
# 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):
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).
self.cancel = None
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.
"""
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 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()
|
