From aa47a0f7fbbd7f91a333822286e685b01c3f5d33 Mon Sep 17 00:00:00 2001 From: autocommit Date: Sun, 26 Apr 2026 19:42:50 -0700 Subject: [PATCH] =?UTF-8?q?feat(tools):=20=E2=9C=A8=20Add=20performance=20?= =?UTF-8?q?analysis=20tool=20to=20calculate=20time-to-peak=20units=20for?= =?UTF-8?q?=20benchmarking?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Lilith Autocommit --- tools/time-to-peak-unit.py | 142 +++++++++++++++++++++++++++++++++++++ 1 file changed, 142 insertions(+) create mode 100755 tools/time-to-peak-unit.py diff --git a/tools/time-to-peak-unit.py b/tools/time-to-peak-unit.py new file mode 100755 index 00000000..41850048 --- /dev/null +++ b/tools/time-to-peak-unit.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +"""time-to-peak-unit.py — when does each AI clan first hit its peak unit tier? + +For each game in a batch dir, walks `turn_stats.jsonl` to find the FIRST turn +each player's `peak_unit_tier` reached its end-of-game max. Useful for +answering: "what's the highest tier AI builds, and how long does it take?" + +Usage: + python3 tools/time-to-peak-unit.py [...] +""" +from __future__ import annotations + +import json +import statistics +import sys +from pathlib import Path + + +def scan_game(game_dir: Path) -> list[dict]: + """Return one record per player: {clan, max_peak_unit, turn_first_max, end_turn}.""" + ts = game_dir / "turn_stats.jsonl" + meta = game_dir / "meta.json" + if not ts.exists() or not meta.exists(): + return [] + try: + meta_data = json.loads(meta.read_text()) + player_clans = meta_data.get("player_clans") or {} + difficulty = meta_data.get("game_settings", {}).get("difficulty", "?") + except Exception: + return [] + + # First pass: find each player's end-of-game max peak_unit_tier. + max_per_player: dict[str, int] = {} + end_turn = 0 + try: + for line in ts.read_text().splitlines(): + if not line.strip(): + continue + entry = json.loads(line) + end_turn = entry.get("turn", end_turn) + for pid, stats in (entry.get("player_stats") or {}).items(): + pu = stats.get("peak_unit_tier", 0) + if pu > max_per_player.get(pid, 0): + max_per_player[pid] = pu + except Exception: + return [] + + # Second pass: find first turn each player reached their personal max. + first_max_turn: dict[str, int] = {} + try: + for line in ts.read_text().splitlines(): + if not line.strip(): + continue + entry = json.loads(line) + turn = entry.get("turn", 0) + for pid, stats in (entry.get("player_stats") or {}).items(): + pu = stats.get("peak_unit_tier", 0) + if pid not in first_max_turn and pu >= max_per_player.get(pid, 0) and pu > 0: + first_max_turn[pid] = turn + except Exception: + return [] + + out = [] + for pid, max_pu in max_per_player.items(): + if max_pu == 0: + continue + out.append({ + "game": game_dir.name, + "difficulty": difficulty, + "player": pid, + "clan": player_clans.get(pid, "?"), + "max_peak_unit": max_pu, + "turn_first_max": first_max_turn.get(pid, end_turn), + "end_turn": end_turn, + }) + return out + + +def main(argv: list[str]) -> int: + if len(argv) < 2: + print(f"Usage: {argv[0]} [...]", file=sys.stderr) + return 2 + all_records: list[dict] = [] + for arg in argv[1:]: + root = Path(arg) + if not root.is_dir(): + print(f"skip non-dir: {root}", file=sys.stderr) + continue + # Support both flat (game_*) and nested (pair/as_clan/game_*) batch layouts. + for game in sorted(root.glob("game_*")) + sorted(root.glob("**/game_*")): + if not game.is_dir(): + continue + all_records.extend(scan_game(game)) + + if not all_records: + print("No completed games found.") + return 1 + + # Per-game table + print(f'{"game":<48} {"diff":<8} {"clan":<13} {"max_unit":<9} {"turn_max":<9} {"end":<5}') + print("-" * 96) + for r in all_records: + print( + f'{r["game"][:46]:<48} {r["difficulty"]:<8} {r["clan"][:12]:<13} ' + f'{r["max_peak_unit"]:<9} {r["turn_first_max"]:<9} {r["end_turn"]:<5}' + ) + + # Aggregate by clan + by_clan: dict[str, list[dict]] = {} + for r in all_records: + by_clan.setdefault(r["clan"], []).append(r) + print() + print(f'{"clan":<13} {"n":<4} {"med_max_unit":<13} {"med_turn_max":<13} {"max_seen":<9}') + print("-" * 60) + for clan in sorted(by_clan): + rs = by_clan[clan] + if not rs: + continue + med_max = statistics.median([r["max_peak_unit"] for r in rs]) + med_turn = statistics.median([r["turn_first_max"] for r in rs]) + max_seen = max(r["max_peak_unit"] for r in rs) + print(f'{clan[:12]:<13} {len(rs):<4} {med_max:<13.1f} {med_turn:<13.0f} {max_seen:<9}') + + # Aggregate by difficulty + by_diff: dict[str, list[dict]] = {} + for r in all_records: + by_diff.setdefault(r["difficulty"], []).append(r) + print() + print(f'{"diff":<8} {"n":<4} {"med_max_unit":<13} {"med_turn_max":<13} {"max_seen":<9}') + print("-" * 55) + for diff in sorted(by_diff): + rs = by_diff[diff] + med_max = statistics.median([r["max_peak_unit"] for r in rs]) + med_turn = statistics.median([r["turn_first_max"] for r in rs]) + max_seen = max(r["max_peak_unit"] for r in rs) + print(f'{diff:<8} {len(rs):<4} {med_max:<13.1f} {med_turn:<13.0f} {max_seen:<9}') + + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv))