Spaces:
Running
Running
| """Prove the Space's embeddings match the corpus before anything ships. | |
| Run inside an environment that has laion_clap (audio-brief's venv locally, | |
| the Space's own env when run there), from a *staged* directory — i.e. the | |
| exact set of files that gets uploaded: | |
| python verify.py --stage DIR --parity AUDIO --parity-ref VECTOR.npy \ | |
| [--analyse AUDIO] [--out RESULT.json] | |
| `--parity` re-embeds a track whose vector is already in the corpus (the | |
| canonical path being audio-brief's clap_worker via WorkerBackend) and reports | |
| the cosine against the stored vector. Anything below 0.9999 means this code | |
| has drifted from the worker and every similarity the app shows would be | |
| quietly wrong. | |
| `--analyse` runs a full headless analysis so the market ranking, snippets and | |
| tags can be eyeballed without a browser. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import sys | |
| from pathlib import Path | |
| import numpy as np | |
| PARITY_FLOOR = 0.9999 | |
| def main() -> int: | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--stage", required=True) | |
| ap.add_argument("--parity", default=None) | |
| ap.add_argument("--parity-ref", default=None) | |
| ap.add_argument("--analyse", default=None) | |
| ap.add_argument("--out", default=None) | |
| args = ap.parse_args() | |
| sys.path.insert(0, str(Path(args.stage).resolve())) | |
| import analysis # noqa: E402 | |
| import clap_embed # noqa: E402 | |
| import livematch # noqa: E402 | |
| import tags as tagmod # noqa: E402 | |
| corpus = livematch.Corpus.load(args.stage) | |
| emb = clap_embed.Embedder() | |
| result: dict = {"corpus": {"week": corpus.week, "ckpt": corpus.ckpt, | |
| "sounds": len(corpus.sounds), | |
| "regions": len(corpus.regions)}} | |
| if args.parity: | |
| v = emb.embed_track(clap_embed.load_audio(args.parity)) | |
| ref = np.load(args.parity_ref).astype(np.float32).ravel() | |
| ref = ref / np.linalg.norm(ref) | |
| cos = float(np.dot(v, ref)) | |
| result["parity"] = {"file": args.parity, "cosine": round(cos, 7), | |
| "floor": PARITY_FLOOR, "passed": cos >= PARITY_FLOOR} | |
| vocab = tagmod.TagVocab(emb) | |
| if args.analyse: | |
| res = analysis.analyse_track(args.analyse, emb, corpus, vocab=vocab) | |
| result["zero_shot_check"] = tagmod.discrimination_check( | |
| vocab, res["track_vec"], | |
| expect_above=["afrobeats", "afro pop", "afro fusion"], | |
| expect_below=["techno", "country", "k-pop", "ambient"]) | |
| res.pop("track_vec", None) | |
| result["analysis"] = { | |
| "duration_s": res["duration_s"], "week": res["week"], | |
| "beat": res["beat"], "window_count": res["window_count"], | |
| "markets": [{"iso": i, "name": res["regions"][i]["name"], | |
| "best": res["regions"][i]["best"], | |
| "band": livematch.band(res["regions"][i]["best"]), | |
| "matched": res["regions"][i]["matched"], | |
| "pool": res["regions"][i]["pool"], | |
| "top": res["regions"][i]["top"][:3]} | |
| for i in res["shown_markets"]], | |
| "snippets": res["snippets"], | |
| "tags": { | |
| "by_group": {g: rows for g, rows | |
| in (res["tags"]["model"]["by_group"].items() | |
| if res["tags"]["model"] else [])}, | |
| "trend": res["tags"]["trend"], | |
| "copy_line": res["tags"]["copy_line"], | |
| }, | |
| } | |
| text = json.dumps(result, indent=2, default=str) | |
| if args.out: | |
| Path(args.out).write_text(text) | |
| print(text) | |
| parity = result.get("parity") | |
| return 0 if (parity is None or parity["passed"]) else 4 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |