qianrui.mmmy
91 lines
3.1 KiB
Python
91 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
from random import Random
|
|
from threading import RLock
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from texas_holdem.agents import build_agent
|
|
from texas_holdem.engine import TableGame
|
|
|
|
|
|
class GameManager:
|
|
def __init__(self) -> None:
|
|
self._games: dict[str, TableGame] = {}
|
|
self._lock = RLock()
|
|
|
|
def create_game(self, payload: dict[str, Any]) -> TableGame:
|
|
players = payload.get("players")
|
|
if not isinstance(players, list):
|
|
raise ValueError("players must be a list")
|
|
if not 2 <= len(players) <= 12:
|
|
raise ValueError("players must contain 2-12 entries")
|
|
|
|
seed = payload.get("seed")
|
|
rng = Random(seed)
|
|
game_id = str(payload.get("game_id") or uuid4())
|
|
starting_stack = int(payload.get("starting_stack", 1000))
|
|
small_blind = int(payload.get("small_blind", 5))
|
|
big_blind = int(payload.get("big_blind", 10))
|
|
|
|
specs = []
|
|
for seat, raw_spec in enumerate(players):
|
|
if not isinstance(raw_spec, dict):
|
|
raise ValueError("each player must be an object")
|
|
player_id = str(raw_spec.get("id") or raw_spec.get("player_id") or f"p{seat + 1}")
|
|
name = str(raw_spec.get("name") or player_id)
|
|
agent = build_agent(raw_spec.get("agent", raw_spec), rng)
|
|
specs.append((player_id, name, agent))
|
|
|
|
game = TableGame(
|
|
game_id=game_id,
|
|
player_specs=specs,
|
|
starting_stack=starting_stack,
|
|
small_blind=small_blind,
|
|
big_blind=big_blind,
|
|
rng=rng,
|
|
)
|
|
with self._lock:
|
|
if game_id in self._games:
|
|
raise ValueError(f"game already exists: {game_id}")
|
|
self._games[game_id] = game
|
|
return game
|
|
|
|
def get_game(self, game_id: str) -> TableGame:
|
|
with self._lock:
|
|
try:
|
|
return self._games[game_id]
|
|
except KeyError as exc:
|
|
raise KeyError(f"game not found: {game_id}") from exc
|
|
|
|
def list_games(self) -> list[dict[str, object]]:
|
|
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,
|
|
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,
|
|
small_blind=small_blind,
|
|
big_blind=big_blind,
|
|
)
|
|
]
|