diff --git a/.gitignore b/.gitignore index c28ca4b..04cb0fe 100644 --- a/.gitignore +++ b/.gitignore @@ -34,3 +34,6 @@ htmlcov/ .vscode/ *.swp *.swo + +# debug resources +debug/ diff --git a/texas_holdem/engine.py b/texas_holdem/engine.py index 0a409f9..560c661 100644 --- a/texas_holdem/engine.py +++ b/texas_holdem/engine.py @@ -8,6 +8,7 @@ from texas_holdem.cards import Deck from texas_holdem.evaluator import evaluate from texas_holdem.models import ( ActionRecord, + BlindLevel, HandSummary, Observation, PlayerAction, @@ -58,16 +59,44 @@ class TableGame: self.board = [] self.action_history: list[ActionRecord] = [] self.hand_summaries: list[HandSummary] = [] + # ``blind_history`` is an append-only log of every blind level change + # (including the initial one). Each entry's ``hand_number`` is the + # first hand that played under those stakes, which makes it trivial + # to reconstruct the schedule from the outside. + self.blind_history: list[BlindLevel] = [] @property def is_complete(self) -> bool: return len([player for player in self.players if player.stack > 0]) < 2 - def run_hand(self) -> HandSummary: + def run_hand( + self, + small_blind: int | None = None, + big_blind: int | None = None, + ) -> HandSummary: + """Play a single hand. + + ``small_blind`` / ``big_blind`` allow callers to bump the stakes + between hands without rebuilding the table. Either both must be + provided or both omitted (in which case the previously configured + blinds carry over). The resolved blind level is appended to + :attr:`blind_history` whenever it changes (including the very first + hand) so external observers can replay the schedule. + """ if self.is_complete: raise GameComplete("game is complete") + self._apply_blinds_for_hand(small_blind, big_blind) + self.hand_number += 1 + # Stamp the active blind level onto the upcoming summary so a hand + # remains self-describing even after the blinds change later on. + active_blinds = BlindLevel( + hand_number=self.hand_number, + small_blind=self.small_blind, + big_blind=self.big_blind, + ) + self._record_blind_level_if_new(active_blinds) started_at = time() self.board = [] self.action_history = [] @@ -116,6 +145,7 @@ class TableGame: board=list(self.board), actions=list(self.action_history), awards=awards, + blinds=active_blinds, showdown_hands=self._collect_showdown_hands(), started_at=started_at, finished_at=time(), @@ -123,14 +153,32 @@ class TableGame: self.hand_summaries.append(summary) return summary - def run_hands(self, max_hands: int, until_one_left: bool = False) -> list[HandSummary]: + def run_hands( + self, + max_hands: int, + until_one_left: bool = False, + small_blind: int | None = None, + big_blind: int | None = None, + ) -> list[HandSummary]: + """Play up to ``max_hands`` hands using a single blind configuration. + + Passing ``small_blind`` / ``big_blind`` bumps the stakes starting + with the first hand of this call; subsequent calls can raise them + again. Leaving them ``None`` keeps the current level unchanged. + """ if max_hands <= 0: raise ValueError("max_hands must be positive") summaries = [] for _ in range(max_hands): if self.is_complete: break - summaries.append(self.run_hand()) + # Only the first hand of the batch needs to apply the blind + # override; after that the engine reuses the stored values. + summaries.append( + self.run_hand(small_blind=small_blind, big_blind=big_blind) + ) + small_blind = None + big_blind = None if until_one_left and self.is_complete: break return summaries @@ -143,8 +191,18 @@ class TableGame: "button_seat": None if self.button_index is None else self.players[self.button_index].seat, + # ``small_blind`` / ``big_blind`` mirror the *current* level so + # legacy callers keep working. New consumers should prefer the + # structured ``blinds`` block which carries the full schedule. "small_blind": self.small_blind, "big_blind": self.big_blind, + "blinds": { + "current": { + "small_blind": self.small_blind, + "big_blind": self.big_blind, + }, + "history": [level.to_dict() for level in self.blind_history], + }, "starting_stack": self.starting_stack, "players": [player.public_dict() for player in self.players], # ``hands`` exposes every finished hand (each entry is the same @@ -153,6 +211,47 @@ class TableGame: "hands": [summary.to_dict() for summary in self.hand_summaries], } + def _apply_blinds_for_hand( + self, + small_blind: int | None, + big_blind: int | None, + ) -> None: + """Validate and apply optional per-hand blind overrides. + + Splitting this out keeps :meth:`run_hand` focused on the table flow + while letting us reuse the validation rules originally enforced by + ``__init__``. We require both values to be supplied together so the + configuration cannot drift into an inconsistent half-update. + """ + if small_blind is None and big_blind is None: + return + if small_blind is None or big_blind is None: + raise ValueError( + "small_blind and big_blind must be provided together" + ) + if small_blind <= 0 or big_blind <= 0 or small_blind > big_blind: + raise ValueError("blinds must satisfy 0 < small_blind <= big_blind") + self.small_blind = int(small_blind) + self.big_blind = int(big_blind) + + def _record_blind_level_if_new(self, level: BlindLevel) -> None: + """Append ``level`` to :attr:`blind_history` when it differs. + + Comparing against the latest entry (rather than blindly appending) + keeps the log compact: stretches of unchanged stakes only contribute + a single record. The very first hand always seeds an entry because + the history starts empty. + """ + if not self.blind_history: + self.blind_history.append(level) + return + latest = self.blind_history[-1] + if ( + latest.small_blind != level.small_blind + or latest.big_blind != level.big_blind + ): + self.blind_history.append(level) + def _advance_button(self) -> None: if self.button_index is None: self.button_index = self._next_index(0, lambda index: self.players[index].stack > 0) diff --git a/texas_holdem/models.py b/texas_holdem/models.py index 49bad67..aceaba8 100644 --- a/texas_holdem/models.py +++ b/texas_holdem/models.py @@ -130,6 +130,28 @@ class Observation: } +@dataclass(slots=True) +class BlindLevel: + """A snapshot of the blind configuration that took effect at a given hand. + + The structure is intentionally append-only: every time the blinds change + (or the very first hand seeds the initial values) we push a new + ``BlindLevel`` so callers can reconstruct how the stakes evolved over the + course of the game without losing any prior state. + """ + + hand_number: int + small_blind: int + big_blind: int + + def to_dict(self) -> dict[str, object]: + return { + "hand_number": self.hand_number, + "small_blind": self.small_blind, + "big_blind": self.big_blind, + } + + @dataclass(slots=True) class PotAward: amount: int @@ -152,6 +174,10 @@ class HandSummary: board: list[Card] actions: list[ActionRecord] awards: list[PotAward] + # ``blinds`` records the exact blind level used by this hand. Storing it + # on the summary (rather than only on the game) guarantees historical + # hands remain self-describing even after the blinds are raised later. + blinds: BlindLevel | None = None showdown_hands: dict[str, list[Card]] = field(default_factory=dict) started_at: float = field(default_factory=time) finished_at: float = field(default_factory=time) @@ -161,6 +187,7 @@ class HandSummary: "game_id": self.game_id, "hand_number": self.hand_number, "button_seat": self.button_seat, + "blinds": self.blinds.to_dict() if self.blinds else None, "board": [str(card) for card in self.board], "actions": [record.to_dict() for record in self.actions], "awards": [award.to_dict() for award in self.awards], diff --git a/texas_holdem/server.py b/texas_holdem/server.py index 03eca02..d6f1544 100644 --- a/texas_holdem/server.py +++ b/texas_holdem/server.py @@ -43,14 +43,28 @@ class PokerRequestHandler(BaseHTTPRequestHandler): body = self._read_json() count = int(body.get("count", 1)) until_one_left = bool(body.get("until_one_left", False)) - summaries = MANAGER.run_hands(path[1], count, until_one_left) + small_blind, big_blind = self._extract_blinds(body) + summaries = MANAGER.run_hands( + path[1], + count, + until_one_left, + small_blind=small_blind, + big_blind=big_blind, + ) self._json({"hands": summaries, "game": MANAGER.get_game(path[1]).to_dict()}) return if len(path) == 4 and path[0] == "games" and path[2] == "hands" and path[3] == "run": body = self._read_json() count = int(body.get("count", 1)) until_one_left = bool(body.get("until_one_left", False)) - summaries = MANAGER.run_hands(path[1], count, until_one_left) + small_blind, big_blind = self._extract_blinds(body) + summaries = MANAGER.run_hands( + path[1], + count, + until_one_left, + small_blind=small_blind, + big_blind=big_blind, + ) self._json({"hands": summaries, "game": MANAGER.get_game(path[1]).to_dict()}) return self._json({"error": "not found"}, HTTPStatus.NOT_FOUND) @@ -78,6 +92,25 @@ class PokerRequestHandler(BaseHTTPRequestHandler): raise ValueError("request body must be a JSON object") return payload + @staticmethod + def _extract_blinds(body: dict[str, Any]) -> tuple[int | None, int | None]: + """Parse optional blind overrides from a /hands POST body. + + Callers may omit both keys (keep current level), or supply both to + raise the blinds for the upcoming batch. Providing only one is + treated as a client error and surfaced via ``ValueError`` so the + handler can reply with 400. + """ + raw_small = body.get("small_blind") + raw_big = body.get("big_blind") + if raw_small is None and raw_big is None: + return None, None + if raw_small is None or raw_big is None: + raise ValueError( + "small_blind and big_blind must be provided together" + ) + return int(raw_small), int(raw_big) + def _json(self, payload: dict[str, Any], status: HTTPStatus = HTTPStatus.OK) -> None: body = json.dumps(payload, ensure_ascii=True).encode("utf-8") self.send_response(status) diff --git a/texas_holdem/service.py b/texas_holdem/service.py index e0704ad..4623161 100644 --- a/texas_holdem/service.py +++ b/texas_holdem/service.py @@ -62,10 +62,29 @@ class GameManager: with self._lock: return [game.to_dict() for game in self._games.values()] - def run_hands(self, game_id: str, count: int = 1, until_one_left: bool = False) -> list[dict[str, object]]: + def run_hands( + self, + game_id: str, + count: int = 1, + until_one_left: bool = False, + small_blind: int | None = None, + big_blind: int | None = None, + ) -> list[dict[str, object]]: + """Run ``count`` hands, optionally raising the blinds first. + + ``small_blind`` / ``big_blind`` are forwarded to the engine so the + blinds can change between batches. Leaving them as ``None`` keeps + the previously configured level, which preserves the original + no-argument behaviour. + """ game = self.get_game(game_id) with self._lock: return [ summary.to_dict() - for summary in game.run_hands(count, until_one_left=until_one_left) + for summary in game.run_hands( + count, + until_one_left=until_one_left, + small_blind=small_blind, + big_blind=big_blind, + ) ]