Modules/text_encoder.py

import math
from typing import Optional, Union
import torch
from torch import nn
from torch.nn import functional as F
from torch.nn.utils.parametrizations import weight_norm
from einops import rearrange
import einops._torch_specific
from torchaudio.models import Conformer


def sequence_mask(length, max_length=None):
    if max_length is None:
        max_length = length.max()
    x = torch.arange(max_length, dtype=length.dtype, device=length.device)
    return x.unsqueeze(0) < length.unsqueeze(1)


def get_padding(kernel_size, dilation=1):
    return int((kernel_size * dilation - dilation) / 2)


class LayerNorm(nn.Module):
    def __init__(self, channels, eps=1e-4):
        super().__init__()
        self.channels = channels
        self.eps = eps

        self.gamma = torch.nn.Parameter(torch.ones(channels))
        self.beta = torch.nn.Parameter(torch.zeros(channels))

    def forward(self, x):
        n_dims = len(x.shape)
        mean = torch.mean(x, 1, keepdim=True)
        variance = torch.mean((x - mean) ** 2, 1, keepdim=True)

        x = (x - mean) * torch.rsqrt(variance + self.eps)

        x = x * self.gamma.view(1, -1, 1) + self.beta.view(1, -1, 1)
        return x


class ConvReluNorm(nn.Module):
    def __init__(

        self,

        in_channels,

        hidden_channels,

        out_channels,

        kernel_size,

        n_layers,

        p_dropout,

    ):
        super().__init__()
        self.in_channels = in_channels
        self.hidden_channels = hidden_channels
        self.out_channels = out_channels
        self.kernel_size = kernel_size
        self.n_layers = n_layers
        self.p_dropout = p_dropout

        self.conv_layers = torch.nn.ModuleList()
        self.norm_layers = torch.nn.ModuleList()
        self.conv_layers.append(
            torch.nn.Conv1d(
                in_channels, hidden_channels, kernel_size, padding=kernel_size // 2
            )
        )
        self.norm_layers.append(LayerNorm(hidden_channels))
        self.relu_drop = torch.nn.Sequential(
            torch.nn.ReLU(), torch.nn.Dropout(p_dropout)
        )
        for _ in range(n_layers - 1):
            self.conv_layers.append(
                torch.nn.Conv1d(
                    hidden_channels,
                    hidden_channels,
                    kernel_size,
                    padding=kernel_size // 2,
                )
            )
            self.norm_layers.append(LayerNorm(hidden_channels))
        self.proj = torch.nn.Conv1d(hidden_channels, out_channels, 1)
        self.proj.weight.data.zero_()
        self.proj.bias.data.zero_()

    def forward(self, x, x_mask):
        x_org = x
        for i in range(self.n_layers):
            x = self.conv_layers[i](x * x_mask)
            x = self.norm_layers[i](x)
            x = self.relu_drop(x)
        x = x_org + self.proj(x)
        return x * x_mask


class RotaryPositionalEmbeddings(nn.Module):
    """

    Rotary positional embeddings (RoPE) helper.

    """

    def __init__(self, d: int, base: int = 10_000):
        super().__init__()
        self.base = base
        self.d = int(d)

    def _build_cache(self, x: torch.Tensor):
        seq_len = x.shape[0]
        theta = 1.0 / (self.base ** (torch.arange(0, self.d, 2).float() / self.d)).to(
            x.device
        )
        seq_idx = torch.arange(seq_len, device=x.device).float().to(x.device)
        idx_theta = torch.einsum("n,d->nd", seq_idx, theta)
        idx_theta2 = torch.cat([idx_theta, idx_theta], dim=1)
        return idx_theta2.cos()[:, None, None, :], idx_theta2.sin()[:, None, None, :]

    def _neg_half(self, x: torch.Tensor):
        d_2 = self.d // 2
        return torch.cat([-x[:, :, :, d_2:], x[:, :, :, :d_2]], dim=-1)

    def forward(self, x: torch.Tensor):
        # input shape expected: [batch, n_heads, seq_len, d] (this matches arrange_heads usage)
        x = torch.permute(x, (2, 0, 1, 3))  # -> [seq_len, batch, n_heads, d]
        cos_cached, sin_cached = self._build_cache(x)

        x_rope, x_pass = x[..., : self.d], x[..., self.d :]
        neg_half_x = self._neg_half(x_rope)

        x_rope = (x_rope * cos_cached[: x.shape[0]]) + (
            neg_half_x * sin_cached[: x.shape[0]]
        )

        result = torch.cat((x_rope, x_pass), dim=-1)
        return torch.permute(result, (1, 2, 0, 3))  # -> [batch, n_heads, seq_len, d]


