Create app.py

app.py

@@ -0,0 +1,373 @@
+# Калі запускаеце ў чыстым асяроддзі (раскаментуйце):
+# !pip install -q gradio spaces huggingface_hub torch scipy tqdm gitpython
+
+import os
+import sys
+import time
+import tempfile
+import subprocess
+import inspect
+import re
+
+import spaces
+import gradio as gr
+import torch
+from huggingface_hub import hf_hub_download
+from scipy.io.wavfile import write
+import numpy as np
+
+# ---------------------------------------------------------
+# 1) Клануем і падключаем coqui-ai-TTS (fork з падтрымкай BE)
+# ---------------------------------------------------------
+REPO_URL = "https://github.com/tuteishygpt/coqui-ai-TTS.git"
+REPO_DIR = "coqui-ai-TTS"
+
+if not os.path.exists(REPO_DIR):
+    subprocess.run(["git", "clone", REPO_URL, REPO_DIR], check=True)
+
+repo_root = os.path.abspath(REPO_DIR)
+if repo_root not in sys.path:
+    sys.path.insert(0, repo_root)
+
+from TTS.tts.configs.xtts_config import XttsConfig
+from TTS.tts.models.xtts import Xtts
+from TTS.tts.layers.xtts.tokenizer import VoiceBpeTokenizer
+
+# ---------------------------------------------------------
+# 2) Файлы мадэлі
+# ---------------------------------------------------------
+repo_id = "archivartaunik/BE_XTTS_V2_10ep250k"
+model_dir = "./model"
+os.makedirs(model_dir, exist_ok=True)
+
+checkpoint_file = os.path.join(model_dir, "model.pth")
+config_file = os.path.join(model_dir, "config.json")
+vocab_file = os.path.join(model_dir, "vocab.json")
+default_voice_file = os.path.join(model_dir, "voice.wav")
+
+if not os.path.exists(checkpoint_file):
+    hf_hub_download(repo_id, filename="model.pth", local_dir=model_dir)
+if not os.path.exists(config_file):
+    hf_hub_download(repo_id, filename="config.json", local_dir=model_dir)
+if not os.path.exists(vocab_file):
+    hf_hub_download(repo_id, filename="vocab.json", local_dir=model_dir)
+if not os.path.exists(default_voice_file):
+    hf_hub_download(repo_id, filename="voice.wav", local_dir=model_dir)
+
+# ---------------------------------------------------------
+# 3) Загрузка мадэлі і токенайзера
+# ---------------------------------------------------------
+config = XttsConfig()
+config.load_json(config_file)
+XTTS_MODEL = Xtts.init_from_config(config)
+XTTS_MODEL.load_checkpoint(
+    config,
+    checkpoint_path=checkpoint_file,
+    vocab_path=vocab_file,
+    use_deepspeed=False,
+)
+
+device = "cuda:0" if torch.cuda.is_available() else "cpu"
+XTTS_MODEL.to(device).eval()
+sampling_rate = int(XTTS_MODEL.config.audio["sample_rate"])
+
+tokenizer = VoiceBpeTokenizer(vocab_file=vocab_file)
+XTTS_MODEL.tokenizer = tokenizer
+
+# ---------------------------------------------------------
+# 4) Патокавая TTS па "токенах" з мінімальнай затрымкай
+#    (натыйўны стримінг -> fallback інкрементальны прэфікс)
+# ---------------------------------------------------------
+
+MIN_BUFFER_MS = 0.05   # мэтавы мінімальны буфер ~50 ms
+FADE_MS = 0.008        # кароткі cross-fade паміж чанкамі
+TOKENS_PER_STEP = 4    # памер кроку «токенаў» у fallback (BPE/субсловы)
+
+def _seconds_to_samples(sec: float, sr: int) -> int:
+    return max(1, int(sec * sr))
+
+def _crossfade_concat(a: np.ndarray, b: np.ndarray, sr: int, fade_ms: float) -> np.ndarray:
+    """Плыўна зліць два кавалкі без клікаў."""
+    if a.size == 0:
+        return b.astype(np.float32, copy=False)
+    if b.size == 0:
+        return a.astype(np.float32, copy=False)
+    a = a.astype(np.float32, copy=False)
+    b = b.astype(np.float32, copy=False)
+    fade_n = _seconds_to_samples(fade_ms, sr)
+    fade_n = min(fade_n, a.size, b.size)
+    if fade_n <= 1:
+        return np.concatenate([a, b], axis=0)
+    fade_out = np.linspace(1.0, 0.0, fade_n, endpoint=True, dtype=np.float32)
+    fade_in  = 1.0 - fade_out
+    head = a[:-fade_n]
+    tail = (a[-fade_n:] * fade_out) + (b[:fade_n] * fade_in)
+    rest = b[fade_n:]
+    return np.concatenate([head, tail, rest], axis=0)
+
+def _merge_chunks_with_crossfade(chunks: list[np.ndarray], sr: int) -> np.ndarray:
+    merged = np.zeros((0,), dtype=np.float32)
+    for c in chunks:
+        if c is None or np.asarray(c).size == 0:
+            continue
+        merged = _crossfade_concat(merged, np.asarray(c, dtype=np.float32), sr, FADE_MS)
+    return merged
+
+def _yield_buffered_chunks(chunks: list[np.ndarray], sr: int, target_ms: float):
+    """
+    Збіраем маленькія кавалкі пакуль не назапасім ~target_ms,
+    пасля чаго yield (sr, buffer) і спім роўна на працягласць buffer.
+    """
+    target_samples = _seconds_to_samples(target_ms, sr)
+    buf = np.zeros((0,), dtype=np.float32)
+    for c in chunks:
+        if c is None:
+            continue
+        c = np.asarray(c, dtype=np.float32)
+        if c.size == 0:
+            continue
+        if buf.size == 0:
+            buf = c
+        else:
+            buf = _crossfade_concat(buf, c, sr, FADE_MS)
+        if buf.size >= target_samples:
+            yield (sr, buf)
+            # даём плэеру «дагуляць» без накладання
+            time.sleep(buf.size / float(sr))
+            buf = np.zeros((0,), dtype=np.float32)
+    if buf.size:
+        yield (sr, buf)
+        time.sleep(buf.size / float(sr))
+
+def _bpe_prefixes(text: str, lang: str, step_tokens: int):
+    """
+    Вяртае паступовыя прэфіксы тэксту па BPE/субсловах, калі атрымліваецца.
+    Інакш — fallback па «псэўда-токенах» (словы+прабелы/пунктуацыя).
+    """
+    # 1) Спроба праз VoiceBpeTokenizer (калі ёсць encode/decode)
+    try:
+        # у вашым форку можа быць encode(text, lang=...), decode(ids, lang=...)
+        ids = tokenizer.encode(text, lang=lang)
+        n = len(ids)
+        for k in range(step_tokens, n + 1, step_tokens):
+            prefix = tokenizer.decode(ids[:k], lang=lang)
+            yield prefix
+        if n % step_tokens != 0:
+            yield tokenizer.decode(ids, lang=lang)
+        return
+    except Exception:
+        pass
+
+    # 2) Падстрахоўка: разбіць на «словы+знакі»
+    pseudo_tokens = re.findall(r"\S+|\s+", text)
+    buff = ""
+    for i in range(0, len(pseudo_tokens), step_tokens):
+        buff = "".join(pseudo_tokens[: i + step_tokens])
+        yield buff
+    if buff.strip() != text.strip():
+        yield text
+
+def _stream_from_model_native(text: str, gpt_cond_latent, speaker_embedding, sr: int, lang: str):
+    """
+    Калі ў форку ёсць натыўны струмень (inference_stream) — карыстаемся ім.
+    Павінен yield'іць PCM фрагменты па меры дэкавання.
+    """
+    # Сфармуем kwargs у залежнасці ад подпісу функцыі
+    common_kwargs = dict(
+        text=text,
+        language=lang,
+        gpt_cond_latent=gpt_cond_latent,
+        speaker_embedding=speaker_embedding,
+        temperature=0.1,
+        length_penalty=1.0,
+        repetition_penalty=10.0,
+        top_k=10,
+        top_p=0.3,
+    )
+    sig = inspect.signature(XTTS_MODEL.inference_stream)
+    if "stream_chunk_size_s" in sig.parameters:
+        common_kwargs["stream_chunk_size_s"] = MIN_BUFFER_MS
+
+    gen = XTTS_MODEL.inference_stream(**common_kwargs)
+    raw_chunks = []
+    for out in gen:
+        cur = out["wav"] if isinstance(out, dict) and "wav" in out else np.asarray(out, dtype=np.float32)
+        cur = cur.astype(np.float32, copy=False)
+        raw_chunks.append(cur)
+        # выдаём дробнымі порцыямі з невялікім буферам
+        yield from _yield_buffered_chunks([cur], sr, MIN_BUFFER_MS)
+
+    # Сабраць «хвост» у адзін WAV з лёгкім cross-fade
+    if raw_chunks:
+        final_full = _merge_chunks_with_crossfade(raw_chunks, sr)
+        yield ("__FINAL__", final_full)
+
+def _stream_fallback_incremental(text: str, gpt_cond_latent, speaker_embedding, sr: int, lang: str):
+    """
+    Fallback: павялічваем прэфікс тэксту па токенах і кожны раз
+    генеруем гукавыя дадаткі (толькі «хвост» новай версіі).
+    """
+    emitted = 0
+    last_full = np.zeros((0,), dtype=np.float32)
+
+    for prefix in _bpe_prefixes(text, lang, TOKENS_PER_STEP):
+        with torch.no_grad():
+            wav = XTTS_MODEL.inference(
+                text=prefix,
+                language=lang,
+                gpt_cond_latent=gpt_cond_latent,
+                speaker_embedding=speaker_embedding,
+                temperature=0.1,
+                length_penalty=1.0,
+                repetition_penalty=10.0,
+                top_k=10,
+                top_p=0.3,
+            )["wav"].astype(np.float32)
+
+        # бярэм толькі новую частку адносна ўжо аддадзенага
+        new_part = wav[emitted:]
+        if new_part.size > 0:
+            yield from _yield_buffered_chunks([new_part], sr, MIN_BUFFER_MS)
+            emitted = wav.size
+            last_full = wav
+
+    if last_full.size:
+        yield ("__FINAL__", last_full)
+
+@spaces.GPU(duration=60)
+def text_to_speech(belarusian_story, speaker_audio_file=None):
+    """
+    Патокавы вывад з мінімальнай затрымкай.
+    - Крок 1: латэнты голасу.
+    - Крок 2: спроба натыўнага streaming з мадэлі.
+    - Крок 3: fallback — інкрементальны прэфікс (токен-крокі).
+    Выхад для gr.Audio: шмат (sr, chunk) + у фінале шлях да поўнага WAV.
+    """
+    if not belarusian_story or str(belarusian_story).strip() == "":
+        raise gr.Error("Увядзі хоць нейкі тэкст 🙂")
+
+    # Голас па змаўчанні
+    if not speaker_audio_file or (
+        not isinstance(speaker_audio_file, str)
+        and getattr(speaker_audio_file, "name", "") == ""
+    ):
+        speaker_audio_file = default_voice_file
+
+    # Conditioning latents
+    try:
+        gpt_cond_latent, speaker_embedding = XTTS_MODEL.get_conditioning_latents(
+            audio_path=speaker_audio_file,
+            gpt_cond_len=XTTS_MODEL.config.gpt_cond_len,
+            max_ref_length=XTTS_MODEL.config.max_ref_len,
+            sound_norm_refs=XTTS_MODEL.config.sound_norm_refs,
+        )
+    except Exception as e:
+        raise gr.Error(f"Памылка пры атрыманні латэнтаў голасу: {e}")
+
+    lang = "be"
+    full_audio = None
+
+    # 1) Спачатку — натыўны паток, калі ёсць
+    try:
+        if hasattr(XTTS_MODEL, "inference_stream"):
+            for out in _stream_from_model_native(
+                belarusian_story, gpt_cond_latent, speaker_embedding, sampling_rate, lang
+            ):
+                if isinstance(out, tuple) and out and out[0] == "__FINAL__":
+                    full_audio = out[1]
+                else:
+                    yield out
+        else:
+            raise AttributeError("No native inference_stream in this build.")
+    except Exception:
+        # 2) fallback — інкрементальны прэфікс (токен-крокі)
+        for out in _stream_fallback_incremental(
+            belarusian_story, gpt_cond_latent, speaker_embedding, sampling_rate, lang
+        ):
+            if isinstance(out, tuple) and out and out[0] == "__FINAL__":
+                full_audio = out[1]
+            else:
+                yield out
+
+    if full_audio is None:
+        raise gr.Error("Нічога не згенеравана. Праверце ўваходныя даныя або лагі.")
+
+    # Фінальны WAV у temp-файл
+    try:
+        temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".wav")
+        write(temp_file.name, sampling_rate, full_audio.astype(np.float32))
+        yield temp_file.name
+    except Exception as e:
+        raise gr.Error(f"Памылка пры запісе фінальнага WAV: {e}")
+
+# ---------------------------------------------------------
+# 5) Прыклады (тэкст + файл голасу)
+# ---------------------------------------------------------
+examples = [
+    [
+        "Такім чынам, клуб стаў уладальнікам усіх існых на сёння міжнародных трафеяў паўднёваамерыканскага футболу.",
+        "Nestarka.wav",
+    ],
+    [
+        "Яму не ўдалося палепшыць фінансавае становішча каралеўства, а, наадварот, прыйшлося распрадаваць каштоўнасці чэшскай кароны.",
+        "muzh.wav",
+    ],
+    [
+        "Кампілятарамі называюць праграмы, якія пераўтвараюць код вышэйшага ўзроўню ў код ніжэйшага ўзроўню.",
+        "chunk_100.wav",
+    ],
+    [
+        "Акрамя таго, ліхачы аддаюць перавагу рэгі, хіп-хопу і класічнай музыцы.",
+        "d1015.mp3",
+    ],
+    [
+        "Позірк можа быць уважлівым, зацікаўленым, захопленым, але бывае і нахабным, задзірлівым, пагардлівым, напышлівым.",
+        "donarka_ench.wav",
+    ],
+    [
+        "Такі нават шчыры, ці што: родная мова народу – трасянка, а беларуская яму чужая!",
+        "muzhcynski.wav",
+    ],
+]
+
+analytics_script = """
+<script async src="https://www.googletagmanager.com/gtag/js?id=G-TKDCRCQ7FK"></script>
+<script>
+  window.dataLayer = window.dataLayer || [];
+  function gtag(){dataLayer.push(arguments);}
+  gtag('js', new Date());
+  gtag('config', 'G-TKDCRCQ7FK');
+</script>
+"""
+
+# ---------------------------------------------------------
+# 6) Gradio UI (аўтапрайграванне, мінімальная затрымка)
+# ---------------------------------------------------------
+with gr.Blocks() as demo:
+    gr.HTML(analytics_script)
+    gr.Interface(
+        fn=text_to_speech,  # генератар
+        inputs=[
+            gr.Textbox(lines=5, label="Тэкст на беларускай мове"),
+            gr.Audio(
+                type="filepath",
+                label="Прыклад голасу (без іншых гукаў) не карацей 7 секунд",
+                interactive=True,
+            ),
+        ],
+        outputs=gr.Audio(
+            type="filepath",   # прымае (sr, ndarray) падчас стриму і фінальны шлях у фінале
+            label="Згенераванае аўдыя (па токенах, мінімальная затрымка)",
+            autoplay=True,
+        ),
+        title="Belarusian TTS — Token Streaming (мінімальная затрымка)",
+        description="""
+        <p>Вывод гуку <b>па токенах</b> з буферам ~50&nbsp;мс. Калі мадэль падтрымлівае <code>inference_stream</code> — выкарыстоўваецца ён; інакш працуе інкрементальны fallback па BPE.</p>
+        """,
+        examples=examples,
+        cache_examples=False,
+        allow_flagging="never",
+    )
+
+if __name__ == "__main__":
+    demo.launch()