Download src/nida/model/align.py from Bilal7Dev/nida-tts: direct link, hf CLI and curl.
- Browser
- Download file 13.1 kB
-
https://huggingface.co/Bilal7Dev/nida-tts/resolve/main/src/nida/model/align.py
- Command line
-
hf download hf://Bilal7Dev/nida-tts/src/nida/model/align.py
-
curl -L -o align.py https://huggingface.co/Bilal7Dev/nida-tts/resolve/main/src/nida/model/align.py
13.1 kB
| """Explicit-duration text path (E11, ADR-0008): duration head + windowed Gaussian alignment. | |
| Instead of padding the characters to the latent length (F5 style, ``text_path="implicit"``), | |
| every latent frame attends softly to the characters of its own unit and the two | |
| neighbouring units (a unit is a word or an event tag, see ``nida.data.timings``), under a | |
| learned Gaussian prior on the distance between continuous positions (EMA-TTS report §3, | |
| docs/experiments/E11-aligned.md). Notation (``B`` items, ``L`` characters, ``U`` units, | |
| ``T`` frames, ``H`` alignment heads): | |
| * ``char_unit`` ``(B, L)`` long — unit index ``w(i)`` of every character; | |
| * ``unit_first_char`` ``(B, U)`` long — index of the first character of every unit; | |
| * ``unit_bounds`` ``(B, U + 1)`` float — frame boundaries: unit ``u`` covers frames | |
| ``[unit_bounds[u], unit_bounds[u + 1])``; padded units have zero length at the item end; | |
| * masks are bool with **True = real** (character / unit / frame). | |
| Coordinates are "unit index + fraction inside the unit", so the last character of one word | |
| and the first of the next are close (they must interact for coarticulation): | |
| * frame ``t``: ``g^f_t = w(t) + (t + 0.5 − s_{w(t)}) / D_{w(t)}``; | |
| * character ``i``: ``g^c_i = w(i) + φ_i`` with | |
| ``φ_i = (Σ_{j<i, w(j)=w(i)} d_j + d_i / 2) / Σ_{w(j)=w(i)} d_j`` (durations ``d`` from | |
| the duration head during training, so the flow loss reaches the within-unit division). | |
| """ | |
| from __future__ import annotations | |
| import itertools | |
| import math | |
| from collections.abc import Sequence | |
| import torch | |
| import torch.nn.functional as F | |
| from torch import Tensor, nn | |
| __all__ = [ | |
| "DurationHead", | |
| "EmotionFiLM", | |
| "GaussianAligner", | |
| "char_phase", | |
| "even_char_phase", | |
| "frame_coordinates", | |
| "inference_bounds", | |
| ] | |
| #: Floor added to unit duration sums so an all-zero unit never divides by zero. | |
| _DUR_EPS = 1e-4 | |
| class DurationHead(nn.Module): | |
| """``l_i = log(1 + d_i) = softplus(MLP(h_i)) >= 0`` per character. | |
| Shapes: ``h`` ``(B, L, D)`` → ``(B, L)`` log-durations (frames). ``d = expm1(l) >= 0``. | |
| The softplus keeps the duration non-negative, so positions built from it are ordered. | |
| """ | |
| def __init__(self, dim: int) -> None: | |
| super().__init__() | |
| self.net = nn.Sequential(nn.Linear(dim, dim), nn.SiLU(), nn.Linear(dim, 1)) | |
| def forward(self, h: Tensor) -> Tensor: | |
| out: Tensor = F.softplus(self.net(h).squeeze(-1).float()) | |
| return out | |
| class EmotionFiLM(nn.Module): | |
| """FiLM of the duration-head input on the emotion condition (E13, ``dur_cond="emotion"``). | |
| ``h' = h * (1 + gamma(e)) + beta(e)`` with ``e = E[emotion] + g(intensity)`` and | |
| ``(gamma, beta) = W e`` (one ``Linear(dim, 2 * dim)``, **zero-initialised**, so a fresh | |
| module is the identity and the model starts exactly equal to ``dur_cond="none"``). | |
| ``E`` has ``n_emotions + 1`` rows, the last one being the null id (emotion dropped for | |
| classifier-free guidance); ``g`` is the same small MLP as the AdaLN intensity embedding | |
| but without a learned null: an unknown intensity contributes nothing. | |
| Shapes: ``h`` ``(B, L, D)``; ``emotion`` ``(B,)`` long in ``[0, n_emotions]``; | |
| ``intensity`` ``(B,)`` float; ``intensity_known`` ``(B,)`` bool → ``(B, L, D)``. | |
| """ | |
| def __init__(self, dim: int, n_emotions: int, *, embed_std: float = 0.02) -> None: | |
| super().__init__() | |
| self.emotion_embed = nn.Embedding(n_emotions + 1, dim) | |
| nn.init.normal_(self.emotion_embed.weight, std=embed_std) | |
| self.intensity_mlp = nn.Sequential(nn.Linear(1, dim), nn.SiLU(), nn.Linear(dim, dim)) | |
| self.proj = nn.Linear(dim, 2 * dim) | |
| nn.init.zeros_(self.proj.weight) | |
| nn.init.zeros_(self.proj.bias) | |
| def forward( | |
| self, h: Tensor, emotion: Tensor, intensity: Tensor, intensity_known: Tensor | |
| ) -> Tensor: | |
| e = self.emotion_embed(emotion) | |
| # Unknown intensities may be NaN placeholders: zero them BEFORE the MLP, otherwise | |
| # where() still back-propagates 0 * NaN into the MLP weights. | |
| level = torch.where(intensity_known, intensity, torch.zeros_like(intensity)) | |
| known = self.intensity_mlp(level[:, None].to(e.dtype)) | |
| e = e + torch.where(intensity_known[:, None], known, torch.zeros_like(known)) | |
| gamma, beta = self.proj(e)[:, None, :].chunk(2, dim=-1) | |
| out: Tensor = h * (1 + gamma) + beta | |
| return out | |
| def _segment_prefix(values: Tensor, char_unit: Tensor, unit_first_char: Tensor) -> Tensor: | |
| """Exclusive prefix sum of ``values`` ``(B, L)`` restarted at every unit's first char.""" | |
| cumsum = torch.cumsum(values, dim=1) | |
| exclusive = cumsum - values | |
| # exclusive sum at the first character of each unit, gathered back to its characters | |
| start = torch.gather(exclusive, 1, unit_first_char.clamp(min=0)) # (B, U) | |
| return exclusive - torch.gather(start, 1, char_unit.clamp(min=0)) | |
| def char_phase( | |
| durations: Tensor, char_unit: Tensor, unit_first_char: Tensor, char_mask: Tensor | |
| ) -> Tensor: | |
| """Within-unit position ``φ`` ``(B, L)`` in ``[0, 1]`` of each character's midpoint. | |
| Args: | |
| durations: ``(B, L)`` non-negative character durations (frames). | |
| char_unit, unit_first_char: see module docstring (contiguous units). | |
| char_mask: ``(B, L)`` bool, True = real character (padding gets ``φ = 0``). | |
| """ | |
| d = torch.where(char_mask, durations.float(), torch.zeros((), device=durations.device)) | |
| prefix = _segment_prefix(d, char_unit, unit_first_char) | |
| total = torch.zeros( | |
| d.shape[0], unit_first_char.shape[1], device=d.device, dtype=d.dtype | |
| ).scatter_add(1, char_unit.clamp(min=0), d) | |
| unit_total = torch.gather(total, 1, char_unit.clamp(min=0)) + _DUR_EPS | |
| phase = (prefix + 0.5 * d) / unit_total | |
| return torch.where(char_mask, phase.clamp(0.0, 1.0), torch.zeros((), device=d.device)) | |
| def even_char_phase(char_unit: Tensor, unit_first_char: Tensor, char_mask: Tensor) -> Tensor: | |
| """``φ_i = (k + 0.5) / n_{w(i)}`` (even spacing inside a unit; inference, EMA §3.6).""" | |
| ones = char_mask.float() | |
| return char_phase(ones, char_unit, unit_first_char, char_mask) | |
| def frame_coordinates(unit_bounds: Tensor, num_frames: int) -> tuple[Tensor, Tensor]: | |
| """Unit index ``w(t)`` ``(B, T)`` long and coordinate ``g^f_t`` ``(B, T)`` float. | |
| Frames at or beyond the last boundary (batch padding) get the last unit and a | |
| coordinate clamped inside it; they are masked by the caller. | |
| """ | |
| batch, n_bounds = unit_bounds.shape | |
| centres = torch.arange(num_frames, device=unit_bounds.device, dtype=torch.float32) + 0.5 | |
| centres = centres[None, :].expand(batch, num_frames).contiguous() | |
| bounds = unit_bounds.float().contiguous() | |
| unit = torch.searchsorted(bounds, centres, right=True) - 1 | |
| unit = unit.clamp(0, n_bounds - 2) | |
| start = torch.gather(bounds, 1, unit) | |
| end = torch.gather(bounds, 1, unit + 1) | |
| frac = ((centres - start) / (end - start).clamp(min=1e-6)).clamp(0.0, 1.0) | |
| return unit, unit.float() + frac | |
| def inference_bounds( | |
| known_bounds: Sequence[float], unit_frames: Sequence[float], keep_units: int, total: int | |
| ) -> list[int]: | |
| """Unit boundaries for generation: the first ``keep_units`` units keep their known | |
| boundaries (the speaker prompt), the rest share ``total − known_bounds[keep_units]`` | |
| frames in proportion to ``unit_frames`` (predicted durations), rounded so that every | |
| boundary is an int, non-decreasing, and the last one is exactly ``total``. | |
| Args: | |
| known_bounds: ``U + 1`` reference boundaries (only ``0..keep_units`` are used). | |
| unit_frames: ``U`` non-negative predicted unit durations. | |
| keep_units: number of leading units to keep (``0..U``). | |
| total: frame count of the item. | |
| Raises: | |
| ValueError: inconsistent lengths, negative durations, or a kept boundary outside | |
| ``[0, total]``. | |
| """ | |
| units = len(unit_frames) | |
| if len(known_bounds) != units + 1 or not 0 <= keep_units <= units: | |
| raise ValueError("need U + 1 known bounds and 0 <= keep_units <= U") | |
| if any(d < 0 or not math.isfinite(d) for d in unit_frames): | |
| raise ValueError("unit durations must be finite and >= 0") | |
| kept = [round(b) for b in known_bounds[: keep_units + 1]] | |
| if any(b < 0 or b > total for b in kept) or any(b < a for a, b in itertools.pairwise(kept)): | |
| raise ValueError("kept boundaries must be non-decreasing within [0, total]") | |
| start = kept[-1] | |
| rest = list(unit_frames[keep_units:]) | |
| room = total - start | |
| weight = sum(rest) | |
| shares = ( | |
| [room / len(rest)] * len(rest) | |
| if rest and weight <= 0 | |
| else [room * d / weight for d in rest] | |
| ) | |
| out = kept[:] | |
| acc = float(start) | |
| for share in shares: | |
| acc += share | |
| out.append(round(acc)) | |
| out[-1] = total | |
| return out | |
| def _rotate(x: Tensor, positions: Tensor, base: float) -> Tensor: | |
| """RoPE at continuous ``positions``: ``x`` ``(..., N, H, Dh)``, ``positions`` ``(..., N)``.""" | |
| dim = x.shape[-1] | |
| half = dim // 2 | |
| freqs = base ** (-torch.arange(half, device=x.device, dtype=torch.float32) / half) | |
| angles = positions.float()[..., None, None] * freqs # (..., N, 1, half) | |
| cos, sin = angles.cos(), angles.sin() | |
| x1, x2 = x[..., :half].float(), x[..., half:].float() | |
| return torch.cat([x1 * cos - x2 * sin, x1 * sin + x2 * cos], dim=-1) | |
| class GaussianAligner(nn.Module): | |
| """Windowed Gaussian cross-attention from frames to characters (E11 §Değişiklik 4). | |
| ``A^{(m)}_{t,i} = softmax_i(τ ⟨q_t, k_i⟩ / √d_h − β_m (g^f_t − g^c_i)² / (2 sigma_m^2) | |
| + M_{t,i})`` with ``q_t = R(gamma g^f_t) q`` (one learned query, no frame content), | |
| ``k_i = R(gamma g^c_i) | |
| W_k h_i``, ``M = 0`` iff ``|w(i) − w(t)| <= window`` (and ``i`` is a real character), | |
| else ``−inf``. Output ``z_t = W_o concat_m Σ_i A^{(m)}_{t,i} W_v h_i`` ``(B, T, D)``. | |
| """ | |
| def __init__( | |
| self, | |
| dim: int, | |
| heads: int, | |
| *, | |
| sigma_init: tuple[float, float] = (0.18, 1.2), | |
| gamma: float = 8.0, | |
| rope_base: float = 100.0, | |
| window: int = 1, | |
| ) -> None: | |
| super().__init__() | |
| if dim % heads or (dim // heads) % 2: | |
| raise ValueError("dim must split into heads of even size") | |
| if not 0.0 < sigma_init[0] <= sigma_init[1]: | |
| raise ValueError("sigma_init must be 0 < low <= high") | |
| self.heads = heads | |
| self.head_dim = dim // heads | |
| self.gamma = gamma | |
| self.rope_base = rope_base | |
| self.window = window | |
| self.query = nn.Parameter(torch.randn(heads, self.head_dim) / math.sqrt(self.head_dim)) | |
| self.k_proj = nn.Linear(dim, dim) | |
| self.v_proj = nn.Linear(dim, dim) | |
| self.out = nn.Linear(dim, dim) | |
| sig = torch.linspace(math.log(sigma_init[0]), math.log(sigma_init[1]), heads) | |
| self.log_sigma = nn.Parameter(sig) | |
| self.log_beta = nn.Parameter(torch.zeros(heads)) | |
| self.log_tau = nn.Parameter(torch.zeros(())) | |
| def forward( | |
| self, | |
| h: Tensor, | |
| char_mask: Tensor, | |
| char_unit: Tensor, | |
| char_coord: Tensor, | |
| frame_unit: Tensor, | |
| frame_coord: Tensor, | |
| ) -> Tensor: | |
| """``h`` ``(B, L, D)``; char tensors ``(B, L)``; frame tensors ``(B, T)`` → ``(B, T, D)``. | |
| Every frame keeps at least one admissible character when its unit has one; a | |
| frame whose window holds no real character (an empty item, or batch padding) gets | |
| a zero output instead of a NaN. | |
| """ | |
| batch, length, dim = h.shape | |
| hd = self.head_dim | |
| k = self.k_proj(h).view(batch, length, self.heads, hd) | |
| v = self.v_proj(h).view(batch, length, self.heads, hd) | |
| k = _rotate(k, self.gamma * char_coord, self.rope_base) # (B, L, H, hd) float | |
| frames = frame_coord.shape[1] | |
| q = self.query[None, None].expand(batch, frames, self.heads, hd) | |
| q = _rotate(q, self.gamma * frame_coord, self.rope_base) # (B, T, H, hd) | |
| content = torch.einsum("bthd,blhd->bhtl", q, k) / math.sqrt(hd) | |
| dist2 = (frame_coord[:, None, :, None].float() - char_coord[:, None, None, :].float()) ** 2 | |
| sigma2 = torch.exp(2 * self.log_sigma)[None, :, None, None] | |
| beta = torch.exp(self.log_beta)[None, :, None, None] | |
| logits = torch.exp(self.log_tau) * content - beta * dist2 / (2 * sigma2) | |
| near = (frame_unit[:, :, None] - char_unit[:, None, :]).abs() <= self.window | |
| allowed = (near & char_mask[:, None, :])[:, None] # (B, 1, T, L) | |
| logits = logits.masked_fill(~allowed, float("-inf")) | |
| has_any = allowed.any(dim=-1, keepdim=True) # (B, 1, T, 1) | |
| logits = torch.where(has_any, logits, torch.zeros((), device=h.device)) | |
| attn = torch.softmax(logits, dim=-1) * has_any | |
| z = torch.einsum("bhtl,blhd->bthd", attn.to(v.dtype), v).reshape(batch, frames, dim) | |
| out: Tensor = self.out(z) | |
| return out | |