🎡 Implement mini game runner

codenames/duet/state.py

@@ -83,7 +83,7 @@ class DuetSideState(DuetSpymasterState):
     game_result: GameResult | None = None
 
     @classmethod
-    def from_board(cls, board: DuetBoard) -> DuetSideState:
+    def from_board(cls, board: DuetBoard) -> Self:
         score = Score.new(green=len(board.green_cards))
         return cls(board=board, score=score)
 

codenames/generic/runner.py

@@ -16,6 +16,9 @@ class TeamPlayers:
     spymaster: Spymaster
     operative: Operative
 
+    def __iter__(self):
+        return iter([self.spymaster, self.operative])
+
     def __post_init__(self):
         if self.spymaster.team != self.operative.team:
             raise ValueError("Spymaster and Operative must be on the same team")

codenames/mini/runner.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import logging
+
+from codenames.duet.board import DuetBoard
+from codenames.duet.score import GameResult
+from codenames.generic.exceptions import InvalidGuess
+from codenames.generic.move import GivenGuess
+from codenames.generic.player import Operative, PlayerRole, Spymaster
+from codenames.generic.runner import (
+    SEPARATOR,
+    ClueGivenSubscriber,
+    GuessGivenSubscriber,
+    TeamPlayers,
+)
+from codenames.mini.state import MiniGameState
+
+log = logging.getLogger(__name__)
+
+
+class MiniGameRunner:
+    def __init__(self, players: TeamPlayers, state: MiniGameState | None = None, board: DuetBoard | None = None):
+        self.players = players
+        if (not state and not board) or (state and board):
+            raise ValueError("Exactly one of state or board must be provided.")
+        self.state = state or MiniGameState.from_board(board=board)  # type: ignore[arg-type]
+        self.clue_given_subscribers: list[ClueGivenSubscriber] = []
+        self.guess_given_subscribers: list[GuessGivenSubscriber] = []
+
+    @property
+    def spymaster(self) -> Spymaster:
+        return self.players.spymaster
+
+    @property
+    def operative(self) -> Operative:
+        return self.players.operative
+
+    @property
+    def current_role(self) -> PlayerRole:
+        return self.state.current_player_role
+
+    def run_game(self) -> GameResult:
+        self._notify_game_starts()
+        result = self._run_rounds()
+        suffix = "win!" if result.win else "lose!"
+        log.info(f"{SEPARATOR}{result.reason}, you {suffix}")
+        return result
+
+    def _notify_game_starts(self):
+        self.spymaster.on_game_start(board=self.state.board)
+        self.operative.on_game_start(board=self.state.board.censored)
+
+    def _run_rounds(self) -> GameResult:
+        while not self.state.is_game_over:
+            self._run_turn()
+        return self.state.game_result  # type: ignore
+
+    def _run_turn(self):
+        if not self.state.is_sudden_death:
+            self._get_clue_from(spymaster=self.spymaster)
+        while not self.state.is_game_over and self.current_role == PlayerRole.OPERATIVE:
+            self._get_guess_from(operative=self.operative)
+
+    def _get_clue_from(self, spymaster: Spymaster):
+        clue = spymaster.give_clue(game_state=self.state.spymaster_state)
+        for subscriber in self.clue_given_subscribers:
+            subscriber(spymaster, clue)
+        given_clue = self.state.process_clue(clue=clue)
+        if given_clue is None:
+            return
+        for player in self.players:
+            player.on_clue_given(given_clue=given_clue)
+
+    def _get_guess_from(self, operative: Operative):
+        given_guess = self._get_guess_until_valid(operative)
+        if given_guess is None:
+            return
+        for player in self.players:
+            player.on_guess_given(given_guess=given_guess)
+
+    def _get_guess_until_valid(self, operative: Operative) -> GivenGuess | None:
+        while True:
+            guess = operative.guess(game_state=self.state.operative_state)
+            try:
+                given_guess = self.state.process_guess(guess=guess)
+            except InvalidGuess:
+                continue
+            for subscriber in self.guess_given_subscribers:
+                subscriber(operative, guess)
+            return given_guess

