#!/usr/bin/env python3 """ Read-only report: how many CLIP-scored assets pass at various min_similarity cutoffs. Uses ``data/clip_asset_scores.json`` (same file as ``clip_align_kink_assets.py`` output). Example:: python scripts/clip_threshold_report.py python scripts/clip_threshold_report.py --db data/store.db With ``--db``, also reports how many ``Asset`` URLs are missing from the scores file (coverage gap). """ from __future__ import annotations import argparse import json import statistics import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) def main() -> int: ap = argparse.ArgumentParser(description="CLIP similarity distribution & coverage report") ap.add_argument("--scores", type=Path, default=ROOT / "data" / "clip_asset_scores.json") ap.add_argument("--db", type=Path, default=None, help="Optional SQLite DB to compare Asset URLs vs scores file") args = ap.parse_args() path = args.scores.resolve() if not path.is_file(): print(f"No scores file at {path}", file=sys.stderr) return 1 raw = json.loads(path.read_text(encoding="utf-8")) sims_accepted: list[float] = [] sims_all: list[float] = [] n_false = 0 for payload in raw.values(): if not isinstance(payload, dict): continue try: s = float(payload.get("similarity", 0.0)) except (TypeError, ValueError): continue sims_all.append(s) if payload.get("accepted") is True: sims_accepted.append(s) elif payload.get("accepted") is False: n_false += 1 print(f"Scores file: {path}") print(f" entries: {len(raw)} accepted_true: {len(sims_accepted)} accepted_false: {n_false}") if sims_all: print(f" similarity (all rows): min={min(sims_all):.4f} median={statistics.median(sims_all):.4f} max={max(sims_all):.4f}") thresholds = [0.18, 0.20, 0.22, 0.24, 0.26, 0.28, 0.30] print(" hypothetical passes if min_similarity were set to T (count of rows with sim >= T):") for t in thresholds: n = sum(1 for s in sims_all if s >= t) pct = 100.0 * n / len(sims_all) if sims_all else 0.0 print(f" T={t:.2f} → {n:6d} ({pct:5.1f}% of scored rows)") if args.db: from sqlmodel import Session, create_engine, select from models import Asset engine = create_engine(f"sqlite:///{args.db.resolve()}") with Session(engine) as session: urls = [r.asset_url for r in session.exec(select(Asset)).all()] in_file = sum(1 for u in urls if u in raw) missing = len(urls) - in_file print(f"\nDB {args.db}: {len(urls)} asset URLs; {in_file} in scores file; {missing} missing from file") if urls: print(f" coverage: {100.0 * in_file / len(urls):.1f}%") if missing and missing <= 25: for u in urls: if u not in raw: print(f" missing: {u}") return 0 if __name__ == "__main__": raise SystemExit(main())