# ╔══════════════════════════════════════════════════════════════════════╗
# ║ XTTS v2 Voice Studio — Professional Edition v11 ║
# ║ SINGLE-FILE BUILD (app.py + requirements.txt + Dockerfile) ║
# ║ • CPU-only (16 vCPU / 104 GB RAM) • @GPU defined, never used ║
# ║ • Faster-Whisper • Soft Clipping • Real Progress • Cancel Jobs ║
# ║ • Audio Post-Editor • Full Mobile Responsive UI ║
# ╚══════════════════════════════════════════════════════════════════════╝
#
# Run: python app.py
# This file now contains everything the old Dockerfile / requirements.txt
# did. On start-up it (1) sets the environment, (2) installs system
# packages when running as root, (3) installs the pinned Python packages
# once, (4) creates the runtime folders, then (5) starts the server on
# port 7860. Set SKIP_BOOTSTRAP=1 to skip steps 2-3.
#
# Minimal Dockerfile that is still needed (a script cannot create its own
# container):
# FROM python:3.10-slim
# RUN useradd -m -u 1000 user && mkdir -p /data && chown 1000:1000 /data
# USER user
# WORKDIR /home/user/app
# COPY --chown=user app.py .
# EXPOSE 7860
# CMD ["python", "app.py"]
import os, sys, re, time, json, uuid, shutil, asyncio, threading
import subprocess, hashlib, tempfile, ctypes.util, traceback, zlib, math, collections, types
# ══════════════════════════════════════════════════════════════════════
# ZeroGPU COMPATIBILITY
# Hugging Face ZeroGPU Spaces kill the app at start-up with
# "No @spaces.GPU function detected" unless at least one function is
# decorated with @spaces.GPU. `import spaces` must happen BEFORE torch.
# The stub below is never called — the app itself still runs on CPU only.
# (If you switch the Space hardware to a normal CPU tier this is harmless.)
# ══════════════════════════════════════════════════════════════════════
try:
import spaces
@spaces.GPU
def _zerogpu_stub():
return True
ZEROGPU = True
except Exception as _e: # not on Spaces / package missing
spaces, ZEROGPU = None, False
print(f"[i] `spaces` not used: {_e}", flush=True)
# ══════════════════════════════════════════════════════════════════════
# HARDWARE PROFILE — 16 CPU cores · 104 GB RAM · NO GPU
# ══════════════════════════════════════════════════════════════════════
CPU_THREADS = int(os.environ.get("CPU_THREADS", "16"))
# ══════════════════════════════════════════════════════════════════════
# ENVIRONMENT (from Dockerfile ENV block)
# ══════════════════════════════════════════════════════════════════════
os.environ["PATH"] = os.path.expanduser("~/.local/bin") + os.pathsep + os.environ.get("PATH", "")
os.environ["COQUI_TOS_AGREED"] = "1" # accept Coqui TOS automatically
os.environ["TOKENIZERS_PARALLELISM"] = "false" # avoid tokenizer warnings
if not ZEROGPU:
os.environ["CUDA_VISIBLE_DEVICES"] = "" # GPU is never used
for _v in ("OMP_NUM_THREADS", "MKL_NUM_THREADS", "OPENBLAS_NUM_THREADS",
"NUMEXPR_NUM_THREADS", "VECLIB_MAXIMUM_THREADS"):
os.environ[_v] = str(CPU_THREADS)
# HF model cache lives on the persistent /data mount when it is writable
if os.path.isdir("/data") and os.access("/data", os.W_OK):
os.environ.setdefault("HF_HOME", "/data/huggingface")
else:
os.environ.setdefault("HF_HOME", os.path.expanduser("~/.cache/huggingface"))
# ══════════════════════════════════════════════════════════════════════
# @GPU — placeholder decorator (like `spaces.GPU`).
# It is defined only for compatibility and is INTENTIONALLY NEVER APPLIED
# anywhere in this file. Everything runs on CPU.
# ══════════════════════════════════════════════════════════════════════
def GPU(func=None, **_kwargs):
if func is None:
return lambda f: f
return func
# ══════════════════════════════════════════════════════════════════════
# REQUIREMENTS (from requirements.txt)
# ══════════════════════════════════════════════════════════════════════
REQUIREMENTS = """
coqui-tts==0.27.5
transformers==4.57.6
torch==2.8.0
torchaudio==2.8.0
soundfile>=0.12.0
scipy>=1.10.0
faster-whisper>=1.0.0
deep-translator>=1.11.0
fastapi>=0.111.0
uvicorn[standard]>=0.29.0
python-multipart>=0.0.9
"""
SYSTEM_PACKAGES = { # apt package -> "is it present?" check
"build-essential": lambda: shutil.which("gcc"),
"libsndfile1": lambda: ctypes.util.find_library("sndfile"),
"ffmpeg": lambda: shutil.which("ffmpeg"),
"git": lambda: shutil.which("git"),
"libgomp1": lambda: ctypes.util.find_library("gomp"),
}
def _sh(cmd):
print("[bootstrap] $", " ".join(cmd), flush=True)
return subprocess.run(cmd).returncode
def _bootstrap():
"""Everything the Dockerfile used to do at build time."""
if os.environ.get("SKIP_BOOTSTRAP") == "1":
return
if os.environ.get("SPACE_ID") and os.environ.get("FORCE_BOOTSTRAP") != "1":
print("[bootstrap] Running on Hugging Face Spaces — dependencies come "
"from requirements.txt, skipping runtime pip.", flush=True)
return
# ── System deps (only possible as root with apt) ──────────────────
missing = [p for p, ok in SYSTEM_PACKAGES.items() if not ok()]
if missing:
if hasattr(os, "geteuid") and os.geteuid() == 0 and shutil.which("apt-get"):
_sh(["apt-get", "update"])
_sh(["apt-get", "install", "-y", "--no-install-recommends"] + missing)
else:
print(f"[bootstrap] Not root — cannot apt-install: {', '.join(missing)} "
"(the pip wheels bundle what the app needs).", flush=True)
# ── Python deps (once per requirements version) ───────────────────
tag = hashlib.md5(REQUIREMENTS.encode()).hexdigest()[:10]
marker = os.path.join(os.path.dirname(os.path.abspath(__file__)), f".deps_ok_{tag}")
if os.path.exists(marker):
return
with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False) as f:
f.write(REQUIREMENTS)
req_path = f.name
try:
_sh([sys.executable, "-m", "pip", "install", "--no-cache-dir", "--upgrade", "pip"])
rc = _sh([sys.executable, "-m", "pip", "install", "--no-cache-dir", "-r", req_path])
if rc == 0:
open(marker, "w").close()
else:
print("[bootstrap] pip install failed — continuing anyway.", flush=True)
finally:
try: os.remove(req_path)
except OSError: pass
def _make_dirs():
"""Runtime folders (from Dockerfile mkdir lines) — best effort."""
here = os.path.dirname(os.path.abspath(__file__))
for d in ("/data/huggingface", "/data/voice_library", "/data/outputs",
os.path.join(here, "voice_library"), os.path.join(here, "outputs")):
try: os.makedirs(d, exist_ok=True)
except OSError: pass
_bootstrap()
_make_dirs()
# ══════════════════════════════════════════════════════════════════════
# APPLICATION
# ══════════════════════════════════════════════════════════════════════
import numpy as np
import soundfile as sf
import uvicorn
from fastapi import FastAPI, Form, File, UploadFile, HTTPException, BackgroundTasks
from fastapi.responses import HTMLResponse, FileResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
import torch
# ══ VOICE QUALITY ENGINE (embedded — keeps the app a single file) ══
VQ_SOURCE = r'''# ═══════════════════════════════════════════════════════════════════════
# VOICE QUALITY ENGINE
# 1. enhance_reference_set : clean / trim / select the best part of the sample
# 2. normalize_text : make text digestible for XTTS (esp. Arabic)
# 3. split_for_xtts : split by the model's REAL per-language char limit
# 4. synth_best : N candidate takes → pick the best one (ASR check)
# 5. assemble / finalize : natural pauses, no tanh distortion
# Pure numpy/scipy/soundfile — model & ASR are injected, so it is testable.
# ═══════════════════════════════════════════════════════════════════════
import os, re, io, json, math, hashlib, subprocess, tempfile, difflib, unicodedata
import numpy as np
import soundfile as sf
from scipy import signal
from scipy.ndimage import uniform_filter, binary_closing, binary_opening
REF_SR = 22050 # XTTS conditioning sample-rate
XTTS_SR = 24000 # XTTS output sample-rate
VQ_VERSION = "2"
# ── generic helpers ───────────────────────────────────────────────────
def _db(x): return 20.0 * np.log10(np.maximum(x, 1e-9))
def load_mono(path, sr=REF_SR, max_sec=180):
"""→ (float32 mono @sr, original_sample_rate)."""
try:
data, fs = sf.read(path, dtype="float32", always_2d=True)
except Exception:
tmp = tempfile.mktemp(suffix=".wav")
try:
subprocess.run(["ffmpeg", "-y", "-loglevel", "error", "-i", path,
"-ac", "1", tmp], check=True, timeout=120)
data, fs = sf.read(tmp, dtype="float32", always_2d=True)
finally:
if os.path.exists(tmp): os.remove(tmp)
x = data.mean(axis=1).astype(np.float32)
x = x[: int(fs * max_sec)]
if fs != sr:
g = math.gcd(int(fs), int(sr))
x = signal.resample_poly(x, int(sr) // g, int(fs) // g).astype(np.float32)
return x, int(fs)
def frame_db(x, sr, win=0.025, hop=0.010):
w, h = int(sr * win), int(sr * hop)
if len(x) < w:
x = np.pad(x, (0, w - len(x)))
n = 1 + (len(x) - w) // h
idx = np.arange(w)[None, :] + h * np.arange(n)[:, None]
rms = np.sqrt(np.mean(x[idx] ** 2, axis=1) + 1e-12)
return _db(rms), w, h
def _highpass(x, sr, fc=70.0, order=4):
sos = signal.butter(order, fc, btype="highpass", fs=sr, output="sos")
return signal.sosfiltfilt(sos, x).astype(np.float32)
def _fade(x, sr, fin_ms=8, fout_ms=25):
x = x.copy()
a, b = int(sr * fin_ms / 1000), int(sr * fout_ms / 1000)
if a and len(x) > a: x[:a] *= np.linspace(0, 1, a, dtype=np.float32)
if b and len(x) > b: x[-b:] *= np.linspace(1, 0, b, dtype=np.float32)
return x
def soft_limit(x, ceiling=0.89, knee=0.70):
"""Transparent below `knee`, smooth tanh-shaped compression above it
(no colouring of normal signal — unlike a global tanh(x*1.8))."""
y = x.copy()
a = np.abs(y)
m = a > knee
if m.any():
over = (a[m] - knee) / max(ceiling - knee, 1e-6)
y[m] = np.sign(y[m]) * (knee + (ceiling - knee) * np.tanh(over))
return y.astype(np.float32)
# ═══════════════════════════════════════════════════════════════════════
# 1. REFERENCE ENHANCEMENT
# ═══════════════════════════════════════════════════════════════════════
def spectral_denoise(x, sr, strength=1.0):
"""Spectral-subtraction gate; noise profile = quietest 15 % of frames.
Smoothed in time+frequency to avoid 'musical noise'."""
nper, nov = 1024, 768
_, _, X = signal.stft(x, sr, nperseg=nper, noverlap=nov)
mag = np.abs(X)
fr = _db(mag.mean(axis=0))
quiet = fr <= np.percentile(fr, 15)
if quiet.sum() < 3:
return x
mu = mag[:, quiet].mean(axis=1, keepdims=True)
alpha = 1.0 + 1.2 * strength
gain = np.clip((mag - alpha * mu) / (mag + 1e-9), 0.0, 1.0)
floor = 10 ** (-(10 + 10 * strength) / 20) # −20 … −30 dB
gain = np.maximum(gain, floor)
gain = uniform_filter(gain, size=(3, 5), mode="nearest")
_, y = signal.istft(X * gain, sr, nperseg=nper, noverlap=nov)
y = y[: len(x)]
if len(y) < len(x): y = np.pad(y, (0, len(x) - len(y)))
return y.astype(np.float32)
def _voiced_mask(db, min_gap=15, min_seg=30):
noise, peak = np.percentile(db, 10), np.percentile(db, 95)
thr = max(noise + 10.0, peak - 28.0)
v = db > thr
v = binary_closing(v, structure=np.ones(min_gap, bool)) # bridge ≤150 ms
v = binary_opening(v, structure=np.ones(min_seg, bool)) # drop < 300 ms
return v
def _segments(mask):
d = np.diff(np.concatenate([[0], mask.astype(np.int8), [0]]))
return list(zip(np.where(d == 1)[0], np.where(d == -1)[0]))
def _analyze(x_native_norm, fs_native, sr=REF_SR):
"""Diagnostics on a (peak-normalised) signal."""
db, w, h = frame_db(x_native_norm, sr)
noise, peak = float(np.percentile(db, 10)), float(np.percentile(db, 95))
f, p = signal.welch(x_native_norm, sr, nperseg=2048)
cum = np.cumsum(p) / max(p.sum(), 1e-12)
rolloff = float(f[np.searchsorted(cum, 0.995)])
return dict(noise_floor_db=round(noise, 1), snr_db=round(peak - noise, 1),
bandwidth_hz=int(min(rolloff, fs_native / 2)), native_sr=int(fs_native))
def _stretches(mask, keep_gap=90):
"""Group voiced segments separated by pauses ≤ 0.9 s into 'stretches', so natural
phrasing (sentence-to-sentence pauses) stays exactly as recorded."""
out = []
for s, e in _segments(mask):
s, e = int(s), int(e)
if out and s - out[-1][1] <= keep_gap: out[-1][1] = e
else: out.append([s, e])
return out
def _split_stretch(db, s, e, max_frames):
"""Cut an over-long stretch at its deepest-energy frames (= inside pauses).
Nothing is ever joined afterwards, so a cut can never become an audible splice."""
smooth = uniform_filter(db, size=7, mode="nearest")
cuts, pos = [], s
while e - pos > max_frames:
lo, hi = pos + int(0.55 * max_frames), pos + max_frames
c = lo + int(np.argmin(smooth[lo:hi]))
cuts.append((pos, c)); pos = c
cuts.append((pos, e))
return cuts
def _score_window(x, db, w, h, a, b, mask):
seg, v = db[a:b], mask[a:b]
snr = float(seg[v].mean() - np.percentile(seg, 10)) if v.any() else 0.0
sd = float(seg[v].std()) if v.sum() > 3 else 12.0
clip = float(np.mean(np.abs(x[a * h: b * h + w]) > 0.98))
sp = float(v.sum() * 0.01)
return snr - 0.5 * sd - 300.0 * clip + 0.5 * min(sp, 20.0), sp
def _finish_segment(x, sr):
"""Level-match one CONTIGUOUS segment (voiced RMS → −20 dBFS) and fade its ends
inside the surrounding silence."""
z = x.astype(np.float32).copy()
db, w, h = frame_db(z, sr)
m = _voiced_mask(db)
vs = [z[s * h: e * h + w] for s, e in _segments(m)]
lvl = float(np.sqrt(np.mean(np.square(np.concatenate(vs)))) + 1e-9) if vs else float(np.sqrt(np.mean(z ** 2)) + 1e-9)
z = soft_limit(z * (10 ** (-20 / 20) / lvl), 0.95, 0.85)
return _fade(z, sr, 15, 45)
def enhance_reference(path, strength=1.0, denoise=True, max_seg_sec=28.0, min_speech_sec=1.5):
"""Clean one recording and return CONTIGUOUS candidate segments (no cutting-and-
pasting): → (candidates [{wav, speech, score}], report)."""
raw, fs = load_mono(path)
if len(raw) < 0.5 * REF_SR:
raise ValueError("العينة قصيرة جداً (أقل من نصف ثانية).")
dur_in = len(raw) / REF_SR
clip_pct = float(np.mean(np.abs(raw) >= 0.995) * 100)
raw = raw - float(np.mean(raw))
x = raw / (float(np.max(np.abs(raw))) + 1e-9) * 0.7
before = _analyze(x, fs)
x = _highpass(x, REF_SR, 70)
did_denoise = False
if denoise and before["noise_floor_db"] > -62:
s = float(np.clip((before["noise_floor_db"] + 62) / 20.0, 0.35, 1.0)) * strength
x = spectral_denoise(x, REF_SR, s); did_denoise = True
after = _analyze(x, fs)
db, w, h = frame_db(x, REF_SR)
mask = _voiced_mask(db)
speech_sec_all = float(mask.sum() * 0.01)
cands = []
lead, tail = int(0.15 * REF_SR), int(0.30 * REF_SR)
for s, e in _stretches(mask):
for a, b in _split_stretch(db, s, e, int(max_seg_sec * 100)):
sc, sp = _score_window(x, db, w, h, a, b, mask)
if sp < min_speech_sec: continue
i0 = max(0, a * h - (lead if a == s else 0))
i1 = min(len(x), b * h + w + (tail if b == e else 0))
cands.append(dict(wav=_finish_segment(x[i0:i1], REF_SR), speech=sp, score=sc))
if not cands: # no clear phrasing: keep one contiguous voiced span
idx = np.where(mask)[0]
i0, i1 = (max(0, idx[0] * h - lead), min(len(x), idx[-1] * h + w + tail)) if len(idx) else (0, len(x))
i1 = min(i1, i0 + int(max_seg_sec * REF_SR))
cands.append(dict(wav=_finish_segment(x[i0:i1], REF_SR), speech=speech_sec_all, score=0.0))
warns, tips = [], []
if speech_sec_all < 6:
warns.append(f"الكلام الفعلي في العينة {speech_sec_all:.1f} ث فقط."); tips.append("استخدم عينة من 10–25 ثانية كلام متصل.")
if clip_pct > 0.5:
warns.append(f"تشويه قص (clipping) {clip_pct:.1f}٪ — لا يمكن إصلاحه."); tips.append("أعد التسجيل بمستوى أقل.")
if before["snr_db"] < 18:
warns.append(f"ضجيج عالٍ (SNR≈{before['snr_db']:.0f} dB)."); tips.append("سجّل في غرفة هادئة قريباً من الميكروفون.")
if before["bandwidth_hz"] < 6000:
warns.append(f"جودة ترددية منخفضة (حتى {before['bandwidth_hz']} Hz) — غالباً هاتف/ضغط شديد."); tips.append("استخدم WAV/FLAC أو تسجيلاً بجودة أعلى.")
if before["native_sr"] < 16000:
warns.append(f"معدل العينة الأصلي {before['native_sr']} Hz منخفض.")
score = 100.0 - max(0, 20 - before["snr_db"]) * 1.5 - min(30, clip_pct * 10) \
- max(0, 6 - speech_sec_all) * 5 - (15 if before["bandwidth_hz"] < 6000 else 0)
report = dict(duration_in=round(dur_in, 1), speech_sec=round(speech_sec_all, 1), clipping_pct=round(clip_pct, 2),
denoised=did_denoise, before=before, after=after, score=int(np.clip(score, 0, 100)),
warnings=warns, tips=tips, candidates=len(cands))
return cands, report
def _file_sha(path):
h = hashlib.sha1()
with open(path, "rb") as f:
for b in iter(lambda: f.read(1 << 20), b""): h.update(b)
return h.hexdigest()
def enhance_reference_set(paths, cache_dir, strength=1.0, target_sec=24.0,
max_segments=4, denoise=True):
"""Clean the recordings and pick the best CONTIGUOUS segments (≤ max_segments, ≈target_sec
of speech in total). Segments are written as SEPARATE files and never concatenated —
each one's conditioning latents are computed on their own and averaged by the caller.
→ ([segment_wav_paths], [per-file reports], summary)."""
os.makedirs(cache_dir, exist_ok=True)
key = hashlib.sha1(("|".join(_file_sha(p) for p in paths) +
f"|{strength}|{target_sec}|{max_segments}|{denoise}|{VQ_VERSION}").encode()).hexdigest()[:20]
meta = os.path.join(cache_dir, f"ref_{key}.json")
if os.path.exists(meta):
d = json.load(open(meta, encoding="utf-8"))
if d["paths"] and all(os.path.exists(p) for p in d["paths"]):
return d["paths"], d["reports"], d["summary"]
pool, reports = [], []
for fi, p in enumerate(paths):
try:
cands, rep = enhance_reference(p, strength, denoise)
rep["file"] = os.path.basename(p)[-40:]
reports.append(rep)
for c in cands: c["fscore"] = rep["score"]; pool.append(c)
except Exception as e:
reports.append(dict(file=os.path.basename(p)[-40:], error=str(e), score=0,
warnings=[f"تعذّرت معالجة الملف: {e}"], tips=[]))
if not pool:
raise ValueError("تعذّرت معالجة كل ملفات العينة: " + "; ".join(r.get("error", "?") for r in reports))
chosen, total = [], 0.0
for c in sorted(pool, key=lambda c: c["score"] + 0.1 * c["fscore"], reverse=True):
if len(chosen) >= max_segments or total >= target_sec: break
chosen.append(c); total += c["speech"]
out_paths = []
for i, c in enumerate(chosen):
op = os.path.join(cache_dir, f"ref_{key}_{i}.wav")
sf.write(op, c["wav"], REF_SR, subtype="PCM_16"); out_paths.append(op)
good = [r for r in reports if "error" not in r]
best = max(good, key=lambda r: r["score"])
summary = dict(score=int(np.mean([r["score"] for r in good])),
duration=round(sum(len(c["wav"]) for c in chosen) / REF_SR, 1),
speech=round(total, 1), segments=len(chosen), files=len(paths), used=len(good),
warnings=[w for r in good for w in r["warnings"]][:6],
tips=list(dict.fromkeys(t for r in good for t in r["tips"]))[:4],
snr_before=best["before"]["snr_db"], snr_after=best["after"]["snr_db"],
denoised=any(r["denoised"] for r in good))
json.dump(dict(paths=out_paths, reports=reports, summary=summary), open(meta, "w", encoding="utf-8"), ensure_ascii=False)
return out_paths, reports, summary
# ═══════════════════════════════════════════════════════════════════════
# 2. TEXT NORMALISATION
# ═══════════════════════════════════════════════════════════════════════
_AR_DIAC = re.compile("[\u0610-\u061A\u064B-\u065F\u0670\u06D6-\u06ED]")
_INVIS = re.compile("[\u200b\u200c\u200d\u200e\u200f\u202a-\u202e\u2066-\u2069\ufeff]")
_EASTERN = str.maketrans("٠١٢٣٤٥٦٧٨٩۰۱۲۳۴۵۶۷۸۹", "01234567890123456789")
def normalize_text(text, lang="ar", strip_diacritics=True):
t = unicodedata.normalize("NFKC" if lang.startswith("ar") else "NFC", text)
t = _INVIS.sub("", t).translate(_EASTERN)
if lang.startswith("ar"):
t = t.replace("\u0640", "")
if strip_diacritics: t = _AR_DIAC.sub("", t)
t = re.sub(r"[“”„«»\"‹›]", "", t)
t = t.replace("’", "'").replace("‘", "'")
t = re.sub(r"[*_#`~|^<>\[\]{}]", " ", t)
comma = "، " if lang.startswith("ar") else ", "
t = re.sub(r"\s*[—–]\s*|\s+-\s+", comma, t)
t = re.sub(r"\s*[()]\s*", comma, t)
t = re.sub(r"\.{4,}", "...", t)
t = re.sub(r"([!?؟])\1+", r"\1", t)
t = re.sub(r"[ \t]+", " ", t)
t = re.sub(r"\s*([,،;؛:])\s*(?=[,،;؛:.!?؟])", r"\1", t) # ", ," / ", ."
t = re.sub(r"[,،]\s*([.!?؟])", r"\1", t)
t = re.sub(r"\n{3,}", "\n\n", t).strip()
if t and not re.search(r"[.!?؟…。！？]$", t):
t += "。" if lang in ("zh", "zh-cn", "ja") else "."
return t
# ═══════════════════════════════════════════════════════════════════════
# 3. SPLITTING (real per-language limits)
# ═══════════════════════════════════════════════════════════════════════
DEFAULT_CHAR_LIMITS = {"en": 250, "de": 253, "fr": 273, "es": 239, "it": 213, "pt": 203,
"pl": 224, "zh": 82, "ar": 166, "cs": 186, "ru": 182, "nl": 251,
"tr": 226, "ja": 71, "hu": 224, "ko": 95, "hi": 150}
PAUSE_MS = {".": 400, "!": 380, "?": 420, "؟": 420, "…": 520, "。": 400, "！": 380, "？": 420,
"none": 300, "clause": 170, "para": 760}
_CLAUSE = re.compile(r"(?<=[,،;؛:])\s+")
def _split_long(s, limit):
"""Split an over-long sentence on clause marks, else on spaces."""
if len(s) <= limit: return [s]
mid = len(s) // 2
cands = [m.end() for m in _CLAUSE.finditer(s) if 15 <= m.end() <= len(s) - 15]
if not cands:
cands = [m.end() for m in re.finditer(r"\s+", s) if 15 <= m.end() <= len(s) - 15]
if not cands:
return [s[:limit]] + _split_long(s[limit:], limit)
cut = min(cands, key=lambda c: abs(c - mid))
return _split_long(s[:cut].strip(), limit) + _split_long(s[cut:].strip(), limit)
def split_for_xtts(text, lang="ar", char_limits=None, safety=0.85, min_chunk=28):
"""→ [(chunk_text, pause_after_ms)] (pause 0 for the very last chunk)."""
lim = dict(DEFAULT_CHAR_LIMITS); lim.update(char_limits or {})
base = "zh" if lang.startswith("zh") else lang.split("-")[0]
limit = max(30, int(lim.get(base, 250) * safety))
out = []
paras = [p for p in re.split(r"\n\s*\n", text.strip()) if p.strip()]
for pi, para in enumerate(paras):
sents = [s.strip() for s in re.split(r"(?<=[.!?؟…])\s+|(?<=[。！？])|\n", para) if s and s.strip()]
items = []
for s in sents:
end = s[-1] if s[-1] in PAUSE_MS else "none"
pieces = _split_long(s, limit)
for j, pc in enumerate(pieces):
items.append((pc, PAUSE_MS[end] if j == len(pieces) - 1 else PAUSE_MS["clause"]))
# merge tiny fragments into a neighbour (very short chunks sound odd)
merged = []
for txt, pz in items:
if merged and len(merged[-1][0]) < min_chunk and len(merged[-1][0]) + 1 + len(txt) <= limit:
merged[-1] = (merged[-1][0] + " " + txt, pz)
else:
merged.append((txt, pz))
if len(merged) > 1 and len(merged[-1][0]) < min_chunk and len(merged[-2][0]) + 1 + len(merged[-1][0]) <= limit:
merged[-2] = (merged[-2][0] + " " + merged[-1][0], merged[-1][1]); merged.pop()
if merged and pi < len(paras) - 1:
merged[-1] = (merged[-1][0], PAUSE_MS["para"])
out.extend(merged)
if out: out[-1] = (out[-1][0], 0)
return out or [(text.strip(), 0)]
# ═══════════════════════════════════════════════════════════════════════
# 4. TAKE SELECTION
# ═══════════════════════════════════════════════════════════════════════
def trim_take(wav, sr=XTTS_SR, rel_db=-42.0, lead_ms=40, tail_ms=70):
wav = np.asarray(wav, np.float32).reshape(-1)
if len(wav) < int(0.1 * sr): return wav
db, w, h = frame_db(wav, sr)
idx = np.where(db > db.max() + rel_db)[0]
if len(idx) == 0: return wav
s = max(0, idx[0] * h - int(lead_ms * sr / 1000))
e = min(len(wav), idx[-1] * h + w + int(tail_ms * sr / 1000))
return _fade(wav[s:e], sr, 8, 25)
_NORM_RE = re.compile(r"[^\w]+", re.UNICODE)
def norm_for_compare(s):
s = _AR_DIAC.sub("", unicodedata.normalize("NFKC", s)).replace("\u0640", "")
s = re.sub("[إأآ]", "ا", s).replace("ى", "ي").replace("ة", "ه")
return _NORM_RE.sub("", s.lower().translate(_EASTERN))
def text_similarity(a, b):
a, b = norm_for_compare(a), norm_for_compare(b)
if not a or not b: return 0.0
return difflib.SequenceMatcher(None, a, b, autojunk=False).ratio()
CPS_RANGE = {"zh": (1.5, 14.0), "ja": (2.0, 16.0), "ko": (3.0, 20.0)} # chars/sec sanity bounds
def take_info(wav, sr, text, asr_fn=None, lang="ar"):
dur = len(wav) / sr
peak = float(np.max(np.abs(wav))) if len(wav) else 0.0
cps = len(norm_for_compare(text)) / max(dur, 1e-3)
lo, hi = CPS_RANGE.get(lang.split("-")[0], (4.0, 28.0))
sim = None
if asr_fn is not None and dur > 0.2:
try: sim = text_similarity(text, asr_fn(wav, sr))
except Exception: sim = None
ok = peak > 0.02 and lo <= cps <= hi
return dict(dur=dur, cps=cps, peak=peak, sim=sim, ok=ok, secs=None,
clip=float(np.mean(np.abs(wav) > 0.99)) if len(wav) else 0.0)
def _take_score(i, med):
s = (i["sim"] * 100.0) if i["sim"] is not None else 60.0
if i.get("secs") is not None:
s += 80.0 * (i["secs"] - 0.6) # speaker-identity consistency
s -= 25.0 * abs(math.log(max(i["dur"], 1e-3) / max(med, 1e-3)))
s -= 300.0 * i["clip"]
s -= 0.0 if i["ok"] else 30.0
return s
def synth_best(gen_fn, text, attempts=1, asr_fn=None, sr=XTTS_SR, seed=1234,
accept_sim=0.90, secs_fn=None, lang="ar", guard_retries=2, min_secs=0.45):
"""gen_fn(seed:int) → float32 wav. Returns (wav, info, n_generated).
attempts==1 : cheap self-healing guard — regenerate (≤ guard_retries) only if the
take is obviously broken (silent / absurd speech-rate / clipped / off-speaker).
attempts>1 : best-of-N ranked by ASR similarity + speaker similarity + duration.
`secs_fn` (speaker-encoder cosine similarity) is a single extra forward pass — cheap
enough to run even when attempts==1, and it is the only check that catches a take
that drifted away from the reference speaker's timbre (silence/clip/rate checks don't)."""
multi = attempts > 1
takes, k = [], 0
while True:
wav = trim_take(gen_fn(seed + 7919 * k), sr)
info = take_info(wav, sr, text, asr_fn if multi else None, lang)
if secs_fn is not None and len(wav) > int(0.3 * sr):
try: info["secs"] = float(secs_fn(wav, sr))
except Exception: info["secs"] = None
takes.append((wav, info)); k += 1
if multi:
if k >= attempts: break
if info["ok"] and info["sim"] is not None and info["sim"] >= accept_sim: break
else:
secs_ok = info.get("secs") is None or info["secs"] >= min_secs
if (info["ok"] and info["clip"] < 0.005 and secs_ok) or k > guard_retries: break
if len(takes) == 1: return takes[0][0], takes[0][1], 1
med = float(np.median([i["dur"] for _, i in takes]))
best = max(takes, key=lambda t: _take_score(t[1], med))
return best[0], best[1], len(takes)
# ═══════════════════════════════════════════════════════════════════════
# 5. ASSEMBLY / FINALISATION
# ═══════════════════════════════════════════════════════════════════════
def assemble(takes, pauses_ms, sr=XTTS_SR, pause_scale=1.0, seed=7, jitter=0.12):
"""Join takes with punctuation-aware pauses (seeded jitter, ±12 %)."""
rng = np.random.default_rng(seed)
parts = []
for wav, pz in zip(takes, pauses_ms):
parts.append(np.asarray(wav, np.float32))
if pz > 0:
ms = pz * pause_scale * (1.0 + rng.uniform(-jitter, jitter))
parts.append(np.zeros(int(sr * ms / 1000), np.float32))
return np.concatenate(parts) if parts else np.zeros(sr, np.float32)
def finalize_audio(wav, sr=XTTS_SR, target_rms_db=-19.0):
wav = _highpass(np.asarray(wav, np.float32), sr, 60, 2)
db, w, h = frame_db(wav, sr)
v = db > max(np.percentile(db, 10) + 10, np.percentile(db, 95) - 30)
if v.any():
lvl = 10 ** (np.percentile(db[v], 50) / 20)
gain = np.clip(10 ** (target_rms_db / 20) / max(lvl, 1e-6), 0.25, 6.0)
wav = wav * gain
return soft_limit(wav, 0.95, 0.80)
# ═══════════════════════════════════════════════════════════════════════
# 6. DEPTH / SPACE ("مجسّم") — stereo room, body, air, room-tone
# XTTS output is DRY, MONO, 24 kHz and gapped with digital silence: that
# is what reads as "flat". This adds the missing dimensions:
# · body/presence EQ · harmonic "air" above XTTS's 12 kHz limit
# · stereo early-reflections + decorrelated late tail (IR convolution)
# · continuous room-tone bed (no dead digital silence between phrases)
# ═══════════════════════════════════════════════════════════════════════
SPACE_PRESETS = {
# rt60 predelay er damp wet_db body presence air tone_db
"studio": dict(rt60=0.30, pre=8, er=0.55, damp=0.70, wet_db=-17, body=1.5, pres=1.2, air=1.0, tone=-68),
"room": dict(rt60=0.50, pre=14, er=0.60, damp=0.50, wet_db=-14, body=1.5, pres=1.0, air=0.8, tone=-64),
"hall": dict(rt60=0.95, pre=22, er=0.45, damp=0.40, wet_db=-12, body=1.0, pres=0.5, air=0.8, tone=-62),
}
DEPTH_SR = 48000
def _rbj(kind, fs, f0, gain_db, Q=0.9):
A = 10 ** (gain_db / 40.0); w0 = 2 * math.pi * f0 / fs
cw, sw = math.cos(w0), math.sin(w0); alpha = sw / (2 * Q)
if kind == "peak":
b = [1 + alpha * A, -2 * cw, 1 - alpha * A]; a = [1 + alpha / A, -2 * cw, 1 - alpha / A]
else:
sA = 2 * math.sqrt(A) * alpha
if kind == "hishelf":
b = [A * ((A + 1) + (A - 1) * cw + sA), -2 * A * ((A - 1) + (A + 1) * cw), A * ((A + 1) + (A - 1) * cw - sA)]
a = [(A + 1) - (A - 1) * cw + sA, 2 * ((A - 1) - (A + 1) * cw), (A + 1) - (A - 1) * cw - sA]
else: # loshelf
b = [A * ((A + 1) - (A - 1) * cw + sA), 2 * A * ((A - 1) - (A + 1) * cw), A * ((A + 1) - (A - 1) * cw - sA)]
a = [(A + 1) + (A - 1) * cw + sA, -2 * ((A - 1) + (A + 1) * cw), (A + 1) + (A - 1) * cw - sA]
return np.array(b) / a[0], np.array(a) / a[0]
def _eq(x, fs, kind, f0, gain_db, Q=0.9):
if abs(gain_db) < 0.05: return x
b, a = _rbj(kind, fs, f0, gain_db, Q)
return signal.lfilter(b, a, x)
def _band_split_noise(n, fs, rng, rt60, damp):
"""Noise with frequency-dependent decay: lows ring longer, highs die first."""
noise = rng.standard_normal(n)
t = np.arange(n) / fs
out = np.zeros(n)
for lo, hi, k in ((None, 500, 1.15), (500, 3000, 1.0), (3000, None, max(0.22, 1.0 - 0.7 * damp))):
if lo is None: sos = signal.butter(2, hi, "lowpass", fs=fs, output="sos")
elif hi is None: sos = signal.butter(2, lo, "highpass", fs=fs, output="sos")
else: sos = signal.butter(2, [lo, hi], "bandpass", fs=fs, output="sos")
out += signal.sosfilt(sos, noise) * np.exp(-6.9078 * t / (rt60 * k))
return out
def make_room_ir(fs=DEPTH_SR, rt60=0.5, pre=12, er=0.5, damp=0.5, seed=1):
"""Synthetic stereo room impulse response WITHOUT the direct sound.
→ (irL, irR), each unit-energy. Early reflections differ per ear."""
rng = np.random.default_rng(seed)
n = int(fs * min(max(rt60 * 1.2, 0.25), 1.8))
sos_hp = signal.butter(2, 180, "highpass", fs=fs, output="sos") # keep bass centred
er_ms = [6.3, 9.7, 14.1, 19.6, 26.2, 33.5, 41.8]
er_g = [0.9, 0.75, 0.60, 0.50, 0.40, 0.30, 0.22]
irs = []
for ch in range(2):
tail = _band_split_noise(n, fs, rng, rt60, damp)
tail *= np.minimum(1.0, np.arange(n) / (0.025 * fs)) # bloom after the ERs
p = int(fs * pre / 1000)
ir = np.zeros(n); ir[p:] = tail[: n - p]
ir = signal.sosfilt(sos_hp, ir)
e = np.sqrt(np.sum(ir ** 2)) + 1e-12
ir = ir / e # unit-energy tail
sos_lp = signal.butter(2, 7000, "lowpass", fs=fs, output="sos")
er_ir = np.zeros(n)
for tm, g in zip(er_ms, er_g):
j = tm * (1.0 + (0.16 if ch else -0.16)) + rng.uniform(-0.8, 0.8) # L/R differ
idx = int(fs * (pre + j) / 1000)
if idx < n: er_ir[idx] += g * er * (1 if rng.random() > 0.35 else -1)
er_ir = signal.sosfilt(sos_lp, er_ir)
er_ir = signal.sosfilt(sos_hp, er_ir) * 0.12 # ERs relative to tail
irs.append((ir + er_ir).astype(np.float64))
return irs[0], irs[1]
def _rms_db(x):
return 20 * math.log10(float(np.sqrt(np.mean(np.square(x))) + 1e-9))
def _exciter(x, fs, amount):
"""Regenerate 'air' above XTTS's ~12 kHz ceiling from its 5.5 kHz+ band."""
if amount < 0.02: return np.zeros_like(x)
hp = signal.sosfilt(signal.butter(4, 5500, "highpass", fs=fs, output="sos"), x)
y = np.tanh(4.0 * hp)
y = signal.sosfilt(signal.butter(4, 9000, "highpass", fs=fs, output="sos"), y)
ry, rh = np.sqrt(np.mean(y ** 2)) + 1e-9, np.sqrt(np.mean(hp ** 2)) + 1e-9
return y / ry * rh * 0.22 * amount
def _room_tone(n, fs, rng, level_db):
"""Pink-ish, low-passed, decorrelated L/R bed — the 'air of the room'."""
out = []
sos = signal.butter(2, 5500, "lowpass", fs=fs, output="sos")
for _ in range(2):
w = rng.standard_normal(n); F = np.fft.rfft(w); F /= np.sqrt(np.arange(1, len(F) + 1))
p = np.fft.irfft(F, n=n); p = signal.sosfilt(sos, p)
p = p / (np.sqrt(np.mean(p ** 2)) + 1e-9) * 10 ** (level_db / 20)
out.append(p)
fade = int(0.35 * fs)
env = np.ones(n); env[:fade] = np.linspace(0, 1, fade); env[-fade:] = np.linspace(1, 0, fade)
return out[0] * env, out[1] * env
def spatialize(wav, sr, preset="studio", amount=0.5, seed=7):
"""mono float wav @sr → (stereo float32 [n,2], DEPTH_SR). amount 0…1 (0 = bypass)."""
wav = np.asarray(wav, np.float64).reshape(-1)
a = float(np.clip(amount, 0.0, 1.0))
if a < 0.02 or len(wav) == 0:
return np.stack([wav, wav], 1).astype(np.float32), sr
P = SPACE_PRESETS.get(preset, SPACE_PRESETS["studio"])
fs = DEPTH_SR
if sr != fs:
g = math.gcd(int(sr), fs); wav = signal.resample_poly(wav, fs // g, int(sr) // g)
ref_lvl = _rms_db(wav)
rng = np.random.default_rng(seed)
# tone: warmth / de-mud / presence / air
x = _eq(wav, fs, "loshelf", 190, P["body"] * a, 0.8)
x = _eq(x, fs, "peak", 380, -1.0 * a, 0.9)
x = _eq(x, fs, "peak", 3200, P["pres"] * a, 0.8)
x = x + _exciter(x, fs, a * P["air"])
# space: stereo IR convolution, tail rings out after the last word
irL, irR = make_room_ir(fs, P["rt60"], P["pre"], P["er"], P["damp"], seed)
tail = int(fs * min(P["rt60"] * 0.9, 1.2))
xp = np.concatenate([x, np.zeros(tail)])
wetL = signal.fftconvolve(xp, irL)[: len(xp)]
wetR = signal.fftconvolve(xp, irR)[: len(xp)]
wg = 10 ** (P["wet_db"] / 20) * a * 2.2 * (np.sqrt(np.mean(x ** 2)) / (np.sqrt(np.mean(wetL ** 2)) + 1e-9)) * 1.0
L, R = xp + wg * wetL, xp + wg * wetR
# room tone under everything (incl. pauses)
tl, tr = _room_tone(len(xp), fs, rng, P["tone"] + 6 * (1 - a))
L, R = L + tl, R + tr
st = np.stack([L, R], 1)
st *= 10 ** ((ref_lvl - _rms_db(st)) / 20) # keep the original loudness
st[-int(0.15 * fs):] *= np.linspace(1, 0, int(0.15 * fs))[:, None]
st = soft_limit(st, 0.95, 0.80)
return st.astype(np.float32), fs
'''
vq = types.ModuleType("voice_quality")
exec(compile(VQ_SOURCE, "voice_quality", "exec"), vq.__dict__)
torch.set_num_threads(CPU_THREADS) # use all 16 cores for inference
try:
torch.set_num_interop_threads(2)
except RuntimeError:
pass
# ── Persistent Storage ────────────────────────────────────────────────
PERSISTENT_DIR = "/data" if os.path.exists("/data") else os.path.join(os.getcwd(), "app_data")
VOICE_LIB = os.path.join(PERSISTENT_DIR, "voice_library")
OUTPUT_DIR = os.path.join(PERSISTENT_DIR, "outputs")
HISTORY_FILE = os.path.join(PERSISTENT_DIR, "history.json")
MAX_CHARS = 5000
for d in [VOICE_LIB, OUTPUT_DIR]:
os.makedirs(d, exist_ok=True)
HF_TOKEN = os.environ.get("HF_TOKEN")
DATASET_REPO_ID = os.environ.get("DATASET_REPO_ID")
try:
from huggingface_hub import HfApi, snapshot_download
HUB_AVAILABLE = True
if HF_TOKEN and DATASET_REPO_ID:
print("[*] Restoring data from Hugging Face Hub …")
try:
snapshot_download(repo_id=DATASET_REPO_ID, repo_type="dataset",
local_dir=PERSISTENT_DIR, token=HF_TOKEN)
print("[✓] Data restoration complete.")
except Exception as e:
print(f"[!] Warning: Could not download backup ({e})")
except ImportError:
HUB_AVAILABLE = False
_backup_lock = threading.Lock()
_backup_state = {"running": False, "pending": False, "last": 0.0}
BACKUP_MIN_INTERVAL = 90 # seconds between uploads
def trigger_cloud_backup():
"""Single-flight + coalesced + rate-limited. (Previously every save/delete spawned
its own thread uploading the whole data folder, so uploads could pile up.)"""
if not (HUB_AVAILABLE and HF_TOKEN and DATASET_REPO_ID): return
with _backup_lock:
if _backup_state["running"]:
_backup_state["pending"] = True
return
_backup_state["running"] = True
def _run():
while True:
wait = BACKUP_MIN_INTERVAL - (time.time() - _backup_state["last"])
if wait > 0: time.sleep(wait)
try:
HfApi(token=HF_TOKEN).upload_folder(
folder_path=PERSISTENT_DIR, repo_id=DATASET_REPO_ID, repo_type="dataset",
commit_message=f"Auto-backup: {int(time.time())}",
ignore_patterns=["*.tmp", "*.corrupt", ".cache/*"])
except Exception as e:
print(f"[!] Cloud backup failed: {e}", flush=True)
with _backup_lock:
_backup_state["last"] = time.time()
if _backup_state["pending"]:
_backup_state["pending"] = False
continue
_backup_state["running"] = False
return
threading.Thread(target=_run, daemon=True, name="cloud-backup").start()
# ── XTTS Model ────────────────────────────────────────────────────────
from TTS.api import TTS
device = "cpu" # GPU intentionally disabled — CPU only
# ── Background model loading (server starts immediately) ──────────────
xtts = None
xtts_ready = False
xtts_loading = True
xtts_error = None
def _load_model():
global xtts, xtts_ready, xtts_loading, xtts_error
try:
print(f"[*] Loading XTTS v2 on {device.upper()} …")
xtts = TTS("tts_models/multilingual/multi-dataset/xtts_v2").to(device)
xtts_ready = True
print("[✓] XTTS ready.")
except Exception as e:
xtts_error = f"{type(e).__name__}: {e}"
print(f"[!] XTTS loading failed: {xtts_error}", flush=True)
traceback.print_exc()
finally:
xtts_loading = False
threading.Thread(target=_load_model, daemon=True).start()
# tts_lock is created lazily inside the event loop to avoid asyncio issues
_tts_lock = None
def get_tts_lock():
global _tts_lock
if _tts_lock is None:
_tts_lock = asyncio.Lock()
return _tts_lock
def _startup_cleanup():
for f in os.listdir("/tmp"):
if any(f.startswith(p) for p in ("ch_","ref_","sts_","dub_")):
try: os.remove(os.path.join("/tmp",f))
except: pass
_startup_cleanup()
# ── Job Management: Cancellation & Progress ───────────────────────────
job_progress: dict = {} # job_id -> {current, total, status, ts}
cancel_events: dict = {} # job_id -> threading.Event
JOB_TTL_SEC = 3600
def _gc_jobs():
cutoff = time.time() - JOB_TTL_SEC
for k in [k for k,v in list(job_progress.items())
if v.get("status") != "running" and v.get("ts",0) < cutoff]:
job_progress.pop(k, None)
# ── Faster-Whisper (replaces openai-whisper) ──────────────────────────
whisper_mdl = None
WHISPER_OK = False
def _load_whisper():
"""Load Faster-Whisper in the background. If the default HF cache
(e.g. the /data volume) gives an I/O error, retry from local disk."""
global whisper_mdl, WHISPER_OK
try:
from faster_whisper import WhisperModel
except Exception as e:
print(f"[!] Faster-Whisper not installed: {e}", flush=True)
return
for root in (None, "/tmp/whisper_models"):
try:
if root: os.makedirs(root, exist_ok=True)
whisper_mdl = WhisperModel("small", device="cpu", compute_type="int8",
cpu_threads=CPU_THREADS, download_root=root)
WHISPER_OK = True
print(f"[✓] Faster-Whisper ready (small · int8 · CPU) [{root or 'default cache'}].", flush=True)
return
except Exception as e:
print(f"[!] Faster-Whisper load failed ({root or 'default cache'}): {e}", flush=True)
print("[!] Faster-Whisper not available.", flush=True)
threading.Thread(target=_load_whisper, daemon=True).start()
def transcribe_audio(path: str) -> str:
"""Transcribe using Faster-Whisper; returns plain text."""
if whisper_mdl is None:
raise RuntimeError("Faster-Whisper not loaded")
segments, _ = whisper_mdl.transcribe(path, beam_size=5)
return " ".join(s.text for s in segments).strip()
# ── Translator ────────────────────────────────────────────────────────
try:
from deep_translator import GoogleTranslator
TRANSLATOR_OK = True
except Exception:
TRANSLATOR_OK = False
# ── History helpers ───────────────────────────────────────────────────
_history_lock = threading.RLock()
MAX_HISTORY = 300
MAX_OUTPUTS = 200
def load_history():
with _history_lock:
try:
if not os.path.exists(HISTORY_FILE): return []
with open(HISTORY_FILE, encoding="utf-8") as f:
return json.load(f)
except Exception:
try: shutil.copyfile(HISTORY_FILE, HISTORY_FILE + ".corrupt") # keep it, don't silently overwrite
except Exception: pass
return []
def save_history(h):
with _history_lock: # atomic: write temp → replace
tmp = HISTORY_FILE + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(h, f, ensure_ascii=False, indent=2)
os.replace(tmp, HISTORY_FILE)
trigger_cloud_backup()
def _prune_outputs():
try:
wavs = sorted(
[f for f in os.listdir(OUTPUT_DIR) if f.endswith(".wav")],
key=lambda f: os.path.getmtime(os.path.join(OUTPUT_DIR, f))
)
for old in wavs[:-MAX_OUTPUTS]:
try: os.remove(os.path.join(OUTPUT_DIR, old))
except Exception: pass
except Exception:
pass
def append_history(filename, text, language, mode, char_count=0):
with _history_lock: # read-modify-write must be one unit
h = load_history()
h.append(dict(filename=filename, text=text[:150], language=language,
mode=mode, char_count=char_count, ts=int(time.time())))
if len(h) > MAX_HISTORY:
h = h[-MAX_HISTORY:]
save_history(h)
_prune_outputs()
# ── Parameter Engineering ─────────────────────────────────────────────
def stability_to_temperature(s): return round(0.85 - (s * 0.40), 3)
def clarity_to_top_p(c): return round(0.70 + (c * 0.25), 3)
def style_to_rep_penalty(st): return round(2.00 + (st * 2.00), 2)
# ── Expression Presets ────────────────────────────────────────────────
# Each preset overrides stability/temperature/top_p/rep_penalty to push
# XTTS toward a specific emotional character.
EXPRESSION_PRESETS = {
"natural": dict(temperature=0.72, top_p=0.88, rep_penalty=2.0, speed_mult=1.00),
"warm": dict(temperature=0.78, top_p=0.90, rep_penalty=1.80, speed_mult=0.97),
"excited": dict(temperature=0.88, top_p=0.92, rep_penalty=1.65, speed_mult=1.10),
"sad": dict(temperature=0.65, top_p=0.85, rep_penalty=2.20, speed_mult=0.88),
"serious": dict(temperature=0.55, top_p=0.82, rep_penalty=2.50, speed_mult=0.95),
"story": dict(temperature=0.82, top_p=0.93, rep_penalty=1.75, speed_mult=0.96),
"news": dict(temperature=0.60, top_p=0.84, rep_penalty=2.40, speed_mult=1.02),
"child": dict(temperature=0.90, top_p=0.95, rep_penalty=1.60, speed_mult=1.05),
}
def apply_expression_preset(preset_name: str, speed: float) -> dict:
p = EXPRESSION_PRESETS.get(preset_name, EXPRESSION_PRESETS["natural"])
return {
"temperature": p["temperature"],
"top_p": p["top_p"],
"repetition_penalty": p["rep_penalty"],
"speed": round(speed * p["speed_mult"], 3),
}
# ── Text Expressiveness Enhancer ──────────────────────────────────────
# XTTS is very sensitive to punctuation and sentence rhythm.
# Adding deliberate pauses, ellipses, and varied sentence endings
# dramatically improves naturalness and eliminates the "robot reading" feel.
def enhance_text_expression(text: str, lang: str, mode: str) -> str:
"""
Enrich text with punctuation cues that guide XTTS prosody.
Does NOT change words — only adds/adjusts punctuation & whitespace.
"""
if mode == "natural" or not text.strip():
return text
t = text.strip()
# 1. Normalise whitespace
t = re.sub(r'[ \t]+', ' ', t)
# 2. Ensure sentences end with terminal punctuation
# (XTTS pauses longer on . than on nothing)
t = re.sub(r'([^\.\!\?؟،,\n])\n', r'\1.\n', t)
# 3. Mode-specific rhythm adjustments
if mode == "excited":
# Shorten pauses between sentences → rapid delivery
t = re.sub(r'\.\s+', '. ', t)
# Add emphasis markers via comma-pauses before key conjunctions
for conj in ['لكن','بل','إلا','however','but','yet','still']:
t = re.sub(rf'\b({conj})\b', r', \1', t, flags=re.IGNORECASE)
elif mode == "sad":
# Extend pauses → slower, heavier delivery
t = re.sub(r'\.\s+', '… ', t)
t = re.sub(r'،\s+', '، ', t)
elif mode == "story":
# Add brief breath-pause after scene-setting clauses
t = re.sub(r'(كان|كانت|في يوم|ذات يوم|once upon|there was)',
r'\1،', t, flags=re.IGNORECASE)
elif mode == "news":
# Crisp, no trailing breath
t = re.sub(r'\s*…+\s*', '. ', t)
elif mode == "child":
# Short bursts with upward inflection hints
t = re.sub(r'\.\s+([A-Zا-ي])', r'! \1', t)
# 4. Collapse multiple punctuation
t = re.sub(r'([.!?؟،,]){2,}', r'\1', t)
t = re.sub(r'…{2,}', '…', t)
return t
# ╔══════════════════════════════════════════════════════════════════════╗
# ║ P H R A S E - L E V E L P R O S O D Y V A R I A T I O N ║
# ║ Each sentence is analysed for emotional content. ║
# ║ XTTS parameters (temperature, speed, repetition_penalty) shift ║
# ║ per-phrase so the delivery feels alive, not machine-read. ║
# ║ Post-generation volume envelopes reinforce the prosodic contour. ║
# ╚══════════════════════════════════════════════════════════════════════╝
# ── Per-emotion XTTS deltas ───────────────────────────────────────────
# (applied on top of the base params; scaled by user's variation_strength)
EMOTION_DELTAS = {
# temp speed_mult rep_penalty silence_ms
"neutral": ( 0.00, 1.000, 0.00, 110 ),
"question": (+0.07, 0.960, -0.10, 140 ), # slower, more uncertain
"exclaim": (+0.12, 1.080, -0.30, 75 ), # faster, more energy
"emphasis": (-0.06, 0.900, +0.30, 155 ), # deliberate, weighty
"sadness": (-0.10, 0.870, +0.35, 185 ), # slow, heavy
"joy": (+0.10, 1.060, -0.25, 85 ), # bright, forward
"anger": (+0.08, 1.040, -0.20, 95 ), # tense, clipped
"soft": (-0.09, 0.920, +0.20, 160 ), # tender, gentle
"urgency": (+0.11, 1.120, -0.25, 65 ), # fast, alert
"story": (+0.06, 0.950, -0.10, 135 ), # varied, narrative
}
# ── Keyword patterns (Arabic + English) ──────────────────────────────
EMOTION_PATTERNS = {
"joy": [
# Arabic
r'سعيد|فرح|رائع|مذهل|عظيم|ممتاز|برافو|مبروك|ياه|يا لروع|'
r'جميل|بديع|أحب|نجح|وفق|الله|يسعد|سرور|ابتهاج|بهجة|نشوة',
# English
r'happy|wonderful|amazing|great|excellent|love|joy|fantastic|'
r'brilliant|perfect|congrats|hooray|yay|delightful|terrific',
],
"sadness": [
r'حزين|للأسف|مؤسف|وداع|ألم|فقدنا|صعب|مؤلم|بكى|دموع|'
r'خسر|رحل|غاب|اشتاق|أفتقد|حسرة|مأساة|مصيبة|فراق|وجع',
r'sad|unfortunately|sorry|pain|lost|difficult|hard|tears|'
r'miss|regret|grieve|mourn|tragedy|farewell|ache',
],
"anger": [
r'غضب|لا أقبل|مستحيل|يكفي|كفى|ظلم|حرام|لا يجوز|رفض|أبداً|'
r'لن أتحمل|سئمت|ضاق|احتج|رفض|نقمة|سخط|فساد',
r'never|impossible|stop|enough|angry|refuse|unacceptable|'
r'outrage|disgusting|unbelievable|ridiculous|furious',
],
"emphasis": [
r'بالتأكيد|دون شك|قطعاً|أكيد|لا شك|يجب|ضرورة|مهم|انتبه|'
r'اسمع|لا بد|حتماً|بلا شك|من المؤكد|بوضوح|صراحةً|جداً',
r'definitely|absolutely|certainly|must|important|listen|'
r'crucial|critical|clearly|obviously|indeed|precisely',
],
"soft": [
r'ربما|لعل|عسى|أتمنى|أرجو|بلطف|برفق|هدوء|سلام|'
r'هناء|رويداً|بهدوء|بتأن|رفيق|حنان|لطيف|رحمة|أمل',
r'perhaps|maybe|hopefully|gently|softly|peace|calm|'
r'wish|hope|kindly|tenderly|quietly|sweetly',
],
"urgency": [
r'الآن|فوراً|سريعاً|عاجل|انتبه|احذر|خطر|أسرع|بسرعة|'
r'عاجلاً|في الحال|لا تتأخر|إنذار|تحذير|يستعجل|عجلة',
r'now|immediately|quickly|urgent|warning|danger|hurry|'
r'fast|alert|emergency|asap|right away|no time',
],
"story": [
r'كان يا مكان|ذات يوم|في قديم|يحكى أن|وفي ليلة|'
r'ثم قال|فأجاب|نظر|قرر|انطلق|توجه|وصل|لقي|رأى',
r'once upon|there was|long ago|he said|she replied|'
r'suddenly|then he|and so|meanwhile|the next day',
],
}
# ── Phrase emotion detector ───────────────────────────────────────────
def analyze_phrase_emotion(phrase: str, lang: str) -> tuple[str, float]:
"""
Detect the dominant emotion in a phrase.
Returns (emotion_name, confidence_0_to_1).
Confidence drives how strongly the deltas are applied.
"""
p = phrase.strip()
# 1. Punctuation-first rules (fastest, highest confidence)
if re.search(r'[؟?]$', p):
return "question", 0.90
count_exclaim = len(re.findall(r'[!]', p))
if count_exclaim >= 2:
return "exclaim", 0.95
if count_exclaim == 1 and len(p) < 60:
return "exclaim", 0.75
# 2. Keyword scan
scores: dict[str, float] = {}
for emotion, patterns in EMOTION_PATTERNS.items():
for pat in patterns:
matches = re.findall(pat, p, re.IGNORECASE)
if matches:
# Score = matches / words — more keywords → higher confidence
words = max(1, len(p.split()))
scores[emotion] = scores.get(emotion, 0) + len(matches) / words
if scores:
best = max(scores, key=scores.get)
conf = min(0.92, scores[best] * 3.0) # cap at 0.92
return best, conf
# 3. Structural heuristics
if len(p) < 20:
return "emphasis", 0.35 # short phrases tend to be punchy
if p.endswith('…') or p.endswith('...'):
return "soft", 0.50 # trailing ellipsis = trailing/wistful
return "neutral", 0.0
# ── Per-phrase XTTS param builder ─────────────────────────────────────
def make_phrase_params(base: dict, emotion: str, conf: float,
variation: float, pos_ratio: float) -> dict:
"""
variation : user slider 0-2 (0=none, 1=natural, 2=dramatic)
pos_ratio : 0.0 = first phrase, 1.0 = last phrase
Returns a NEW params dict with emotion-driven adjustments.
"""
if variation < 0.02:
return base.copy()
d = EMOTION_DELTAS.get(emotion, EMOTION_DELTAS["neutral"])
temp_d, speed_m, rep_d, _ = d
strength = conf * variation # 0 … ~1.84
p = base.copy()
p["temperature"] = float(np.clip(
base["temperature"] + temp_d * strength, 0.40, 0.96))
p["speed"] = float(np.clip(
base["speed"] * (1.0 + (speed_m - 1.0) * strength), 0.60, 1.65))
p["repetition_penalty"] = float(np.clip(
base["repetition_penalty"] + rep_d * strength, 1.30, 3.80))
# ── Position-based natural drift ────────────────────────────────
# Opening: slightly more controlled (lower temp, slower)
if pos_ratio < 0.08:
p["temperature"] = float(np.clip(p["temperature"] - 0.04 * variation, 0.40, 0.96))
p["speed"] = float(np.clip(p["speed"] * (1.0 - 0.04 * variation), 0.60, 1.65))
# Closing: natural deceleration (the "landing" of a speech)
if pos_ratio > 0.88:
p["speed"] = float(np.clip(p["speed"] * (1.0 - 0.06 * variation), 0.60, 1.65))
p["temperature"] = float(np.clip(p["temperature"] - 0.03 * variation, 0.40, 0.96))
# ── Micro-random jitter: prevent identical params on consecutive ─
# neutral phrases (which would sound machine-even)
if emotion == "neutral" and variation > 0.3:
# deterministic pseudo-random from phrase content hash
h = (zlib.crc32(base.get("_phrase_hint", "").encode("utf-8")) % 1000) / 1000.0
p["temperature"] += (h - 0.5) * 0.06 * variation
p["speed"] *= 1.0 + (h - 0.5) * 0.04 * variation
p["temperature"] = float(np.clip(p["temperature"], 0.40, 0.96))
p["speed"] = float(np.clip(p["speed"], 0.60, 1.65))
return p
# ── Post-generation phrase volume envelope ────────────────────────────
def apply_phrase_dynamics(data: np.ndarray, sr: int,
emotion: str, variation: float) -> np.ndarray:
"""
Shape amplitude envelope after XTTS generation.
This adds prosodic *weight* that XTTS parameter variation alone can't provide.
variation : 0-2
"""
if variation < 0.02 or len(data) == 0:
return data
n = len(data)
t = np.linspace(0.0, 1.0, n, dtype=np.float32)
env = np.ones(n, dtype=np.float32)
v = min(variation, 2.0)
if emotion == "question":
# Rising volume in last 25% — mimics upward intonation contour
rise_start = 0.75
env += np.where(t > rise_start,
(t - rise_start) / (1.0 - rise_start) * 0.07 * v, 0)
elif emotion == "exclaim":
# Forward energy burst — louder in first half, taper
env += (1.0 - t) * 0.05 * v
env += np.sin(t * np.pi * 2) * 0.03 * v # subtle tremolo feel
elif emotion == "emphasis":
# Bell curve — maximum weight in the centre
env += np.sin(t * np.pi) * 0.10 * v
elif emotion == "sadness":
# Gradual fade throughout — speech drains away
env -= t * 0.08 * v
env = np.clip(env, 0.70, 1.0)
elif emotion == "joy":
# Bright forward motion
env += t * 0.04 * v + 0.02 * v
elif emotion == "anger":
# Sharp, compressed — no gentle tails
attack = np.where(t < 0.05, t / 0.05, 1.0)
env *= attack
env += 0.04 * v
elif emotion == "soft":
# Gentle, backed-off
env -= 0.06 * v
elif emotion == "urgency":
# Flat high energy — no decay
env += 0.06 * v
elif emotion == "story":
# Slow arc: quiet start, rise, settle
env += np.sin(t * np.pi * 0.7) * 0.05 * v
# Smooth envelope (25 ms window to avoid clicks)
k = max(1, int(0.025 * sr))
env = np.convolve(env, np.ones(k) / k, mode='same').astype(np.float32)
env = np.clip(env, 0.50, 1.30)
result = data * env
# Soft-limit peaks
peak = np.max(np.abs(result))
if peak > 0.95:
result = result / peak * 0.93
return result.astype(np.float32)
def silence_samples(sr: int, ms: int = 120) -> np.ndarray:
"""Short silence gap inserted between sentence chunks."""
return np.zeros(int(sr * ms / 1000), dtype=np.float32)
# ── Soft Clipping (tanh) — prevents digital distortion ────────────────
def apply_speaker_boost(path):
"""Loudness lift + transparent soft-knee limiter.
(The old tanh(x*1.8) saturated the whole signal → harsh, 'digital' timbre.)"""
data, sr = sf.read(path)
if data.ndim > 1: data = data.mean(axis=1)
data = vq.finalize_audio(data.astype(np.float32), sr, target_rms_db=-17.0)
sf.write(path, data, sr, subtype='PCM_16')
# ╔══════════════════════════════════════════════════════════════════════╗
# ║ H U M A N I Z E — Pure-NumPy/SciPy DSP Chain ║
# ║ Simulates recording in a real acoustic environment. ║
# ║ No extra libraries. No IR files. All synthesized algorithmically. ║
# ╚══════════════════════════════════════════════════════════════════════╝
# ── Environment presets ───────────────────────────────────────────────
# Each dict drives the parameters of every DSP stage below.
HUMANIZE_PRESETS = {
"studio": dict(
# Clean, slightly padded room — professional voiceover booth
reverb_room=0.06, reverb_damping=0.85, reverb_wet=0.04,
noise_floor=-68, noise_color=0.5,
mic_hp_hz=90, mic_presence_db=1.5, mic_presence_hz=5500,
mic_air_db=-1.0, mic_air_hz=14000,
saturation=0.04, pitch_drift_cents=0.4, jitter_ms=0.3,
breathing=False,
),
"bedroom": dict(
# Untreated small room — YouTube / indie podcast feel
reverb_room=0.22, reverb_damping=0.60, reverb_wet=0.09,
noise_floor=-62, noise_color=0.7,
mic_hp_hz=80, mic_presence_db=2.0, mic_presence_hz=4800,
mic_air_db=-0.5, mic_air_hz=13000,
saturation=0.08, pitch_drift_cents=0.8, jitter_ms=0.6,
breathing=False,
),
"podcast": dict(
# Dynamic-mic, slightly driven — warm, intimate
reverb_room=0.10, reverb_damping=0.78, reverb_wet=0.05,
noise_floor=-65, noise_color=0.6,
mic_hp_hz=100, mic_presence_db=2.5, mic_presence_hz=3800,
mic_air_db=-2.5, mic_air_hz=12000,
saturation=0.14, pitch_drift_cents=0.6, jitter_ms=0.4,
breathing=False,
),
"radio": dict(
# Classic broadcast: bandpass 200–8 kHz, compressed
reverb_room=0.04, reverb_damping=0.90, reverb_wet=0.02,
noise_floor=-72, noise_color=0.3,
mic_hp_hz=200, mic_presence_db=3.0, mic_presence_hz=3000,
mic_air_db=-8.0, mic_air_hz=8000,
saturation=0.18, pitch_drift_cents=0.3, jitter_ms=0.2,
breathing=False,
),
"phone": dict(
# GSM / telephone: narrow band 300–3400 Hz, heavy saturation
reverb_room=0.02, reverb_damping=0.95, reverb_wet=0.01,
noise_floor=-55, noise_color=0.9,
mic_hp_hz=300, mic_presence_db=1.0, mic_presence_hz=2200,
mic_air_db=-20.0, mic_air_hz=3500,
saturation=0.28, pitch_drift_cents=1.2, jitter_ms=1.2,
breathing=False,
),
"cafe": dict(
# Mid-sized reverberant room, background murmur
reverb_room=0.42, reverb_damping=0.45, reverb_wet=0.16,
noise_floor=-54, noise_color=0.8,
mic_hp_hz=70, mic_presence_db=1.0, mic_presence_hz=4000,
mic_air_db=-1.0, mic_air_hz=14000,
saturation=0.06, pitch_drift_cents=1.0, jitter_ms=0.8,
breathing=False,
),
"outdoor": dict(
# Open space — very short reverb, wind-ish noise
reverb_room=0.08, reverb_damping=0.92, reverb_wet=0.03,
noise_floor=-50, noise_color=0.95,
mic_hp_hz=60, mic_presence_db=0.5, mic_presence_hz=3500,
mic_air_db=0.0, mic_air_hz=16000,
saturation=0.05, pitch_drift_cents=1.5, jitter_ms=1.5,
breathing=False,
),
}
# ── DSP helpers ───────────────────────────────────────────────────────
def _biquad_hp(data: np.ndarray, sr: int, fc: float) -> np.ndarray:
"""2nd-order Butterworth high-pass filter — clamped to safe range."""
from scipy.signal import butter, sosfilt
fc = float(np.clip(fc, 1.0, sr * 0.49)) # never hit Nyquist
sos = butter(2, fc / (sr / 2), btype='high', output='sos')
out = sosfilt(sos, data.astype(np.float64))
if not np.all(np.isfinite(out)):
return data.astype(np.float32)
return out.astype(np.float32)
def _biquad_peak(data: np.ndarray, sr: int, fc: float,
gain_db: float, Q: float = 1.4) -> np.ndarray:
"""Parametric peak EQ biquad — clamped and guarded."""
from scipy.signal import sosfilt
fc = float(np.clip(fc, 20.0, sr * 0.45))
gain_db = float(np.clip(gain_db, -40.0, 20.0))
Q = float(np.clip(Q, 0.1, 20.0))
w0 = 2 * np.pi * fc / sr
A = 10 ** (gain_db / 40.0)
cos = np.cos(w0); sin = np.sin(w0)
alpha = sin / (2 * Q)
b0 = 1 + alpha * A; b1 = -2 * cos; b2 = 1 - alpha * A
a0 = 1 + alpha / A; a1 = -2 * cos; a2 = 1 - alpha / A
if abs(a0) < 1e-10:
return data.astype(np.float32)
sos = np.array([[b0/a0, b1/a0, b2/a0, 1.0, a1/a0, a2/a0]], dtype=np.float64)
if not np.all(np.isfinite(sos)):
return data.astype(np.float32)
out = sosfilt(sos, data.astype(np.float64))
if not np.all(np.isfinite(out)):
return data.astype(np.float32)
return out.astype(np.float32)
def _biquad_shelf_hi(data: np.ndarray, sr: int, fc: float,
gain_db: float) -> np.ndarray:
"""
High-shelf filter (air band roll-off / boost).
Fixed: clamp fc below Nyquist and guard alpha against overflow/NaN.
"""
from scipy.signal import sosfilt
# Clamp cutoff to a safe range (never exceed 90% of Nyquist)
fc = float(np.clip(fc, 20.0, sr * 0.45))
gain_db = float(np.clip(gain_db, -40.0, 20.0))
w0 = 2 * np.pi * fc / sr
A = 10 ** (gain_db / 40.0)
cos = np.cos(w0)
sin = np.sin(w0)
sqA = float(np.sqrt(max(A, 1e-12)))
S = 1.0
# Guard the sqrt argument — clamp to avoid negative/NaN
sqrt_arg = (A + 1.0 / max(A, 1e-12)) * (1.0 / max(S, 1e-9) - 1.0) + 2.0
sqrt_arg = max(sqrt_arg, 0.0)
alpha = sin / 2.0 * np.sqrt(sqrt_arg)
b0 = A * ((A + 1) + (A - 1) * cos + 2 * sqA * alpha)
b1 = -2 * A * ((A - 1) + (A + 1) * cos)
b2 = A * ((A + 1) + (A - 1) * cos - 2 * sqA * alpha)
a0 = (A + 1) - (A - 1) * cos + 2 * sqA * alpha
a1 = 2 * ((A - 1) - (A + 1) * cos)
a2 = (A + 1) - (A - 1) * cos - 2 * sqA * alpha
# Guard against degenerate filter (a0 near zero)
if abs(a0) < 1e-10:
return data.astype(np.float32)
sos = np.array([[b0/a0, b1/a0, b2/a0, 1.0, a1/a0, a2/a0]], dtype=np.float64)
# Final guard: if any coefficient is NaN/Inf, bypass
if not np.all(np.isfinite(sos)):
return data.astype(np.float32)
result = sosfilt(sos, data.astype(np.float64))
# Guard output (should never happen now, but safety net)
if not np.all(np.isfinite(result)):
return data.astype(np.float32)
return result.astype(np.float32)
def _pink_noise(n: int, rng: np.random.Generator) -> np.ndarray:
"""FFT-based pink noise — O(N log N), no sample loops."""
white = rng.standard_normal(n).astype(np.float64)
fft_w = np.fft.rfft(white)
freqs = np.arange(1, len(fft_w) + 1, dtype=np.float64)
fft_w /= np.sqrt(freqs) # 1/sqrt(f) → pink (1/f power)
pink = np.fft.irfft(fft_w, n=n)
peak = np.max(np.abs(pink))
if peak > 0: pink /= peak
return pink.astype(np.float32)
def _comb_reverb(data: np.ndarray, sr: int,
room: float, damping: float, wet: float) -> np.ndarray:
"""Room reverb via a synthetic impulse response (dense, non-metallic).
Replaces the old 4-comb Schroeder network, which rang at 30-44 ms delays."""
if wet < 0.001:
return data
from scipy.signal import fftconvolve
irL, irR = vq.make_room_ir(sr, rt60=0.12 + room * 0.9, pre=8, er=0.5, damp=damping)
rev = fftconvolve(data.astype(np.float64), (irL + irR) / 2.0)[: len(data)]
rev *= (np.sqrt(np.mean(data ** 2)) + 1e-9) / (np.sqrt(np.mean(rev ** 2)) + 1e-9)
return (data * (1.0 - wet) + rev.astype(np.float32) * wet).astype(np.float32)
def _micro_pitch_drift(data: np.ndarray, sr: int,
cents: float, rng: np.random.Generator) -> np.ndarray:
"""
Subtle continuous pitch drift ±cents via nearest-neighbour resampling.
Creates the organic wobble of a real human voice.
"""
if cents < 0.05:
return data
# Generate a slow-moving drift signal (0.1–0.5 Hz sinusoid + noise)
t = np.arange(len(data)) / sr
freq_hz = 0.2 + rng.uniform(0, 0.15)
drift = np.sin(2 * np.pi * freq_hz * t)
drift += 0.3 * rng.standard_normal(len(data)) # add gentle randomness
# Smooth with a 200 ms window
k = max(1, int(0.2 * sr))
drift = np.convolve(drift, np.ones(k)/k, mode='same')
# Scale to ±ratio
ratio = 2 ** (cents / 1200) # cents → semitone ratio
factor = 1.0 + (ratio - 1.0) * drift # ±ratio around 1.0
factor = np.clip(factor, 0.90, 1.10)
# Nearest-neighbour reindex
indices = np.cumsum(factor)
indices = indices / indices[-1] * (len(data) - 1)
indices = np.clip(indices.astype(np.int32), 0, len(data) - 1)
return data[indices].astype(np.float32)
def _timing_jitter(data: np.ndarray, sr: int, jitter_ms: float,
rng: np.random.Generator) -> np.ndarray:
"""
Apply block-level timing jitter: randomly shift 20 ms blocks by ±jitter_ms.
Breaks the perfectly even machine cadence of synthetic speech.
"""
if jitter_ms < 0.1:
return data
block = int(0.020 * sr) # 20 ms blocks
out = data.copy()
shift_max = int(jitter_ms * sr / 1000)
for start in range(0, len(data) - block, block):
shift = rng.integers(-shift_max, shift_max + 1)
src_start = max(0, start + shift)
src_end = min(len(data), start + block + shift)
dst_len = min(block, len(data) - start)
src_len = src_end - src_start
n = min(dst_len, src_len)
if n > 0:
out[start:start + n] = data[src_start:src_start + n]
return out
def _soft_saturate(data: np.ndarray, amount: float) -> np.ndarray:
"""
Soft harmonic saturation — adds 2nd/3rd harmonics like analog tape/tube.
amount: 0 = bypass, 1 = heavy saturation
"""
if amount < 0.005:
return data
drive = 1.0 + amount * 6.0 # 1x … 7x overdrive
driven = data * drive
# tanh saturation, then scale back down
sat = np.tanh(driven) / np.tanh(drive)
return (data * (1.0 - amount) + sat * amount).astype(np.float32)
# ── Master humanize function ──────────────────────────────────────────
def humanize_audio(src_path: str, out_path: str, preset_name: str,
intensity: float = 1.0, seed: int = 42):
"""
Apply full humanization DSP chain to a WAV file.
intensity: 0.0 = almost dry, 1.0 = full preset, >1.0 = exaggerated.
Each stage is wrapped in a try/except so a failure degrades gracefully
instead of corrupting the output file.
"""
p = HUMANIZE_PRESETS.get(preset_name, HUMANIZE_PRESETS["bedroom"])
rng = np.random.default_rng(seed)
data, sr = sf.read(src_path)
if data.ndim > 1:
data = data.mean(axis=1)
data = data.astype(np.float32)
if len(data) == 0:
raise ValueError("ملف الصوت فارغ.")
mix = float(np.clip(intensity, 0.0, 2.0))
def _safe(fn, label, *args, **kwargs):
"""Run fn(*args, **kwargs), return original data on failure."""
nonlocal data
try:
result = fn(*args, **kwargs)
# Reject NaN / Inf output — fall back to unchanged
if result is not None and len(result) > 0 and np.all(np.isfinite(result)):
data = result.astype(np.float32)
else:
print(f"[humanize] {label}: non-finite output — stage skipped")
except Exception as e:
print(f"[humanize] {label}: {e} — stage skipped")
# 1. Mic coloring ─────────────────────────────────────────────────
_safe(_biquad_hp, "HP filter", data, sr, p["mic_hp_hz"])
if abs(p["mic_presence_db"]) > 0.1:
_safe(_biquad_peak, "Peak EQ", data, sr,
p["mic_presence_hz"], p["mic_presence_db"] * mix, Q=1.2)
if abs(p["mic_air_db"]) > 0.1:
_safe(_biquad_shelf_hi, "Hi-shelf", data, sr,
p["mic_air_hz"], p["mic_air_db"] * mix)
# 2. Harmonic saturation ──────────────────────────────────────────
_safe(_soft_saturate, "Saturation", data, p["saturation"] * mix)
# 3/4. Timing jitter & micro pitch drift — REMOVED: measured to add
# -11…-2 dB (jitter) / -40…-32 dB (drift) of non-harmonic distortion
# (block copies without crossfade; nearest-neighbour resampling).
# 5. Room reverb ──────────────────────────────────────────────────
_safe(_comb_reverb, "Reverb", data, sr,
p["reverb_room"], p["reverb_damping"], p["reverb_wet"] * mix)
# 6. Noise floor ──────────────────────────────────────────────────
try:
floor_db = p["noise_floor"] + (1.0 - mix) * 20
floor_db = float(np.clip(floor_db, -96.0, -30.0))
noise_amp = 10 ** (floor_db / 20.0)
color = float(np.clip(p["noise_color"], 0.0, 1.0))
pink = _pink_noise(len(data), rng)
white = rng.standard_normal(len(data)).astype(np.float32)
noise_mix = pink * color + white * (1.0 - color)
nm_peak = np.max(np.abs(noise_mix))
if nm_peak > 1e-9:
noise_mix = noise_mix / nm_peak * noise_amp
candidate = data + noise_mix.astype(np.float32)
if np.all(np.isfinite(candidate)):
data = candidate
except Exception as e:
print(f"[humanize] Noise floor: {e} — stage skipped")
# 7. Final normalize + soft ceiling ──────────────────────────────
peak = float(np.max(np.abs(data)))
if peak > 0:
data = data / peak * 0.92
# tanh soft-limiter (can't produce NaN on finite input)
data = vq.soft_limit(data, 0.92, 0.75)
sf.write(out_path, data, sr, subtype='PCM_16')
# ── Text Splitting for Real Progress ─────────────────────────────────
def split_text_sentences(text: str, max_chunk: int = 280) -> list:
"""Split text into sentence-sized chunks with a max character limit."""
parts = re.split(r'(?<=[.!?؟\n،])\s+', text.strip())
chunks, cur = [], ""
for p in parts:
if not p.strip():
continue
if len(cur) + len(p) + 1 <= max_chunk:
cur = (cur + " " + p).strip() if cur else p
else:
if cur:
chunks.append(cur)
if len(p) > max_chunk:
# Sub-split on comma/semicolon
sub_parts = re.split(r'(?<=[،,;])\s+', p)
sc = ""
for s in sub_parts:
if len(sc) + len(s) + 1 <= max_chunk:
sc = (sc + " " + s).strip() if sc else s
else:
if sc: chunks.append(sc)
sc = s
if sc: chunks.append(sc)
cur = ""
else:
cur = p
if cur:
chunks.append(cur)
return chunks if chunks else [text]
# ── Core TTS call ─────────────────────────────────────────────────────
def run_tts(text, refs, lang, tts_params, split, out_path):
if not xtts_ready or xtts is None:
raise RuntimeError("النموذج لا يزال يُحمَّل، يُرجى الانتظار قليلاً ثم المحاولة مجدداً.")
p = {k: v for k, v in tts_params.items() if not k.startswith("_")}
xtts.tts_to_file(
text=text, speaker_wav=refs, language=lang, file_path=out_path,
enable_text_splitting=split, **p,
)
def _run_tts_chunked_legacy(text, refs, lang, base_params,
out_path, job_id,
silence_ms: int = 110,
variation: float = 0.0,
) -> bool:
"""
Split text → per-phrase emotion analysis → per-phrase XTTS params
→ per-phrase dynamics shaping → silence gaps → merge.
variation=0 → all chunks use base_params (original behaviour).
Returns True on success, False if cancelled.
"""
cancel_ev = cancel_events.get(job_id)
chunks = split_text_sentences(text) if len(text) > 280 else [text]
total = len(chunks)
job_progress[job_id] = {"current": 0, "total": total, "status": "running"}
# ── Single-chunk fast path ────────────────────────────────────────
if total == 1:
if cancel_ev and cancel_ev.is_set():
job_progress[job_id]["status"] = "cancelled"; return False
phrase_p = base_params
sc_emotion = "neutral"
if variation > 0.02:
sc_emotion, sc_conf = analyze_phrase_emotion(text, lang) # ONE call
phrase_p = make_phrase_params(dict(base_params, _phrase_hint=text[:30]),
sc_emotion, sc_conf, variation, 0.5)
run_tts(text, refs, lang, phrase_p, True, out_path)
if variation > 0.02 and sc_emotion != "neutral":
data, sr = sf.read(out_path)
if data.ndim > 1: data = data.mean(axis=1)
data = apply_phrase_dynamics(data.astype(np.float32), sr, sc_emotion, variation)
sf.write(out_path, data, sr, subtype="PCM_16")
job_progress[job_id] = {"current": 1, "total": 1, "status": "done", "ts": time.time()}
return True
# ── Multi-chunk expressive path ───────────────────────────────────
emotions = [analyze_phrase_emotion(c, lang) for c in chunks]
tmp_files, arrays, sr_out = [], [], None
try:
for i, (chunk, (emotion, conf)) in enumerate(zip(chunks, emotions)):
if cancel_ev and cancel_ev.is_set():
job_progress[job_id]["status"] = "cancelled"; return False
pos = i / max(total - 1, 1)
phrase_p = make_phrase_params(
dict(base_params, _phrase_hint=chunk[:30]),
emotion, conf, variation, pos)
gap_ms = EMOTION_DELTAS.get(emotion, EMOTION_DELTAS["neutral"])[3]
actual_gap = int(silence_ms * 0.4 + gap_ms * 0.6) if variation > 0.1 else silence_ms
tmp = f"/tmp/ch_{job_id}_{i}.wav"
run_tts(chunk, refs, lang, phrase_p, False, tmp)
tmp_files.append(tmp)
data, sr = sf.read(tmp)
if sr_out is None: sr_out = sr
if data.ndim > 1: data = data.mean(axis=1)
data = data.astype(np.float32)
if variation > 0.02:
data = apply_phrase_dynamics(data, sr, emotion, variation)
arrays.append(data)
arrays.append(silence_samples(sr, actual_gap))
job_progress[job_id]["current"] = i + 1
combined = np.concatenate(arrays) if arrays else np.zeros(sr_out or 22050, np.float32)
sf.write(out_path, combined, sr_out or 22050, subtype="PCM_16")
job_progress[job_id]["status"] = "done"
job_progress[job_id]["ts"] = time.time()
return True
finally:
for tf in tmp_files:
try: os.remove(tf)
except: pass
# ╔══════════════════════════════════════════════════════════════════════╗
# ║ QUALITY PIPELINE (reference cleaning · real char limits · best-of-N) ║
# ╚══════════════════════════════════════════════════════════════════════╝
REF_CACHE_DIR = os.path.join(tempfile.gettempdir(), "vs_refcache")
_LAT_CACHE = collections.OrderedDict()
def _tts_model():
return getattr(getattr(xtts, "synthesizer", None), "tts_model", None)
def _model_defaults(model):
c = getattr(model, "config", None)
g = lambda k, d: type(d)(getattr(c, k, d)) if c is not None else d
return dict(temperature=g("temperature", 0.75), length_penalty=g("length_penalty", 1.0),
repetition_penalty=g("repetition_penalty", 5.0), top_k=g("top_k", 50),
top_p=g("top_p", 0.85))
def prepare_refs(refs, enhance=True):
"""Clean + merge the reference samples → ([path], summary|None)."""
if not enhance:
return list(refs), None
try:
paths, _reports, summary = vq.enhance_reference_set(refs, REF_CACHE_DIR)
return list(paths), summary
except Exception as e:
print(f"[!] Reference enhancement failed — using the raw sample: {e}", flush=True)
return list(refs), None
def get_latents(model, ref_paths):
"""Conditioning latents for one or several reference files.
With several files each one is encoded ON ITS OWN and the results are averaged
(duration-weighted). XTTS' own multi-file path glues the waveforms together with
torch.cat before encoding — an audible splice that the model then imitates."""
key = tuple((p, os.path.getsize(p), int(os.path.getmtime(p))) for p in ref_paths)
if key in _LAT_CACHE:
_LAT_CACHE.move_to_end(key)
return _LAT_CACHE[key]
c = model.config
kw = dict(gpt_cond_len=int(getattr(c, "gpt_cond_len", 30)),
gpt_cond_chunk_len=int(getattr(c, "gpt_cond_chunk_len", 4)),
max_ref_length=int(getattr(c, "max_ref_len", 30)),
sound_norm_refs=bool(getattr(c, "sound_norm_refs", False)))
paths = list(ref_paths)
if len(paths) == 1:
lat = model.get_conditioning_latents(audio_path=paths, **kw)
else:
parts, weights = [], []
for p in paths:
parts.append(model.get_conditioning_latents(audio_path=[p], **kw))
try: dur = float(sf.info(p).duration)
except Exception: dur = float(kw["gpt_cond_len"])
weights.append(max(0.5, min(dur, float(kw["gpt_cond_len"]))))
wsum = sum(weights)
lat = (sum(w * g for w, (g, _) in zip(weights, parts)) / wsum,
sum(w * s for w, (_, s) in zip(weights, parts)) / wsum)
_LAT_CACHE[key] = lat
while len(_LAT_CACHE) > 6:
_LAT_CACHE.popitem(last=False)
return lat
def _gen_kwargs(p, d):
"""UI/preset params → safe XTTS sampling args, centred on the model's own defaults.
XTTS 'speed' stretches latents: beyond ±15 % it degrades, so the excess
is applied afterwards as a time-stretch."""
speed = float(np.clip(p.get("speed", 1.0), 0.5, 2.0))
native = float(np.clip(speed, 0.85, 1.20))
rep0 = d["repetition_penalty"]
rep = float(np.clip(rep0 * float(p.get("repetition_penalty", 2.0)) / 2.0, rep0 * 0.7, 12.0))
kw = dict(temperature=float(np.clip(p.get("temperature", d["temperature"]), 0.60, 0.90)),
top_p=float(np.clip(p.get("top_p", d["top_p"]), 0.80, 0.95)),
top_k=d["top_k"], length_penalty=d["length_penalty"],
repetition_penalty=rep, speed=native)
return kw, speed / native
def _tame_phrase_params(base, p):
"""Per-sentence emotion tweaks must stay small — big jumps in temperature/speed
between neighbouring sentences make the voice sound like different takes."""
q = dict(p)
q["temperature"] = float(np.clip(p["temperature"], base["temperature"] - 0.06, base["temperature"] + 0.06))
q["speed"] = float(np.clip(p["speed"], base["speed"] * 0.94, base["speed"] * 1.06))
q["repetition_penalty"] = float(np.clip(p["repetition_penalty"],
base["repetition_penalty"] - 0.3, base["repetition_penalty"] + 0.3))
return q
def _residual_stretch(wav, rate):
if abs(rate - 1.0) < 0.03:
return wav
try:
import librosa
return librosa.effects.time_stretch(np.asarray(wav, np.float32), rate=rate)
except Exception:
return wav
def _asr_for_scoring(lang):
"""Whisper round-trip used only to compare candidate takes."""
if not (WHISPER_OK and whisper_mdl is not None):
return None
code = "zh" if lang.startswith("zh") else lang.split("-")[0]
from scipy.signal import resample_poly
def fn(wav, sr):
a = np.asarray(wav, np.float32)
if sr != 16000:
g = math.gcd(int(sr), 16000)
a = resample_poly(a, 16000 // g, int(sr) // g).astype(np.float32)
segs, _ = whisper_mdl.transcribe(a, language=code, beam_size=1, vad_filter=False,
condition_on_previous_text=False)
return " ".join(s.text for s in segs)
return fn
DRY_CACHE_DIR = os.path.join(tempfile.gettempdir(), "vs_drycache")
os.makedirs(DRY_CACHE_DIR, exist_ok=True)
os.makedirs(REF_CACHE_DIR, exist_ok=True)
RECENT_JOBS = collections.deque(maxlen=20)
def _prune_dir(path, max_files, max_bytes):
try:
fs = [os.path.join(path, f) for f in os.listdir(path)]
fs = sorted((f for f in fs if os.path.isfile(f)), key=os.path.getmtime)
total = sum(os.path.getsize(f) for f in fs)
while fs and (len(fs) > max_files or total > max_bytes):
f = fs.pop(0); total -= os.path.getsize(f); os.remove(f)
except Exception:
pass
_prune_dir(REF_CACHE_DIR, 40, 300 << 20)
_prune_dir(DRY_CACHE_DIR, 60, 400 << 20)
def _dry_cache_key(txt, lang, ref_paths, base_params, silence_ms, variation, attempts, asr_on):
"""Everything that changes the synthesised (dry) audio. Depth / boost are post-FX
and deliberately NOT part of the key → changing them re-renders instantly."""
payload = dict(v=vq.VQ_VERSION + "/dry2", t=txt, l=lang, r=[vq._file_sha(p) for p in ref_paths],
p={k: round(float(v), 4) for k, v in base_params.items() if isinstance(v, (int, float))},
s=int(silence_ms), var=round(float(variation), 3), a=int(attempts), asr=bool(asr_on))
return hashlib.sha1(json.dumps(payload, sort_keys=True, ensure_ascii=False).encode()).hexdigest()[:24]
def _record_job(stats_out, job_id, **kw):
rec = dict(job=str(job_id)[:8], ts=int(time.time()), **kw)
if rec.get("audio_sec") and rec.get("wall") is not None:
rec["rtf"] = round(rec["wall"] / rec["audio_sec"], 2)
for k in ("wall", "audio_sec"):
if k in rec: rec[k] = round(float(rec[k]), 1)
RECENT_JOBS.append(rec)
if stats_out is not None: stats_out.update(rec)
print(f"[job {rec['job']}] " + ", ".join(f"{k}={v}" for k, v in rec.items() if k not in ("job", "ts")), flush=True)
def _secs_fn(model, ref_spk):
"""Cosine similarity between a take and the reference, using XTTS's own speaker encoder."""
ref = ref_spk.detach().flatten().float().cpu()
warned = []
def fn(wav, sr):
try:
with torch.inference_mode():
e = model.get_speaker_embedding(torch.from_numpy(np.ascontiguousarray(wav, dtype=np.float32))[None, :], sr)
return float(torch.nn.functional.cosine_similarity(e.detach().flatten().float().cpu(), ref, dim=0))
except Exception as ex:
if not warned: # log once, never spam
warned.append(1)
print(f"[!] speaker-similarity scoring unavailable ({type(ex).__name__}: {ex}) — ranking without it.", flush=True)
raise
return fn
def run_tts_chunked(text, refs, lang, base_params, out_path, job_id,
silence_ms: int = 110, variation: float = 0.0,
attempts: int = 1, enhance_ref: bool = True,
clean_text: bool = True, ref_summary_out: dict = None,
stats_out: dict = None) -> bool:
"""Quality pipeline. Falls back to the legacy pipeline if anything unexpected
goes wrong, so a bug here can never block generation."""
model = _tts_model()
if model is None or not hasattr(model, "get_conditioning_latents"):
return _run_tts_chunked_legacy(text, refs, lang, base_params, out_path, job_id, silence_ms, variation)
try:
return _run_tts_quality(model, text, refs, lang, base_params, out_path, job_id,
silence_ms, variation, attempts, enhance_ref, clean_text, ref_summary_out, stats_out)
except Exception:
print("[!] Quality pipeline failed — falling back to legacy pipeline:", flush=True)
traceback.print_exc()
return _run_tts_chunked_legacy(text, refs, lang, base_params, out_path, job_id, silence_ms, variation)
def _run_tts_quality(model, text, refs, lang, base_params, out_path, job_id,
silence_ms, variation, attempts, enhance_ref, clean_text, ref_summary_out, stats_out=None):
cancel_ev = cancel_events.get(job_id)
job_progress[job_id] = {"current": 0, "total": 1, "status": "running"}
t_start = time.time()
ref_paths, summary = prepare_refs(refs, enhance_ref)
if summary and ref_summary_out is not None:
ref_summary_out.update(summary)
txt = vq.normalize_text(text, lang) if clean_text else text.strip()
attempts = int(np.clip(attempts, 1, 4))
asr = _asr_for_scoring(lang) if attempts > 1 else None
cache_path = os.path.join(DRY_CACHE_DIR, _dry_cache_key(
txt, lang, ref_paths, base_params, silence_ms, variation, attempts, asr is not None) + ".wav")
if os.path.exists(cache_path): # identical request → skip synthesis entirely
shutil.copyfile(cache_path, out_path)
try: os.utime(cache_path, None)
except Exception: pass
job_progress[job_id] = {"current": 1, "total": 1, "status": "done", "ts": time.time()}
_record_job(stats_out, job_id, cached=True, audio_sec=sf.info(out_path).duration,
wall=time.time() - t_start)
return True
lat_gpt, lat_spk = get_latents(model, ref_paths)
limits = getattr(getattr(model, "tokenizer", None), "char_limits", None)
chunks = vq.split_for_xtts(txt, lang, limits)
total = len(chunks)
job_progress[job_id] = {"current": 0, "total": total, "status": "running"}
d = _model_defaults(model)
try:
# One speaker-encoder forward pass per take — cheap, so run it even at attempts==1;
# this used to be gated behind attempts>1, which meant the default UI setting
# (attempts=1) never actually checked whether a chunk still sounded like the
# reference speaker. That gap is the main source of "sounds like a different
# speaker sometimes" on normal (non best-of-N) generations.
secs_fn = _secs_fn(model, lat_spk)
except Exception as ex: # scoring is a bonus — never block generation
print(f"[!] speaker-similarity scoring disabled: {ex}", flush=True)
secs_fn = None
retries = 0
use_var = variation > 0.02
emotions = [analyze_phrase_emotion(c, lang) for c, _ in chunks] if use_var else [("neutral", 0.0)] * total
seed0 = zlib.crc32(txt.encode("utf-8")) & 0x7FFFFFFF
scale = silence_ms / 110.0
takes, pauses = [], []
for i, ((chunk, pz), (emo, conf)) in enumerate(zip(chunks, emotions)):
if cancel_ev and cancel_ev.is_set():
job_progress[job_id]["status"] = "cancelled"; return False
p = dict(base_params)
if use_var:
p = make_phrase_params(dict(base_params, _phrase_hint=chunk[:30]),
emo, conf, variation, i / max(total - 1, 1))
p = _tame_phrase_params(base_params, p)
kw, resid = _gen_kwargs(p, d)
def gen(seed, chunk=chunk, kw=kw):
torch.manual_seed(seed)
out = model.inference(chunk, lang, lat_gpt, lat_spk, enable_text_splitting=False, **kw)
return np.asarray(out["wav"], np.float32).reshape(-1)
wav, _info, _n = vq.synth_best(gen, chunk, attempts, asr, vq.XTTS_SR, seed0 + 101 * i,
secs_fn=secs_fn, lang=lang)
retries += _n - 1
wav = _residual_stretch(wav, resid)
# `conf` (how confident analyze_phrase_emotion was) was computed above but never
# reached these two effects — they used to apply full-strength dynamics/pause
# changes even on a low-confidence (often false-positive) keyword match, which is
# what made ordinary neutral sentences suddenly jump in loudness/pacing next to
# their neighbours. Scaling by `conf` makes weak matches barely audible and keeps
# strong matches (question marks, "!!", clear keywords) at full strength.
if use_var and emo != "neutral":
wav = apply_phrase_dynamics(wav, vq.XTTS_SR, emo, variation * 0.5 * conf)
gap = EMOTION_DELTAS.get(emo, EMOTION_DELTAS["neutral"])[3]
emo_factor = 1.0 + (gap / 110.0 - 1.0) * 0.5 * min(variation, 1.0) * conf if use_var else 1.0
takes.append(wav); pauses.append(pz * scale * emo_factor)
job_progress[job_id]["current"] = i + 1
audio = vq.finalize_audio(vq.assemble(takes, pauses, vq.XTTS_SR, 1.0, seed0), vq.XTTS_SR)
sf.write(out_path, audio, vq.XTTS_SR, subtype="PCM_16")
try:
shutil.copyfile(out_path, cache_path)
_prune_dir(DRY_CACHE_DIR, 60, 400 << 20)
except Exception:
pass
job_progress[job_id]["status"] = "done"
job_progress[job_id]["ts"] = time.time()
_record_job(stats_out, job_id, cached=False, chunks=total, attempts=attempts, extra_takes=retries,
audio_sec=len(audio) / vq.XTTS_SR, wall=time.time() - t_start)
return True
def apply_depth(path, amount=0.5, space="studio"):
"""Stereo room + body + air + room-tone (see vq.spatialize). Never breaks the file."""
try:
amount = float(np.clip(amount, 0.0, 1.0))
if amount < 0.02:
return
data, sr = sf.read(path, dtype="float32")
if data.ndim > 1: data = data.mean(axis=1)
st, out_sr = vq.spatialize(data, sr, space, amount)
sf.write(path, st, out_sr, subtype="PCM_16")
except Exception:
print("[!] depth processing failed — file left dry:", flush=True)
traceback.print_exc()
MAX_UPLOAD_BYTES = 30 * 1024 * 1024 # per file
MAX_REQUEST_BYTES = 130 * 1024 * 1024 # per request (Content-Length)
MAX_REF_FILES = 4
_AUDIO_EXT = (".wav", ".mp3", ".flac", ".ogg", ".m4a", ".opus", ".webm", ".aac")
def _copy_limited(fobj, dest, limit=MAX_UPLOAD_BYTES):
"""Copy an upload to disk, refusing anything above `limit` bytes."""
size = 0
try:
with open(dest, "wb") as out:
while True:
chunk = fobj.read(1 << 20)
if not chunk: break
size += len(chunk)
if size > limit:
raise HTTPException(413, f"الملف أكبر من {limit >> 20} MB.")
out.write(chunk)
except Exception:
try: os.remove(dest)
except Exception: pass
raise
_VOICE_NAME_RE = re.compile(r"[^\w\- ]", re.UNICODE)
def safe_voice_name(name) -> str:
return _VOICE_NAME_RE.sub("_", (name or "").strip()).strip(" ._")[:40]
def voice_dir(name, must_exist=False) -> str:
"""Resolve a voice-library folder, guaranteed to live directly inside VOICE_LIB
(blocks '..', absolute paths and separators). Names saved by older versions
are still found via an exact-name lookup."""
lib = os.path.realpath(VOICE_LIB)
raw = (name or "").strip()
found = None
for cand in dict.fromkeys([raw, safe_voice_name(raw)]):
if not cand or cand in (".", "..") or any(c in cand for c in "/\\\x00"):
continue
p = os.path.realpath(os.path.join(lib, cand))
if os.path.dirname(p) != lib:
continue
if os.path.isdir(p): return p
found = found or p
if found is None: raise HTTPException(400, "اسم الصوت غير صالح.")
if must_exist: raise HTTPException(404, "صوت غير موجود.")
return found
def _output_path(filename: str) -> str:
safe = os.path.basename(filename or "")
if not safe or safe != filename or ".." in safe:
raise HTTPException(400, "اسم ملف غير صالح.")
return os.path.join(OUTPUT_DIR, safe)
def collect_refs(files, voice_name):
paths = []
try:
for f in files:
if not f or not getattr(f, "filename", None): continue
if len(paths) >= MAX_REF_FILES: break
base = re.sub(r"[^\w.\-]", "_", os.path.basename(f.filename), flags=re.UNICODE)[-60:] or "ref.wav"
p = f"/tmp/ref_{uuid.uuid4().hex}_{base}"
_copy_limited(f.file, p)
paths.append(p)
if voice_name and voice_name not in ("", "null", "none", "undefined"):
lib_dir = voice_dir(voice_name)
if os.path.isdir(lib_dir):
paths += [os.path.join(lib_dir, fn) for fn in sorted(os.listdir(lib_dir))
if fn.lower().endswith(_AUDIO_EXT)]
except Exception:
cleanup(paths); raise
return paths
def cleanup(paths):
for p in paths:
if p and p.startswith("/tmp/"):
try: os.remove(p)
except: pass
# ── FastAPI ───────────────────────────────────────────────────────────
app = FastAPI(title="XTTS Voice Studio CPU Optimized")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
@app.middleware("http")
async def _limit_request_size(request, call_next):
cl = request.headers.get("content-length", "")
if request.method in ("POST", "PUT") and cl.isdigit() and int(cl) > MAX_REQUEST_BYTES:
return JSONResponse({"detail": f"الطلب أكبر من {MAX_REQUEST_BYTES >> 20} MB."}, status_code=413)
return await call_next(request)
@app.exception_handler(Exception)
async def _unhandled_error(request, exc):
print(f"[!] Unhandled error on {request.method} {request.url.path}:", flush=True)
traceback.print_exception(type(exc), exc, exc.__traceback__)
return JSONResponse({"detail": f"خطأ داخلي ({type(exc).__name__}). راجع سجلات الخادم."}, status_code=500)
LANGUAGES = {
"ar":"العربية","en":"English","es":"Español","fr":"Français",
"de":"Deutsch","it":"Italiano","pt":"Português","ru":"Русский",
"zh-cn":"中文","ja":"日本語","ko":"한국어","tr":"Türkçe",
"nl":"Nederlands","pl":"Polski","cs":"Čeština","hi":"हिन्दी",
}
LANG_MAP = {"zh-cn":"zh-CN"}
# ══════════════════════════════════════════════════════════════════════
# F R O N T E N D
# ══════════════════════════════════════════════════════════════════════
HTML = r"""
XTTS Voice Studio
"""
# ══════════════════════════════════════════════════════════════════════
# R O U T E S
# ══════════════════════════════════════════════════════════════════════
@app.get("/", response_class=HTMLResponse)
async def ui():
return (HTML
.replace("DEVICE_PH", device)
.replace("LANGS_PH", json.dumps(LANGUAGES, ensure_ascii=False)))
# ── Model status endpoint ─────────────────────────────────────────────
@app.get("/model_status")
def model_status():
return JSONResponse({
"loading": xtts_loading,
"ready": xtts_ready,
"error": xtts_error,
"device": device,
"whisper": bool(WHISPER_OK),
"zerogpu": bool(ZEROGPU),
})
_STARTED_AT = time.time()
@app.get("/health")
def health():
"""Always HTTP 200 (so health checks never restart a model that is still loading)."""
return {"status": "ok" if xtts_ready else ("loading" if xtts_loading else "error"),
"model_ready": bool(xtts_ready), "whisper_ready": bool(WHISPER_OK),
"busy": get_tts_lock().locked(), "uptime_s": int(time.time() - _STARTED_AT)}
@app.get("/diagnostics")
def diagnostics():
import importlib.metadata as md
def ver(n):
try: return md.version(n)
except Exception: return "n/a"
def folder(d):
try:
fs = [os.path.join(d, f) for f in os.listdir(d)]
return {"files": len(fs), "mb": round(sum(os.path.getsize(f) for f in fs if os.path.isfile(f)) / 1e6, 1)}
except Exception:
return None
return {
"versions": {k: ver(k) for k in ("torch", "coqui-tts", "transformers", "faster-whisper", "numpy", "spaces")},
"device": device, "zerogpu": bool(ZEROGPU), "cpu_threads": CPU_THREADS,
"model_ready": bool(xtts_ready), "model_error": xtts_error, "whisper_ready": bool(WHISPER_OK),
"caches": {"reference": folder(REF_CACHE_DIR), "dry_audio": folder(DRY_CACHE_DIR), "outputs": folder(OUTPUT_DIR)},
"latent_cache_entries": len(_LAT_CACHE),
"engine": "voice-quality v" + vq.VQ_VERSION,
"recent_jobs": list(RECENT_JOBS),
}
# ── Job control endpoints ─────────────────────────────────────────────
@app.get("/progress/{job_id}")
def get_progress(job_id: str):
_gc_jobs()
p = job_progress.get(job_id, {"current": 0, "total": 1, "status": "unknown"})
return JSONResponse(p)
@app.post("/cancel/{job_id}")
def cancel_job(job_id: str):
ev = cancel_events.get(job_id)
if ev:
ev.set()
return {"cancelled": True}
# ── Audio post-edit endpoint ──────────────────────────────────────────
@app.post("/edit_audio")
async def edit_audio(
filename: str = Form(...),
volume_db: float = Form(0.0),
speed_factor: float = Form(1.0),
trim_start: float = Form(0.0),
trim_end: float = Form(0.0),
fade_in: float = Form(0.0), # seconds
fade_out: float = Form(0.0), # seconds
normalize: bool = Form(False), # peak-normalize to 0 dBFS
):
"""Edit a generated audio file without re-running TTS.
Always writes a NEW file — the original is never modified."""
src = _output_path(filename)
if not os.path.exists(src):
raise HTTPException(404, "الملف غير موجود.")
def _edit():
data, sr = sf.read(src)
if data.ndim > 1:
data = data.mean(axis=1)
# ── Trim start / end ──────────────────────────────────────────
if trim_start > 0:
data = data[min(int(trim_start * sr), len(data)):]
if trim_end > 0:
end_sample = max(0, len(data) - int(trim_end * sr))
data = data[:end_sample]
if len(data) == 0:
raise ValueError("الملف فارغ بعد القطع — قلّل قيم القطع.")
# ── Speed via polyphase resampling ────────────────────────────
if abs(speed_factor - 1.0) > 0.01:
try:
from scipy.signal import resample_poly
from math import gcd
# Represent speed as a rational fraction p/q
p = round(speed_factor * 1000)
q = 1000
g = gcd(p, q)
# resample_poly(data, up, down): output length = len * up/down
# We want output_len = input_len / speed ⟹ up=q, down=p
data = resample_poly(data, q // g, p // g).astype(np.float32)
except ImportError:
# Fallback: crude integer resampling via scipy.signal.resample
try:
from scipy.signal import resample as sp_resample
new_len = max(1, int(len(data) / speed_factor))
data = sp_resample(data, new_len).astype(np.float32)
except ImportError:
pass # scipy not available
# ── Normalize (peak to 0 dBFS) ────────────────────────────────
if normalize:
peak = np.max(np.abs(data))
if peak > 0:
data = data / peak * 0.98
# ── Volume (dB) ───────────────────────────────────────────────
if abs(volume_db) > 0.01:
factor = 10 ** (volume_db / 20.0)
data = data * factor
# Soft clip via tanh to prevent digital distortion
data = np.tanh(data * 1.2) * (1.0 / np.tanh(1.2))
# ── Fade In (linear ramp) ─────────────────────────────────────
if fade_in > 0:
n = min(int(fade_in * sr), len(data))
data[:n] *= np.linspace(0.0, 1.0, n, dtype=np.float32)
# ── Fade Out (linear ramp) ────────────────────────────────────
if fade_out > 0:
n = min(int(fade_out * sr), len(data))
data[-n:] *= np.linspace(1.0, 0.0, n, dtype=np.float32)
out = os.path.join(OUTPUT_DIR, f"edit_{uuid.uuid4().hex[:8]}.wav")
sf.write(out, data, sr, subtype='PCM_16')
return os.path.basename(out)
try:
new_filename = await asyncio.to_thread(_edit)
except ValueError as e:
raise HTTPException(400, str(e))
return {"filename": new_filename}
# ── Humanize endpoint ─────────────────────────────────────────────────
@app.post("/humanize")
async def humanize_endpoint(
filename: str = Form(...),
preset: str = Form("bedroom"),
intensity: float = Form(1.0),
seed: int = Form(42),
):
"""Apply acoustic environment simulation to a generated WAV.
Returns a brand-new file — original untouched."""
src = _output_path(filename)
if not os.path.exists(src):
raise HTTPException(404, "الملف غير موجود.")
if preset not in HUMANIZE_PRESETS:
raise HTTPException(400, f"بيئة غير معروفة: {preset}")
intensity = float(np.clip(intensity, 0.05, 2.0))
out = os.path.join(OUTPUT_DIR, f"hum_{uuid.uuid4().hex[:8]}.wav")
try:
await asyncio.to_thread(humanize_audio, src, out, preset, intensity, seed)
except Exception as e:
if os.path.exists(out):
os.remove(out)
raise HTTPException(500, f"خطأ في المعالجة: {e}")
return {"filename": os.path.basename(out)}
@app.get("/humanize/presets")
def humanize_presets():
return list(HUMANIZE_PRESETS.keys())
@app.post("/generate")
async def generate(
background_tasks: BackgroundTasks,
text: str = Form(...),
job_id: str = Form(default=""),
language: str = Form("ar"),
stability: float = Form(0.5),
clarity: float = Form(0.75),
style_exaggeration: float = Form(0.0),
speed: float = Form(1.0),
speaker_boost: bool = Form(False),
enable_text_splitting: bool = Form(True),
expression_mode: str = Form("natural"),
variation_strength: float = Form(1.0),
enhance_ref: bool = Form(True),
clean_text: bool = Form(True),
attempts: int = Form(1),
depth: float = Form(0.5),
space: str = Form("studio"),
voice_name: str = Form(None),
files: list[UploadFile] = File(default=[]),
):
if xtts_loading: raise HTTPException(503, "⏳ النموذج لا يزال يُحمَّل، يُرجى الانتظار قليلاً ثم المحاولة مجدداً.")
if not xtts_ready: raise HTTPException(503, f"❌ فشل تحميل النموذج: {xtts_error}")
if not text.strip(): raise HTTPException(400, "النص فارغ.")
if len(text) > MAX_CHARS: raise HTTPException(400, f"النص يتجاوز {MAX_CHARS} حرف.")
if language not in LANGUAGES: raise HTTPException(400, f"لغة غير مدعومة: {language}")
if not job_id:
job_id = uuid.uuid4().hex
enhanced_text = enhance_text_expression(text, language, expression_mode)
tts_params = apply_expression_preset(expression_mode, speed)
if expression_mode == "natural":
tts_params = dict(
temperature=stability_to_temperature(stability),
top_p=clarity_to_top_p(clarity),
repetition_penalty=style_to_rep_penalty(style_exaggeration),
speed=speed,
)
refs = collect_refs(files, voice_name)
if not refs: raise HTTPException(400, "يجب تحديد عينة صوتية مرجعية.")
out = os.path.join(OUTPUT_DIR, f"gen_{uuid.uuid4().hex[:10]}.wav")
cancel_events[job_id] = threading.Event()
silence_ms = {"story":160,"sad":180,"news":80,"excited":70}.get(expression_mode, 110)
variation = float(np.clip(variation_strength, 0.0, 2.0))
ref_summary, job_stats = {}, {}
try:
async with get_tts_lock():
success = await asyncio.to_thread(
run_tts_chunked, enhanced_text, refs, language,
tts_params, out, job_id, silence_ms, variation,
attempts, enhance_ref, clean_text, ref_summary, job_stats
)
if not success:
if os.path.exists(out): os.remove(out)
raise HTTPException(499, "تم إلغاء التوليد.")
if speaker_boost:
await asyncio.to_thread(apply_speaker_boost, out)
await asyncio.to_thread(apply_depth, out, depth, space)
finally:
cleanup(refs)
cancel_events.pop(job_id, None)
append_history(os.path.basename(out), text, language, "tts", len(text))
return {"filename": os.path.basename(out), "ref_report": ref_summary or None,
"cached": bool(job_stats.get("cached")), "stats": job_stats or None}
# ── Reference check: clean the sample and let the user LISTEN to it ───
@app.post("/enhance_ref")
async def enhance_ref_endpoint(
voice_name: str = Form(None),
files: list[UploadFile] = File(default=[]),
):
refs = collect_refs(files, voice_name)
if not refs: raise HTTPException(400, "حدد عينة صوتية أولاً.")
try:
paths, _reports, summary = await asyncio.to_thread(vq.enhance_reference_set, refs, REF_CACHE_DIR)
primary = paths[0] # best contiguous segment (others are used too, never joined)
fn = "refclean_" + os.path.basename(primary)[4:-4] + ".wav"
shutil.copyfile(primary, os.path.join(OUTPUT_DIR, fn))
return {"url": f"/audio/{fn}", "report": summary, "segments": len(paths)}
except ValueError as e:
raise HTTPException(422, str(e))
finally:
cleanup(refs)
# ── STS ───────────────────────────────────────────────────────────────
@app.post("/sts")
async def sts(
background_tasks: BackgroundTasks,
source_audio: UploadFile = File(...),
job_id: str = Form(default=""),
language: str = Form("ar"),
stability: float = Form(0.5),
clarity: float = Form(0.75),
style_exaggeration: float = Form(0.0),
speed: float = Form(1.0),
speaker_boost: bool = Form(False),
variation_strength_sts: float = Form(1.0),
depth: float = Form(0.5),
space: str = Form("studio"),
voice_name: str = Form(None),
ref_files: list[UploadFile] = File(default=[]),
):
if xtts_loading: raise HTTPException(503, "⏳ النموذج لا يزال يُحمَّل، يُرجى الانتظار.")
if not xtts_ready: raise HTTPException(503, f"❌ فشل تحميل النموذج: {xtts_error}")
if not WHISPER_OK: raise HTTPException(503, "Faster-Whisper غير جاهز بعد (لا يزال يُحمَّل أو فشل تحميله)، أعد المحاولة بعد قليل.")
if not job_id: job_id = uuid.uuid4().hex
src = f"/tmp/sts_{uuid.uuid4().hex}.wav"
_copy_limited(source_audio.file, src)
refs = []
cancel_events[job_id] = threading.Event()
try:
text = await asyncio.to_thread(transcribe_audio, src)
if not text: raise HTTPException(422, "لم يتمكن Whisper من نسخ الصوت.")
refs = collect_refs(ref_files, voice_name)
if not refs: raise HTTPException(400, "حدد صوت الهدف.")
out = os.path.join(OUTPUT_DIR, f"sts_{uuid.uuid4().hex[:10]}.wav")
sts_params = dict(
temperature=stability_to_temperature(stability),
top_p=clarity_to_top_p(clarity),
repetition_penalty=style_to_rep_penalty(style_exaggeration),
speed=speed,
)
var_sts = float(np.clip(variation_strength_sts, 0.0, 2.0))
async with get_tts_lock():
success = await asyncio.to_thread(
run_tts_chunked, text, refs, language, sts_params, out, job_id, 110, var_sts
)
if not success:
if os.path.exists(out): os.remove(out)
raise HTTPException(499, "تم إلغاء التحويل.")
if speaker_boost:
await asyncio.to_thread(apply_speaker_boost, out)
await asyncio.to_thread(apply_depth, out, depth, space)
finally:
cleanup([src]); cleanup(refs)
cancel_events.pop(job_id, None)
append_history(os.path.basename(out), text, language, "sts", len(text))
return {"filename": os.path.basename(out), "transcribed": text}
# ── DUB ───────────────────────────────────────────────────────────────
@app.post("/dub")
async def dub(
background_tasks: BackgroundTasks,
source_audio: UploadFile = File(...),
job_id: str = Form(default=""),
target_language: str = Form("en"),
stability: float = Form(0.5),
clarity: float = Form(0.75),
style_exaggeration: float = Form(0.0),
speed: float = Form(1.0),
speaker_boost: bool = Form(False),
variation_strength_dub: float = Form(1.0),
depth: float = Form(0.5),
space: str = Form("studio"),
voice_name: str = Form(None),
ref_files: list[UploadFile] = File(default=[]),
):
if xtts_loading: raise HTTPException(503, "⏳ النموذج لا يزال يُحمَّل، يُرجى الانتظار.")
if not xtts_ready: raise HTTPException(503, f"❌ فشل تحميل النموذج: {xtts_error}")
if not WHISPER_OK: raise HTTPException(503, "Faster-Whisper غير جاهز بعد (لا يزال يُحمَّل أو فشل تحميله)، أعد المحاولة بعد قليل.")
if not TRANSLATOR_OK: raise HTTPException(503, "deep-translator غير مثبّت.")
if not job_id: job_id = uuid.uuid4().hex
src = f"/tmp/dub_{uuid.uuid4().hex}.wav"
_copy_limited(source_audio.file, src)
refs = []
cancel_events[job_id] = threading.Event()
try:
original = await asyncio.to_thread(transcribe_audio, src)
if not original: raise HTTPException(422, "لم يتمكن Whisper من نسخ الصوت.")
tgt_code = LANG_MAP.get(target_language, target_language)
translated = await asyncio.to_thread(
GoogleTranslator(source="auto", target=tgt_code).translate, original)
if not translated: raise HTTPException(422, "فشلت الترجمة.")
refs = collect_refs(ref_files, voice_name)
if not refs: raise HTTPException(400, "حدد صوت الدبلجة.")
out = os.path.join(OUTPUT_DIR, f"dub_{uuid.uuid4().hex[:10]}.wav")
dub_params = dict(
temperature=stability_to_temperature(stability),
top_p=clarity_to_top_p(clarity),
repetition_penalty=style_to_rep_penalty(style_exaggeration),
speed=speed,
)
var_dub = float(np.clip(variation_strength_dub, 0.0, 2.0))
async with get_tts_lock():
success = await asyncio.to_thread(
run_tts_chunked, translated, refs, target_language, dub_params, out, job_id, 110, var_dub
)
if not success:
if os.path.exists(out): os.remove(out)
raise HTTPException(499, "تم إلغاء الدبلجة.")
if speaker_boost:
await asyncio.to_thread(apply_speaker_boost, out)
await asyncio.to_thread(apply_depth, out, depth, space)
finally:
cleanup([src]); cleanup(refs)
cancel_events.pop(job_id, None)
append_history(os.path.basename(out), translated, target_language, "dub", len(translated))
return {"filename": os.path.basename(out), "original": original, "translated": translated}
# ── SFX ───────────────────────────────────────────────────────────────
@app.post("/sfx")
async def sfx(
background_tasks: BackgroundTasks,
description: str = Form(...),
job_id: str = Form(default=""),
stability: float = Form(0.1),
clarity: float = Form(0.5),
style_exaggeration: float = Form(0.9),
speed: float = Form(0.85),
speaker_boost: bool = Form(False),
voice_name: str = Form(None),
ref_file: UploadFile = File(default=None),
):
if xtts_loading: raise HTTPException(503, "⏳ النموذج لا يزال يُحمَّل، يُرجى الانتظار.")
if not xtts_ready: raise HTTPException(503, f"❌ فشل تحميل النموذج: {xtts_error}")
if not description.strip(): raise HTTPException(400, "الوصف فارغ.")
if not job_id: job_id = uuid.uuid4().hex
ref_list = [ref_file] if (ref_file and getattr(ref_file, "filename", None)) else []
refs = collect_refs(ref_list, voice_name)
if not refs: raise HTTPException(400, "حدد صوتاً مرجعياً.")
out = os.path.join(OUTPUT_DIR, f"sfx_{uuid.uuid4().hex[:10]}.wav")
cancel_events[job_id] = threading.Event()
sfx_params = dict(
temperature=stability_to_temperature(stability),
top_p=clarity_to_top_p(clarity),
repetition_penalty=style_to_rep_penalty(style_exaggeration),
speed=speed,
)
try:
async with get_tts_lock():
success = await asyncio.to_thread(
run_tts_chunked, description, refs, "en", sfx_params, out, job_id
)
if not success:
if os.path.exists(out): os.remove(out)
raise HTTPException(499, "تم إلغاء التوليد.")
if speaker_boost:
await asyncio.to_thread(apply_speaker_boost, out)
finally:
cleanup(refs)
cancel_events.pop(job_id, None)
append_history(os.path.basename(out), description, "en", "sfx", len(description))
return {"filename": os.path.basename(out)}
# ── Audio / Voice endpoints ───────────────────────────────────────────
@app.get("/audio/{filename}")
def audio(filename: str):
safe = os.path.basename(filename)
if safe != filename or ".." in safe:
raise HTTPException(400, "اسم ملف غير صالح.")
p = os.path.join(OUTPUT_DIR, safe)
if not os.path.exists(p): raise HTTPException(404, "ملف غير موجود.")
return FileResponse(p, media_type="audio/wav",
headers={"Accept-Ranges": "bytes", "Cache-Control": "no-cache"})
@app.get("/history")
def get_history():
return JSONResponse(load_history())
@app.get("/voices")
def list_voices():
if not os.path.isdir(VOICE_LIB): return []
return [d for d in os.listdir(VOICE_LIB) if os.path.isdir(os.path.join(VOICE_LIB, d))]
@app.post("/voices/save")
async def save_voice(background_tasks: BackgroundTasks,
name: str = Form(...),
file: UploadFile = File(...),
file2: UploadFile = File(default=None)):
safe = safe_voice_name(name)
if not safe: raise HTTPException(400, "اسم الصوت غير صالح.")
uploads = [f for f in ([file, file2] if file2 and file2.filename else [file]) if f and f.filename]
names = []
for i, f in enumerate(uploads):
fn = re.sub(r"[^\w.\-]", "_", os.path.basename(f.filename), flags=re.UNICODE)[:80] or f"ref{i}.wav"
if not fn.lower().endswith(_AUDIO_EXT):
raise HTTPException(400, "صيغة الملف غير مدعومة (wav / mp3 / flac / ogg / m4a).")
names.append(fn)
if not uploads: raise HTTPException(400, "لا يوجد ملف صوتي.")
lib = voice_dir(safe)
os.makedirs(lib, exist_ok=True)
for f, fn in zip(uploads, names):
_copy_limited(f.file, os.path.join(lib, fn))
background_tasks.add_task(trigger_cloud_backup)
return {"name": os.path.basename(lib)}
@app.delete("/voices/{name}")
def delete_voice(background_tasks: BackgroundTasks, name: str):
lib = voice_dir(name, must_exist=True)
shutil.rmtree(lib)
background_tasks.add_task(trigger_cloud_backup)
return {"deleted": name}
@app.get("/voices/{name}/preview")
def preview_voice(name: str):
lib = voice_dir(name, must_exist=True)
for fn in sorted(os.listdir(lib)):
if fn.lower().endswith(_AUDIO_EXT):
mt = {"wav": "audio/wav", "mp3": "audio/mpeg", "flac": "audio/flac", "ogg": "audio/ogg",
"m4a": "audio/mp4", "opus": "audio/ogg", "webm": "audio/webm", "aac": "audio/aac"}[fn.rsplit(".", 1)[-1].lower()]
return FileResponse(os.path.join(lib, fn), media_type=mt)
raise HTTPException(404, "لا توجد ملفات صوتية.")
# ══════════════════════════════════════════════════════════════════════
if __name__ == "__main__":
# ZeroGPU: the `spaces` package only reports its @spaces.GPU functions to
# the platform from inside gradio's Blocks.launch(). This app serves with
# uvicorn directly, so we send that start-up report ourselves; otherwise
# the Space is killed with "No @spaces.GPU function detected during startup".
if ZEROGPU:
try:
_startup = getattr(spaces.zero, "startup", None) # only exists on ZeroGPU
if callable(_startup):
_startup()
print("[✓] ZeroGPU start-up report sent.", flush=True)
else:
print("[i] ZeroGPU not active on this hardware — no start-up report needed.", flush=True)
except Exception:
print("[!] ZeroGPU start-up report failed:", flush=True)
traceback.print_exc()
uvicorn.run(app, host="0.0.0.0", port=7860, log_level="info")