lspcloud/prolific-preferences-personalized

1"""2Simulate N users going through the study and verify all 50 items get covered.3 4Usage:5    cd /dfs/scratch1/echoi1/prolific_preferences6    HF_TOKEN=hf_... python scripts/test_coverage.py7"""8import sys9import uuid10from pathlib import Path11 12sys.path.insert(0, str(Path(__file__).resolve().parent.parent))13 14from src.config import load_config15from src.data import (16    ensure_datasets,17    assign_items,18    release_reservation,19    record_completion,20    _load_pool,21    _pool_path,22    _data_dir,23)24 25 26def simulate_user(cfg: dict, complete: bool = True) -> dict:27    user_id = str(uuid.uuid4())28    items   = assign_items(cfg, user_id)29    if complete:30        release_reservation(user_id, cfg)31        record_completion(user_id, items, cfg)32    item_ids = [(item.get("pair_id") or item.get("item_id", ""), item.get("category", ""))33                for item in items]34    return {"user_id": user_id, "items": item_ids, "raw_items": items, "completed": complete}35 36 37def clear_local_state(cfg: dict):38    data_dir = _data_dir(cfg)39    for pattern in ["reservations*", "completion_cache*", "local_completions*",40                    "variant_counter*", "alternation_counter*"]:41        for f in data_dir.glob(pattern):42            f.unlink()43 44 45def analyse_coverage(results: list, cfg: dict) -> bool:46    cats      = [c["name"] for c in cfg["categories"]]47    all_passed = True48 49    print()50    print("=" * 60)51    print("COVERAGE ANALYSIS")52    print("=" * 60)53 54    for cat in cats:55        pool     = _load_pool(str(_pool_path(cat, cfg)))56        pool_ids = [p.get("pair_id") or p.get("item_id", "") for p in pool]57        covered  = {pid: 0 for pid in pool_ids}58 59        for result in results:60            if not result["completed"]:61                continue62            for item_id, item_cat in result["items"]:63                if item_cat == cat and item_id in covered:64                    covered[item_id] += 165 66        covered_once  = sum(1 for c in covered.values() if c >= 1)67        never_covered = [pid[:8] for pid, c in covered.items() if c == 0]68        over_covered  = [pid[:8] for pid, c in covered.items() if c > 1]69 70        print(f"\nCategory: {cat}")71        print(f"  Pool size:     {len(pool)}")72        print(f"  Covered >= 1x: {covered_once} / {len(pool)}")73        print(f"  Never covered: {len(never_covered)} {never_covered[:5]}")74        print(f"  Over-covered:  {len(over_covered)} {over_covered[:5]}")75 76        if covered_once == len(pool):77            print(f"  ✅ PASS — all {len(pool)} items covered")78        else:79            print(f"  ❌ FAIL — {len(pool) - covered_once} items not covered")80            all_passed = False81 82    print()83    print("=" * 60)84    print("OVERALL:", "✅ PASS" if all_passed else "❌ FAIL")85    print("=" * 60)86    return all_passed87 88 89def run_simulation(label: str, n_users: int, dropout_indices: list = None):90    dropout_indices = dropout_indices or []91    cfg = load_config()92    ensure_datasets(cfg)93    clear_local_state(cfg)94 95    print(f"\n── {label} ──")96    print(f"[TEST] {n_users} users, dropouts at: {dropout_indices}")97 98    results = []99    for i in range(n_users):100        complete = i not in dropout_indices101        result   = simulate_user(cfg, complete=complete)102        results.append(result)103        status = "✅ completed" if complete else "❌ abandoned"104        print(f"  User {i+1:2d} ({status}): "105              f"indices = {[r[0][:8] for r in result['items']]}")106 107    return analyse_coverage(results, cfg)108 109 110if __name__ == "__main__":111    # Test 1: perfect run — all 10 users complete, all 50 items covered exactly once112    run_simulation("Test 1: Perfect run", n_users=10)113 114    # Test 2: 2 dropouts — abandoned items should be picked up by extra users115    # The new sort_key means uncovered+reserved items are preferred over covered+unreserved116    # so items 35-39 (abandoned) get picked up by users 11-12 instead of re-covering 0-9117    run_simulation("Test 2: 2 dropouts, 12 users", n_users=12, dropout_indices=[7, 3])118 119    # Test 3: first user drops out — 11 users needed to cover all 50120    run_simulation("Test 3: First user drops out", n_users=11, dropout_indices=[0])