class MultiHeadAttention(nn.Module):
    def __init__(

        self,

        channels,

        out_channels,

        n_heads,

        heads_share=True,

        p_dropout=0.0,

        proximal_bias=False,

        proximal_init=False,

        use_sdpa=True,

    ):
        super().__init__()
        assert channels % n_heads == 0

        self.channels = channels
        self.out_channels = out_channels
        self.n_heads = n_heads
        self.heads_share = heads_share
        self.proximal_bias = proximal_bias
        self.p_dropout = p_dropout
        self.attn = None

        self.k_channels = channels // n_heads
        self.conv_q = torch.nn.Conv1d(channels, channels, 1)
        self.conv_k = torch.nn.Conv1d(channels, channels, 1)
        self.conv_v = torch.nn.Conv1d(channels, channels, 1)

        # rotary embedding expects an integer d; int() is used internally
        self.query_rotary_pe = RotaryPositionalEmbeddings(self.k_channels * 0.5)
        self.key_rotary_pe = RotaryPositionalEmbeddings(self.k_channels * 0.5)

        self.conv_o = torch.nn.Conv1d(channels, out_channels, 1)
        self.drop = torch.nn.Dropout(p_dropout)

        torch.nn.init.xavier_uniform_(self.conv_q.weight)
        torch.nn.init.xavier_uniform_(self.conv_k.weight)
        if proximal_init:
            self.conv_k.weight.data.copy_(self.conv_q.weight.data)
            self.conv_k.bias.data.copy_(self.conv_q.bias.data)
        torch.nn.init.xavier_uniform_(self.conv_v.weight)
        self.use_sdpa = use_sdpa

    def forward(self, x, c, attn_mask=None):
        torch._check(x.shape[2] > 0)
        q = self.conv_q(x)
        k = self.conv_k(c)
        v = self.conv_v(c)

        x, _ = self.attention(q, k, v, mask=attn_mask)

        x = self.conv_o(x)
        return x

    def arrange_heads(self, x):
        # x: [batch, channels, time] -> [batch, n_heads, time, head_dim]
        x = x.chunk(chunks=self.n_heads, dim=1)
        x = torch.stack(x)
        x = x.permute(1, 0, 3, 2)
        return x

    def attention(self, query, key, value, mask=None):
        b, d, t_s, t_t = (key.size(0), key.size(1), key.size(2), query.size(2))

        query = self.arrange_heads(query)  # [b, h, t_q, head_dim]
        key = self.arrange_heads(key)
        value = self.arrange_heads(value)

        query = self.query_rotary_pe(query)
        key = self.key_rotary_pe(key)

        if self.use_sdpa:
            final_attn_mask = None
            if self.proximal_bias:
                assert t_s == t_t, "Proximal bias is only available for self-attention."
                bias_val = self._attention_bias_proximal(t_s).to(
                    device=query.device, dtype=query.dtype
                )
                final_attn_mask = bias_val

            if mask is not None:
                expanded_bool_mask = mask
                additive_external_mask = torch.zeros_like(
                    expanded_bool_mask, dtype=query.dtype
                )
                additive_external_mask.masked_fill_(
                    ~(expanded_bool_mask.to(bool)), -1e4
                )

                if final_attn_mask is not None:
                    final_attn_mask = final_attn_mask + additive_external_mask
                else:
                    final_attn_mask = additive_external_mask

            output = F.scaled_dot_product_attention(
                query,
                key,
                value,
                attn_mask=final_attn_mask,
                dropout_p=self.p_dropout if self.training else 0.0,
                is_causal=False,
            )

            output = output.transpose(2, 3).contiguous().view(b, d, t_t)
            return output, None
        else:
            scores = torch.matmul(query, key.transpose(2, 3)) / math.sqrt(
                self.k_channels
            )

            if self.proximal_bias:
                assert t_s == t_t, "Proximal bias is only available for self-attention."
                scores = scores + self._attention_bias_proximal(t_s).to(
                    device=scores.device, dtype=scores.dtype
                )
            if mask is not None:
                scores = scores.masked_fill(mask == 0, -1e4)
            p_attn = torch.nn.functional.softmax(scores, dim=-1)
            p_attn = self.drop(p_attn)
            output = torch.matmul(p_attn, value)
            output = output.transpose(2, 3).contiguous().view(b, d, t_t)
            return output, p_attn

    @staticmethod
    def _attention_bias_proximal(length):
        r = torch.arange(length, dtype=torch.float32)
        diff = torch.unsqueeze(r, 0) - torch.unsqueeze(r, 1)
        return torch.unsqueeze(torch.unsqueeze(-torch.log1p(torch.abs(diff)), 0), 0)


