from __future__ import annotations import io import os import base64 import sys import random from pathlib import Path from typing import Dict, List, Optional, Tuple import gradio as gr import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt import matplotlib.patches as mpatches import numpy as np import pretty_midi import torch import torch.nn as nn import torch.nn.functional as F from huggingface_hub import hf_hub_download from PIL import Image ROOT = Path(__file__).resolve().parent SRC_DIR = ROOT / "src" if str(SRC_DIR) not in sys.path: sys.path.insert(0, str(SRC_DIR)) from compound import AXIS_SIZES, N_AXES, SENTINELS, STEP_BOS, STEP_EOS, decode_compound from compound_model import CompoundGPT, CompoundGPTConfig, default_compound_config # ── Config ──────────────────────────────────────────────────────────────────── HF_REPO_ID = os.getenv("HF_REPO_ID", "Prajanya23/Coda") HF_TOKEN = os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACE_HUB_TOKEN") TMP_MIDI = "/tmp/coda_output.mid" N_PREFIX = 8 CLAP_DIM = 256 CTX_WINDOW = 64 # sliding window: keeps cost constant per step # Checkpoint paths inside the HF repo (set via Space secrets to override) GPT_FILE = os.getenv("GPT_CKPT", "checkpoints/compound_best.pt") CLAP_FILE = os.getenv("CLAP_CKPT", "checkpoints/clap_compound_best.pt") PREFIX_FILE = os.getenv("PREFIX_CKPT", "checkpoints/prefix_projector_best.pt") EXAMPLES = [ "a slow melancholic piano piece in a minor key with sparse flowing notes", "an upbeat jazz trio with piano bass and drums syncopated and energetic", "ambient electronic music with synthesizer pads slow and atmospheric", "fast energetic rock band with electric guitar and drums", "a gentle classical piece for piano and strings moderate tempo", "a funky groove with bass guitar and brass instruments", "a soft acoustic guitar piece fingerpicked quiet and introspective", "an orchestral piece with strings and brass building to a climax", ] VOICE_COLORS = ["#534AB7", "#0F6E56", "#BA7517", "#993C1D", "#185FA5", "#639922", "#A32D2D", "#D4537E"] # ── Model cache ─────────────────────────────────────────────────────────────── _CACHE: Dict[str, object] = {} def _dl(filename: str) -> str: return hf_hub_download(repo_id=HF_REPO_ID, filename=filename, token=HF_TOKEN) # ── GPT loader ──────────────────────────────────────────────────────────────── def _load_gpt() -> CompoundGPT: if "gpt" in _CACHE: return _CACHE["gpt"] # type: ignore[return-value] ckpt = torch.load(_dl(GPT_FILE), map_location="cpu", weights_only=True) cfg = default_compound_config() raw = ckpt.get("config") if isinstance(ckpt, dict) else None if isinstance(raw, dict): for k, v in raw.items(): if hasattr(cfg, k): setattr(cfg, k, v) model = CompoundGPT(cfg) model.load_state_dict(ckpt.get("model_state_dict", ckpt), strict=False) model.eval() _CACHE["gpt"] = model return model # ── CLAP text encoder (lightweight — no CompoundGPT inside) ────────────────── class _TextEncoder(nn.Module): """Sentence-transformer + CLAP text projection, reconstructed from checkpoint.""" def __init__(self, clap_state: dict): super().__init__() from sentence_transformers import SentenceTransformer self._st = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2") # Extract text_projection weights from CLAP state dict proj = {k[len("text_projection."):]: v for k, v in clap_state.items() if k.startswith("text_projection.")} # Only 2D tensors = Linear weights (excludes LayerNorm/BN gamma which are 1D) weight_keys = sorted(k for k in proj if k.endswith(".weight") and proj[k].dim() == 2) print(f"[CODA] text_projection keys found: {weight_keys}") if not weight_keys: print(f"[CODA] WARNING: no 2D weights in text_projection. Available: {sorted(proj.keys())[:10]}") self._proj = nn.Identity() else: layers: List[nn.Module] = [] for i, wk in enumerate(weight_keys): w = proj[wk] bk = wk.replace(".weight", ".bias") lin = nn.Linear(w.shape[1], w.shape[0], bias=(bk in proj)) lin.weight.data.copy_(w) if bk in proj: lin.bias.data.copy_(proj[bk]) layers.append(lin) if i < len(weight_keys) - 1: layers.append(nn.ReLU()) self._proj = nn.Sequential(*layers) @torch.no_grad() def encode(self, text: str) -> torch.Tensor: raw = self._st.encode([text], convert_to_tensor=True, show_progress_bar=False) emb = self._proj(raw.float()) return F.normalize(emb, dim=-1) # (1, 256) # ── Prefix projector (reconstructed from checkpoint) ───────────────────────── class _PrefixProjector(nn.Module): def __init__(self, state: dict, n_prefix: int = N_PREFIX): super().__init__() self.n_prefix = n_prefix weight_keys = sorted(k for k in state if k.endswith(".weight") and state[k].dim() == 2) print(f"[CODA] prefix_projector keys found: {weight_keys}") self._gpt_dim = state[weight_keys[-1]].shape[0] // n_prefix layers: List[nn.Module] = [] for i, wk in enumerate(weight_keys): w = state[wk] bk = wk.replace(".weight", ".bias") lin = nn.Linear(w.shape[1], w.shape[0], bias=(bk in state)) lin.weight.data.copy_(w) if bk in state: lin.bias.data.copy_(state[bk]) layers.append(lin) if i < len(weight_keys) - 1: layers.append(nn.GELU()) self._net = nn.Sequential(*layers) # out_ln matches PrefixProjector.out_ln in prefix_projector.py self._out_ln = nn.LayerNorm(self._gpt_dim) ln_w = state.get("out_ln.weight") ln_b = state.get("out_ln.bias") if ln_w is not None: self._out_ln.weight.data.copy_(ln_w) if ln_b is not None: self._out_ln.bias.data.copy_(ln_b) @torch.no_grad() def forward(self, text_emb: torch.Tensor) -> torch.Tensor: # (1,256) → (1,8,768) out = self._net(text_emb).view(text_emb.shape[0], self.n_prefix, self._gpt_dim) return self._out_ln(out) def _load_clap() -> Tuple[Optional[_TextEncoder], Optional[_PrefixProjector]]: if "text_enc" in _CACHE: return _CACHE.get("text_enc"), _CACHE.get("prefix_proj") # type: ignore try: # weights_only=False needed because args is argparse.Namespace clap_ckpt = torch.load(_dl(CLAP_FILE), map_location="cpu", weights_only=False) prefix_ckpt = torch.load(_dl(PREFIX_FILE), map_location="cpu", weights_only=False) clap_state = clap_ckpt.get("model_state_dict", clap_ckpt) # Prefix projector may be saved under different keys depending on training script prefix_state = None for key in ["model_state_dict", "projector_state_dict", "prefix_state_dict", "state_dict"]: if key in prefix_ckpt: prefix_state = prefix_ckpt[key] print(f"[CODA] prefix ckpt key used: '{key}'") break if prefix_state is None: # Top-level might already be the state dict — check if values are tensors if any(hasattr(v, "dim") for v in prefix_ckpt.values()): prefix_state = prefix_ckpt print("[CODA] prefix ckpt: using top-level as state dict") else: print(f"[CODA] prefix ckpt keys: {list(prefix_ckpt.keys())}") raise KeyError("Cannot find projector weights in prefix checkpoint") text_enc = _TextEncoder(_LightStateDict(clap_state)) prefix_proj = _PrefixProjector(_LightStateDict(prefix_state)) text_enc.eval(); prefix_proj.eval() _CACHE["text_enc"] = text_enc _CACHE["prefix_proj"] = prefix_proj print("[CODA] CLAP text conditioning loaded ✓") return text_enc, prefix_proj except Exception as exc: print(f"[CODA] CLAP not available ({exc}). Falling back to unconditioned generation.") _CACHE["text_enc"] = _CACHE["prefix_proj"] = None return None, None class _LightStateDict(dict): """Passthrough so _TextEncoder / _PrefixProjector constructors work with raw state dicts.""" pass # ── Compound step embeddings (needed to prepend prefix) ─────────────────────── def _compound_embeds(model: CompoundGPT, step_ids: torch.Tensor) -> torch.Tensor: """ Sum per-axis embeddings to get (B, T, n_embd) float tensor. Searches common attribute names used in CompoundGPT implementations. """ B, T, _ = step_ids.shape d = model.config.d_model emb = torch.zeros(B, T, d) axis_list = None if hasattr(model, "input_embeds"): axis_list = model.input_embeds else: for attr in ["axis_embeds", "axis_embed", "embed_axes", "wtes"]: if hasattr(model, attr): axis_list = getattr(model, attr); break if hasattr(model, "transformer") and hasattr(model.transformer, attr): axis_list = getattr(model.transformer, attr); break if axis_list is None: raise AttributeError( "Cannot find axis embedding layers in CompoundGPT (tried 'input_embeds' " "and common fallbacks). Check compound_model.py." ) for i, layer in enumerate(axis_list): emb = emb + layer(step_ids[:, :, i]) return emb # ── Sampling ────────────────────────────────────────────────────────────────── def _sample(logits: torch.Tensor, temp: float, top_k: int, top_p: float) -> int: scaled = logits / max(temp, 1e-6) # top-k if 0 < top_k < scaled.numel(): v, _ = torch.topk(scaled, top_k) scaled = scaled.masked_fill(scaled < v[-1], float("-inf")) probs = F.softmax(scaled, dim=-1) # top-p (nucleus) if 0.0 < top_p < 1.0: sp, si = torch.sort(probs, descending=True) cum = torch.cumsum(sp, dim=-1) sp[cum - sp > top_p] = 0.0 probs = torch.zeros_like(scaled).scatter_(0, si, sp) probs = probs / probs.sum().clamp(min=1e-8) return int(torch.multinomial(probs, 1).item()) # ── Generation ──────────────────────────────────────────────────────────────── @torch.no_grad() def _generate( model: CompoundGPT, prefix_embs: Optional[torch.Tensor], # (1, N_PREFIX, n_embd) or None temperature: float, top_k: int, top_p: float, max_steps: int, ) -> List[List[int]]: bos = list(SENTINELS); bos[0] = STEP_BOS generated: List[List[int]] = [bos] conditioned = prefix_embs is not None for _ in range(max_steps): # Sliding window: only feed last CTX_WINDOW steps to keep cost constant ctx = generated[-CTX_WINDOW:] if len(generated) > CTX_WINDOW else generated step_ids = torch.tensor([ctx], dtype=torch.long) # (1, ≤CTX_WINDOW, 7) T = step_ids.shape[1] n_pre = prefix_embs.shape[1] if conditioned else 0 if T + n_pre > model.config.block_size: break if conditioned: try: step_e = _compound_embeds(model, step_ids) full_e = torch.cat([prefix_embs, step_e], dim=1) # (1, 8+T, d) pos_ids = torch.arange(full_e.shape[1]).unsqueeze(0) logits = model(inputs_embeds=full_e, position_ids=pos_ids) except (AttributeError, TypeError): conditioned = False prefix_embs = None if not conditioned: pos_ids = torch.arange(T).unsqueeze(0) logits = model(idx=step_ids, position_ids=pos_ids) next_step = [ _sample(ax[0, -1, :], temperature, top_k, top_p) for ax in logits ] if next_step[0] == STEP_EOS: break generated.append(next_step) return generated # ── Piano roll ──────────────────────────────────────────────────────────────── def _piano_roll(pm: pretty_midi.PrettyMIDI) -> Image.Image: notes = [ (n.start, n.end, n.pitch, n.velocity, i) for i, inst in enumerate(pm.instruments) for n in inst.notes ] fig, ax = plt.subplots(figsize=(12, 3.2), dpi=140) fig.patch.set_facecolor("#F9F8F5") ax.set_facecolor("#F9F8F5") if notes: min_p = max(0, min(n[2] for n in notes) - 3) max_p = min(127, max(n[2] for n in notes) + 3) max_t = max(n[1] for n in notes) for start, end, pitch, vel, vi in notes: ax.broken_barh( [(start, max(0.02, end - start))], (pitch - 0.42, 0.84), facecolors=VOICE_COLORS[vi % len(VOICE_COLORS)], alpha=0.45 + 0.55 * vel / 127, linewidth=0, ) ax.set_xlim(0, max_t + 0.2) ax.set_ylim(min_p, max_p) n_inst = len(pm.instruments) if n_inst > 1: patches = [ mpatches.Patch(color=VOICE_COLORS[i % len(VOICE_COLORS)], label=pm.instruments[i].name or f"voice {i+1}") for i in range(n_inst) ] ax.legend(handles=patches, fontsize=7, loc="upper right", framealpha=0.8, edgecolor="none", facecolor="#F9F8F5") else: ax.text(0.5, 0.5, "No notes generated", ha="center", va="center", transform=ax.transAxes, color="#888780", fontsize=11) ax.set_xlabel("Time (s)", fontsize=9, color="#5F5E5A") ax.set_ylabel("Pitch", fontsize=9, color="#5F5E5A") ax.tick_params(labelsize=8, colors="#5F5E5A") for sp in ax.spines.values(): sp.set_visible(False) ax.spines["bottom"].set_visible(True) ax.spines["left"].set_visible(True) ax.spines["bottom"].set_color("#D3D1C7"); ax.spines["bottom"].set_linewidth(0.5) ax.spines["left"].set_color("#D3D1C7"); ax.spines["left"].set_linewidth(0.5) ax.grid(axis="y", alpha=0.12, linewidth=0.5, color="#888780") buf = io.BytesIO() fig.tight_layout(pad=0.6) fig.savefig(buf, format="png", facecolor="#F9F8F5") plt.close(fig) buf.seek(0) return Image.open(buf).convert("RGB") # ── MIDI player HTML ────────────────────────────────────────────────────────── def _midi_player_html(midi_path: str) -> str: """ Inline MIDI player using @tonejs/midi + Web Audio oscillators. Converts MIDI to audio in the browser. Loads the parser from cdn.jsdelivr.net (matches Hugging Face Spaces allow_list — unpkg is typically blocked). """ with open(midi_path, "rb") as f: b64 = base64.b64encode(f.read()).decode() return f"""