"""Per-game measurements taken from the event log, and their aggregation over many games.""" import statistics from collections import Counter from collections.abc import Iterable, Sequence from dataclasses import dataclass from dojo.board import Pos, quadrant_of from dojo.events import ( Event, ExamTaken, FightKind, GameOver, KumiteFought, Moved, SentToStart, Source, Teleported, TrackChanged, TurnStarted, ) from dojo.rules import RulesConfig # Table-time estimate (plan, resolved questions). TURN_SECONDS = 45 KUMITE_SECONDS = 30 @dataclass(frozen=True) class GameMetrics: """One game, flat so it can be written as a CSV row. Kumite counts exclude the Exam.""" scenario: str seed: int lineup: str winner_seat: int | None winner_bot: str | None rounds: int turns: int minutes: float kumite: int upsets: int expected_wins: int max_chain: int exam_attempts: int exam_passes: int turns_on_14: int longest_14_streak: int income_square: int income_card: int income_kumite: int income_exam: int square_income_home: int square_income_away: int lead_changes: int final_spread: int def measure( events: Sequence[Event], config: RulesConfig, corners: Sequence[int], bots: Sequence[str], *, scenario: str, seed: int, ) -> GameMetrics: """Replay ``events`` of one game. ``corners[seat]`` and ``bots[seat]`` describe each seat.""" players = len(corners) track = [config.start_position] * players position: list[Pos | None] = [None] * players blue_top = config.brown_position - 1 streak = [0] * players longest_streak = 0 turns_on_14 = 0 income = Counter[Source]() home = away = 0 kumite = upsets = expected = 0 chain = max_chain = 0 exam_attempts = exam_passes = 0 turns = 0 rounds = 0 winner: int | None = None leader: int | None = None lead_changes = 0 active = 0 for event in events: match event: case TurnStarted(player=player): turns += 1 active = player chain = 0 if track[player] == blue_top: turns_on_14 += 1 streak[player] += 1 longest_streak = max(longest_streak, streak[player]) else: streak[player] = 0 case Moved(player=player, path=path) if path: position[player] = path[-1] case Teleported(player=player, destination=destination): position[player] = destination case SentToStart(player=player, square=square): position[player] = square case KumiteFought(exam=False, attacker=attacker, kind=kind): kumite += 1 upsets += kind is FightKind.UPSET expected += kind is FightKind.EXPECTED if attacker == active: chain += 1 max_chain = max(max_chain, chain) case ExamTaken(passed=passed): exam_attempts += 1 exam_passes += passed case TrackChanged(player=player, before=before, after=after, source=source): delta = after - before income[source] += delta square = position[player] if source is Source.SQUARE and square is not None: if quadrant_of(square) == corners[player]: home += delta else: away += delta track[player] = after top = max(track) leaders = [p for p in range(players) if track[p] == top] if len(leaders) == 1 and leaders[0] != leader: lead_changes += leader is not None leader = leaders[0] case GameOver(winner=game_winner, rounds=game_rounds): winner, rounds = game_winner, game_rounds case _: pass fights = kumite + exam_attempts return GameMetrics( scenario=scenario, seed=seed, lineup=",".join(bots), winner_seat=winner, winner_bot=None if winner is None else bots[winner], rounds=rounds, turns=turns, minutes=round((turns * TURN_SECONDS + fights * KUMITE_SECONDS) / 60, 1), kumite=kumite, upsets=upsets, expected_wins=expected, max_chain=max_chain, exam_attempts=exam_attempts, exam_passes=exam_passes, turns_on_14=turns_on_14, longest_14_streak=longest_streak, income_square=income[Source.SQUARE], income_card=income[Source.CARD], income_kumite=income[Source.KUMITE], income_exam=income[Source.EXAM], square_income_home=home, square_income_away=away, lead_changes=lead_changes, final_spread=max(track) - min(track), ) @dataclass(frozen=True) class Spread: mean: float p10: float median: float p90: float @classmethod def of(cls, values: Sequence[float]) -> Spread: if len(values) < 2: value = float(values[0]) if values else 0.0 return cls(value, value, value, value) deciles = statistics.quantiles(values, n=10, method="inclusive") return cls(statistics.fmean(values), deciles[0], statistics.median(values), deciles[-1]) @dataclass(frozen=True) class Summary: """Aggregate of one scenario. Rates are per turn unless named otherwise.""" scenario: str games: int finished: float rounds: Spread minutes: Spread kumite_per_round: float upset_rate: float expected_rate: float max_chain: Spread exam_attempts_per_game: float exam_success: float turns_on_14_per_game: float longest_14_streak: int income_per_turn: dict[Source, float] square_income_home_per_turn: float square_income_away_per_turn: float lead_changes: Spread final_spread: Spread win_rate_by_seat: tuple[float, ...] win_rate_by_bot: dict[str, float] def summarize(scenario: str, games: Iterable[GameMetrics]) -> Summary: rows = list(games) if not rows: raise ValueError(f"{scenario}: no games to summarize") turns = sum(r.turns for r in rows) or 1 rounds = sum(r.rounds for r in rows) or 1 kumite = sum(r.kumite for r in rows) attempts = sum(r.exam_attempts for r in rows) seats = len(rows[0].lineup.split(",")) appearances = Counter[str]() wins = Counter[str]() for row in rows: appearances.update(row.lineup.split(",")) if row.winner_bot is not None: wins[row.winner_bot] += 1 seat_wins = Counter(r.winner_seat for r in rows if r.winner_seat is not None) incomes = { Source.SQUARE: sum(r.income_square for r in rows), Source.CARD: sum(r.income_card for r in rows), Source.KUMITE: sum(r.income_kumite for r in rows), Source.EXAM: sum(r.income_exam for r in rows), } return Summary( scenario=scenario, games=len(rows), finished=sum(r.winner_seat is not None for r in rows) / len(rows), rounds=Spread.of([r.rounds for r in rows]), minutes=Spread.of([r.minutes for r in rows]), kumite_per_round=kumite / rounds, upset_rate=sum(r.upsets for r in rows) / (kumite or 1), expected_rate=sum(r.expected_wins for r in rows) / (kumite or 1), max_chain=Spread.of([r.max_chain for r in rows]), exam_attempts_per_game=attempts / len(rows), exam_success=sum(r.exam_passes for r in rows) / (attempts or 1), turns_on_14_per_game=sum(r.turns_on_14 for r in rows) / len(rows), longest_14_streak=max(r.longest_14_streak for r in rows), income_per_turn={source: total / turns for source, total in incomes.items()}, square_income_home_per_turn=sum(r.square_income_home for r in rows) / turns, square_income_away_per_turn=sum(r.square_income_away for r in rows) / turns, lead_changes=Spread.of([r.lead_changes for r in rows]), final_spread=Spread.of([r.final_spread for r in rows]), win_rate_by_seat=tuple(seat_wins[s] / len(rows) for s in range(seats)), # Wins per seat occupied: an average bot scores 1 / players. win_rate_by_bot={bot: wins[bot] / count for bot, count in sorted(appearances.items())}, )