codenames/mini/state.py

@@ -1,68 +1,47 @@
 import logging
 from typing import Self
 
-from pydantic import BaseModel, model_validator
+from pydantic import model_validator
 
-from codenames.duet.board import DuetBoard
-from codenames.duet.score import (
-    MISTAKE_LIMIT_REACHED,
-    TIMER_TOKENS_DEPLETED,
-    GameResult,
-)
-from codenames.duet.state import DuetSideState
-from codenames.duet.types import DuetGivenClue, DuetGivenGuess
-from codenames.generic.move import Clue, Guess
+from codenames.duet.score import MISTAKE_LIMIT_REACHED, TIMER_TOKENS_DEPLETED
+from codenames.duet.state import DuetOperativeState, DuetSideState, DuetSpymasterState
+from codenames.duet.types import DuetGivenGuess
+from codenames.generic.move import Guess
 from codenames.generic.player import PlayerRole
 
 log = logging.getLogger(__name__)
 
 
-class MiniGameState(BaseModel):
-    side_state: DuetSideState
+class MiniGameState(DuetSideState):
     timer_tokens: int = 5
     allowed_mistakes: int = 4
 
-    @classmethod
-    def from_board(cls, board: DuetBoard) -> Self:
-        if not board.is_clean:
-            raise ValueError("Board must be clean.")
-        side_state = DuetSideState.from_board(board)
-        return cls(side_state=side_state)
+    @property
+    def spymaster_state(self) -> DuetSpymasterState:
+        return self.get_spymaster_state(None)
 
     @property
-    def game_result(self) -> GameResult | None:
-        # If the timer runs out, the game is lost
-        if self.timer_tokens < 0:
-            return TIMER_TOKENS_DEPLETED
-        if self.allowed_mistakes == 0:
-            return MISTAKE_LIMIT_REACHED
-        return self.side_state.game_result
+    def operative_state(self) -> DuetOperativeState:
+        return self.get_operative_state(None)
 
     @property
     def is_sudden_death(self) -> bool:
         return self.timer_tokens == 0
 
-    @property
-    def is_game_over(self) -> bool:
-        return self.game_result is not None
-
     @model_validator(mode="after")
     def validate_allowed_mistakes(self) -> Self:
         if self.allowed_mistakes > self.timer_tokens:
             raise ValueError("Allowed mistakes cannot be greater than timer tokens.")
         return self
 
-    def process_clue(self, clue: Clue) -> DuetGivenClue | None:
-        return self.side_state.process_clue(clue)
-
     def process_guess(self, guess: Guess) -> DuetGivenGuess | None:
-        given_guess = self.side_state.process_guess(guess)
+        given_guess = super().process_guess(guess)
         # If the guess is wrong or passed the turn, the timer is updated
         if not given_guess or not given_guess.correct:
             self._update_tokens(mistake=given_guess is not None)
             return given_guess
         # If we reached our target score, and we are not in "sudden death", we consume a timer token
-        if self.side_state.is_game_over and not self.is_sudden_death:
+        if self.is_game_over and not self.is_sudden_death:
             self._update_tokens(mistake=False)
         return given_guess
 
@@ -71,9 +50,14 @@ def _update_tokens(self, mistake: bool) -> None:
             self.timer_tokens -= 1
         if self.timer_tokens == 0:
             log.info("Timer tokens depleted! Entering sudden death")
-            self.side_state.current_player_role = PlayerRole.OPERATIVE
+            self.current_player_role = PlayerRole.OPERATIVE
+        elif self.timer_tokens < 0:
+            self.game_result = TIMER_TOKENS_DEPLETED
+            log.info("Timer tokens depleted (after sudden death)!")
         if not mistake:
             return
         self.allowed_mistakes -= 1
         if self.allowed_mistakes == 0:
             log.info("Mistake limit reached!")
+            self.game_result = MISTAKE_LIMIT_REACHED
+            return

tests/mini/conftest.py

