| """PyTorch inference for LGTM. |
| |
| from lgtm import LGTMTTS |
| tts = LGTMTTS.from_pretrained("path/or/hf-repo") # downloads from the Hub if needed |
| wav = tts.synthesize("Xin chào!", lang="vi", voice="F1") |
| tts.save_wav(wav, "out.wav") |
| voice = tts.clone_voice("reference.wav") # zero-shot voice from 5-15 s of audio |
| wav = tts.synthesize("Hello there.", lang="en", voice=voice) |
| """ |
| import json |
| import os |
| import re |
|
|
| import numpy as np |
| import soundfile as sf |
| import torch |
| import torch.nn.functional as F |
| import torchaudio |
| from safetensors.torch import load_file |
|
|
| from .model import LGTM |
| from .text import AVAILABLE_LANGS, TextProcessor |
|
|
| SAMPLE_RATE = 44100 |
|
|
|
|
| def load_voice_style(path, device="cpu"): |
| d = json.load(open(path)) |
| ttl = torch.tensor(np.array(d["style_ttl"]["data"], np.float32).reshape(d["style_ttl"]["dims"])) |
| dp = torch.tensor(np.array(d["style_dp"]["data"], np.float32).reshape(d["style_dp"]["dims"])) |
| return ttl.to(device), dp.to(device) |
|
|
|
|
| def save_voice_style(path, voice): |
| ttl, dp = voice |
| d = {"style_ttl": {"data": ttl.detach().float().cpu().numpy().tolist(), "dims": list(ttl.shape), "type": "float32"}, |
| "style_dp": {"data": dp.detach().float().cpu().numpy().tolist(), "dims": list(dp.shape), "type": "float32"}} |
| json.dump(d, open(path, "w")) |
|
|
|
|
| def _prepare_reference(w, top_db=40.0, pad_s=0.1, target_rms_db=-23.0, peak=0.95): |
| """Trim leading/trailing silence and normalise loudness (same as the training data).""" |
| hop = SAMPLE_RATE // 100 |
| n = len(w) // hop |
| if n >= 3: |
| db = 10 * torch.log10(w[: n * hop].view(n, hop).pow(2).mean(1) + 1e-10) |
| active = torch.nonzero(db > db.max() - top_db).flatten() |
| if len(active): |
| pad = int(pad_s * SAMPLE_RATE) |
| w = w[max(int(active[0]) * hop - pad, 0): min((int(active[-1]) + 1) * hop + pad, len(w))] |
| w = w * (10 ** (target_rms_db / 20) / (w.pow(2).mean().sqrt() + 1e-6)) |
| m = w.abs().max() |
| return w * (peak / m) if m > peak else w |
|
|
|
|
| def split_text(text, max_len=300): |
| """Split long text into sentence chunks of at most max_len characters.""" |
| chunks = [] |
| for para in [p.strip() for p in re.split(r"\n\s*\n+", text.strip()) if p.strip()]: |
| cur = "" |
| for sent in re.split(r"(?<=[.!?。！？])\s+", para): |
| if cur and len(cur) + len(sent) + 1 > max_len: |
| chunks.append(cur) |
| cur = sent |
| else: |
| cur = f"{cur} {sent}".strip() |
| if cur: |
| chunks.append(cur) |
| return chunks |
|
|
|
|
| class LGTMTTS: |
| def __init__(self, model_dir, device=None): |
| self.device = device or ("cuda" if torch.cuda.is_available() else "cpu") |
| cfg = json.load(open(os.path.join(model_dir, "config.json"))) |
| self.model = LGTM(cfg) |
| self.model.load_state_dict(load_file(os.path.join(model_dir, "pytorch", "model.safetensors"))) |
| self.model.to(self.device).eval() |
| self.tp = TextProcessor(os.path.join(model_dir, "unicode_indexer.json")) |
| self.voice_dir = os.path.join(model_dir, "voice_styles") |
|
|
| @classmethod |
| def from_pretrained(cls, repo_or_dir, device=None): |
| if not os.path.isdir(repo_or_dir): |
| from huggingface_hub import snapshot_download |
|
|
| repo_or_dir = snapshot_download(repo_or_dir, allow_patterns=["config.json", "unicode_indexer.json", "voice_styles/*", "pytorch/*"]) |
| return cls(repo_or_dir, device) |
|
|
| @property |
| def voices(self): |
| return sorted(f[:-5] for f in os.listdir(self.voice_dir) if f.endswith(".json")) |
|
|
| def _voice(self, voice): |
| if isinstance(voice, str): |
| path = voice if voice.endswith(".json") else os.path.join(self.voice_dir, f"{voice}.json") |
| return load_voice_style(path, self.device) |
| return voice[0].to(self.device), voice[1].to(self.device) |
|
|
| @torch.no_grad() |
| def clone_voice(self, wav_path, max_seconds=15.0): |
| """Voice style from a reference recording (5-15 s of clean speech works best).""" |
| wav, sr = sf.read(wav_path, dtype="float32") |
| if wav.ndim == 2: |
| wav = wav.mean(1) |
| w = torch.from_numpy(wav) |
| if sr != SAMPLE_RATE: |
| w = torchaudio.functional.resample(w, sr, SAMPLE_RATE) |
| w = _prepare_reference(w)[: int(max_seconds * SAMPLE_RATE)] |
| lat = self.model.ae.encode_ttl(w[None].to(self.device)) |
| mask = torch.ones(1, 1, lat.shape[-1], device=self.device) |
| return self.model.ttl.style_encoder(lat, mask), self.model.dp.style_encoder(lat, mask) |
|
|
| @torch.no_grad() |
| def _synth_batch(self, texts, langs, voice, steps, speed, cfg_scale): |
| s_ttl, s_dp = voice |
| b = len(texts) |
| ids, mask = self.tp.batch(texts, langs, device=self.device) |
| s_ttl, s_dp = s_ttl.expand(b, -1, -1), s_dp.expand(b, -1, -1) |
| m = self.model |
| dur = m.dp(ids, mask, s_dp) / speed |
| text_emb = m.ttl.encode_text(ids, mask, s_ttl) |
| chunk = m.ae.hop * m.ae.ccf |
| wav_len = (dur * SAMPLE_RATE).long() |
| lat_len = (wav_len + chunk - 1) // chunk |
| L = int(lat_len.max()) |
| lmask = (torch.arange(L, device=self.device)[None] < lat_len[:, None]).float().unsqueeze(1) |
| x = torch.randn(b, m.ae.ldim * m.ae.ccf, L, device=self.device) * lmask |
| total = torch.full((b,), float(steps), device=self.device) |
| for i in range(steps): |
| x = m.ttl.euler_step(x, torch.full((b,), float(i), device=self.device), total, text_emb, mask, s_ttl, lmask, cfg_scale) |
| wav = m.ae.decode_ttl(x) |
| return [wav[k, : wav_len[k]].float().cpu().numpy() for k in range(b)] |
|
|
| def synthesize(self, text, lang="en", voice="F1", steps=8, speed=1.05, cfg_scale=4.0, silence=0.3): |
| """Text -> 44.1 kHz float32 waveform. Long text is split into sentence chunks. |
| voice: preset name (F1..F5, M1..M5), a .json path, or the output of clone_voice().""" |
| if lang not in AVAILABLE_LANGS: |
| raise ValueError(f"unsupported language {lang!r}") |
| v = self._voice(voice) |
| chunks = split_text(text, 120 if lang in ("ja", "ko") else 300) |
| wavs = self._synth_batch(chunks, [lang] * len(chunks), v, steps, speed, cfg_scale) |
| gap = np.zeros(int(silence * SAMPLE_RATE), np.float32) |
| out = [] |
| for i, w in enumerate(wavs): |
| out += [w] + ([gap] if i < len(wavs) - 1 else []) |
| return np.concatenate(out) |
|
|
| @staticmethod |
| def save_wav(wav, path): |
| sf.write(path, wav, SAMPLE_RATE) |
|
|
