Upload semantic.py with huggingface_hub

semantic.py

@@ -0,0 +1,180 @@
+"""Zero-shot semantic listening with CLAP.
+
+The DSP layer knows a band is 4 dB hot. It does not know the sound is a
+reese. CLAP scores the audio against a bank of sound-design descriptors, so
+the report can say "gritty distorted reese bass, over-compressed drums"
+instead of only quoting numbers.
+
+Loads in a background thread so the Space boots immediately, and fails soft:
+if the model never arrives, everything else still works.
+"""
+
+from __future__ import annotations
+
+import threading
+
+import numpy as np
+
+MODEL_ID = "laion/clap-htsat-unfused"
+CLAP_SR = 48_000
+
+# Grouped so the report can show one line per axis rather than a flat top-k.
+BANK: dict[str, list[str]] = {
+    "character": [
+        "a gritty distorted reese bass",
+        "a clean deep sine sub bass",
+        "a metallic screaming growl bass",
+        "an aggressive detuned saw lead",
+        "a warm analog pad",
+        "a plucky short synth stab",
+        "a bright supersaw chord stack",
+        "a wobbling filtered bass",
+        "a soft mellow electric piano",
+        "an acoustic guitar",
+        "a male vocal",
+        "a female vocal",
+    ],
+    "drums": [
+        "a punchy tight kick drum",
+        "a boomy undamped kick drum",
+        "a sharp cracking snare",
+        "a boxy resonant snare",
+        "crisp hi hats",
+        "a heavily compressed drum break",
+        "a loose live drum kit",
+    ],
+    "problem": [
+        "a muddy boomy cluttered mix",
+        "a harsh sibilant painful mix",
+        "an over-compressed lifeless mix",
+        "a thin tinny weak mix",
+        "a clipping distorted overloaded mix",
+        "a clean balanced professional mix",
+        "a hissy noisy recording",
+        "a phasey hollow comb-filtered sound",
+    ],
+    "space": [
+        "a dry close-miked sound with no reverb",
+        "a tight small room reverb",
+        "a huge cavernous hall reverb",
+        "a long washed-out ambient reverb tail",
+        "a slapback delay",
+    ],
+    "energy": [
+        "a quiet sparse intro section",
+        "a building tense riser",
+        "a full loud drop section",
+        "a calm breakdown section",
+    ],
+}
+
+_FLAT: list[tuple[str, str]] = [(g, t) for g, items in BANK.items() for t in items]
+
+
+def _features(raw, projection):
+    """Normalise CLAP's feature output across transformers versions.
+
+    4.x returns the projected tensor directly. 5.x returns a
+    BaseModelOutputWithPooling, so the projection has to be applied here —
+    which is exactly what 4.x did internally.
+    """
+    if hasattr(raw, "shape"):
+        return raw
+    pooled = getattr(raw, "pooler_output", None)
+    if pooled is None:
+        pooled = raw.last_hidden_state[:, 0]
+    return projection(pooled)
+
+
+class _Semantic:
+    def __init__(self) -> None:
+        self.ready = False
+        self.error: str | None = None
+        self._model = None
+        self._processor = None
+        self._text_emb = None
+        self._lock = threading.Lock()
+
+    def start(self) -> None:
+        threading.Thread(target=self._load, daemon=True).start()
+
+    def _load(self) -> None:
+        try:
+            import torch
+            from transformers import ClapModel, ClapProcessor
+
+            torch.set_num_threads(2)
+            model = ClapModel.from_pretrained(MODEL_ID)
+            model.eval()
+            processor = ClapProcessor.from_pretrained(MODEL_ID)
+
+            texts = [t for _, t in _FLAT]
+            with torch.no_grad():
+                inputs = processor(text=texts, return_tensors="pt", padding=True)
+                emb = _features(model.get_text_features(**inputs), model.text_projection)
+                emb = emb / emb.norm(dim=-1, keepdim=True)
+
+            self._model, self._processor, self._text_emb = model, processor, emb
+            self.ready = True
+        except Exception as exc:  # noqa: BLE001 - fail soft, the app still works
+            self.error = f"{type(exc).__name__}: {exc}"
+
+    def status(self) -> str:
+        if self.ready:
+            return "ready"
+        if self.error:
+            return f"unavailable ({self.error})"
+        return "warming up"
+
+    def describe(self, mono48: np.ndarray, top_k: int = 2) -> dict[str, list[tuple[str, float]]]:
+        """Score the clip against every descriptor, grouped by axis."""
+        if not self.ready or mono48.size < CLAP_SR // 2:
+            return {}
+
+        import torch
+
+        # CLAP was trained on 10 s windows; take the loudest one.
+        want = CLAP_SR * 10
+        if mono48.size > want:
+            hop = CLAP_SR
+            best_s, best_e = 0, -1.0
+            for s in range(0, mono48.size - want + 1, hop):
+                e = float(np.mean(mono48[s : s + want] ** 2))
+                if e > best_e:
+                    best_e, best_s = e, s
+            mono48 = mono48[best_s : best_s + want]
+
+        clip = mono48.astype(np.float32)
+        with self._lock, torch.no_grad():
+            # transformers 4.x takes `audios`, 5.x renamed it to `audio`.
+            try:
+                inputs = self._processor(audio=clip, sampling_rate=CLAP_SR,
+                                         return_tensors="pt")
+            except TypeError:
+                inputs = self._processor(audios=clip, sampling_rate=CLAP_SR,
+                                         return_tensors="pt")
+            audio_emb = _features(self._model.get_audio_features(**inputs),
+                                  self._model.audio_projection)
+            audio_emb = audio_emb / audio_emb.norm(dim=-1, keepdim=True)
+            sims = (audio_emb @ self._text_emb.T).squeeze(0).cpu().numpy()
+
+        grouped: dict[str, list[tuple[str, float]]] = {}
+        for group in BANK:
+            idx = [i for i, (g, _) in enumerate(_FLAT) if g == group]
+            local = sims[idx]
+            # Softmax within the group — cross-group absolute scores are not
+            # comparable, ranking inside a group is.
+            e = np.exp((local - local.max()) * 20.0)
+            probs = e / e.sum()
+            order = np.argsort(-probs)[:top_k]
+            grouped[group] = [(_FLAT[idx[o]][1], float(probs[o])) for o in order]
+        return grouped
+
+
+SEMANTIC = _Semantic()
+
+
+def tags_line(grouped: dict[str, list[tuple[str, float]]], min_conf: float = 0.30) -> str:
+    """Flatten the grouped scores into one readable sentence."""
+    picks = [items[0][0] for items in grouped.values() if items and items[0][1] >= min_conf]
+    return ", ".join(picks) if picks else ""