diff --git a/README.md b/README.md index e6aca91..c7a7cf6 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,23 @@ uv run dojo board validate --quadrant other.txt --legend other.toml # check an `validate` exits with `0` when the board is fine (warnings allowed), `1` on rule violations, `2` when the file is malformed or missing. +## Simulation + +Plays many seeded games with bots and prints metrics per scenario. Every combination of the +listed values is a scenario; game `i` of each scenario uses seed `--seed + i`, and the lineup +rotates one seat per game. Bots: `random`, `balanced`, `farmer`, `aggressive`. + +```sh +uv run dojo sim # variants A and B, 4 players, 200 games each +uv run dojo sim --games 1000 --csv games.csv # more games; one CSV row per game +uv run dojo sim --variant B --players 3 4 6 8 --start 0 3 # player counts x start position +uv run dojo sim --kiai 2 1 # Kiai +2 vs +1 +uv run dojo sim --lineup balanced,random balanced # several lineups (comma-separated seats) +uv run dojo sim --seed 42 --jobs 1 # other dice; single process +``` + +Minutes are estimated as 45 s per turn plus 30 s per Kumite. + ## Development ```sh diff --git a/agents/plans/2026-09-25-dojo-board-and-sim.md b/agents/plans/2026-09-25-dojo-board-and-sim.md index a63aea5..e5a643d 100644 --- a/agents/plans/2026-09-25-dojo-board-and-sim.md +++ b/agents/plans/2026-09-25-dojo-board-and-sim.md @@ -101,16 +101,29 @@ Deliverables: ### Phase 5 — Simulator and metrics -- [ ] 5.1 Monte Carlo runner over a config matrix; seeded and reproducible. -- [ ] 5.2 Metrics: - - [ ] game length in rounds and estimated minutes (turn and fight duration as parameters); - - [ ] Kumite per round, upset rate, chain length; - - [ ] win rate by seat and by bot profile; - - [ ] exam attempts per game, exam success rate; - - [ ] rounds spent stuck on 14 (variant B deadlock check); - - [ ] stripe income split: squares vs cards vs Kumite; - - [ ] lead changes per game, final spread. -- [ ] 5.3 Report output (plain text / CSV). +- [x] 5.1 Monte Carlo runner over a config matrix; seeded and reproducible. `src/dojo/sim.py`: `Scenario` (variant × players × Kiai bonus × start position × lineup), `matrix`, `run(files, scenarios, games, seed, jobs)`. Game `i` uses seed `base + i` in every scenario (same dice across scenarios); the lineup rotates one seat per game; process pool, results independent of `jobs`. +- [x] 5.2 Metrics (`src/dojo/metrics.py`, computed from the event log; Kumite counts exclude the Exam): + - [x] game length in rounds and estimated minutes (45 s per turn + 30 s per Kumite); + - [x] Kumite per round, upset rate, higher-beats-lower rate, longest chain per game (Kumite attacked by the active player in one turn); + - [x] win rate by seat and by bot profile (per seat occupied); + - [x] exam attempts per game, exam success rate; + - [x] turns started on 14, longest streak (variant B deadlock check); + - [x] net stripe income per turn: squares vs cards vs Kumite vs Exam; square income in the player's own quadrant vs elsewhere (home farming); + - [x] lead changes per game (strict leader), final spread. +- [x] 5.3 Report output: `uv run dojo sim` prints a text summary per scenario; `--csv` writes one row per game (`src/dojo/report.py`). Commands in `README.md`. +- Baseline (1,000 games per variant, 4 players, lineup balanced/farmer/aggressive, seed 0), for Phase 6: + + | Metric | A | B | + |---|---|---| + | Rounds mean (p10–p90) | 18.4 (14–23) | 13.1 (9–17) | + | Minutes mean (p10–p90) | 66 (50–85) | 46 (33–62) | + | Kumite per round | 1.25 | 1.16 | + | Upsets / higher beats lower | 29% / 55% | 27% / 52% | + | Exam attempts per game, pass rate | 1.26, 79% | 1.22, 82% | + | Turns on 14 per game, longest streak | 1.26, 4 | 1.73, 7 | + | Income per turn: square (home / away), card, Kumite | +0.52 (0.38 / 0.14), 0.00, +0.04 | +0.54 (0.39 / 0.14), 0.00, +0.22 | + | Lead changes, final spread | 5.6, 9.3 | 3.5, 11.5 | + | Wins by seat | 23–27% | 23–27% | ### Phase 6 — Balance experiments @@ -147,14 +160,14 @@ Deliverables: ### Open - Q2 Board asymmetry: keep the quadrant as source of truth, or later switch to the full board as source of truth with a symmetry *report* (warnings) instead of generation? *(Deferred by the developer.)* -- Q4 Target net income per turn — derived from 1–2 h and ~15–20 turns per player; confirm with first simulations. +- Q4 Target net income per turn — derived from 1–2 h and ~15–20 turns per player; confirm with first simulations. Baseline measured net income ≈ 0.56 (A) / 0.75 (B) stripes per turn. ### Risks - Movement enumeration: self-avoiding walks of length up to 12 can explode in open rooms. Keep rooms small; prune by outcome (destination + ordered opponents crossed + Dojo entry). - Engine speed ~1,400 turns/s (random and heuristic bots, real board). If Phase 5 needs more: precompute neighbour lists in `Board`. - Simulation conclusions depend on bot quality; compare several profiles before trusting a result. -- Board draft 1 puts 4 Allenamento in each 3x3 room next to the Safe Room: possible "home farming" (see the Phase 2 review). Measure stripe income by region in the simulator. +- Board draft 1 puts 4 Allenamento in each 3x3 room next to the Safe Room: possible "home farming" (see the Phase 2 review). Measured (Phase 5 baseline): square income is +0.38/turn in the own quadrant vs +0.14/turn elsewhere; 6.4 decides whether that is a problem. ## External documentation diff --git a/src/dojo/cli.py b/src/dojo/cli.py index 1c3c805..029b671 100644 --- a/src/dojo/cli.py +++ b/src/dojo/cli.py @@ -1,9 +1,14 @@ import argparse +import os import sys +from itertools import groupby from pathlib import Path from dojo.board import QUADRANTS, Board, BoardError, load_board from dojo.grid import GridFormatError, render_grid +from dojo.metrics import summarize +from dojo.report import format_summary, write_games_csv +from dojo.sim import BOTS, VARIANTS, BoardFiles, matrix, run from dojo.validate import Report, validate DEFAULT_QUADRANT = Path("board/quadrant.txt") @@ -13,6 +18,8 @@ EXIT_OK = 0 EXIT_DESIGN_ERRORS = 1 EXIT_FORMAT_ERRORS = 2 +DEFAULT_LINEUP = "balanced,farmer,aggressive" + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="dojo") @@ -32,12 +39,36 @@ def main(argv: list[str] | None = None) -> int: "-o", "--output", type=Path, help="write to this file instead of stdout" ) + sim_parser = commands.add_parser( + "sim", help="simulate many games and print metrics per scenario" + ) + _add_board_arguments(sim_parser) + sim_parser.add_argument("--games", type=int, default=200, help="games per scenario") + sim_parser.add_argument("--seed", type=int, default=0, help="seed of the first game") + sim_parser.add_argument("--variant", nargs="+", choices=list(VARIANTS), default=["A", "B"]) + sim_parser.add_argument("--players", nargs="+", type=int, default=[4]) + sim_parser.add_argument("--kiai", nargs="+", type=int, default=[2], help="Kiai bonus") + sim_parser.add_argument("--start", nargs="+", type=int, default=[0], help="start position") + sim_parser.add_argument( + "--lineup", + nargs="+", + default=[DEFAULT_LINEUP], + help=f"comma-separated bots, repeated to fill the seats; bots: {', '.join(BOTS)}", + ) + sim_parser.add_argument( + "--jobs", type=int, default=os.cpu_count() or 1, help="worker processes" + ) + sim_parser.add_argument("--csv", type=Path, help="write one row per game to this file") + args = parser.parse_args(argv) legend = _resolve_legend(args.quadrant, args.legend) board = _load(args.quadrant, legend) if board is None: return EXIT_FORMAT_ERRORS + if args.command == "sim": + return _simulate(args, BoardFiles(args.quadrant, legend), parser) + if args.board_command == "validate": report = validate(board) print(_format_report(args.quadrant, legend, report)) @@ -51,6 +82,27 @@ def main(argv: list[str] | None = None) -> int: return EXIT_OK +def _simulate(args: argparse.Namespace, files: BoardFiles, parser: argparse.ArgumentParser) -> int: + if args.games < 1: + parser.error("--games must be at least 1") + try: + scenarios = matrix( + args.variant, + args.players, + args.kiai, + args.start, + [tuple(lineup.split(",")) for lineup in args.lineup], + ) + except ValueError as error: + parser.error(str(error)) + games = run(files, scenarios, args.games, seed=args.seed, jobs=max(1, args.jobs)) + for scenario, rows in groupby(games, key=lambda game: game.scenario): + print(format_summary(summarize(scenario, rows)), end="\n\n") + if args.csv is not None: + write_games_csv(args.csv, games) + return EXIT_OK + + def _add_board_arguments(parser: argparse.ArgumentParser) -> None: parser.add_argument("--quadrant", type=Path, default=DEFAULT_QUADRANT) parser.add_argument( diff --git a/src/dojo/metrics.py b/src/dojo/metrics.py new file mode 100644 index 0000000..8be23f6 --- /dev/null +++ b/src/dojo/metrics.py @@ -0,0 +1,253 @@ +"""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())}, + ) diff --git a/src/dojo/report.py b/src/dojo/report.py new file mode 100644 index 0000000..a9c9152 --- /dev/null +++ b/src/dojo/report.py @@ -0,0 +1,53 @@ +"""Simulation output: a plain-text summary per scenario and one CSV row per game.""" + +import csv +import dataclasses +from collections.abc import Iterable +from pathlib import Path + +from dojo.events import Source +from dojo.metrics import GameMetrics, Spread, Summary + + +def format_summary(summary: Summary) -> str: + income = summary.income_per_turn + seats = ", ".join(f"{seat} {rate:.0%}" for seat, rate in enumerate(summary.win_rate_by_seat)) + bots = ", ".join(f"{bot} {rate:.0%}" for bot, rate in summary.win_rate_by_bot.items()) + lines = [ + f"== {summary.scenario} ({summary.games} games, {summary.finished:.0%} finished)", + f"Rounds {_spread(summary.rounds)}", + f"Minutes {_spread(summary.minutes)}", + f"Kumite {summary.kumite_per_round:.2f}/round; upsets {summary.upset_rate:.0%}, " + f"higher beats lower {summary.expected_rate:.0%}", + f"Longest chain {_spread(summary.max_chain)} (Kumite in one turn, per game)", + f"Exam {summary.exam_attempts_per_game:.2f} attempts/game, " + f"{summary.exam_success:.0%} passed", + f"Stuck on 14 {summary.turns_on_14_per_game:.2f} turns/game, " + f"longest streak {summary.longest_14_streak} turns", + f"Income/turn square {income[Source.SQUARE]:+.2f} " + f"(home {summary.square_income_home_per_turn:+.2f}, " + f"away {summary.square_income_away_per_turn:+.2f}), " + f"card {income[Source.CARD]:+.2f}, kumite {income[Source.KUMITE]:+.2f}, " + f"exam {income[Source.EXAM]:+.2f}", + f"Lead changes {_spread(summary.lead_changes)}", + f"Final spread {_spread(summary.final_spread)}", + f"Wins by seat {seats}", + f"Wins by bot {bots}", + ] + return "\n".join(lines) + + +def write_games_csv(path: Path, games: Iterable[GameMetrics]) -> None: + fields = [field.name for field in dataclasses.fields(GameMetrics)] + with path.open("w", encoding="utf-8", newline="") as file: + writer = csv.DictWriter(file, fieldnames=fields) + writer.writeheader() + for game in games: + writer.writerow(dataclasses.asdict(game)) + + +def _spread(spread: Spread) -> str: + return ( + f"mean {spread.mean:.1f} p10 {spread.p10:.1f} median {spread.median:.1f} " + f"p90 {spread.p90:.1f}" + ) diff --git a/src/dojo/sim.py b/src/dojo/sim.py new file mode 100644 index 0000000..eccae57 --- /dev/null +++ b/src/dojo/sim.py @@ -0,0 +1,128 @@ +"""Monte Carlo runner: many seeded games over a matrix of rule configurations. + +Game ``i`` of every scenario uses seed ``base + i``, so scenarios are compared on the same +dice. The lineup rotates one seat per game, separating seat effects from bot effects. +Results do not depend on the number of worker processes. +""" + +import itertools +import random +from collections.abc import Sequence +from concurrent.futures import ProcessPoolExecutor +from dataclasses import dataclass, replace +from functools import cache +from pathlib import Path + +from dojo.board import Board, load_board +from dojo.bots import PROFILES, HeuristicAgent, RandomAgent +from dojo.engine import Agent, Game, RandomSource, seat_corners +from dojo.metrics import GameMetrics, measure +from dojo.rules import VARIANT_A, VARIANT_B, RulesConfig + +VARIANTS = {"A": VARIANT_A, "B": VARIANT_B} +RANDOM_BOT = "random" +BOTS = (RANDOM_BOT, *PROFILES) + + +@dataclass(frozen=True) +class Scenario: + variant: str + players: int + kiai_bonus: int + start_position: int + # Bot names, repeated in order to fill the seats. + lineup: tuple[str, ...] + + def __post_init__(self) -> None: + if self.variant not in VARIANTS: + raise ValueError(f"unknown variant {self.variant!r}: expected one of {list(VARIANTS)}") + unknown = [bot for bot in self.lineup if bot not in BOTS] + if unknown or not self.lineup: + raise ValueError(f"unknown bots {unknown}: expected some of {list(BOTS)}") + seat_corners(self.players) + + @property + def name(self) -> str: + return ( + f"{self.variant} p{self.players} kiai+{self.kiai_bonus} start{self.start_position} " + + ",".join(self.lineup) + ) + + @property + def config(self) -> RulesConfig: + return replace( + VARIANTS[self.variant], + kiai_bonus=self.kiai_bonus, + start_position=self.start_position, + ) + + def seating(self, game: int) -> tuple[str, ...]: + seats = [self.lineup[seat % len(self.lineup)] for seat in range(self.players)] + shift = game % self.players + return tuple(seats[shift:] + seats[:shift]) + + +def matrix( + variants: Sequence[str], + players: Sequence[int], + kiai_bonuses: Sequence[int], + start_positions: Sequence[int], + lineups: Sequence[tuple[str, ...]], +) -> list[Scenario]: + return [ + Scenario(*axes) + for axes in itertools.product(variants, players, kiai_bonuses, start_positions, lineups) + ] + + +@dataclass(frozen=True) +class BoardFiles: + quadrant: Path + legend: Path | None + + +def make_agent(bot: str, seed: int, seat: int) -> Agent: + if bot == RANDOM_BOT: + return RandomAgent(random.Random(f"bot:{seed}:{seat}")) + return HeuristicAgent(PROFILES[bot]) + + +def play(board: Board, scenario: Scenario, index: int, seed: int) -> GameMetrics: + bots = scenario.seating(index) + agents = [make_agent(bot, seed, seat) for seat, bot in enumerate(bots)] + game = Game(board, scenario.config, agents, RandomSource(seed)) + game.run() + return measure( + game.events, + scenario.config, + seat_corners(scenario.players), + bots, + scenario=scenario.name, + seed=seed, + ) + + +def run( + files: BoardFiles, + scenarios: Sequence[Scenario], + games: int, + seed: int = 0, + jobs: int = 1, +) -> list[GameMetrics]: + """Play ``games`` games of each scenario; results are ordered by scenario, then game.""" + tasks = [(s, i, seed + i) for s in scenarios for i in range(games)] + if jobs == 1: + board = _board(files) + return [play(board, *task) for task in tasks] + with ProcessPoolExecutor(jobs) as pool: + chunk = max(1, len(tasks) // (jobs * 8)) + return list(pool.map(_play_task, itertools.repeat(files), tasks, chunksize=chunk)) + + +def _play_task(files: BoardFiles, task: tuple[Scenario, int, int]) -> GameMetrics: + return play(_board(files), *task) + + +@cache +def _board(files: BoardFiles) -> Board: + return load_board(files.quadrant, files.legend) diff --git a/tests/test_cli.py b/tests/test_cli.py index d19a899..97223c4 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -56,3 +56,28 @@ def test_legend_next_to_the_quadrant_is_used_by_default( assert main(["board", "validate", "--quadrant", str(quadrant)]) == EXIT_FORMAT_ERRORS assert "legend links tunnel '9'" in capsys.readouterr().err + + +def test_sim_prints_a_summary_per_scenario_and_writes_games( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + games = tmp_path / "games.csv" + argv = ["sim", "--games", "2", "--variant", "A", "B", "--players", "3", "--jobs", "1"] + + assert main([*argv, "--csv", str(games)]) == EXIT_OK + + output = capsys.readouterr().out + assert "== A p3 kiai+2 start0 balanced,farmer,aggressive (2 games" in output + assert "== B p3 kiai+2 start0" in output + rows = games.read_text(encoding="utf-8").splitlines() + assert rows[0].startswith("scenario,seed,lineup,winner_seat") + assert len(rows) == 1 + 4 + + +@pytest.mark.parametrize("bad", [["--lineup", "balanced,cheater"], ["--games", "0"]]) +def test_sim_rejects_bad_arguments(bad: list[str], capsys: pytest.CaptureFixture[str]) -> None: + with pytest.raises(SystemExit) as exit_info: + main(["sim", *bad]) + + assert exit_info.value.code == 2 + assert "error:" in capsys.readouterr().err diff --git a/tests/test_metrics.py b/tests/test_metrics.py new file mode 100644 index 0000000..fba641c --- /dev/null +++ b/tests/test_metrics.py @@ -0,0 +1,108 @@ +from dataclasses import replace + +import pytest + +from dojo.board import Pos +from dojo.events import ( + Event, + ExamTaken, + FightKind, + GameOver, + KumiteFought, + Moved, + Source, + TrackChanged, + TurnStarted, +) +from dojo.metrics import KUMITE_SECONDS, TURN_SECONDS, GameMetrics, measure, summarize +from dojo.rules import VARIANT_A + +CORNERS = (0, 1) +BOTS = ("balanced", "random") +HOME_0 = Pos(0, 1) # quadrant 0 +HOME_1 = Pos(0, 14) # quadrant 1 + + +def fought(attacker: int, defender: int, kind: FightKind, *, exam: bool = False) -> KumiteFought: + return KumiteFought(attacker, defender, 0, 0, 4, 3, True, kind, exam, cards=()) + + +def run(events: list[Event]) -> GameMetrics: + return measure(events, VARIANT_A, CORNERS, BOTS, scenario="test", seed=3) + + +def test_measure_counts_fights_income_and_leads() -> None: + metrics = run( + [ + TurnStarted(1, 0, 7), + Moved(0, (HOME_0,)), + TrackChanged(0, 0, 1, Source.SQUARE), + TurnStarted(1, 1, 5), + Moved(1, (HOME_0,)), + TrackChanged(1, 0, 1, Source.SQUARE), + fought(1, 0, FightKind.EQUAL), + TrackChanged(1, 1, 4, Source.KUMITE), + TrackChanged(0, 1, 0, Source.KUMITE), + fought(1, 0, FightKind.UPSET), + TurnStarted(2, 0, 6), + fought(1, 0, FightKind.EXPECTED), # Player 1 defends: not part of a chain. + TrackChanged(0, 0, 5, Source.CARD), + fought(0, 1, FightKind.EQUAL, exam=True), + ExamTaken(0, 1, passed=True), + GameOver(0, 2), + ] + ) + + assert (metrics.kumite, metrics.upsets, metrics.expected_wins) == (3, 1, 1) + assert metrics.max_chain == 2 + assert (metrics.exam_attempts, metrics.exam_passes) == (1, 1) + assert (metrics.income_square, metrics.income_kumite, metrics.income_card) == (2, 2, 5) + assert (metrics.square_income_home, metrics.square_income_away) == (1, 1) + # Leader: 0 (1-0), tie (1-1) keeps it, 1 (0-4), 0 (5-4). + assert metrics.lead_changes == 2 + assert metrics.final_spread == 1 + assert (metrics.winner_seat, metrics.winner_bot, metrics.rounds) == (0, "balanced", 2) + assert metrics.minutes == pytest.approx(round((3 * TURN_SECONDS + 4 * KUMITE_SECONDS) / 60, 1)) + + +def test_measure_tracks_turns_started_on_14() -> None: + metrics = run( + [ + TrackChanged(0, 0, 14, Source.CARD), + TurnStarted(1, 0, 7), + TurnStarted(1, 1, 7), + TurnStarted(2, 0, 7), + TrackChanged(0, 14, 15, Source.KUMITE), + TurnStarted(2, 1, 7), + TurnStarted(3, 0, 7), + TrackChanged(0, 15, 14, Source.SQUARE), + TurnStarted(4, 0, 7), + GameOver(None, 4), + ] + ) + + assert metrics.turns_on_14 == 3 + assert metrics.longest_14_streak == 2 + assert metrics.winner_seat is None + + +def test_summarize_splits_wins_by_seat_and_bot() -> None: + base = run([GameOver(None, 10)]) + rows = [ + replace(base, winner_seat=0, winner_bot="balanced"), + replace(base, lineup="random,balanced", winner_seat=1, winner_bot="balanced"), + replace(base, lineup="random,balanced", winner_seat=0, winner_bot="random"), + base, + ] + + summary = summarize("test", rows) + + assert summary.finished == 0.75 + assert summary.win_rate_by_seat == (0.5, 0.25) + assert summary.win_rate_by_bot == {"balanced": 0.5, "random": 0.25} + assert summary.rounds.median == 10 + + +def test_summarize_needs_games() -> None: + with pytest.raises(ValueError, match="no games"): + summarize("test", []) diff --git a/tests/test_sim.py b/tests/test_sim.py new file mode 100644 index 0000000..5928cc9 --- /dev/null +++ b/tests/test_sim.py @@ -0,0 +1,58 @@ +from pathlib import Path + +import pytest + +from dojo.sim import BoardFiles, Scenario, matrix, run + +BOARD = Path(__file__).parent.parent / "board" +FILES = BoardFiles(BOARD / "quadrant.txt", BOARD / "legend.toml") +LINEUP = ("balanced", "random") + + +def test_matrix_crosses_every_axis() -> None: + scenarios = matrix(["A", "B"], [3, 4], [2, 1], [0], [LINEUP]) + + assert len(scenarios) == 8 + assert scenarios[0].name == "A p3 kiai+2 start0 balanced,random" + assert scenarios[-1].config.kiai_bonus == 1 + assert scenarios[-1].config.brown_requires_kumite + + +def test_lineup_fills_the_seats_and_rotates_each_game() -> None: + scenario = Scenario("A", 3, 2, 0, LINEUP) + + assert scenario.seating(0) == ("balanced", "random", "balanced") + assert scenario.seating(1) == ("random", "balanced", "balanced") + assert scenario.seating(3) == scenario.seating(0) + + +@pytest.mark.parametrize( + ("variant", "players", "lineup", "message"), + [ + ("C", 4, LINEUP, "unknown variant"), + ("A", 4, ("balanced", "cheater"), "unknown bots"), + ("A", 4, (), "unknown bots"), + ("A", 9, LINEUP, "supported range"), + ], +) +def test_invalid_scenarios_are_rejected( + variant: str, players: int, lineup: tuple[str, ...], message: str +) -> None: + with pytest.raises(ValueError, match=message): + Scenario(variant, players, 2, 0, lineup) + + +def test_runs_are_reproducible_and_independent_of_worker_count() -> None: + scenarios = matrix(["A", "B"], [3], [2], [0], [LINEUP]) + + serial = run(FILES, scenarios, games=2, seed=5, jobs=1) + parallel = run(FILES, scenarios, games=2, seed=5, jobs=2) + + assert serial == parallel + assert [(g.scenario, g.seed) for g in serial] == [ + (scenarios[0].name, 5), + (scenarios[0].name, 6), + (scenarios[1].name, 5), + (scenarios[1].name, 6), + ] + assert serial[1].lineup == "random,balanced,balanced"