@@ -0,0 +1,9 @@
+import pytest
+
+from codenames.duet.board import DuetBoard
+from tests.duet.utils import constants
+
+
+@pytest.fixture()
+def board_10() -> DuetBoard:
+    return constants.board_10()

tests/mini/test_game_runner.py

@@ -0,0 +1,74 @@
+from codenames.duet.board import DuetBoard
+from codenames.duet.score import (
+    MISTAKE_LIMIT_REACHED,
+    TARGET_REACHED,
+    TIMER_TOKENS_DEPLETED,
+)
+from codenames.duet.state import DuetSide
+from codenames.generic.move import PASS_GUESS, Clue
+from codenames.mini.runner import MiniGameRunner
+from codenames.mini.state import MiniGameState
+from tests.duet.utils.runner import build_players
+from tests.utils.players.dictated import DictatedTurn
+
+
+def test_happy_flow(board_10: DuetBoard):
+    turns_by_side = {
+        DuetSide.SIDE_A: [
+            DictatedTurn(clue=Clue(word="A", card_amount=3), guesses=[0, 1, 4]),  # Green, Green, Neutral
+            DictatedTurn(clue=Clue(word="B", card_amount=2), guesses=[2, PASS_GUESS]),  # Green, pass
+            DictatedTurn(clue=Clue(word="C", card_amount=2), guesses=[3]),  # Green
+        ]
+    }
+    players = build_players(turns_by_side=turns_by_side)
+    state = MiniGameState.from_board(board=board_10)
+    runner = MiniGameRunner(players=players.team_a, state=state)
+    runner.run_game()
+
+    assert runner.state.game_result == TARGET_REACHED
+    assert runner.state.timer_tokens == 2
+    assert runner.state.allowed_mistakes == 3
+    assert len(runner.state.given_clues) == 3
+    assert len(runner.state.given_guesses) == 5
+
+
+def test_timer_token_depleted(board_10: DuetBoard):
+    turns_by_side = {
+        DuetSide.SIDE_A: [
+            DictatedTurn(clue=Clue(word="A", card_amount=3), guesses=[4]),  # Neutral
+            DictatedTurn(clue=Clue(word="B", card_amount=2), guesses=[PASS_GUESS]),  # pass
+            DictatedTurn(clue=Clue(word="NONE", card_amount=2), guesses=[4, 1, 5]),  # Green, Neutral
+        ]
+    }
+    players = build_players(turns_by_side=turns_by_side)
+    state = MiniGameState.from_board(board=board_10)
+    state.timer_tokens = 2
+    runner = MiniGameRunner(players=players.team_a, state=state)
+    runner.run_game()
+
+    assert runner.state.game_result == TIMER_TOKENS_DEPLETED
+    assert runner.state.timer_tokens == -1
+    assert runner.state.allowed_mistakes == 2
+    assert len(runner.state.given_clues) == 2
+    assert len(runner.state.given_guesses) == 3
+
+
+def test_mistake_limit_reached(board_10: DuetBoard):
+    turns_by_side = {
+        DuetSide.SIDE_A: [
+            DictatedTurn(clue=Clue(word="A", card_amount=3), guesses=[4]),  # Neutral
+            DictatedTurn(clue=Clue(word="B", card_amount=2), guesses=[PASS_GUESS]),  # pass
+            DictatedTurn(clue=Clue(word="C", card_amount=2), guesses=[0, 1, 5]),  # Green, Green, Neutral
+        ]
+    }
+    players = build_players(turns_by_side=turns_by_side)
+    state = MiniGameState.from_board(board=board_10)
+    state.allowed_mistakes = 2
+    runner = MiniGameRunner(players=players.team_a, state=state)
+    runner.run_game()
+
+    assert runner.state.game_result == MISTAKE_LIMIT_REACHED
+    assert runner.state.timer_tokens == 2
+    assert runner.state.allowed_mistakes == 0
+    assert len(runner.state.given_clues) == 3
+    assert len(runner.state.given_guesses) == 4