class FFN(nn.Module):
    def __init__(

        self, in_channels, out_channels, filter_channels, kernel_size, p_dropout=0.0

    ):
        super().__init__()
        self.in_channels = in_channels
        self.out_channels = out_channels
        self.filter_channels = filter_channels
        self.kernel_size = kernel_size
        self.p_dropout = p_dropout

        self.conv_1 = torch.nn.Conv1d(
            in_channels, filter_channels, kernel_size, padding=kernel_size // 2
        )
        self.conv_2 = torch.nn.Conv1d(
            filter_channels, out_channels, kernel_size, padding=kernel_size // 2
        )
        self.drop = torch.nn.Dropout(p_dropout)

    def forward(self, x, x_mask):
        x = self.conv_1(x * x_mask)
        x = torch.relu(x)
        x = self.drop(x)
        x = self.conv_2(x * x_mask)
        return x * x_mask


class AdaLayerNorm(nn.Module):
    def __init__(self, style_dim, channels, eps=1e-5):
        super().__init__()
        self.channels = channels
        self.eps = eps

        self.fc = nn.Linear(style_dim, channels*2)

    def forward(self, x, s):
        x = x.transpose(1, 2) # [B, C, T] -> [B, T, C]
                
        h = self.fc(s)
        h = h.view(h.size(0), h.size(1), 1)
        gamma, beta = torch.chunk(h, chunks=2, dim=1)
        # gamma, beta: [B, C, 1] -> [B, 1, C]
        gamma, beta = gamma.transpose(1, -1), beta.transpose(1, -1)
        
        x = F.layer_norm(x, (self.channels,), eps=self.eps)
        x = (1 + gamma) * x + beta
        return x.transpose(1, 2) # [B, T, C] -> [B, C, T]

class Encoder(nn.Module):
    def __init__(

        self,

        hidden_channels,

        filter_channels,

        n_heads,

        n_layers,

        kernel_size=1,

        p_dropout=0.0,

        style_dim=0,

        **kwargs,

    ):
        super().__init__()
        self.hidden_channels = hidden_channels
        self.filter_channels = filter_channels
        self.n_heads = n_heads
        self.n_layers = n_layers
        self.kernel_size = kernel_size
        self.p_dropout = p_dropout
        self.style_dim = style_dim

        self.drop = torch.nn.Dropout(p_dropout)
        self.attn_layers = torch.nn.ModuleList()
        self.norm_layers_1 = torch.nn.ModuleList()
        self.ffn_layers = torch.nn.ModuleList()
        self.norm_layers_2 = torch.nn.ModuleList()
        self.adain_layers = torch.nn.ModuleList() if style_dim > 0 else None
        
        for _ in range(self.n_layers):
            self.attn_layers.append(
                MultiHeadAttention(
                    hidden_channels, hidden_channels, n_heads, p_dropout=p_dropout
                )
            )
            self.norm_layers_1.append(LayerNorm(hidden_channels))
            self.ffn_layers.append(
                FFN(
                    hidden_channels,
                    hidden_channels,
                    filter_channels,
                    kernel_size,
                    p_dropout=p_dropout,
                )
            )
            self.norm_layers_2.append(LayerNorm(hidden_channels))
            if style_dim > 0:
                self.adain_layers.append(AdaLayerNorm(style_dim, hidden_channels))

    def forward(self, x, x_mask, style=None):
        attn_mask = x_mask.unsqueeze(2) * x_mask.unsqueeze(-1)
        for i in range(self.n_layers):
            x = x * x_mask
            y = self.attn_layers[i](x, x, attn_mask)
            y = self.drop(y)
            x = self.norm_layers_1[i](x + y)
            y = self.ffn_layers[i](x, x_mask)
            y = self.drop(y)
            x = self.norm_layers_2[i](x + y)
            if self.adain_layers is not None and style is not None:
                x = self.adain_layers[i](x, style)
        x = x * x_mask
        return x


