diff --git a/mlx_audio/convert.py b/mlx_audio/convert.py index 7aaf5256..29d9971b 100644 --- a/mlx_audio/convert.py +++ b/mlx_audio/convert.py @@ -243,10 +243,11 @@ def get_model_path(path_or_hf_repo: str, revision: Optional[str] = None) -> Path def load_config(model_path: Path) -> dict: """Load model configuration from a path.""" - config_path = model_path / "config.json" - if config_path.exists(): - with open(config_path, "r", encoding="utf-8") as f: - return json.load(f) + for name in ("config.json", "model_config.json"): + config_path = model_path / name + if config_path.exists(): + with open(config_path, "r", encoding="utf-8") as f: + return json.load(f) raise FileNotFoundError(f"Config not found at {model_path}") diff --git a/mlx_audio/tts/generate.py b/mlx_audio/tts/generate.py index 205eb24c..2fa925db 100644 --- a/mlx_audio/tts/generate.py +++ b/mlx_audio/tts/generate.py @@ -596,6 +596,24 @@ def parse_args(): action="store_true", help="Optional model-specific zero speaker embedding mode (e.g., Ming Omni).", ) + parser.add_argument( + "--duration", + type=float, + default=None, + help="Audio duration in seconds (for diffusion models like Stable Audio 3).", + ) + parser.add_argument( + "--sampler", + type=str, + default=None, + help="Sampler type (e.g., 'pingpong', 'euler' for Stable Audio 3).", + ) + parser.add_argument( + "--seed", + type=int, + default=None, + help="Random seed for reproducible generation.", + ) parser.add_argument("--top_p", type=float, default=0.9, help="Top-p for the model") parser.add_argument("--top_k", type=int, default=50, help="Top-k for the model") parser.add_argument( diff --git a/mlx_audio/tts/models/stable_audio_3/__init__.py b/mlx_audio/tts/models/stable_audio_3/__init__.py new file mode 100644 index 00000000..fdf6dc59 --- /dev/null +++ b/mlx_audio/tts/models/stable_audio_3/__init__.py @@ -0,0 +1,3 @@ +from .stable_audio_3 import Model, ModelConfig + +__all__ = ["Model", "ModelConfig"] diff --git a/mlx_audio/tts/models/stable_audio_3/dit.py b/mlx_audio/tts/models/stable_audio_3/dit.py new file mode 100644 index 00000000..8e00020a --- /dev/null +++ b/mlx_audio/tts/models/stable_audio_3/dit.py @@ -0,0 +1,368 @@ +"""Diffusion Transformer (DiT) for Stable Audio 3. + +20-layer transformer with AdaLN conditioning, self-attention with RoPE, +cross-attention for text tokens, and optional local additive conditioning. +""" + +import math +from dataclasses import dataclass +from typing import Optional + +import mlx.core as mx +import mlx.nn as nn + + +@dataclass +class DiTConfig: + io_channels: int = 256 + embed_dim: int = 1024 + depth: int = 20 + num_heads: int = 16 + cond_token_dim: int = 768 + global_cond_dim: int = 768 + local_add_cond_dim: int = 257 + num_memory_tokens: int = 64 + ff_mult: float = 4.0 + + +class RMSNorm(nn.Module): + def __init__(self, dim: int): + super().__init__() + self.gamma = mx.ones((dim,)) + + def __call__(self, x: mx.array) -> mx.array: + x_f32 = x.astype(mx.float32) + rms = mx.sqrt(mx.mean(x_f32 * x_f32, axis=-1, keepdims=True) + 1e-6) + return (self.gamma * (x_f32 / rms)).astype(x.dtype) + + +class RotaryEmbedding(nn.Module): + def __init__(self, dim: int): + super().__init__() + self.inv_freq = 1.0 / ( + 10000.0 ** (mx.arange(0, dim, 2).astype(mx.float32) / dim) + ) + + def __call__(self, seq_len: int) -> mx.array: + t = mx.arange(seq_len, dtype=mx.float32) + return mx.outer(t, self.inv_freq) + + +def rotate_half(x: mx.array) -> mx.array: + x1 = x[..., : x.shape[-1] // 2] + x2 = x[..., x.shape[-1] // 2 :] + return mx.concatenate([-x2, x1], axis=-1) + + +def apply_rotary_emb(x: mx.array, freqs: mx.array) -> mx.array: + # freqs has dim/2 entries; only first rope_dim dimensions get rotated + rope_dim = freqs.shape[-1] * 2 + cos = mx.cos(freqs) + sin = mx.sin(freqs) + cos = mx.concatenate([cos, cos], axis=-1) + sin = mx.concatenate([sin, sin], axis=-1) + + while cos.ndim < x.ndim: + cos = mx.expand_dims(cos, axis=0) + sin = mx.expand_dims(sin, axis=0) + + x_rope = x[..., :rope_dim] + x_pass = x[..., rope_dim:] + x_rotated = x_rope * cos + rotate_half(x_rope) * sin + return mx.concatenate([x_rotated, x_pass], axis=-1) + + +class ExpoFourierFeatures(nn.Module): + def __init__( + self, dim: int = 256, min_freq: float = 0.5, max_freq: float = 10000.0 + ): + super().__init__() + self._dim = dim + self._log_min = math.log(min_freq) + self._log_max = math.log(max_freq) + + def __call__(self, t: mx.array) -> mx.array: + if t.ndim == 1: + t = mx.expand_dims(t, -1) + half = self._dim // 2 + ramp = mx.linspace(0.0, 1.0, half) + freqs = mx.exp(ramp * (self._log_max - self._log_min) + self._log_min) + args = t * freqs * (2.0 * math.pi) + return mx.concatenate([mx.cos(args), mx.sin(args)], axis=-1) + + +class _GatedProj(nn.Module): + def __init__(self, dim_in: int, dim_out: int): + super().__init__() + self.proj = nn.Linear(dim_in, dim_out * 2, bias=True) + + def __call__(self, x: mx.array) -> mx.array: + x = self.proj(x) + x, gate = mx.split(x, 2, axis=-1) + return x * nn.silu(gate) + + +class GLU(nn.Module): + def __init__(self, dim: int, mult: float = 4.0): + super().__init__() + inner = int(dim * mult) + self.ff = [ + _GatedProj(dim, inner), + None, + nn.Linear(inner, dim), + ] + + def __call__(self, x: mx.array) -> mx.array: + x = self.ff[0](x) + x = self.ff[2](x) + return x + + +class SelfAttention(nn.Module): + def __init__(self, dim: int, num_heads: int): + super().__init__() + self.num_heads = num_heads + self.head_dim = dim // num_heads + self.to_qkv = nn.Linear(dim, dim * 3, bias=False) + self.to_out = nn.Linear(dim, dim, bias=False) + self.q_norm = RMSNorm(self.head_dim) + self.k_norm = RMSNorm(self.head_dim) + self.scale = self.head_dim**-0.5 + + def __call__(self, x: mx.array, freqs: mx.array) -> mx.array: + B, L, _ = x.shape + qkv = self.to_qkv(x) + q, k, v = mx.split(qkv, 3, axis=-1) + + q = q.reshape(B, L, self.num_heads, self.head_dim).transpose(0, 2, 1, 3) + k = k.reshape(B, L, self.num_heads, self.head_dim).transpose(0, 2, 1, 3) + v = v.reshape(B, L, self.num_heads, self.head_dim).transpose(0, 2, 1, 3) + + q = self.q_norm(q) + k = self.k_norm(k) + + if freqs is not None: + q = apply_rotary_emb(q, freqs) + k = apply_rotary_emb(k, freqs) + + attn = (q @ k.transpose(0, 1, 3, 2)) * self.scale + attn = mx.softmax(attn, axis=-1) + out = (attn @ v).transpose(0, 2, 1, 3).reshape(B, L, -1) + return self.to_out(out) + + +class CrossAttention(nn.Module): + def __init__(self, dim: int, num_heads: int, cond_dim: int): + super().__init__() + self.num_heads = num_heads + self.head_dim = dim // num_heads + self.to_q = nn.Linear(dim, dim, bias=False) + self.to_kv = nn.Linear(cond_dim, dim * 2, bias=False) + self.to_out = nn.Linear(dim, dim, bias=False) + self.q_norm = RMSNorm(self.head_dim) + self.k_norm = RMSNorm(self.head_dim) + self.scale = self.head_dim**-0.5 + + def __call__( + self, + x: mx.array, + context: mx.array, + mask: Optional[mx.array] = None, + ) -> mx.array: + B, L, _ = x.shape + _, S, _ = context.shape + + q = self.to_q(x) + kv = self.to_kv(context) + k, v = mx.split(kv, 2, axis=-1) + + q = q.reshape(B, L, self.num_heads, self.head_dim).transpose(0, 2, 1, 3) + k = k.reshape(B, S, self.num_heads, self.head_dim).transpose(0, 2, 1, 3) + v = v.reshape(B, S, self.num_heads, self.head_dim).transpose(0, 2, 1, 3) + + q = self.q_norm(q) + k = self.k_norm(k) + + attn = (q @ k.transpose(0, 1, 3, 2)) * self.scale + if mask is not None: + attn = attn + mask + attn = mx.softmax(attn, axis=-1) + out = (attn @ v).transpose(0, 2, 1, 3).reshape(B, L, -1) + return self.to_out(out) + + +class TransformerBlock(nn.Module): + def __init__(self, config: DiTConfig): + super().__init__() + dim = config.embed_dim + + self.to_scale_shift_gate = mx.zeros((6 * dim,)) + + self.pre_norm = RMSNorm(dim) + self.self_attn = SelfAttention(dim, config.num_heads) + + self.cross_attend_norm = RMSNorm(dim) + self.cross_attn = CrossAttention(dim, config.num_heads, dim) + + if config.local_add_cond_dim > 0: + self.to_local_embed = [ + nn.Linear(config.local_add_cond_dim, dim, bias=True), + None, + nn.Linear(dim, dim, bias=True), + ] + + self.ff_norm = RMSNorm(dim) + self.ff = GLU(dim, config.ff_mult) + + def __call__( + self, + x: mx.array, + freqs: mx.array, + cond_tokens: mx.array, + adaln_params: mx.array, + local_cond: Optional[mx.array] = None, + cond_mask: Optional[mx.array] = None, + ) -> mx.array: + adaln_params = adaln_params + self.to_scale_shift_gate + scale_self, shift_self, gate_self, scale_ff, shift_ff, gate_ff = mx.split( + adaln_params, 6, axis=-1 + ) + + residual = x + x = self.pre_norm(x) + x = x * (1 + scale_self) + shift_self + x = self.self_attn(x, freqs) + x = x * mx.sigmoid(1 - gate_self) + x = x + residual + + x = x + self.cross_attn(self.cross_attend_norm(x), cond_tokens, mask=cond_mask) + + if local_cond is not None and hasattr(self, "to_local_embed"): + lc = self.to_local_embed[0](local_cond) + lc = nn.silu(lc) + lc = self.to_local_embed[2](lc) + x = x + lc + + residual = x + x = self.ff_norm(x) + x = x * (1 + scale_ff) + shift_ff + x = self.ff(x) + x = x * mx.sigmoid(1 - gate_ff) + x = x + residual + + return x + + +class ContinuousTransformer(nn.Module): + def __init__(self, config: DiTConfig): + super().__init__() + dim = config.embed_dim + + self.project_in = nn.Linear(config.io_channels, dim, bias=False) + self.project_out = nn.Linear(dim, config.io_channels, bias=False) + + self.memory_tokens = mx.zeros((config.num_memory_tokens, dim)) + self.rotary_pos_emb = RotaryEmbedding(dim // config.num_heads // 2) + + self.global_cond_embedder = [ + nn.Linear(dim, dim, bias=True), + None, + nn.Linear(dim, dim * 6, bias=True), + ] + + self.layers = [TransformerBlock(config) for _ in range(config.depth)] + + def __call__( + self, + x: mx.array, + cond_tokens: mx.array, + global_embed: mx.array, + local_cond: Optional[mx.array] = None, + cond_mask: Optional[mx.array] = None, + ) -> mx.array: + B = x.shape[0] + x = self.project_in(x) + + mem = mx.broadcast_to( + self.memory_tokens, + (B, self.memory_tokens.shape[0], self.memory_tokens.shape[1]), + ) + x = mx.concatenate([mem, x], axis=1) + + if local_cond is not None: + mem_pad = mx.zeros((B, mem.shape[1], local_cond.shape[-1])) + local_cond = mx.concatenate([mem_pad, local_cond], axis=1) + + freqs = self.rotary_pos_emb(x.shape[1]) + + adaln = self.global_cond_embedder[0](global_embed) + adaln = nn.silu(adaln) + adaln = self.global_cond_embedder[2](adaln) + adaln = mx.expand_dims(adaln, 1) + + for layer in self.layers: + x = layer(x, freqs, cond_tokens, adaln, local_cond, cond_mask) + + x = x[:, mem.shape[1] :] + x = self.project_out(x) + return x + + +class DiffusionTransformer(nn.Module): + def __init__(self, config: DiTConfig): + super().__init__() + + self.preprocess_conv = nn.Conv1d( + config.io_channels, config.io_channels, 1, bias=False + ) + self.postprocess_conv = nn.Conv1d( + config.io_channels, config.io_channels, 1, bias=False + ) + + self.timestep_features = ExpoFourierFeatures(256) + self.to_timestep_embed = [ + nn.Linear(256, config.embed_dim, bias=True), + None, + nn.Linear(config.embed_dim, config.embed_dim, bias=True), + ] + + self.to_cond_embed = [ + nn.Linear(config.cond_token_dim, config.embed_dim, bias=False), + None, + nn.Linear(config.embed_dim, config.embed_dim, bias=False), + ] + self.to_global_embed = [ + nn.Linear(config.global_cond_dim, config.embed_dim, bias=False), + None, + nn.Linear(config.embed_dim, config.embed_dim, bias=False), + ] + + self.transformer = ContinuousTransformer(config) + + def __call__( + self, + x: mx.array, + t: mx.array, + cond_tokens: mx.array, + global_cond: mx.array, + local_cond: Optional[mx.array] = None, + cond_mask: Optional[mx.array] = None, + ) -> mx.array: + x = x.transpose(0, 2, 1) + x = self.preprocess_conv(x) + x + + t_emb = self.to_timestep_embed[0](self.timestep_features(t)) + t_emb = self.to_timestep_embed[2](nn.silu(t_emb)) + + cond_tokens = self.to_cond_embed[0](cond_tokens) + cond_tokens = self.to_cond_embed[2](nn.silu(cond_tokens)) + + g_emb = self.to_global_embed[0](global_cond) + g_emb = self.to_global_embed[2](nn.silu(g_emb)) + + global_embed = t_emb + g_emb + + x = self.transformer(x, cond_tokens, global_embed, local_cond, cond_mask) + x = self.postprocess_conv(x) + x + + return x.transpose(0, 2, 1) diff --git a/mlx_audio/tts/models/stable_audio_3/same.py b/mlx_audio/tts/models/stable_audio_3/same.py new file mode 100644 index 00000000..e096b4f1 --- /dev/null +++ b/mlx_audio/tts/models/stable_audio_3/same.py @@ -0,0 +1,308 @@ +"""SAME (Stereo Audio Masked autoEncoder) decoder for Stable Audio 3. + +Decodes 256-dim latent sequences to stereo 44.1kHz audio via: + SoftNormBottleneck → TransformerResamplingBlock (16x upsample) + → Conv1d mapping → patch unpacking (256x upsample) → stereo audio + +Total downsampling ratio: 4096 (256 patch_size × 16 stride). +""" + +from dataclasses import dataclass, field +from typing import Optional + +import mlx.core as mx +import mlx.nn as nn + +from .dit import RotaryEmbedding, apply_rotary_emb + + +@dataclass +class SAMEConfig: + latent_dim: int = 256 + patch_size: int = 256 + audio_channels: int = 2 + encoder_channels: int = 128 + encoder_c_mults: list = field(default_factory=lambda: [6]) + encoder_strides: list = field(default_factory=lambda: [16]) + encoder_depths: list = field(default_factory=lambda: [6]) + dim_heads: int = 64 + downsampling_ratio: int = 4096 + ff_mult: int = 3 + chunk_size: int = 32 + chunk_midpoint_shift: bool = True + conv_mapping: bool = True + differential: bool = True + decoder_out_channels: int = 512 + + +class SAMEBottleneck(nn.Module): + def __init__(self, dim: int): + super().__init__() + self.scaling_factor = mx.ones((1, dim, 1)) + self.bias = mx.zeros((1, dim, 1)) + # running_std is learned during training; decode multiplies by it + self.running_std = mx.ones((1,)) + + def decode(self, x: mx.array) -> mx.array: + return x * self.running_std + + +class DynamicTanh(nn.Module): + def __init__(self, dim: int): + super().__init__() + self.alpha = mx.array([4.0]) + self.gamma = mx.ones((dim,)) + self.beta = mx.zeros((dim,)) + + def __call__(self, x: mx.array) -> mx.array: + return self.gamma * mx.tanh(self.alpha * x) + self.beta + + +class _NoOp(nn.Module): + def __call__(self, x: mx.array) -> mx.array: + return x + + +class SAMEGLU(nn.Module): + def __init__(self, dim_in: int, dim_out: int): + super().__init__() + self.proj = nn.Linear(dim_in, dim_out * 2) + + def __call__(self, x: mx.array) -> mx.array: + x = self.proj(x) + x, gate = mx.split(x, 2, axis=-1) + return x * nn.silu(gate) + + +class SAMEFeedForward(nn.Module): + def __init__(self, dim: int, mult: int = 3): + super().__init__() + inner_dim = dim * mult + self.ff = [SAMEGLU(dim, inner_dim), _NoOp(), nn.Linear(inner_dim, dim), _NoOp()] + + def __call__(self, x: mx.array) -> mx.array: + x = self.ff[0](x) + x = self.ff[2](x) + return x + + +class SAMEDiffAttention(nn.Module): + """Differential attention: attn(q,k,v) - attn(q_diff,k_diff,v).""" + + def __init__(self, dim: int, dim_heads: int = 64): + super().__init__() + self.dim_heads = dim_heads + self.num_heads = dim // dim_heads + self.to_qkv = nn.Linear(dim, dim * 5, bias=False) + self.to_out = nn.Linear(dim, dim, bias=False) + self.q_norm = DynamicTanh(dim_heads) + self.k_norm = DynamicTanh(dim_heads) + + def __call__(self, x: mx.array, freqs: Optional[mx.array] = None) -> mx.array: + B, L, _ = x.shape + h, dh = self.num_heads, self.dim_heads + + qkv = self.to_qkv(x) + q, k, v, q_diff, k_diff = mx.split(qkv, 5, axis=-1) + + def reshape(t): + return t.reshape(B, L, h, dh).transpose(0, 2, 1, 3) + + q, k, v = reshape(q), reshape(k), reshape(v) + q_diff, k_diff = reshape(q_diff), reshape(k_diff) + + q, k = self.q_norm(q), self.k_norm(k) + q_diff, k_diff = self.q_norm(q_diff), self.k_norm(k_diff) + + if freqs is not None: + q = apply_rotary_emb(q, freqs) + k = apply_rotary_emb(k, freqs) + q_diff = apply_rotary_emb(q_diff, freqs) + k_diff = apply_rotary_emb(k_diff, freqs) + + scale = dh**-0.5 + + attn = mx.softmax((q @ k.transpose(0, 1, 3, 2)) * scale, axis=-1) + out = attn @ v + + attn_diff = mx.softmax((q_diff @ k_diff.transpose(0, 1, 3, 2)) * scale, axis=-1) + out_diff = attn_diff @ v + + out = (out - out_diff).transpose(0, 2, 1, 3).reshape(B, L, -1) + return self.to_out(out) + + +class SAMETransformerBlock(nn.Module): + def __init__(self, dim: int, dim_heads: int = 64, ff_mult: int = 3): + super().__init__() + self.pre_norm = DynamicTanh(dim) + self.self_attn = SAMEDiffAttention(dim, dim_heads) + self.ff_norm = DynamicTanh(dim) + self.ff = SAMEFeedForward(dim, ff_mult) + # Only first half of dim_heads gets RoPE (partial rotation) + self.rope = RotaryEmbedding(dim_heads // 2) + + def __call__(self, x: mx.array) -> mx.array: + freqs = self.rope(x.shape[1]) + x = x + self.self_attn(self.pre_norm(x), freqs) + x = x + self.ff(self.ff_norm(x)) + return x + + +class SAMEResamplingBlock(nn.Module): + """Upsamples via token expansion: each input token spawns stride new tokens.""" + + def __init__(self, config: SAMEConfig): + super().__init__() + inner_dim = config.encoder_channels * config.encoder_c_mults[0] + out_channels = config.decoder_out_channels + stride = config.encoder_strides[0] + depth = config.encoder_depths[0] + self.stride = stride + self.chunk_size = config.chunk_size + self.chunk_midpoint_shift = config.chunk_midpoint_shift + + kernel_size = 3 if config.conv_mapping else 1 + self.mapping = nn.Conv1d( + inner_dim, out_channels, kernel_size, padding=kernel_size // 2 + ) + self.new_tokens = mx.zeros((1, 1, inner_dim)) + self.transformers = [ + SAMETransformerBlock(inner_dim, config.dim_heads, config.ff_mult) + for _ in range(depth) + ] + + def __call__(self, x: mx.array, stride: Optional[int] = None) -> mx.array: + if stride is None: + stride = self.stride + B = x.shape[0] + + x = x.transpose(0, 2, 1) + + pad_modulo = self.chunk_size // stride + L = x.shape[1] + if L % pad_modulo != 0: + pad_len = pad_modulo - (L % pad_modulo) + x = mx.pad(x, [(0, 0), (0, pad_len), (0, 0)]) + + n = x.shape[1] + x = x.reshape(B * n, 1, -1) + new_tokens = mx.broadcast_to(self.new_tokens, (B * n, stride, x.shape[-1])) + x = mx.concatenate([x, new_tokens], axis=1) + sub_chunk_size = stride + 1 + x = x.reshape(B, n * sub_chunk_size, -1) + + effective_chunk_size = self.chunk_size + pad_modulo + + if self.chunk_midpoint_shift: + split = len(self.transformers) // 2 + shift = effective_chunk_size // 2 + nc = x.shape[1] // effective_chunk_size + + x = x.reshape(B * nc, effective_chunk_size, -1) + for layer in self.transformers[:split]: + x = layer(x) + x = x.reshape(B, nc * effective_chunk_size, -1) + + x = mx.concatenate([x[:, :shift, :], x, x[:, -shift:, :]], axis=1) + nc_shifted = x.shape[1] // effective_chunk_size + x = x.reshape(B * nc_shifted, effective_chunk_size, -1) + for layer in self.transformers[split:]: + x = layer(x) + x = x.reshape(B, nc_shifted * effective_chunk_size, -1) + x = x[:, shift:-shift, :] + else: + nc = x.shape[1] // effective_chunk_size + x = x.reshape(B * nc, effective_chunk_size, -1) + for layer in self.transformers: + x = layer(x) + x = x.reshape(B, nc * effective_chunk_size, -1) + + x = x.reshape(B * n, sub_chunk_size, -1) + x = x[:, -stride:, :] + x = x.reshape(B, -1, x.shape[-1]) + + x = self.mapping(x) + x = x.transpose(0, 2, 1) + return x + + +class _Transpose(nn.Module): + def __call__(self, x: mx.array) -> mx.array: + return x.transpose(0, 2, 1) + + +class SAMEDecoder(nn.Module): + def __init__(self, config: SAMEConfig): + super().__init__() + inner_dim = config.encoder_channels * config.encoder_c_mults[0] + self.layers = [ + _Transpose(), + nn.Linear(config.latent_dim, inner_dim), + _Transpose(), + SAMEResamplingBlock(config), + ] + + def __call__(self, x: mx.array) -> mx.array: + for layer in self.layers: + x = layer(x) + return x + + +class SAMEEncoderResamplingBlock(nn.Module): + """Encoder resampling block — only needed for weight loading.""" + + def __init__(self, config: SAMEConfig): + super().__init__() + inner_dim = config.encoder_channels * config.encoder_c_mults[0] + in_channels = config.decoder_out_channels + self.mapping = nn.Conv1d(in_channels, inner_dim, 1) + self.new_tokens = mx.zeros((1, 1, inner_dim)) + self.transformers = [ + SAMETransformerBlock(inner_dim, config.dim_heads, config.ff_mult) + for _ in range(config.encoder_depths[0]) + ] + + +class SAMEEncoderStub(nn.Module): + """Encoder stub — only needed for weight loading, not used during inference.""" + + def __init__(self, config: SAMEConfig): + super().__init__() + inner_dim = config.encoder_channels * config.encoder_c_mults[0] + self.layers = [ + SAMEEncoderResamplingBlock(config), + _NoOp(), + nn.Linear(inner_dim, config.latent_dim), + ] + + +class SAMEAutoencoder(nn.Module): + def __init__(self, config: SAMEConfig): + super().__init__() + self.bottleneck = SAMEBottleneck(config.latent_dim) + self.encoder = SAMEEncoderStub(config) + self.decoder = SAMEDecoder(config) + + def decode(self, latents: mx.array) -> mx.array: + x = self.bottleneck.decode(latents) + return self.decoder(x) + + +class Pretransform(nn.Module): + """Wraps SAME autoencoder; handles patch unpacking on decode.""" + + def __init__(self, config: SAMEConfig): + super().__init__() + self.model = SAMEAutoencoder(config) + self.patch_size = config.patch_size + self.audio_channels = config.audio_channels + self.downsampling_ratio = config.downsampling_ratio + + def decode(self, latents: mx.array) -> mx.array: + x = self.model.decode(latents) + B, C, T = x.shape + x = x.reshape(B, self.audio_channels, self.patch_size, T) + x = x.transpose(0, 1, 3, 2) + x = x.reshape(B, self.audio_channels, T * self.patch_size) + return x diff --git a/mlx_audio/tts/models/stable_audio_3/stable_audio_3.py b/mlx_audio/tts/models/stable_audio_3/stable_audio_3.py new file mode 100644 index 00000000..f7282d2f --- /dev/null +++ b/mlx_audio/tts/models/stable_audio_3/stable_audio_3.py @@ -0,0 +1,524 @@ +"""Stable Audio 3 — text-to-audio diffusion model for MLX. + +Generates stereo 44.1kHz audio (sound effects up to 30s, music up to 30s) +from text prompts using a DiT backbone, T5Gemma text encoder, and SAME decoder. + +Supports two official variants: + - stabilityai/stable-audio-3-small-sfx (sound effects) + - stabilityai/stable-audio-3-small-music (music) +""" + +import math +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Generator, Optional + +import mlx.core as mx +import mlx.nn as nn +import numpy as np + +from ..base import BaseModelArgs, GenerationResult +from .dit import DiffusionTransformer, DiTConfig +from .same import Pretransform, SAMEConfig + +CONV_SUFFIXES = { + "preprocess_conv.weight", + "postprocess_conv.weight", + "mapping.weight", + "out_mapping.weight", +} + + +@dataclass +class ModelConfig(BaseModelArgs): + dit: dict = field(default_factory=dict) + same: dict = field(default_factory=dict) + sample_rate: int = 44100 + audio_channels: int = 2 + model_type: str = "stable_audio_3" + model_path: str = "" + + def get_dit_config(self) -> DiTConfig: + return DiTConfig(**self.dit) if self.dit else DiTConfig() + + def get_same_config(self) -> SAMEConfig: + return SAMEConfig(**self.same) if self.same else SAMEConfig() + + @classmethod + def from_dict(cls, d: dict) -> "ModelConfig": + model = d.get("model", d) + diff_cfg = model.get("diffusion", {}).get("config", {}) + ae_cfg = model.get("pretransform", {}).get("config", {}) + enc_cfg = ae_cfg.get("encoder", {}).get("config", {}) + dec_cfg = ae_cfg.get("decoder", {}).get("config", {}) + patch_cfg = ae_cfg.get("pretransform", {}).get("config", {}) + + dit = dict( + io_channels=diff_cfg.get("io_channels", 256), + embed_dim=diff_cfg.get("embed_dim", 1024), + depth=diff_cfg.get("depth", 20), + num_heads=diff_cfg.get("num_heads", 16), + cond_token_dim=diff_cfg.get("cond_token_dim", 768), + global_cond_dim=diff_cfg.get("global_cond_dim", 768), + local_add_cond_dim=diff_cfg.get("local_add_cond_dim", 257), + num_memory_tokens=diff_cfg.get("num_memory_tokens", 64), + ff_mult=diff_cfg.get("ff_kwargs", {}).get("mult", 4.0), + ) + + channels = enc_cfg.get("channels", 128) + c_mults = enc_cfg.get("c_mults", [6]) + strides = enc_cfg.get("strides", [16]) + patch_size = patch_cfg.get("patch_size", 256) + audio_channels = patch_cfg.get("channels", 2) + downsampling_ratio = patch_size + for s in strides: + downsampling_ratio *= s + + same = dict( + latent_dim=ae_cfg.get("latent_dim", 256), + patch_size=patch_size, + audio_channels=audio_channels, + encoder_channels=channels, + encoder_c_mults=c_mults, + encoder_strides=strides, + encoder_depths=enc_cfg.get("transformer_depths", [6]), + dim_heads=enc_cfg.get("dim_heads", 64), + downsampling_ratio=downsampling_ratio, + ff_mult=dec_cfg.get("ff_mult", 3), + chunk_size=dec_cfg.get("chunk_size", 32), + chunk_midpoint_shift=dec_cfg.get("chunk_midpoint_shift", True), + conv_mapping=dec_cfg.get("conv_mapping", True), + differential=dec_cfg.get("differential", True), + decoder_out_channels=dec_cfg.get( + "out_channels", patch_size * audio_channels + ), + ) + + return cls( + dit=dit, + same=same, + sample_rate=d.get("sample_rate", 44100), + audio_channels=d.get("audio_channels", 2), + model_type=d.get("model_type", "stable_audio_3"), + model_path=d.get("model_path", ""), + ) + + +class _DiffusionWrapper(nn.Module): + """Matches PyTorch weight prefix: model.model.*""" + + def __init__(self, config: DiTConfig): + super().__init__() + self.model = DiffusionTransformer(config) + + +class _PromptConditioner(nn.Module): + def __init__(self): + super().__init__() + self.padding_embedding = mx.zeros((768,)) + + +class _DurationEmbedder(nn.Module): + def __init__(self): + super().__init__() + self.embedding = [None, nn.Linear(256, 768, bias=True)] + + +class _DurationConditioner(nn.Module): + def __init__(self): + super().__init__() + self.embedder = _DurationEmbedder() + + +class _ConditionerSet(nn.Module): + def __init__(self): + super().__init__() + self.prompt = _PromptConditioner() + self.seconds_total = _DurationConditioner() + + +class _Conditioner(nn.Module): + def __init__(self): + super().__init__() + self.conditioners = _ConditionerSet() + + +class Model(nn.Module): + def __init__(self, config: ModelConfig): + super().__init__() + self.config = config + dit_config = config.get_dit_config() + same_config = config.get_same_config() + + self.model = _DiffusionWrapper(dit_config) + self.conditioner = _Conditioner() + self.pretransform = Pretransform(same_config) + + self._t5_tokenizer = None + self._t5_model = None + + @property + def sample_rate(self) -> int: + return self.config.sample_rate + + def sanitize(self, weights: dict) -> dict: + result = {} + skip_keys = set() + + for key in sorted(weights.keys()): + if key in skip_keys: + continue + + # Merge weight normalization pairs (weight_g + weight_v -> weight) + if key.endswith(".weight_g"): + base = key[: -len(".weight_g")] + v_key = base + ".weight_v" + if v_key in weights: + g = np.array(weights[key]) + v = np.array(weights[v_key]) + norm = np.sqrt(np.sum(v**2, axis=(1, 2), keepdims=True)) + norm = np.maximum(norm, 1e-12) + merged = (g * v / norm).astype(np.float32) + weight_key = base + ".weight" + if ( + any(weight_key.endswith(s) for s in CONV_SUFFIXES) + and len(merged.shape) == 3 + and merged.shape[-1] < merged.shape[-2] + ): + merged = np.transpose(merged, (0, 2, 1)) + result[weight_key] = mx.array(merged) + skip_keys.add(key) + skip_keys.add(v_key) + continue + + # Filter training-only zero params (but keep running_std) + if "noise_scaling_factor" in key: + arr = np.array(weights[key]) + if arr.size <= 1 or np.all(arr == 0): + continue + + value = weights[key] + arr = np.array(value) if not isinstance(value, np.ndarray) else value + + # Transpose Conv1d weights: PyTorch [out, in, K] -> MLX [out, K, in] + # Skip if already in MLX format (shape[-1] >= shape[-2]) + if ( + any(key.endswith(s) for s in CONV_SUFFIXES) + and len(arr.shape) == 3 + and arr.shape[-1] < arr.shape[-2] + ): + arr = np.transpose(arr, (0, 2, 1)) + + result[key] = mx.array(arr) + + return result + + def _load_t5gemma(self, model_path: str): + import torch + from transformers import AutoConfig, AutoTokenizer, T5GemmaEncoderModel + + t5_path = Path(model_path) / "t5gemma-b-b-ul2" + if not t5_path.exists(): + raise FileNotFoundError( + f"T5Gemma tokenizer not found at {t5_path}. " + "Ensure the model was downloaded with the full HF repo." + ) + self._t5_tokenizer = AutoTokenizer.from_pretrained(str(t5_path)) + config = AutoConfig.from_pretrained(str(t5_path)) + config.is_encoder_decoder = False + self._t5_model = T5GemmaEncoderModel.from_pretrained( + str(t5_path), config=config, torch_dtype=torch.float32 + ) + self._t5_model.eval() + + def _encode_text( + self, prompt: str, max_length: int = 256 + ) -> tuple[mx.array, mx.array]: + import torch + + inputs = self._t5_tokenizer( + prompt, + return_tensors="pt", + padding="max_length", + truncation=True, + max_length=max_length, + ) + with torch.no_grad(): + outputs = self._t5_model(**inputs) + embeddings = outputs.last_hidden_state + + text_emb = mx.array(embeddings.numpy()) + attn_mask = mx.array(inputs["attention_mask"].numpy()) + return text_emb, attn_mask + + def _encode_duration(self, duration: float) -> mx.array: + t = mx.array([duration / 384.0]) + half = 128 + freqs = mx.exp(mx.linspace(0.0, 1.0, half) * math.log(10000.0)) + t_expanded = mx.expand_dims(t, -1) + fourier = mx.concatenate( + [mx.sin(t_expanded * freqs), mx.cos(t_expanded * freqs)], axis=-1 + ) + emb = self.conditioner.conditioners.seconds_total.embedder.embedding[1] + return emb(fourier) + + def _build_schedule( + self, + steps: int, + seq_len: int, + sigma_max: float = 1.0, + min_length: int = 256, + max_length: int = 4096, + base_shift: float = 0.5, + max_shift: float = 1.15, + ) -> mx.array: + t = mx.linspace(sigma_max, 0.0, steps + 1) + seq_len = max(min(seq_len, max_length), min_length) + mu = -( + base_shift + + (max_shift - base_shift) + * (seq_len - min_length) + / (max_length - min_length) + ) + t_shifted = 1.0 - math.exp(mu) / ( + math.exp(mu) + (1.0 / (1.0 - t + 1e-10) - 1.0) + ) + t_shifted = mx.where(t <= 0, mx.zeros_like(t_shifted), t_shifted) + t_shifted = mx.where(t >= 1, mx.ones_like(t_shifted), t_shifted) + return mx.concatenate([mx.array([sigma_max]), t_shifted[1:]]) + + def _sample_euler( + self, + noise: mx.array, + sigmas: mx.array, + cond_tokens: mx.array, + global_cond: mx.array, + cfg_scale: float, + negative_cond_tokens: Optional[mx.array], + negative_global_cond: Optional[mx.array], + verbose: bool, + ) -> mx.array: + x = noise + num_steps = sigmas.shape[0] - 1 + dit = self.model.model + + for i in range(num_steps): + t_curr = sigmas[i] + t_next = sigmas[i + 1] + dt = t_next - t_curr + t_batch = mx.broadcast_to(mx.array([t_curr.item()]), (x.shape[0],)) + + if cfg_scale != 1.0 and negative_cond_tokens is not None: + x_double = mx.concatenate([x, x], axis=0) + t_double = mx.concatenate([t_batch, t_batch], axis=0) + cond_double = mx.concatenate( + [cond_tokens, negative_cond_tokens], axis=0 + ) + global_double = mx.concatenate( + [global_cond, negative_global_cond], axis=0 + ) + v_both = dit(x_double, t_double, cond_double, global_double) + v_cond, v_uncond = mx.split(v_both, 2, axis=0) + v = v_uncond + cfg_scale * (v_cond - v_uncond) + else: + v = dit(x, t_batch, cond_tokens, global_cond) + + x = x + dt * v + mx.async_eval(x) + if verbose: + print( + f" Step {i + 1}/{num_steps} " + f"(t={t_curr.item():.3f} -> {t_next.item():.3f})" + ) + return x + + def _sample_pingpong( + self, + noise: mx.array, + sigmas: mx.array, + cond_tokens: mx.array, + global_cond: mx.array, + cfg_scale: float, + negative_cond_tokens: Optional[mx.array], + negative_global_cond: Optional[mx.array], + verbose: bool, + ) -> mx.array: + x = noise + num_steps = sigmas.shape[0] - 1 + dit = self.model.model + + for i in range(num_steps): + t_curr = sigmas[i] + t_next = sigmas[i + 1] + t_batch = mx.broadcast_to(mx.array([t_curr.item()]), (x.shape[0],)) + + if cfg_scale != 1.0 and negative_cond_tokens is not None: + x_double = mx.concatenate([x, x], axis=0) + t_double = mx.concatenate([t_batch, t_batch], axis=0) + cond_double = mx.concatenate( + [cond_tokens, negative_cond_tokens], axis=0 + ) + global_double = mx.concatenate( + [global_cond, negative_global_cond], axis=0 + ) + v_both = dit(x_double, t_double, cond_double, global_double) + v_cond, v_uncond = mx.split(v_both, 2, axis=0) + v = v_uncond + cfg_scale * (v_cond - v_uncond) + else: + v = dit(x, t_batch, cond_tokens, global_cond) + + t_curr_expanded = mx.expand_dims(mx.expand_dims(t_curr, 0), 0) + denoised = x - t_curr_expanded * v + + t_next_expanded = mx.expand_dims(mx.expand_dims(t_next, 0), 0) + x = (1 - t_next_expanded) * denoised + t_next_expanded * mx.random.normal( + x.shape + ) + mx.async_eval(x) + if verbose: + print( + f" Step {i + 1}/{num_steps} " + f"(t={t_curr.item():.3f} -> {t_next.item():.3f})" + ) + return x + + def generate( + self, + text: str, + model_path: Optional[str] = None, + duration: Optional[float] = None, + ddpm_steps: Optional[int] = None, + cfg_scale: Optional[float] = None, + sampler: Optional[str] = None, + seed: Optional[int] = None, + verbose: bool = True, + **kwargs, + ) -> Generator[GenerationResult, None, None]: + if duration is None: + duration = 10.0 + if ddpm_steps is None: + ddpm_steps = 8 + if cfg_scale is None: + cfg_scale = 1.0 + if sampler is None: + sampler = "pingpong" + if seed is None: + seed = 42 + start_time = time.time() + + if self._t5_tokenizer is None: + path = model_path or self.config.model_path + if not path: + raise ValueError( + "model_path is required on first call to load T5Gemma encoder" + ) + if verbose: + print("Loading T5Gemma text encoder...") + self._load_t5gemma(path) + + if verbose: + print(f"Encoding prompt: '{text}'") + text_emb, text_mask = self._encode_text(text) + + padding_emb = self.conditioner.conditioners.prompt.padding_embedding + inv_mask = 1.0 - mx.expand_dims(text_mask, -1) + text_emb = text_emb * mx.expand_dims(text_mask, -1) + padding_emb * inv_mask + + if verbose: + print(f"Encoding duration: {duration}s") + duration_emb = self._encode_duration(duration) + + duration_emb_expanded = mx.expand_dims(duration_emb, 1) + cond_tokens = mx.concatenate([text_emb, duration_emb_expanded], axis=1) + global_cond = duration_emb + + negative_cond_tokens = None + negative_global_cond = None + if cfg_scale != 1.0: + neg_text_emb = mx.broadcast_to(padding_emb, text_emb.shape) + negative_cond_tokens = mx.concatenate( + [neg_text_emb, mx.expand_dims(duration_emb, 1)], axis=1 + ) + negative_global_cond = duration_emb + + dit_config = self.config.get_dit_config() + same_config = self.config.get_same_config() + audio_samples = int(duration * self.config.sample_rate) + latent_len = audio_samples // same_config.downsampling_ratio + padding_tokens = int( + 6.0 * self.config.sample_rate / same_config.downsampling_ratio + ) + latent_len += padding_tokens + + if verbose: + print(f"Latent shape: [1, {dit_config.io_channels}, {latent_len}]") + + mx.random.seed(seed) + noise = mx.random.normal((1, dit_config.io_channels, latent_len)) + sigmas = self._build_schedule(ddpm_steps, seq_len=latent_len) + + if verbose: + print(f"Sampling with {sampler} ({ddpm_steps} steps)...") + + sample_fn = ( + self._sample_pingpong if sampler == "pingpong" else self._sample_euler + ) + latents = sample_fn( + noise, + sigmas, + cond_tokens, + global_cond, + cfg_scale, + negative_cond_tokens, + negative_global_cond, + verbose, + ) + + if verbose: + print("Decoding latents to audio...") + audio = self.pretransform.decode(latents) + mx.eval(audio) + + target_samples = int(duration * self.config.sample_rate) + audio = audio[:, :, :target_samples] + + elapsed = time.time() - start_time + audio_duration_seconds = target_samples / self.config.sample_rate + rtf = elapsed / audio_duration_seconds if audio_duration_seconds > 0 else 0 + + duration_hours = int(audio_duration_seconds // 3600) + duration_mins = int(audio_duration_seconds % 3600 // 60) + duration_secs = int(audio_duration_seconds % 60) + duration_ms = int((audio_duration_seconds % 1) * 1000) + duration_str = ( + f"{duration_hours:02d}:{duration_mins:02d}:" + f"{duration_secs:02d}.{duration_ms:03d}" + ) + + # audio shape is [1, channels, samples]; flatten to [samples, channels] + audio_out = audio[0].transpose(1, 0) + + yield GenerationResult( + audio=audio_out, + samples=target_samples, + sample_rate=self.config.sample_rate, + segment_idx=0, + token_count=latent_len * ddpm_steps, + audio_duration=duration_str, + real_time_factor=round(rtf, 2), + prompt={ + "tokens": latent_len, + "tokens-per-sec": ( + round(latent_len / elapsed, 2) if elapsed > 0 else 0 + ), + }, + audio_samples={ + "samples": target_samples, + "samples-per-sec": ( + round(target_samples / elapsed, 2) if elapsed > 0 else 0 + ), + }, + processing_time_seconds=elapsed, + peak_memory_usage=mx.get_peak_memory() / 1e9, + ) diff --git a/mlx_audio/tts/utils.py b/mlx_audio/tts/utils.py index a511dbf5..b4e1a412 100644 --- a/mlx_audio/tts/utils.py +++ b/mlx_audio/tts/utils.py @@ -17,6 +17,7 @@ ) MODEL_REMAPPING = { + "stable_audio_3": "stable_audio_3", "qwen3_tts": "qwen3_tts", "outetts": "outetts", "spark": "spark", diff --git a/mlx_audio/utils.py b/mlx_audio/utils.py index c6533c51..f1486fd7 100644 --- a/mlx_audio/utils.py +++ b/mlx_audio/utils.py @@ -168,10 +168,11 @@ def load_config(model_path: Union[str, Path], **kwargs) -> dict: if isinstance(model_path, str): model_path = get_model_path(model_path, **kwargs) - config_file = model_path / "config.json" - if config_file.exists(): - with open(config_file, encoding="utf-8") as f: - return json.load(f) + for name in ("config.json", "model_config.json"): + config_file = model_path / name + if config_file.exists(): + with open(config_file, encoding="utf-8") as f: + return json.load(f) raise FileNotFoundError(f"Config not found at {model_path}")