from __future__ import annotations from typing import List, Dict, Optional from pathlib import Path from shortdeck_arena.simulation import Simulation from shortdeck_arena.agent import HumanAgent from shortdeck_arena.game_stage import BlindConfig, GameStage, PlayerState import uuid class ArenaGame: def __init__(self, starting_stack: int = 1000, max_players: int = 6, small_blind: int = 1, big_blind: int = 2): self.agents = [] self.player_names: List[str] = [] self.starting_stack = starting_stack self.max_players = max_players self.sim: Optional[Simulation] = None # 筹码管理 self.stacks: List[int] = [] # 盲注配置 self.blind_config = BlindConfig(small_blind, big_blind, ante=0) # 游戏标识 self.game_id = str(uuid.uuid4()) def join_game(self, name: str) -> int: if len(self.player_names) >= self.max_players: raise ValueError("table full") pid = len(self.player_names) self.player_names.append(name) agent = HumanAgent(pid) self.agents.append(agent) self.stacks.append(self.starting_stack) self.sim = Simulation(self.agents, self.blind_config) return pid def info(self, player_id: Optional[int] = None) -> Dict: """获取游戏状态信息""" if not self.sim: return { "game_id": self.game_id, "players": self.player_names, "stacks": [], "dealer_index": 0, "current_turn": 0, "pot": 0, "stage": "preflop", "actions": {}, "player_cards": [], "board_cards": [], } # 更新栈大小 (扣除已投入底池的金额) updated_stacks = [] for i, base_stack in enumerate(self.stacks): pot_contribution = self.sim.pot[i] if i < len(self.sim.pot) else 0 updated_stacks.append(max(0, base_stack - pot_contribution)) for i in range(len(self.stacks)): if i < len(updated_stacks): self.stacks[i] = updated_stacks[i] player_cards = [] board_cards = [] # 获取玩家手牌 try: if player_id is not None and 0 <= player_id < len(self.agents): player_cards = [str(c) for c in self.sim.player_cards(player_id)] except Exception: player_cards = [] # 获取公共牌 (根据当前阶段) try: current_stage = self.sim.current_stage.value board_cards = [str(c) for c in self.sim.board_cards(current_stage)] except Exception: board_cards = [] # 获取可用动作信息 actions = {} if (player_id is not None and player_id == self.sim.current_turn and self.sim.current_stage != GameStage.FINISHED): call_amount = self.sim.get_call_amount(player_id) available_stack = updated_stacks[player_id] if player_id < len(updated_stacks) else 0 # 更新node_info以包含实际栈信息 node_info = self.sim.node_info() node_info["bet_max"] = available_stack actions = { "call_amount": call_amount, "bet_min": node_info["bet_min"], "bet_max": node_info["bet_max"], "can_check": call_amount == 0, "can_fold": True, "can_call": call_amount > 0 and call_amount <= available_stack, "can_bet": available_stack > call_amount, } return { "game_id": self.game_id, "players": self.player_names, "stacks": updated_stacks, "dealer_index": 0, # 简化:固定庄家位置, (优化轮询) "current_turn": self.sim.current_turn, "pot": self.sim.total_pot, "stage": self.sim.current_stage.value, "actions": actions, "player_cards": player_cards, "board_cards": board_cards, "player_states": [state.value for state in self.sim.player_states], } def apply_action(self, pid: int, action: str, amount: Optional[int] = None): if not self.sim: raise ValueError("no game") # 验证动作合法性 if pid != self.sim.current_turn: raise ValueError(f"not your turn, current turn: {self.sim.current_turn}") if self.sim.current_stage == GameStage.FINISHED: raise ValueError("game already finished") # 获取玩家可用筹码 pot_contribution = self.sim.pot[pid] if pid < len(self.sim.pot) else 0 available_stack = self.stacks[pid] - pot_contribution # 预处理动作和金额 action = action.lower() if action in ("bet", "raise") and amount is not None: # 限制下注金额不超过可用筹码 if amount > available_stack: amount = available_stack # 如果全下,可能需要标记为allin if amount == available_stack and available_stack > 0: self.sim.player_states[pid] = PlayerState.ALLIN elif action == "call": # call动作的金额验证 call_amount = self.sim.get_call_amount(pid) if call_amount > available_stack: # 不够跟注,自动allin amount = available_stack if available_stack > 0: self.sim.player_states[pid] = PlayerState.ALLIN try: self.sim.apply_action(pid, action, amount) except ValueError as e: raise ValueError(f"invalid action: {e}") self.sim.dump_data(Path.cwd() / "shortdeck_arena_history.jsonl") @property def current_turn(self) -> int: if not self.sim: return 0 return self.sim.current_turn @property def pot(self) -> int: if not self.sim: return 0 return self.sim.total_pot @property def history(self) -> List[Dict]: if not self.sim: return [] return self.sim.history