class TextEncoderTransformer(nn.Module):
    """

    TextEncoder with optional language embeddings.



    Example initialization:

      text_encoder = TextEncoder(

          channels=args.hidden_dim,

          language_count=3,

          language_hidden_dim=64,

          depth=args.n_layer,

          kernel_size=5,

          n_symbols=args.n_token

      )



    Forward:

      mu, encoded, x_mask = text_encoder(tokens, lengths, language_id=language_ids)

      - tokens: LongTensor [batch, seq_len]

      - lengths: LongTensor [batch]

      - language_id: None | int | LongTensor([batch])

    """

    def __init__(

        self,

        *,

        channels: int,

        language_count: int = 1,

        language_hidden_dim: int = 0,

        depth: int = 6,

        kernel_size: int = 5,

        n_symbols: int = 256,

        filter_channels: Optional[int] = None,

        heads: int = 8,

        dropout: float = 0.1,

        prenet_dropout: float = 0.1,

        inter_dim: Optional[int] = None,

        prenet_layers: int = 3,

    ):
        super().__init__()

        if language_count > 1 and language_hidden_dim <= 0:
            raise ValueError(
                "language_hidden_dim must be > 0 when language_count > 1"
            )

        self.language_count = int(language_count)
        self.language_hidden_dim = int(language_hidden_dim) if self.language_count > 1 else 0

        self.n_channels = int(channels)
        self.n_symbols = int(n_symbols)

        if filter_channels is None:
            filter_channels = max(self.n_channels * 4, 512)
        self.filter_channels = int(filter_channels)

        if inter_dim is None:
            inter_dim = self.n_channels

        # token embedding
        self.emb = torch.nn.Embedding(self.n_symbols, self.n_channels)
        torch.nn.init.normal_(self.emb.weight, 0.0, self.n_channels ** -0.5)

        # language embeddings 
        if self.language_count > 1:
            self.language_emb = nn.Embedding(self.language_count, self.language_hidden_dim)
            torch.nn.init.normal_(self.language_emb.weight, 0.0, self.language_hidden_dim ** -0.5)
            
        # encoder input channels = token channels
        encoder_in_channels = self.n_channels

        # prenet
        self.prenet = ConvReluNorm(
            encoder_in_channels,
            encoder_in_channels,
            encoder_in_channels,
            kernel_size=kernel_size,
            n_layers=prenet_layers,
            p_dropout=prenet_dropout,
        )

        self.encoder = Encoder(
            encoder_in_channels,
            self.filter_channels,
            heads,
            depth,
            kernel_size,
            dropout,
            style_dim=self.language_hidden_dim
        )

        self.proj_m = torch.nn.Conv1d(encoder_in_channels, inter_dim, 1)

    def forward(self, x: torch.LongTensor, x_lengths: torch.LongTensor, language_id: Optional[Union[int, torch.LongTensor, list]] = None, language_emb: Optional[torch.Tensor] = None):
        if language_emb is None and language_id is not None and self.language_count > 1:
            language_emb = self.language_emb(language_id)
            
        x = self.emb(x) * math.sqrt(self.n_channels)  # [batch, seq_len, channels]
        
        x = torch.transpose(x, 1, -1)  # [batch, channels, seq_len]

        x_mask = torch.unsqueeze(sequence_mask(x_lengths, x.size(2)), 1).to(x.dtype)  # [batch, 1, seq_len]

        x = self.prenet(x, x_mask)  # prenet result: [batch, channels, seq_len]

        x = self.encoder(x, x_mask, style=language_emb)
        mu = self.proj_m(x) * x_mask
        return mu