Spaces:
Running
Running
Antigravity
feat(squarefc): integrate Square FC into squares race pipeline with 80:20 trigger split
fa7a508 | #!/usr/bin/env python3 | |
| """ | |
| Square FC — Candidate Prober & Verification CLI | |
| Usage: | |
| python3 squarefc/cli_probe.py --count 50 --start-seed 1000 | |
| python3 squarefc/cli_probe.py --verify-seed 2023 | |
| python3 squarefc/cli_probe.py --probe 100 --save-run | |
| """ | |
| import argparse | |
| import datetime | |
| import json | |
| import subprocess | |
| import sys | |
| from pathlib import Path | |
| # Ensure root directory is in sys.path | |
| ROOT_DIR = Path(__file__).resolve().parent.parent | |
| if str(ROOT_DIR) not in sys.path: | |
| sys.path.insert(0, str(ROOT_DIR)) | |
| from squarefc.match_evaluator import probe_candidates, score_match, load_scoring_config | |
| def run_verify_seed(seed: int): | |
| """ | |
| Verifies full-match bit-for-bit determinism for a specific seed: | |
| Runs simulation twice headlessly, asserts identical frames and telemetry. | |
| """ | |
| js_prober = Path(__file__).parent / "js" / "probe_matches.js" | |
| # Run A | |
| cmd_a = ["node", str(js_prober), "--seeds", str(seed)] | |
| res_a = subprocess.run(cmd_a, capture_output=True, text=True, check=True) | |
| tel_a = json.loads(res_a.stdout)[0] | |
| # Run B | |
| cmd_b = ["node", str(js_prober), "--seeds", str(seed)] | |
| res_b = subprocess.run(cmd_b, capture_output=True, text=True, check=True) | |
| tel_b = json.loads(res_b.stdout)[0] | |
| str_a = json.dumps(tel_a, sort_keys=True) | |
| str_b = json.dumps(tel_b, sort_keys=True) | |
| is_identical = (str_a == str_b) | |
| eval_a = score_match(tel_a) | |
| print(f"\n{'='*55}") | |
| print(f" SQUARE FC — SEED REPRODUCIBILITY VERIFICATION") | |
| print(f"{'='*55}") | |
| print(f"SEED: {seed}") | |
| print(f"RUN A DURATION: {tel_a.get('durationSec')}s ({tel_a.get('framesCount')} frames)") | |
| print(f"RUN B DURATION: {tel_b.get('durationSec')}s ({tel_b.get('framesCount')} frames)") | |
| print(f"FINAL SCORE: Blue {tel_a['finalScore']['blue']} - {tel_a['finalScore']['red']} Red ({tel_a['winner'].upper()} WINS)") | |
| print(f"GOALS: {len(tel_a['goals'])}") | |
| print(f"NEAR MISSES: {tel_a.get('nearMissesCount')}") | |
| print(f"LEAD CHANGES: {tel_a.get('leadChanges')}") | |
| print(f"MATCH SCORE: {eval_a['total_score']} / 10.0") | |
| print(f"TELEMETRY STREAM: {'IDENTICAL' if is_identical else 'MISMATCH'}") | |
| print(f"DETERMINISM: {'PASS ✅' if is_identical else 'FAIL ❌'}") | |
| print(f"{'='*55}\n") | |
| if not is_identical: | |
| sys.exit(1) | |
| def run_probe_leaderboard(count: int, start_seed: int, min_d: float, max_d: float, save_run: bool = False): | |
| print(f"\n[Square FC] Probing {count} candidates starting at seed {start_seed}...") | |
| top_cand, leaderboard = probe_candidates( | |
| count=count, | |
| start_seed=start_seed, | |
| min_duration=min_d, | |
| max_duration=max_d, | |
| ) | |
| print(f"[Square FC] Evaluation complete. Qualified {len(leaderboard)}/{count} matches within [{min_d}s - {max_d}s].\n") | |
| print(f"{'RANK':<5} {'SEED':<8} {'SCORE':<7} {'DUR':<7} {'RESULT':<12} {'HOOK':<6} {'TENS':<6} {'COMP':<6} {'ACT':<6} {'PAC':<6} {'SIGNATURE':<20}") | |
| print("-" * 95) | |
| for i, c in enumerate(leaderboard[:15]): # Show top 15 | |
| sub = c["sub_scores"] | |
| score_str = f"B:{c['final_score']['blue']} R:{c['final_score']['red']}" | |
| print(f"{i+1:<5} {c['seed']:<8} {c['total_score']:<7.2f} {c['duration_sec']:<7.1f} {score_str:<12} " | |
| f"{sub['hook']:<6.1f} {sub['tension']:<6.1f} {sub['competition']:<6.1f} {sub['action']:<6.1f} {sub['pacing']:<6.1f} " | |
| f"{c['match_signature']:<20}") | |
| if top_cand: | |
| print(f"\n🏆 TOP CANDIDATE SELECTED: Seed {top_cand['seed']} with Score {top_cand['total_score']}/10.0") | |
| print(f" Match: Blue {top_cand['final_score']['blue']} - {top_cand['final_score']['red']} Red in {top_cand['duration_sec']}s") | |
| print(f" Sub-scores: {top_cand['sub_scores']}") | |
| print(f" Metrics: {top_cand['metrics']}\n") | |
| if save_run: | |
| runs_dir = Path(__file__).parent.parent / "out" / "runs" | |
| timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H%M%S") | |
| run_folder = runs_dir / timestamp | |
| run_folder.mkdir(parents=True, exist_ok=True) | |
| with open(run_folder / "leaderboard.json", "w", encoding="utf-8") as f: | |
| # Strip heavy telemetry from leaderboard summary | |
| clean_board = [{k: v for k, v in item.items() if k != "telemetry"} for item in leaderboard] | |
| json.dump(clean_board, f, indent=2) | |
| if top_cand: | |
| with open(run_folder / "winner.json", "w", encoding="utf-8") as f: | |
| json.dump(top_cand, f, indent=2) | |
| print(f"[Square FC] Full candidate leaderboard saved to: {run_folder}") | |
| def main(): | |
| parser = argparse.ArgumentParser(description="Square FC Candidate Prober & Verification CLI") | |
| parser.add_argument("--count", type=int, default=50, help="Number of candidate seeds to probe") | |
| parser.add_argument("--start-seed", type=int, default=1000, help="Starting integer seed") | |
| parser.add_argument("--min-duration", type=float, default=22.0, help="Min match duration in seconds") | |
| parser.add_argument("--max-duration", type=float, default=48.0, help="Max match duration in seconds") | |
| parser.add_argument("--verify-seed", type=int, default=None, help="Verify determinism for a specific seed") | |
| parser.add_argument("--save-run", action="store_true", help="Persist candidate leaderboard and winner to runs/") | |
| args = parser.parse_args() | |
| if args.verify_seed is not None: | |
| run_verify_seed(args.verify_seed) | |
| else: | |
| run_probe_leaderboard(args.count, args.start_seed, args.min_duration, args.max_duration, args.save_run) | |
| if __name__ == "__main__": | |
| main() | |