magicciv/tools/batch-quality-metrics.sh
Natalie 8c5612914d feat(@projects/@magic-civilization): add post-smoke clan evidence analysis
Co-Authored-By: Lilith Autocommit <noreply@atlilith.com>
2026-04-18 10:02:32 -07:00

95 lines
3.1 KiB
Bash
Executable file

#!/usr/bin/env bash
# batch-quality-metrics.sh — aggregate post-p0-25 quality metrics per batch.
#
# Usage:
# tools/batch-quality-metrics.sh <batch_dir>
# tools/batch-quality-metrics.sh apricot:<remote_batch_dir>
#
# Prints:
# median_winner_tier_peak=N
# median_peak_unit_tier=N
# median_wonder_count=N
# median_tier_peak_gap=N
# games_with_any_wonder=M/N
# ...
set -euo pipefail
TARGET="${1:?usage: tools/batch-quality-metrics.sh <batch_dir | apricot:/path>}"
read -r -d '' QUERY <<'EOF' || true
set -e
: "${DIR:?DIR must be set}"
python3 - "$DIR" <<'PY'
import json, pathlib, sys, statistics
root = pathlib.Path(sys.argv[1])
games = sorted(root.glob("game_*"))
winner_tier_peaks = []
peak_unit_tiers = []
wonder_counts = []
tier_peak_gaps = []
any_wonder_games = 0
for g in games:
stats_path = g / "turn_stats.jsonl"
if not stats_path.is_file() or stats_path.stat().st_size == 0:
continue
last = None
with open(stats_path) as f:
for line in f:
if line.strip():
last = line
if not last:
continue
try:
d = json.loads(last)
except Exception:
continue
ps = d.get("player_stats", {})
winner_idx = d.get("winner_index")
all_tp = []
for k, v in ps.items():
tp = v.get("tier_peak")
put = v.get("peak_unit_tier")
wc = v.get("wonder_count", 0)
if isinstance(tp, int) and tp >= 0:
all_tp.append(tp)
if isinstance(put, int) and put >= 0:
peak_unit_tiers.append(put)
if isinstance(wc, int):
wonder_counts.append(wc)
if wc >= 1:
any_wonder_games += 0 # game-level check below
# Winner tier_peak: winner_idx's tier_peak, only for decided games
if winner_idx is not None and winner_idx >= 0:
winner_stats = ps.get(str(winner_idx), {})
wtp = winner_stats.get("tier_peak")
if isinstance(wtp, int) and wtp >= 0:
winner_tier_peaks.append(wtp)
# tier_peak_gap = max - min across players
if len(all_tp) >= 2:
tier_peak_gaps.append(max(all_tp) - min(all_tp))
# game-level "any wonder"
if any(isinstance(v.get("wonder_count"), int) and v["wonder_count"] >= 1 for v in ps.values()):
any_wonder_games += 1
def fmt_median(xs):
return f"{statistics.median(xs):.1f}" if xs else "(no data)"
print(f"total_games_with_stats={sum(1 for g in games if (g/'turn_stats.jsonl').is_file())}")
print(f"decided_games={len(winner_tier_peaks)}")
print(f"median_winner_tier_peak={fmt_median(winner_tier_peaks)} (n={len(winner_tier_peaks)})")
print(f"median_peak_unit_tier={fmt_median(peak_unit_tiers)} (n={len(peak_unit_tiers)})")
print(f"median_tier_peak_gap={fmt_median(tier_peak_gaps)} (n={len(tier_peak_gaps)})")
print(f"median_wonder_count_per_player={fmt_median(wonder_counts)} (n={len(wonder_counts)})")
print(f"games_with_any_wonder={any_wonder_games}/{len(games)}")
PY
EOF
if [[ "$TARGET" == apricot:* ]]; then
REMOTE_PATH="${TARGET#apricot:}"
ssh apricot "DIR='${REMOTE_PATH}' bash -s" <<< "$QUERY"
else
DIR="$TARGET" bash -c "$QUERY"
fi