Commit ac4643d0 by PLN (Algolia)

feat(semantics): validated CLAP vibe-search + live /vibe endpoint (#82,#86)

Katana-first finding: per-one-shot CLAP genre/mood tags are unreliable (every
hit → boom-bap/euphoric — a 0.3s sound has no genre), but the audio EMBEDDINGS
are gold for RELATIVE similarity. 'warm dusty rhodes' → suns_keys gold-keys +
west-coast electric; 'jazzy upright bass' → no_sunshine/come_bass loops; a kick's
nearest neighbours are other kicks (0.96 cross-folder). So we ship similarity,
not fake absolute labels (Principle 1: trust the instrument).

- sample_semantics.py validated on real audio; semantics_embeds.npz = 1490×512-d.
- serve.py: lazy CLAP /vibe?q= (embed any phrase → rank) + /similar?name= (by
  audio-embed cosine). 503 if unbuilt, 400/404 on bad input; static serving
  untouched. Single-user LAN, torch loads once on first hit.
parent 7981dd07
......@@ -7,12 +7,92 @@ the printed LAN URL on a phone on the same wifi.
python3 serve.py --dir tide-table/punkachien --port 8731
"""
import argparse, os, re, socket
import argparse, json, os, re, socket, sys, threading
from functools import partial
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs
# ── optional CLAP vibe-search API (lazy: torch loads on first /vibe hit) ───────
# Serves /vibe?q=<phrase> and /similar?name=folder/stem over the cached sample
# embeddings (semantics_embeds.npz in --dir). Absent/unbuilt → 503 with a hint,
# never breaks static serving. See sample_semantics.py.
_VIBE = {}
_VIBE_LOCK = threading.Lock()
def _vibe_load():
"""Lazy-load CLAP + the cached embed table once. Returns state or raises."""
if "ready" in _VIBE:
return _VIBE
with _VIBE_LOCK:
if "ready" in _VIBE:
return _VIBE
sys.path.insert(0, os.getcwd()) # --dir (tide-table) for imports
import numpy as np
import sample_semantics as S
if not S.EMBEDS.exists():
raise FileNotFoundError(f"{S.EMBEDS.name} not built — run "
"`python3 sample_semantics.py embed`")
z = np.load(S.EMBEDS, allow_pickle=True)
_VIBE.update(S=S, np=np, M=z["embeds"].astype("float32"),
names=[str(x) for x in z["names"]],
idx={str(n): i for i, n in enumerate(z["names"])})
_VIBE["ready"] = True
return _VIBE
def _vibe_rank(qvec, n, drop=-1):
V = _VIBE
sims = V["M"] @ qvec
order = V["np"].argsort(-sims)
out = []
for i in order:
if i == drop:
continue
out.append({"name": V["names"][i], "sim": round(float(sims[i]), 4)})
if len(out) >= n:
break
return out
class RangeHandler(SimpleHTTPRequestHandler):
def do_GET(self):
if self.path.split("?", 1)[0] in ("/vibe", "/similar"):
return self._api()
return super().do_GET()
def _api(self):
u = urlparse(self.path)
q = parse_qs(u.query)
n = int(q.get("n", ["24"])[0])
try:
V = _vibe_load()
except Exception as e:
return self._json({"error": str(e)}, 503)
try:
if u.path == "/vibe":
phrase = (q.get("q", [""])[0]).strip()
if not phrase:
return self._json({"error": "empty query"}, 400)
qe = V["S"]._embed_texts([phrase]).cpu().numpy()[0]
return self._json({"query": phrase, "results": _vibe_rank(qe, n)})
else: # /similar
name = q.get("name", [""])[0]
if name not in V["idx"]:
return self._json({"error": f"unknown sample {name!r}"}, 404)
i = V["idx"][name]
return self._json({"of": name, "results": _vibe_rank(V["M"][i], n, drop=i)})
except Exception as e:
return self._json({"error": repr(e)}, 500)
def _json(self, obj, code=200):
body = json.dumps(obj).encode()
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def end_headers(self):
self.send_header("Accept-Ranges", "bytes")
self.send_header("Cache-Control", "no-cache")
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment