chessai.core.game
1import abc 2import argparse 3import datetime 4import copy 5import logging 6import math 7import os 8import random 9import typing 10 11import edq.util.serial 12 13import chessai.core.action 14import chessai.core.agentinfo 15import chessai.core.board 16import chessai.core.gameparser 17import chessai.core.isolation.level 18import chessai.core.ui 19import chessai.util.alias 20 21DEFAULT_MAX_MOVES: int = -1 22DEFAULT_AGENT_START_TIMEOUT: float = 0.0 23DEFAULT_AGENT_END_TIMEOUT: float = 0.0 24DEFAULT_AGENT_ACTION_TIMEOUT: float = 0.0 25 26DEFAULT_AGENT: str = chessai.util.alias.AGENT_RANDOM.short 27 28class GameInfo(edq.util.serial.DictConverter): 29 """ 30 A simple container that holds common information about a game. 31 """ 32 33 def __init__(self, 34 agent_infos: dict[chessai.core.types.Color, chessai.core.agentinfo.AgentInfo], 35 start_fen: str | None = None, 36 isolation_level: chessai.core.isolation.level.Level = chessai.core.isolation.level.Level.NONE, 37 max_moves: int = DEFAULT_MAX_MOVES, 38 agent_start_timeout: float = DEFAULT_AGENT_START_TIMEOUT, 39 agent_end_timeout: float = DEFAULT_AGENT_END_TIMEOUT, 40 agent_action_timeout: float = DEFAULT_AGENT_ACTION_TIMEOUT, 41 seed: int | None = None, 42 event: str | None = None, 43 site: str | None = None, 44 date: str | None = None, 45 game_round: str | None = None, 46 white_player: str | None = None, 47 black_player: str | None = None, 48 extra_info: dict[str, typing.Any] | None = None, 49 ) -> None: 50 if (seed is None): 51 seed = random.randint(0, 2**64) 52 53 self.seed: int = seed 54 """ The random seed for this game's RNG. """ 55 56 self.agent_infos: dict[chessai.core.types.Color, chessai.core.agentinfo.AgentInfo] = agent_infos 57 """ The required information for creating the agents for this game. """ 58 59 if (len(self.agent_infos) == 0): 60 raise ValueError("No agents provided.") 61 62 if (start_fen is None): 63 start_fen = chessai.core.gamestate.DEFAULT_FEN 64 65 self.start_fen: str = start_fen 66 """ The starting fen for this game. """ 67 68 self.isolation_level: chessai.core.isolation.level.Level = isolation_level 69 """ The isolation level to use for this game. """ 70 71 self.max_moves: int = max_moves 72 """ 73 The total number of moves (between all agents) allowed for this game. 74 If -1, unlimited moves are allowed. 75 """ 76 77 self.agent_start_timeout: float = agent_start_timeout 78 """ 79 The maximum number of seconds an agent is allowed when starting a game. 80 If <= 0, unlimited time is allowed. 81 """ 82 83 self.agent_end_timeout: float = agent_end_timeout 84 """ 85 The maximum number of seconds an agent is allowed when ending a game. 86 If <= 0, unlimited time is allowed. 87 """ 88 89 self.agent_action_timeout: float = agent_action_timeout 90 """ 91 The maximum number of seconds an agent is allowed when getting an action. 92 If <= 0, unlimited time is allowed. 93 """ 94 95 if (event is None): 96 event = '?' 97 98 self.event: str = event 99 """ The tournament or match event name. """ 100 101 if (site is None): 102 site = '?' 103 104 self.site: str = site 105 """ The location of the event. """ 106 107 if (date is None): 108 date = '????.??.??' 109 110 self.date: str = date 111 """ The starting date of the game. """ 112 113 if (game_round is None): 114 game_round = '?' 115 116 self.game_round: str = game_round 117 """ The playing round ordinal. """ 118 119 if (white_player is None): 120 white_player = '?' 121 122 self.white_player: str = white_player 123 """ The name of the white player. """ 124 125 if (black_player is None): 126 black_player = '?' 127 128 self.black_player: str = black_player 129 """ The name of the black player. """ 130 131 if (extra_info is None): 132 extra_info = {} 133 134 self.extra_info: dict[str, typing.Any] = extra_info 135 """ Any additional arguments passed to the game. """ 136 137class GameResult(edq.util.serial.DictConverter): 138 """ The result of running a game. """ 139 140 def __init__(self, 141 game_id: int, 142 game_info: GameInfo, 143 termination_reason: chessai.core.types.TerminationReason | None = None, 144 score: float = 0, 145 end_fen: str | None = None, 146 game_timeout: bool = False, 147 timeout_agent_teams: list[chessai.core.types.Color] | None = None, 148 crash_agent_teams: list[chessai.core.types.Color] | None = None, 149 winning_agent_teams: list[chessai.core.types.Color] | None = None, 150 start_time: edq.util.time.Timestamp | None = None, 151 end_time: edq.util.time.Timestamp | None = None, 152 history: list[chessai.core.agentaction.AgentActionRecord] | None = None, 153 agent_complete_records: dict[chessai.core.types.Color, chessai.core.agentaction.AgentActionRecord] | None = None, 154 **kwargs: typing.Any, 155 ) -> None: 156 self.game_id: int = game_id 157 """ The ID of the game result. """ 158 159 self.game_info: GameInfo = game_info 160 """ The core information about this game. """ 161 162 self.termination_reason: chessai.core.types.TerminationReason | None = termination_reason 163 """ The reason the game ended. """ 164 165 self.score: float = score 166 """ The score of the game from white's perspective. """ 167 168 self.end_fen = end_fen 169 """ The fen the board is in at the end of the game. """ 170 171 if (start_time is None): 172 start_time = edq.util.time.Timestamp.now() 173 174 self.start_time: edq.util.time.Timestamp = start_time 175 """ The time the game started at. """ 176 177 self.end_time: edq.util.time.Timestamp | None = end_time 178 """ The time the game ended at. """ 179 180 if (history is None): 181 history = [] 182 183 self.history: list[chessai.core.agentaction.AgentActionRecord] = history 184 """ The history of actions taken by each agent in this game. """ 185 186 if (agent_complete_records is None): 187 agent_complete_records = {} 188 189 self.agent_complete_records: dict[chessai.core.types.Color, chessai.core.agentaction.AgentActionRecord] = agent_complete_records 190 """ 191 The record recieved from an agent when the game finishes. 192 For agents that learn, this may include information that the agent learned this game. 193 """ 194 195 self.game_timeout: bool = game_timeout 196 """ Indicates that the game has timed out (reached the maximum number of moves). """ 197 198 if (timeout_agent_teams is None): 199 timeout_agent_teams = [] 200 201 self.timeout_agent_teams: list[chessai.core.types.Color] = timeout_agent_teams 202 """ The list of agents that timed out in this game. """ 203 204 if (crash_agent_teams is None): 205 crash_agent_teams = [] 206 207 self.crash_agent_teams: list[chessai.core.types.Color] = crash_agent_teams 208 """ The list of agents that crashed in this game. """ 209 210 if (winning_agent_teams is None): 211 winning_agent_teams = [] 212 213 self.winning_agent_teams: list[chessai.core.types.Color] = winning_agent_teams 214 """ 215 The list of agents that won this game. 216 An empty list indicates a tie. 217 218 Games may interpret this value in different ways. 219 """ 220 221 def get_duration_secs(self) -> float: 222 """ 223 Get the game's duration in seconds. 224 Will return positive infinity if the game has no end time 225 (it is still going or crashed (in very rare cases)). 226 """ 227 228 if (self.end_time is None): 229 return math.inf 230 231 return self.end_time.sub(self.start_time).to_secs() 232 233class Game(abc.ABC): 234 """ 235 A game that can be run in chessai. 236 Games combine the rules, layouts, and agents to run. 237 """ 238 239 def __init__(self, 240 game_info: GameInfo, 241 save_path: str | None = None, 242 is_replay: bool = False, 243 initial_actions: list[chessai.core.action.Action] | None = None, 244 action_history: list[chessai.core.action.Action] | None = None, 245 expected_result: chessai.core.gameparser.PGNResult = chessai.core.gameparser.PGNResult.UNKNOWN, 246 ) -> None: 247 self.game_info: GameInfo = game_info 248 """ The core information about this game. """ 249 250 self._save_path: str | None = save_path 251 """ Where to save the results of this game. """ 252 253 self._is_replay: bool = is_replay 254 """ 255 Indicates that this game is being loaded from a replay. 256 Some behavior, like saving the result, will be modified. 257 """ 258 259 if (initial_actions is None): 260 initial_actions = [] 261 262 self.initial_actions: list[chessai.core.action.Action] = initial_actions 263 """ The moves to be played at the beginning of the game. """ 264 265 if (action_history is None): 266 action_history = [] 267 268 self.action_history: list[chessai.core.action.Action] = action_history 269 """ The actions taken throughout the game. """ 270 271 self.expected_result: chessai.core.gameparser.PGNResult = expected_result 272 """ The expected result from the PGN: '1-0', '0-1', '1/2-1/2', or '*'. """ 273 274 @classmethod 275 def from_pgn(cls, 276 pgn: str, 277 state_class: typing.Type[chessai.core.gamestate.GameState], 278 game_info: GameInfo, 279 save_path: str | None = None, 280 **kwargs: typing.Any) -> typing.Optional['Game']: 281 """ 282 Create a Game from a PGN. 283 284 The game will be initialized with metadata from the PGN headers and will store the moves for replay. 285 286 Returns the game and whether the PGN parsing was successful. 287 """ 288 289 parsed_pgn = chessai.core.gameparser.parse_pgn(pgn, state_class) 290 if (parsed_pgn is None): 291 return None 292 293 # Extract starting FEN from optional headers or use default. 294 start_fen = parsed_pgn.get_starting_fen() 295 if (start_fen is not None): 296 game_info.start_fen = start_fen 297 298 # Extract header information into the game info. 299 game_info.event = parsed_pgn.headers.event 300 game_info.site = parsed_pgn.headers.site 301 game_info.date = parsed_pgn.headers.date 302 game_info.game_round = parsed_pgn.headers.round 303 game_info.white_player = parsed_pgn.headers.white 304 game_info.black_player = parsed_pgn.headers.black 305 game_info.extra_info.update(parsed_pgn.optional_headers) 306 307 return cls( 308 game_info = game_info, 309 save_path = save_path, 310 is_replay = True, 311 initial_actions = parsed_pgn.initial_actions, 312 expected_result = parsed_pgn.result, 313 **kwargs 314 ) 315 316 def to_pgn(self, 317 final_state: chessai.core.gamestate.GameState, 318 termination_reason: chessai.core.types.TerminationReason, 319 ) -> str: 320 """ Returns the PGN representation of the Game. """ 321 322 if (termination_reason in [chessai.core.types.TerminationReason.CHECKMATE, 323 chessai.core.types.TerminationReason.FORFEIT]): 324 winners = final_state.get_winners() 325 if (chessai.core.types.Color.WHITE in winners): 326 result = chessai.core.gameparser.PGNResult.WHITE_WIN 327 elif (chessai.core.types.Color.BLACK in winners): 328 result = chessai.core.gameparser.PGNResult.BLACK_WIN 329 else: 330 result = chessai.core.gameparser.PGNResult.TIE 331 elif (termination_reason in [chessai.core.types.TerminationReason.STALEMATE, 332 chessai.core.types.TerminationReason.INSUFFICIENT_MATERIAL, 333 chessai.core.types.TerminationReason.ACCEPTED_DRAW_PROPOSAL]): 334 result = chessai.core.gameparser.PGNResult.TIE 335 elif (termination_reason == chessai.core.types.TerminationReason.IN_PROGRESS): 336 result = chessai.core.gameparser.PGNResult.IN_PROGRESS 337 else: 338 result = chessai.core.gameparser.PGNResult.UNKNOWN 339 340 headers = chessai.core.gameparser.StandardHeaders( 341 event = self.game_info.event, 342 site = self.game_info.site, 343 date = self.game_info.date, 344 game_round = self.game_info.game_round, 345 white = self.game_info.white_player, 346 black = self.game_info.black_player, 347 result = result) 348 349 return chessai.core.gameparser.to_pgn(headers, self.game_info.extra_info, type(final_state), 350 self.game_info.start_fen, self.action_history) 351 352 def process_args(self, args: argparse.Namespace) -> None: 353 """ Process any special arguments from the command-line. """ 354 355 @abc.abstractmethod 356 def get_initial_state(self, 357 rng: random.Random, 358 fen: str | None = None) -> chessai.core.gamestate.GameState: 359 """ Create the initial state for this game. """ 360 361 def process_turn(self, 362 state: chessai.core.gamestate.GameState, 363 action_record: chessai.core.agentaction.AgentActionRecord, 364 result: GameResult, 365 rng: random.Random) -> chessai.core.gamestate.GameState: 366 """ 367 Process the given agent action and return an updated game state. 368 The returned game state may be a copy or modified version of the passed in game state. 369 """ 370 371 # The agent has timed out. 372 if (action_record.timeout): 373 result.timeout_agent_teams.append(action_record.player) 374 result.termination_reason = chessai.core.types.TerminationReason.AGENT_TIMEOUT 375 state.process_agent_timeout(action_record.player) 376 return state 377 378 # The agent has crashed. 379 if (action_record.crashed): 380 result.crash_agent_teams.append(action_record.player) 381 result.termination_reason = chessai.core.types.TerminationReason.AGENT_CRASHED 382 state.process_agent_crash(action_record.player) 383 return state 384 385 action = action_record.get_action() 386 if (action not in state.get_legal_actions()): 387 raise ValueError(f"Illegal action for agent {action_record.player}: '{action.uci()}' of type '{type(action)}'.") 388 389 # Add the action to the action history. 390 self.action_history.append(action) 391 392 self._call_state_process_turn_full(state, action, rng) 393 394 return state 395 396 def _call_state_process_turn_full(self, 397 state: chessai.core.gamestate.GameState, 398 action: chessai.core.action.Action, 399 rng: random.Random, 400 **kwargs: typing.Any) -> None: 401 """ Call on the game state to process a full turn. """ 402 403 state.process_turn_full(action, rng, **kwargs) 404 405 def check_end(self, state: chessai.core.gamestate.GameState) -> bool: 406 """ 407 Check to see if the game is over. 408 Return True if the game is now over, False otherwise. 409 410 By default, this will check chessai.core.gamestate.GameState.game_over 411 and chessai.core.gamestate.GameState.is_game_over(), 412 but child games can override for more complex functionality. 413 """ 414 415 return (state.game_over or state.is_game_over()) 416 417 def game_complete(self, state: chessai.core.gamestate.GameState, result: GameResult) -> None: 418 """ 419 Make any last adjustments to the game result after the game is over. 420 """ 421 422 def run(self, ui: chessai.core.ui.UI) -> GameResult: 423 """ 424 The main "game loop" for all games. 425 """ 426 427 logging.debug("Starting a game with seed: %d.", self.game_info.seed) 428 429 # Create a new random number generator just for this game. 430 rng = random.Random(self.game_info.seed) 431 432 # Keep track of what happens during this game. 433 game_id = rng.randint(0, 2**64) 434 result = GameResult(game_id, self.game_info) 435 436 # Initialize the agent isolator. 437 isolator = self.game_info.isolation_level.get_isolator() 438 isolator.init_agents(self.game_info.agent_infos) 439 440 # Create the initial game state (and force it's seed). 441 state = self.get_initial_state(rng, self.game_info.start_fen) 442 state.seed = game_id 443 state.game_start() 444 445 # board_highlights: list[chessai.core.board.Highlight] = [] 446 447 # Notify agents about the start of the game. 448 records = isolator.game_start(rng, state, self.game_info.agent_start_timeout) 449 for record in records.values(): 450 if (record.timeout): 451 result.timeout_agent_teams.append(record.player) 452 state.process_agent_timeout(record.player) 453 result.termination_reason = chessai.core.types.TerminationReason.AGENT_TIMEOUT 454 elif (record.crashed): 455 result.crash_agent_teams.append(record.player) 456 state.process_agent_crash(record.player) 457 result.termination_reason = chessai.core.types.TerminationReason.AGENT_CRASHED 458 else: 459 continue 460 # board_highlights += record.get_board_highlights() 461 462 state.agents_game_start(records) 463 464 # Start the UI. 465 ui.game_start(state) 466 467 while (not self.check_end(state)): 468 logging.trace("Turn %d, agent %s.", state.fullmove_number, state.turn) # type: ignore[attr-defined] # pylint: disable=no-member 469 470 if (len(self.initial_actions) > 0): 471 # Get the action from the pre-loaded actions. 472 action = self.initial_actions.pop(0) 473 agent_action = chessai.core.agentaction.AgentAction(action = action) 474 duration = edq.util.time.Duration(0) 475 476 action_record = chessai.core.agentaction.AgentActionRecord(state.turn, agent_action, duration) 477 else: 478 # Get the next action from the agent. 479 action_record = isolator.get_action(state, self.game_info.agent_action_timeout) 480 481 # Execute the next action and update the state. 482 state = self.process_turn(state, action_record, result, rng) 483 484 # Update the UI. 485 ui.update(state) 486 487 # Update the game result and move history. 488 result.history.append(action_record) 489 490 # Check for game ending conditions. 491 if (self.check_end(state)): 492 break 493 494 # Check if this game has ran for the maximum number of moves. 495 if ((self.game_info.max_moves > 0) and (len(result.history) >= self.game_info.max_moves)): 496 state.process_game_timeout() 497 result.game_timeout = True 498 result.termination_reason = chessai.core.types.TerminationReason.GAME_TIMEOUT 499 break 500 501 # Mark the end time of the game. 502 result.end_time = edq.util.time.Timestamp.now() 503 504 # Notify the state about the end of the game. 505 (winners, score) = state.game_complete() 506 result.winning_agent_teams += winners 507 result.score = score 508 509 if (result.termination_reason is None): 510 result.termination_reason = state.get_termination_reason() 511 512 result.end_fen = state.get_fen() 513 514 # Notify agents about the end of this game. 515 result.agent_complete_records = isolator.game_complete(state, self.game_info.agent_end_timeout) 516 517 # All the game to make final updates to the result. 518 self.game_complete(state, result) 519 520 # Update the UI. 521 ui.game_complete(state, result.termination_reason) 522 523 # Cleanup 524 isolator.close() 525 ui.close() 526 527 if ((not self._is_replay) and (self._save_path is not None)): 528 logging.info("Saving results to '%s'.", self._save_path) 529 530 pgn = self.to_pgn(state, result.termination_reason) 531 edq.util.dirent.write_file(self._save_path, pgn) 532 533 return result 534 535def set_cli_args(parser: argparse.ArgumentParser, default_board: str | None = None) -> argparse.ArgumentParser: 536 """ 537 Set common CLI arguments. 538 This is a sibling to init_from_args(), as the arguments set here can be interpreted there. 539 """ 540 541 parser.add_argument('--board', dest = 'board', 542 action = 'store', type = str, default = default_board, 543 help = ('Play on this board (default: %(default)s).' 544 + ' This may be a FEN of the board.' 545 + ' It may also be the full path to a board, or just a filename.' 546 + ' If just a filename, than the `chessai/resources/boards` directory will be checked (using a ".fen" extension.')) 547 548 parser.add_argument('--num-games', dest = 'num_games', 549 action = 'store', type = int, default = 1, 550 help = 'The number of games to play (default: %(default)s).') 551 552 parser.add_argument('--seed', dest = 'seed', 553 action = 'store', type = int, default = None, 554 help = 'The random seed for the game (will be randomly generated if not set).') 555 556 parser.add_argument('--max-moves', dest = 'max_moves', 557 action = 'store', type = int, default = DEFAULT_MAX_MOVES, 558 help = 'The maximum number of moves (total for all agents) allowed in this game (-1 for unlimited) (default: %(default)s).') 559 560 parser.add_argument('--agent-start-timeout', dest = 'agent_start_timeout', 561 action = 'store', type = float, default = DEFAULT_AGENT_START_TIMEOUT, 562 help = ('The maximum number of seconds each agent is allowed when starting a game (<= 0 for unlimited time) (default: %(default)s).' 563 + ' Note that the "none" isolation level cannot enforce timeouts.')) 564 565 parser.add_argument('--agent-end-timeout', dest = 'agent_end_timeout', 566 action = 'store', type = float, default = DEFAULT_AGENT_END_TIMEOUT, 567 help = ('The maximum number of seconds each agent is allowed when ending a game (<= 0 for unlimited time) (default: %(default)s).' 568 + ' Note that the "none" isolation level cannot enforce timeouts.')) 569 570 parser.add_argument('--agent-action-timeout', dest = 'agent_action_timeout', 571 action = 'store', type = float, default = DEFAULT_AGENT_ACTION_TIMEOUT, 572 help = ('The maximum number of seconds each agent is allowed when getting an action (<= 0 for unlimited time) (default: %(default)s).' 573 + ' Note that the "none" isolation level cannot enforce timeouts.')) 574 575 parser.add_argument('--isolation', dest = 'isolation_level', metavar = 'LEVEL', 576 action = 'store', type = str, default = chessai.core.isolation.level.Level.NONE.value, 577 choices = chessai.core.isolation.level.LEVELS, 578 help = ('Set the agent isolation level for this game (default: %(default)s).' 579 + ' Choose one of:' 580 + ' `none` -- Do not make any attempt to isolate the agent code from the game (fastest and least secure),' 581 + ' `process` -- Run the agent code in a separate process' 582 + ' (offers some protection, but still vulnerable to disk or execution exploits),' 583 + ' `tcp` -- Open TCP listeners to communicate with agents (most secure, requires additional work to set up agents).')) 584 585 parser.add_argument('--agent-arg', dest = 'raw_agent_args', metavar = 'ARG', 586 action = 'append', type = str, default = [], 587 help = ('Specify arguments directly to agents (may be used multiple times).' 588 + ' The value for this argument must be formatted as "player::key=value",' 589 + ' for example to set `foo = 9` for white and `bar = a` for black, we can use:' 590 + ' `--agent-arg white::foo=9 --agent-arg black::bar=a`.')) 591 592 parser.add_argument('--save-path', dest = 'save_path', 593 action = 'store', type = str, default = None, 594 help = ('If specified, write the result of this game to the specified location.' 595 + ' This file can be replayed with `--replay-path`.')) 596 597 parser.add_argument('--replay-path', dest = 'replay_path', 598 action = 'store', type = str, default = None, 599 help = ('If specified, replay the game whose result was saved at the specified path with `--save-path`.' 600 + ' This may be a PGN of the game.' 601 + ' It may also be the full path to a game, or just a filename.' 602 + ' If just a filename, than the `chessai/resources/games` directory will be checked (using a ".pgn" extension.')) 603 604 return parser 605 606def init_from_args( 607 args: argparse.Namespace, 608 game_class: typing.Type[Game], 609 state_class: typing.Type[chessai.core.gamestate.GameState], 610 base_agent_infos: dict[chessai.core.types.Color, chessai.core.agentinfo.AgentInfo] | None = None, 611 **kwargs: typing.Any, 612 ) -> argparse.Namespace: 613 """ 614 Take in args from a parser that was passed to set_cli_args(), 615 and initialize the proper components. 616 This will create a number of games (and related resources) 617 based on `--num-games` + `--num-training`. 618 Each of these resources will be placed in their respective list at 619 `args._fens`, `args._agent_infos`, or `args._games`. 620 """ 621 622 if (base_agent_infos is None): 623 base_agent_infos = {} 624 625 if (args.num_games <= 0): 626 raise ValueError(f"At least one game must be played (--num-games), {args.num_games} was specified.") 627 628 # Establish an RNG to generate seeds for each game using the given seed. 629 seed = args.seed 630 if (seed is None): 631 seed = random.randint(0, 2**64) 632 633 logging.debug("Using source seed for games: %d.", seed) 634 rng = random.Random(seed) 635 636 agents = [chessai.core.types.Color.WHITE, chessai.core.types.Color.BLACK] 637 agent_infos = _parse_agent_infos(agents, args.raw_agent_args, base_agent_infos) 638 639 base_save_path = args.save_path 640 641 all_fens: list[str] = [] 642 all_agent_infos = [] 643 all_games = [] 644 645 for i in range(args.num_games): 646 game_seed = rng.randint(0, 2**64) 647 648 all_agent_infos.append(copy.deepcopy(agent_infos)) 649 all_fens.append(args.board) 650 651 white_agent_info = all_agent_infos[-1].get(chessai.core.types.Color.WHITE, None) 652 if (white_agent_info is not None): 653 white_reference = white_agent_info.name 654 if (isinstance(white_reference, chessai.util.reflection.Reference)): 655 white_player = str(white_reference) 656 else: 657 white_player = white_reference 658 else: 659 white_player = '?' 660 661 black_agent_info = all_agent_infos[-1].get(chessai.core.types.Color.BLACK, None) 662 if (black_agent_info is not None): 663 black_reference = black_agent_info.name 664 if (isinstance(black_reference, chessai.util.reflection.Reference)): 665 black_player = str(black_reference) 666 else: 667 black_player = black_reference 668 else: 669 black_player = '?' 670 671 game_info = GameInfo( 672 all_agent_infos[-1], 673 start_fen = all_fens[-1], 674 isolation_level = chessai.core.isolation.level.Level(args.isolation_level), 675 max_moves = args.max_moves, 676 agent_start_timeout = args.agent_start_timeout, 677 agent_end_timeout = args.agent_end_timeout, 678 agent_action_timeout = args.agent_action_timeout, 679 seed = game_seed, 680 event = 'Casual Chessai Game', 681 site = 'https://github.com/ucsc-cse-240/chessai', 682 date = datetime.date.today().isoformat(), 683 game_round = str(i), 684 white_player = white_player, 685 black_player = black_player, 686 ) 687 688 # Suffix the save path if there is more than one game. 689 save_path = base_save_path 690 if ((save_path is not None) and (args.num_games > 1)): 691 parts = os.path.splitext(save_path) 692 save_path = f"{parts[0]}_{i:03d}{parts[1]}" 693 694 if (args.replay_path is None): 695 game_args = { 696 'game_info': game_info, 697 'save_path': save_path, 698 } 699 700 game = game_class(**game_args) 701 else: 702 raw_game = game_class.from_pgn(args.replay_path, state_class, game_info, save_path) 703 if (raw_game is None): 704 raise ValueError(f"Failed to initialize game number {i} from the PGN: '{args.replay_path}'.") 705 706 game = raw_game 707 708 game.process_args(args) 709 710 all_games.append(game) 711 712 setattr(args, '_fens', all_fens) 713 setattr(args, '_agent_infos', all_agent_infos) 714 setattr(args, '_games', all_games) 715 716 return args 717 718def _parse_agent_infos( 719 agent_teams: list[chessai.core.types.Color], 720 raw_args: list[str], 721 base_agent_infos: dict[chessai.core.types.Color, chessai.core.agentinfo.AgentInfo], 722 ) -> dict[chessai.core.types.Color, chessai.core.agentinfo.AgentInfo]: 723 # Initialize with random agents. 724 agent_info = {agent_team: chessai.core.agentinfo.AgentInfo(name = DEFAULT_AGENT) for agent_team in sorted(agent_teams)} 725 726 # Take any args from the base args. 727 for (team, base_agent_info) in base_agent_infos.items(): 728 if (team in agent_info): 729 agent_info[team].update(base_agent_info) 730 731 # Update with CLI args. 732 for raw_arg in raw_args: 733 raw_arg = raw_arg.strip() 734 if (len(raw_arg) == 0): 735 continue 736 737 parts = raw_arg.split('::', 1) 738 if (len(parts) != 2): 739 raise ValueError(f"Improperly formatted CLI agent argument: '{raw_arg}'.") 740 741 raw_team = parts[0].strip().lower() 742 team_color: chessai.core.types.Color | None = None 743 if (raw_team in ('white', 'w')): 744 team_color = chessai.core.types.Color.WHITE 745 elif (raw_team in ('black', 'b')): 746 team_color = chessai.core.types.Color.BLACK 747 else: 748 raise ValueError(f"CLI agent argument has an unknown agent team: {parts[0]}.") 749 750 raw_pair = parts[1] 751 752 parts = raw_pair.split('=', 1) 753 if (len(parts) != 2): 754 raise ValueError(f"Improperly formatted CLI agent argument key/value pair: '{raw_pair}'.") 755 756 key = parts[0].strip() 757 value = parts[1].strip() 758 759 agent_info[team_color].set_from_string(key, value) 760 761 return agent_info
29class GameInfo(edq.util.serial.DictConverter): 30 """ 31 A simple container that holds common information about a game. 32 """ 33 34 def __init__(self, 35 agent_infos: dict[chessai.core.types.Color, chessai.core.agentinfo.AgentInfo], 36 start_fen: str | None = None, 37 isolation_level: chessai.core.isolation.level.Level = chessai.core.isolation.level.Level.NONE, 38 max_moves: int = DEFAULT_MAX_MOVES, 39 agent_start_timeout: float = DEFAULT_AGENT_START_TIMEOUT, 40 agent_end_timeout: float = DEFAULT_AGENT_END_TIMEOUT, 41 agent_action_timeout: float = DEFAULT_AGENT_ACTION_TIMEOUT, 42 seed: int | None = None, 43 event: str | None = None, 44 site: str | None = None, 45 date: str | None = None, 46 game_round: str | None = None, 47 white_player: str | None = None, 48 black_player: str | None = None, 49 extra_info: dict[str, typing.Any] | None = None, 50 ) -> None: 51 if (seed is None): 52 seed = random.randint(0, 2**64) 53 54 self.seed: int = seed 55 """ The random seed for this game's RNG. """ 56 57 self.agent_infos: dict[chessai.core.types.Color, chessai.core.agentinfo.AgentInfo] = agent_infos 58 """ The required information for creating the agents for this game. """ 59 60 if (len(self.agent_infos) == 0): 61 raise ValueError("No agents provided.") 62 63 if (start_fen is None): 64 start_fen = chessai.core.gamestate.DEFAULT_FEN 65 66 self.start_fen: str = start_fen 67 """ The starting fen for this game. """ 68 69 self.isolation_level: chessai.core.isolation.level.Level = isolation_level 70 """ The isolation level to use for this game. """ 71 72 self.max_moves: int = max_moves 73 """ 74 The total number of moves (between all agents) allowed for this game. 75 If -1, unlimited moves are allowed. 76 """ 77 78 self.agent_start_timeout: float = agent_start_timeout 79 """ 80 The maximum number of seconds an agent is allowed when starting a game. 81 If <= 0, unlimited time is allowed. 82 """ 83 84 self.agent_end_timeout: float = agent_end_timeout 85 """ 86 The maximum number of seconds an agent is allowed when ending a game. 87 If <= 0, unlimited time is allowed. 88 """ 89 90 self.agent_action_timeout: float = agent_action_timeout 91 """ 92 The maximum number of seconds an agent is allowed when getting an action. 93 If <= 0, unlimited time is allowed. 94 """ 95 96 if (event is None): 97 event = '?' 98 99 self.event: str = event 100 """ The tournament or match event name. """ 101 102 if (site is None): 103 site = '?' 104 105 self.site: str = site 106 """ The location of the event. """ 107 108 if (date is None): 109 date = '????.??.??' 110 111 self.date: str = date 112 """ The starting date of the game. """ 113 114 if (game_round is None): 115 game_round = '?' 116 117 self.game_round: str = game_round 118 """ The playing round ordinal. """ 119 120 if (white_player is None): 121 white_player = '?' 122 123 self.white_player: str = white_player 124 """ The name of the white player. """ 125 126 if (black_player is None): 127 black_player = '?' 128 129 self.black_player: str = black_player 130 """ The name of the black player. """ 131 132 if (extra_info is None): 133 extra_info = {} 134 135 self.extra_info: dict[str, typing.Any] = extra_info 136 """ Any additional arguments passed to the game. """
A simple container that holds common information about a game.
34 def __init__(self, 35 agent_infos: dict[chessai.core.types.Color, chessai.core.agentinfo.AgentInfo], 36 start_fen: str | None = None, 37 isolation_level: chessai.core.isolation.level.Level = chessai.core.isolation.level.Level.NONE, 38 max_moves: int = DEFAULT_MAX_MOVES, 39 agent_start_timeout: float = DEFAULT_AGENT_START_TIMEOUT, 40 agent_end_timeout: float = DEFAULT_AGENT_END_TIMEOUT, 41 agent_action_timeout: float = DEFAULT_AGENT_ACTION_TIMEOUT, 42 seed: int | None = None, 43 event: str | None = None, 44 site: str | None = None, 45 date: str | None = None, 46 game_round: str | None = None, 47 white_player: str | None = None, 48 black_player: str | None = None, 49 extra_info: dict[str, typing.Any] | None = None, 50 ) -> None: 51 if (seed is None): 52 seed = random.randint(0, 2**64) 53 54 self.seed: int = seed 55 """ The random seed for this game's RNG. """ 56 57 self.agent_infos: dict[chessai.core.types.Color, chessai.core.agentinfo.AgentInfo] = agent_infos 58 """ The required information for creating the agents for this game. """ 59 60 if (len(self.agent_infos) == 0): 61 raise ValueError("No agents provided.") 62 63 if (start_fen is None): 64 start_fen = chessai.core.gamestate.DEFAULT_FEN 65 66 self.start_fen: str = start_fen 67 """ The starting fen for this game. """ 68 69 self.isolation_level: chessai.core.isolation.level.Level = isolation_level 70 """ The isolation level to use for this game. """ 71 72 self.max_moves: int = max_moves 73 """ 74 The total number of moves (between all agents) allowed for this game. 75 If -1, unlimited moves are allowed. 76 """ 77 78 self.agent_start_timeout: float = agent_start_timeout 79 """ 80 The maximum number of seconds an agent is allowed when starting a game. 81 If <= 0, unlimited time is allowed. 82 """ 83 84 self.agent_end_timeout: float = agent_end_timeout 85 """ 86 The maximum number of seconds an agent is allowed when ending a game. 87 If <= 0, unlimited time is allowed. 88 """ 89 90 self.agent_action_timeout: float = agent_action_timeout 91 """ 92 The maximum number of seconds an agent is allowed when getting an action. 93 If <= 0, unlimited time is allowed. 94 """ 95 96 if (event is None): 97 event = '?' 98 99 self.event: str = event 100 """ The tournament or match event name. """ 101 102 if (site is None): 103 site = '?' 104 105 self.site: str = site 106 """ The location of the event. """ 107 108 if (date is None): 109 date = '????.??.??' 110 111 self.date: str = date 112 """ The starting date of the game. """ 113 114 if (game_round is None): 115 game_round = '?' 116 117 self.game_round: str = game_round 118 """ The playing round ordinal. """ 119 120 if (white_player is None): 121 white_player = '?' 122 123 self.white_player: str = white_player 124 """ The name of the white player. """ 125 126 if (black_player is None): 127 black_player = '?' 128 129 self.black_player: str = black_player 130 """ The name of the black player. """ 131 132 if (extra_info is None): 133 extra_info = {} 134 135 self.extra_info: dict[str, typing.Any] = extra_info 136 """ Any additional arguments passed to the game. """
The required information for creating the agents for this game.
The total number of moves (between all agents) allowed for this game. If -1, unlimited moves are allowed.
The maximum number of seconds an agent is allowed when starting a game. If <= 0, unlimited time is allowed.
The maximum number of seconds an agent is allowed when ending a game. If <= 0, unlimited time is allowed.
The maximum number of seconds an agent is allowed when getting an action. If <= 0, unlimited time is allowed.
Inherited Members
138class GameResult(edq.util.serial.DictConverter): 139 """ The result of running a game. """ 140 141 def __init__(self, 142 game_id: int, 143 game_info: GameInfo, 144 termination_reason: chessai.core.types.TerminationReason | None = None, 145 score: float = 0, 146 end_fen: str | None = None, 147 game_timeout: bool = False, 148 timeout_agent_teams: list[chessai.core.types.Color] | None = None, 149 crash_agent_teams: list[chessai.core.types.Color] | None = None, 150 winning_agent_teams: list[chessai.core.types.Color] | None = None, 151 start_time: edq.util.time.Timestamp | None = None, 152 end_time: edq.util.time.Timestamp | None = None, 153 history: list[chessai.core.agentaction.AgentActionRecord] | None = None, 154 agent_complete_records: dict[chessai.core.types.Color, chessai.core.agentaction.AgentActionRecord] | None = None, 155 **kwargs: typing.Any, 156 ) -> None: 157 self.game_id: int = game_id 158 """ The ID of the game result. """ 159 160 self.game_info: GameInfo = game_info 161 """ The core information about this game. """ 162 163 self.termination_reason: chessai.core.types.TerminationReason | None = termination_reason 164 """ The reason the game ended. """ 165 166 self.score: float = score 167 """ The score of the game from white's perspective. """ 168 169 self.end_fen = end_fen 170 """ The fen the board is in at the end of the game. """ 171 172 if (start_time is None): 173 start_time = edq.util.time.Timestamp.now() 174 175 self.start_time: edq.util.time.Timestamp = start_time 176 """ The time the game started at. """ 177 178 self.end_time: edq.util.time.Timestamp | None = end_time 179 """ The time the game ended at. """ 180 181 if (history is None): 182 history = [] 183 184 self.history: list[chessai.core.agentaction.AgentActionRecord] = history 185 """ The history of actions taken by each agent in this game. """ 186 187 if (agent_complete_records is None): 188 agent_complete_records = {} 189 190 self.agent_complete_records: dict[chessai.core.types.Color, chessai.core.agentaction.AgentActionRecord] = agent_complete_records 191 """ 192 The record recieved from an agent when the game finishes. 193 For agents that learn, this may include information that the agent learned this game. 194 """ 195 196 self.game_timeout: bool = game_timeout 197 """ Indicates that the game has timed out (reached the maximum number of moves). """ 198 199 if (timeout_agent_teams is None): 200 timeout_agent_teams = [] 201 202 self.timeout_agent_teams: list[chessai.core.types.Color] = timeout_agent_teams 203 """ The list of agents that timed out in this game. """ 204 205 if (crash_agent_teams is None): 206 crash_agent_teams = [] 207 208 self.crash_agent_teams: list[chessai.core.types.Color] = crash_agent_teams 209 """ The list of agents that crashed in this game. """ 210 211 if (winning_agent_teams is None): 212 winning_agent_teams = [] 213 214 self.winning_agent_teams: list[chessai.core.types.Color] = winning_agent_teams 215 """ 216 The list of agents that won this game. 217 An empty list indicates a tie. 218 219 Games may interpret this value in different ways. 220 """ 221 222 def get_duration_secs(self) -> float: 223 """ 224 Get the game's duration in seconds. 225 Will return positive infinity if the game has no end time 226 (it is still going or crashed (in very rare cases)). 227 """ 228 229 if (self.end_time is None): 230 return math.inf 231 232 return self.end_time.sub(self.start_time).to_secs()
The result of running a game.
141 def __init__(self, 142 game_id: int, 143 game_info: GameInfo, 144 termination_reason: chessai.core.types.TerminationReason | None = None, 145 score: float = 0, 146 end_fen: str | None = None, 147 game_timeout: bool = False, 148 timeout_agent_teams: list[chessai.core.types.Color] | None = None, 149 crash_agent_teams: list[chessai.core.types.Color] | None = None, 150 winning_agent_teams: list[chessai.core.types.Color] | None = None, 151 start_time: edq.util.time.Timestamp | None = None, 152 end_time: edq.util.time.Timestamp | None = None, 153 history: list[chessai.core.agentaction.AgentActionRecord] | None = None, 154 agent_complete_records: dict[chessai.core.types.Color, chessai.core.agentaction.AgentActionRecord] | None = None, 155 **kwargs: typing.Any, 156 ) -> None: 157 self.game_id: int = game_id 158 """ The ID of the game result. """ 159 160 self.game_info: GameInfo = game_info 161 """ The core information about this game. """ 162 163 self.termination_reason: chessai.core.types.TerminationReason | None = termination_reason 164 """ The reason the game ended. """ 165 166 self.score: float = score 167 """ The score of the game from white's perspective. """ 168 169 self.end_fen = end_fen 170 """ The fen the board is in at the end of the game. """ 171 172 if (start_time is None): 173 start_time = edq.util.time.Timestamp.now() 174 175 self.start_time: edq.util.time.Timestamp = start_time 176 """ The time the game started at. """ 177 178 self.end_time: edq.util.time.Timestamp | None = end_time 179 """ The time the game ended at. """ 180 181 if (history is None): 182 history = [] 183 184 self.history: list[chessai.core.agentaction.AgentActionRecord] = history 185 """ The history of actions taken by each agent in this game. """ 186 187 if (agent_complete_records is None): 188 agent_complete_records = {} 189 190 self.agent_complete_records: dict[chessai.core.types.Color, chessai.core.agentaction.AgentActionRecord] = agent_complete_records 191 """ 192 The record recieved from an agent when the game finishes. 193 For agents that learn, this may include information that the agent learned this game. 194 """ 195 196 self.game_timeout: bool = game_timeout 197 """ Indicates that the game has timed out (reached the maximum number of moves). """ 198 199 if (timeout_agent_teams is None): 200 timeout_agent_teams = [] 201 202 self.timeout_agent_teams: list[chessai.core.types.Color] = timeout_agent_teams 203 """ The list of agents that timed out in this game. """ 204 205 if (crash_agent_teams is None): 206 crash_agent_teams = [] 207 208 self.crash_agent_teams: list[chessai.core.types.Color] = crash_agent_teams 209 """ The list of agents that crashed in this game. """ 210 211 if (winning_agent_teams is None): 212 winning_agent_teams = [] 213 214 self.winning_agent_teams: list[chessai.core.types.Color] = winning_agent_teams 215 """ 216 The list of agents that won this game. 217 An empty list indicates a tie. 218 219 Games may interpret this value in different ways. 220 """
The history of actions taken by each agent in this game.
The record recieved from an agent when the game finishes. For agents that learn, this may include information that the agent learned this game.
The list of agents that won this game. An empty list indicates a tie.
Games may interpret this value in different ways.
222 def get_duration_secs(self) -> float: 223 """ 224 Get the game's duration in seconds. 225 Will return positive infinity if the game has no end time 226 (it is still going or crashed (in very rare cases)). 227 """ 228 229 if (self.end_time is None): 230 return math.inf 231 232 return self.end_time.sub(self.start_time).to_secs()
Get the game's duration in seconds. Will return positive infinity if the game has no end time (it is still going or crashed (in very rare cases)).
Inherited Members
234class Game(abc.ABC): 235 """ 236 A game that can be run in chessai. 237 Games combine the rules, layouts, and agents to run. 238 """ 239 240 def __init__(self, 241 game_info: GameInfo, 242 save_path: str | None = None, 243 is_replay: bool = False, 244 initial_actions: list[chessai.core.action.Action] | None = None, 245 action_history: list[chessai.core.action.Action] | None = None, 246 expected_result: chessai.core.gameparser.PGNResult = chessai.core.gameparser.PGNResult.UNKNOWN, 247 ) -> None: 248 self.game_info: GameInfo = game_info 249 """ The core information about this game. """ 250 251 self._save_path: str | None = save_path 252 """ Where to save the results of this game. """ 253 254 self._is_replay: bool = is_replay 255 """ 256 Indicates that this game is being loaded from a replay. 257 Some behavior, like saving the result, will be modified. 258 """ 259 260 if (initial_actions is None): 261 initial_actions = [] 262 263 self.initial_actions: list[chessai.core.action.Action] = initial_actions 264 """ The moves to be played at the beginning of the game. """ 265 266 if (action_history is None): 267 action_history = [] 268 269 self.action_history: list[chessai.core.action.Action] = action_history 270 """ The actions taken throughout the game. """ 271 272 self.expected_result: chessai.core.gameparser.PGNResult = expected_result 273 """ The expected result from the PGN: '1-0', '0-1', '1/2-1/2', or '*'. """ 274 275 @classmethod 276 def from_pgn(cls, 277 pgn: str, 278 state_class: typing.Type[chessai.core.gamestate.GameState], 279 game_info: GameInfo, 280 save_path: str | None = None, 281 **kwargs: typing.Any) -> typing.Optional['Game']: 282 """ 283 Create a Game from a PGN. 284 285 The game will be initialized with metadata from the PGN headers and will store the moves for replay. 286 287 Returns the game and whether the PGN parsing was successful. 288 """ 289 290 parsed_pgn = chessai.core.gameparser.parse_pgn(pgn, state_class) 291 if (parsed_pgn is None): 292 return None 293 294 # Extract starting FEN from optional headers or use default. 295 start_fen = parsed_pgn.get_starting_fen() 296 if (start_fen is not None): 297 game_info.start_fen = start_fen 298 299 # Extract header information into the game info. 300 game_info.event = parsed_pgn.headers.event 301 game_info.site = parsed_pgn.headers.site 302 game_info.date = parsed_pgn.headers.date 303 game_info.game_round = parsed_pgn.headers.round 304 game_info.white_player = parsed_pgn.headers.white 305 game_info.black_player = parsed_pgn.headers.black 306 game_info.extra_info.update(parsed_pgn.optional_headers) 307 308 return cls( 309 game_info = game_info, 310 save_path = save_path, 311 is_replay = True, 312 initial_actions = parsed_pgn.initial_actions, 313 expected_result = parsed_pgn.result, 314 **kwargs 315 ) 316 317 def to_pgn(self, 318 final_state: chessai.core.gamestate.GameState, 319 termination_reason: chessai.core.types.TerminationReason, 320 ) -> str: 321 """ Returns the PGN representation of the Game. """ 322 323 if (termination_reason in [chessai.core.types.TerminationReason.CHECKMATE, 324 chessai.core.types.TerminationReason.FORFEIT]): 325 winners = final_state.get_winners() 326 if (chessai.core.types.Color.WHITE in winners): 327 result = chessai.core.gameparser.PGNResult.WHITE_WIN 328 elif (chessai.core.types.Color.BLACK in winners): 329 result = chessai.core.gameparser.PGNResult.BLACK_WIN 330 else: 331 result = chessai.core.gameparser.PGNResult.TIE 332 elif (termination_reason in [chessai.core.types.TerminationReason.STALEMATE, 333 chessai.core.types.TerminationReason.INSUFFICIENT_MATERIAL, 334 chessai.core.types.TerminationReason.ACCEPTED_DRAW_PROPOSAL]): 335 result = chessai.core.gameparser.PGNResult.TIE 336 elif (termination_reason == chessai.core.types.TerminationReason.IN_PROGRESS): 337 result = chessai.core.gameparser.PGNResult.IN_PROGRESS 338 else: 339 result = chessai.core.gameparser.PGNResult.UNKNOWN 340 341 headers = chessai.core.gameparser.StandardHeaders( 342 event = self.game_info.event, 343 site = self.game_info.site, 344 date = self.game_info.date, 345 game_round = self.game_info.game_round, 346 white = self.game_info.white_player, 347 black = self.game_info.black_player, 348 result = result) 349 350 return chessai.core.gameparser.to_pgn(headers, self.game_info.extra_info, type(final_state), 351 self.game_info.start_fen, self.action_history) 352 353 def process_args(self, args: argparse.Namespace) -> None: 354 """ Process any special arguments from the command-line. """ 355 356 @abc.abstractmethod 357 def get_initial_state(self, 358 rng: random.Random, 359 fen: str | None = None) -> chessai.core.gamestate.GameState: 360 """ Create the initial state for this game. """ 361 362 def process_turn(self, 363 state: chessai.core.gamestate.GameState, 364 action_record: chessai.core.agentaction.AgentActionRecord, 365 result: GameResult, 366 rng: random.Random) -> chessai.core.gamestate.GameState: 367 """ 368 Process the given agent action and return an updated game state. 369 The returned game state may be a copy or modified version of the passed in game state. 370 """ 371 372 # The agent has timed out. 373 if (action_record.timeout): 374 result.timeout_agent_teams.append(action_record.player) 375 result.termination_reason = chessai.core.types.TerminationReason.AGENT_TIMEOUT 376 state.process_agent_timeout(action_record.player) 377 return state 378 379 # The agent has crashed. 380 if (action_record.crashed): 381 result.crash_agent_teams.append(action_record.player) 382 result.termination_reason = chessai.core.types.TerminationReason.AGENT_CRASHED 383 state.process_agent_crash(action_record.player) 384 return state 385 386 action = action_record.get_action() 387 if (action not in state.get_legal_actions()): 388 raise ValueError(f"Illegal action for agent {action_record.player}: '{action.uci()}' of type '{type(action)}'.") 389 390 # Add the action to the action history. 391 self.action_history.append(action) 392 393 self._call_state_process_turn_full(state, action, rng) 394 395 return state 396 397 def _call_state_process_turn_full(self, 398 state: chessai.core.gamestate.GameState, 399 action: chessai.core.action.Action, 400 rng: random.Random, 401 **kwargs: typing.Any) -> None: 402 """ Call on the game state to process a full turn. """ 403 404 state.process_turn_full(action, rng, **kwargs) 405 406 def check_end(self, state: chessai.core.gamestate.GameState) -> bool: 407 """ 408 Check to see if the game is over. 409 Return True if the game is now over, False otherwise. 410 411 By default, this will check chessai.core.gamestate.GameState.game_over 412 and chessai.core.gamestate.GameState.is_game_over(), 413 but child games can override for more complex functionality. 414 """ 415 416 return (state.game_over or state.is_game_over()) 417 418 def game_complete(self, state: chessai.core.gamestate.GameState, result: GameResult) -> None: 419 """ 420 Make any last adjustments to the game result after the game is over. 421 """ 422 423 def run(self, ui: chessai.core.ui.UI) -> GameResult: 424 """ 425 The main "game loop" for all games. 426 """ 427 428 logging.debug("Starting a game with seed: %d.", self.game_info.seed) 429 430 # Create a new random number generator just for this game. 431 rng = random.Random(self.game_info.seed) 432 433 # Keep track of what happens during this game. 434 game_id = rng.randint(0, 2**64) 435 result = GameResult(game_id, self.game_info) 436 437 # Initialize the agent isolator. 438 isolator = self.game_info.isolation_level.get_isolator() 439 isolator.init_agents(self.game_info.agent_infos) 440 441 # Create the initial game state (and force it's seed). 442 state = self.get_initial_state(rng, self.game_info.start_fen) 443 state.seed = game_id 444 state.game_start() 445 446 # board_highlights: list[chessai.core.board.Highlight] = [] 447 448 # Notify agents about the start of the game. 449 records = isolator.game_start(rng, state, self.game_info.agent_start_timeout) 450 for record in records.values(): 451 if (record.timeout): 452 result.timeout_agent_teams.append(record.player) 453 state.process_agent_timeout(record.player) 454 result.termination_reason = chessai.core.types.TerminationReason.AGENT_TIMEOUT 455 elif (record.crashed): 456 result.crash_agent_teams.append(record.player) 457 state.process_agent_crash(record.player) 458 result.termination_reason = chessai.core.types.TerminationReason.AGENT_CRASHED 459 else: 460 continue 461 # board_highlights += record.get_board_highlights() 462 463 state.agents_game_start(records) 464 465 # Start the UI. 466 ui.game_start(state) 467 468 while (not self.check_end(state)): 469 logging.trace("Turn %d, agent %s.", state.fullmove_number, state.turn) # type: ignore[attr-defined] # pylint: disable=no-member 470 471 if (len(self.initial_actions) > 0): 472 # Get the action from the pre-loaded actions. 473 action = self.initial_actions.pop(0) 474 agent_action = chessai.core.agentaction.AgentAction(action = action) 475 duration = edq.util.time.Duration(0) 476 477 action_record = chessai.core.agentaction.AgentActionRecord(state.turn, agent_action, duration) 478 else: 479 # Get the next action from the agent. 480 action_record = isolator.get_action(state, self.game_info.agent_action_timeout) 481 482 # Execute the next action and update the state. 483 state = self.process_turn(state, action_record, result, rng) 484 485 # Update the UI. 486 ui.update(state) 487 488 # Update the game result and move history. 489 result.history.append(action_record) 490 491 # Check for game ending conditions. 492 if (self.check_end(state)): 493 break 494 495 # Check if this game has ran for the maximum number of moves. 496 if ((self.game_info.max_moves > 0) and (len(result.history) >= self.game_info.max_moves)): 497 state.process_game_timeout() 498 result.game_timeout = True 499 result.termination_reason = chessai.core.types.TerminationReason.GAME_TIMEOUT 500 break 501 502 # Mark the end time of the game. 503 result.end_time = edq.util.time.Timestamp.now() 504 505 # Notify the state about the end of the game. 506 (winners, score) = state.game_complete() 507 result.winning_agent_teams += winners 508 result.score = score 509 510 if (result.termination_reason is None): 511 result.termination_reason = state.get_termination_reason() 512 513 result.end_fen = state.get_fen() 514 515 # Notify agents about the end of this game. 516 result.agent_complete_records = isolator.game_complete(state, self.game_info.agent_end_timeout) 517 518 # All the game to make final updates to the result. 519 self.game_complete(state, result) 520 521 # Update the UI. 522 ui.game_complete(state, result.termination_reason) 523 524 # Cleanup 525 isolator.close() 526 ui.close() 527 528 if ((not self._is_replay) and (self._save_path is not None)): 529 logging.info("Saving results to '%s'.", self._save_path) 530 531 pgn = self.to_pgn(state, result.termination_reason) 532 edq.util.dirent.write_file(self._save_path, pgn) 533 534 return result
A game that can be run in chessai. Games combine the rules, layouts, and agents to run.
The moves to be played at the beginning of the game.
The expected result from the PGN: '1-0', '0-1', '1/2-1/2', or '*'.
275 @classmethod 276 def from_pgn(cls, 277 pgn: str, 278 state_class: typing.Type[chessai.core.gamestate.GameState], 279 game_info: GameInfo, 280 save_path: str | None = None, 281 **kwargs: typing.Any) -> typing.Optional['Game']: 282 """ 283 Create a Game from a PGN. 284 285 The game will be initialized with metadata from the PGN headers and will store the moves for replay. 286 287 Returns the game and whether the PGN parsing was successful. 288 """ 289 290 parsed_pgn = chessai.core.gameparser.parse_pgn(pgn, state_class) 291 if (parsed_pgn is None): 292 return None 293 294 # Extract starting FEN from optional headers or use default. 295 start_fen = parsed_pgn.get_starting_fen() 296 if (start_fen is not None): 297 game_info.start_fen = start_fen 298 299 # Extract header information into the game info. 300 game_info.event = parsed_pgn.headers.event 301 game_info.site = parsed_pgn.headers.site 302 game_info.date = parsed_pgn.headers.date 303 game_info.game_round = parsed_pgn.headers.round 304 game_info.white_player = parsed_pgn.headers.white 305 game_info.black_player = parsed_pgn.headers.black 306 game_info.extra_info.update(parsed_pgn.optional_headers) 307 308 return cls( 309 game_info = game_info, 310 save_path = save_path, 311 is_replay = True, 312 initial_actions = parsed_pgn.initial_actions, 313 expected_result = parsed_pgn.result, 314 **kwargs 315 )
Create a Game from a PGN.
The game will be initialized with metadata from the PGN headers and will store the moves for replay.
Returns the game and whether the PGN parsing was successful.
317 def to_pgn(self, 318 final_state: chessai.core.gamestate.GameState, 319 termination_reason: chessai.core.types.TerminationReason, 320 ) -> str: 321 """ Returns the PGN representation of the Game. """ 322 323 if (termination_reason in [chessai.core.types.TerminationReason.CHECKMATE, 324 chessai.core.types.TerminationReason.FORFEIT]): 325 winners = final_state.get_winners() 326 if (chessai.core.types.Color.WHITE in winners): 327 result = chessai.core.gameparser.PGNResult.WHITE_WIN 328 elif (chessai.core.types.Color.BLACK in winners): 329 result = chessai.core.gameparser.PGNResult.BLACK_WIN 330 else: 331 result = chessai.core.gameparser.PGNResult.TIE 332 elif (termination_reason in [chessai.core.types.TerminationReason.STALEMATE, 333 chessai.core.types.TerminationReason.INSUFFICIENT_MATERIAL, 334 chessai.core.types.TerminationReason.ACCEPTED_DRAW_PROPOSAL]): 335 result = chessai.core.gameparser.PGNResult.TIE 336 elif (termination_reason == chessai.core.types.TerminationReason.IN_PROGRESS): 337 result = chessai.core.gameparser.PGNResult.IN_PROGRESS 338 else: 339 result = chessai.core.gameparser.PGNResult.UNKNOWN 340 341 headers = chessai.core.gameparser.StandardHeaders( 342 event = self.game_info.event, 343 site = self.game_info.site, 344 date = self.game_info.date, 345 game_round = self.game_info.game_round, 346 white = self.game_info.white_player, 347 black = self.game_info.black_player, 348 result = result) 349 350 return chessai.core.gameparser.to_pgn(headers, self.game_info.extra_info, type(final_state), 351 self.game_info.start_fen, self.action_history)
Returns the PGN representation of the Game.
353 def process_args(self, args: argparse.Namespace) -> None: 354 """ Process any special arguments from the command-line. """
Process any special arguments from the command-line.
356 @abc.abstractmethod 357 def get_initial_state(self, 358 rng: random.Random, 359 fen: str | None = None) -> chessai.core.gamestate.GameState: 360 """ Create the initial state for this game. """
Create the initial state for this game.
362 def process_turn(self, 363 state: chessai.core.gamestate.GameState, 364 action_record: chessai.core.agentaction.AgentActionRecord, 365 result: GameResult, 366 rng: random.Random) -> chessai.core.gamestate.GameState: 367 """ 368 Process the given agent action and return an updated game state. 369 The returned game state may be a copy or modified version of the passed in game state. 370 """ 371 372 # The agent has timed out. 373 if (action_record.timeout): 374 result.timeout_agent_teams.append(action_record.player) 375 result.termination_reason = chessai.core.types.TerminationReason.AGENT_TIMEOUT 376 state.process_agent_timeout(action_record.player) 377 return state 378 379 # The agent has crashed. 380 if (action_record.crashed): 381 result.crash_agent_teams.append(action_record.player) 382 result.termination_reason = chessai.core.types.TerminationReason.AGENT_CRASHED 383 state.process_agent_crash(action_record.player) 384 return state 385 386 action = action_record.get_action() 387 if (action not in state.get_legal_actions()): 388 raise ValueError(f"Illegal action for agent {action_record.player}: '{action.uci()}' of type '{type(action)}'.") 389 390 # Add the action to the action history. 391 self.action_history.append(action) 392 393 self._call_state_process_turn_full(state, action, rng) 394 395 return state
Process the given agent action and return an updated game state. The returned game state may be a copy or modified version of the passed in game state.
406 def check_end(self, state: chessai.core.gamestate.GameState) -> bool: 407 """ 408 Check to see if the game is over. 409 Return True if the game is now over, False otherwise. 410 411 By default, this will check chessai.core.gamestate.GameState.game_over 412 and chessai.core.gamestate.GameState.is_game_over(), 413 but child games can override for more complex functionality. 414 """ 415 416 return (state.game_over or state.is_game_over())
Check to see if the game is over. Return True if the game is now over, False otherwise.
By default, this will check chessai.core.gamestate.GameState.game_over and chessai.core.gamestate.GameState.is_game_over(), but child games can override for more complex functionality.
418 def game_complete(self, state: chessai.core.gamestate.GameState, result: GameResult) -> None: 419 """ 420 Make any last adjustments to the game result after the game is over. 421 """
Make any last adjustments to the game result after the game is over.
423 def run(self, ui: chessai.core.ui.UI) -> GameResult: 424 """ 425 The main "game loop" for all games. 426 """ 427 428 logging.debug("Starting a game with seed: %d.", self.game_info.seed) 429 430 # Create a new random number generator just for this game. 431 rng = random.Random(self.game_info.seed) 432 433 # Keep track of what happens during this game. 434 game_id = rng.randint(0, 2**64) 435 result = GameResult(game_id, self.game_info) 436 437 # Initialize the agent isolator. 438 isolator = self.game_info.isolation_level.get_isolator() 439 isolator.init_agents(self.game_info.agent_infos) 440 441 # Create the initial game state (and force it's seed). 442 state = self.get_initial_state(rng, self.game_info.start_fen) 443 state.seed = game_id 444 state.game_start() 445 446 # board_highlights: list[chessai.core.board.Highlight] = [] 447 448 # Notify agents about the start of the game. 449 records = isolator.game_start(rng, state, self.game_info.agent_start_timeout) 450 for record in records.values(): 451 if (record.timeout): 452 result.timeout_agent_teams.append(record.player) 453 state.process_agent_timeout(record.player) 454 result.termination_reason = chessai.core.types.TerminationReason.AGENT_TIMEOUT 455 elif (record.crashed): 456 result.crash_agent_teams.append(record.player) 457 state.process_agent_crash(record.player) 458 result.termination_reason = chessai.core.types.TerminationReason.AGENT_CRASHED 459 else: 460 continue 461 # board_highlights += record.get_board_highlights() 462 463 state.agents_game_start(records) 464 465 # Start the UI. 466 ui.game_start(state) 467 468 while (not self.check_end(state)): 469 logging.trace("Turn %d, agent %s.", state.fullmove_number, state.turn) # type: ignore[attr-defined] # pylint: disable=no-member 470 471 if (len(self.initial_actions) > 0): 472 # Get the action from the pre-loaded actions. 473 action = self.initial_actions.pop(0) 474 agent_action = chessai.core.agentaction.AgentAction(action = action) 475 duration = edq.util.time.Duration(0) 476 477 action_record = chessai.core.agentaction.AgentActionRecord(state.turn, agent_action, duration) 478 else: 479 # Get the next action from the agent. 480 action_record = isolator.get_action(state, self.game_info.agent_action_timeout) 481 482 # Execute the next action and update the state. 483 state = self.process_turn(state, action_record, result, rng) 484 485 # Update the UI. 486 ui.update(state) 487 488 # Update the game result and move history. 489 result.history.append(action_record) 490 491 # Check for game ending conditions. 492 if (self.check_end(state)): 493 break 494 495 # Check if this game has ran for the maximum number of moves. 496 if ((self.game_info.max_moves > 0) and (len(result.history) >= self.game_info.max_moves)): 497 state.process_game_timeout() 498 result.game_timeout = True 499 result.termination_reason = chessai.core.types.TerminationReason.GAME_TIMEOUT 500 break 501 502 # Mark the end time of the game. 503 result.end_time = edq.util.time.Timestamp.now() 504 505 # Notify the state about the end of the game. 506 (winners, score) = state.game_complete() 507 result.winning_agent_teams += winners 508 result.score = score 509 510 if (result.termination_reason is None): 511 result.termination_reason = state.get_termination_reason() 512 513 result.end_fen = state.get_fen() 514 515 # Notify agents about the end of this game. 516 result.agent_complete_records = isolator.game_complete(state, self.game_info.agent_end_timeout) 517 518 # All the game to make final updates to the result. 519 self.game_complete(state, result) 520 521 # Update the UI. 522 ui.game_complete(state, result.termination_reason) 523 524 # Cleanup 525 isolator.close() 526 ui.close() 527 528 if ((not self._is_replay) and (self._save_path is not None)): 529 logging.info("Saving results to '%s'.", self._save_path) 530 531 pgn = self.to_pgn(state, result.termination_reason) 532 edq.util.dirent.write_file(self._save_path, pgn) 533 534 return result
The main "game loop" for all games.
536def set_cli_args(parser: argparse.ArgumentParser, default_board: str | None = None) -> argparse.ArgumentParser: 537 """ 538 Set common CLI arguments. 539 This is a sibling to init_from_args(), as the arguments set here can be interpreted there. 540 """ 541 542 parser.add_argument('--board', dest = 'board', 543 action = 'store', type = str, default = default_board, 544 help = ('Play on this board (default: %(default)s).' 545 + ' This may be a FEN of the board.' 546 + ' It may also be the full path to a board, or just a filename.' 547 + ' If just a filename, than the `chessai/resources/boards` directory will be checked (using a ".fen" extension.')) 548 549 parser.add_argument('--num-games', dest = 'num_games', 550 action = 'store', type = int, default = 1, 551 help = 'The number of games to play (default: %(default)s).') 552 553 parser.add_argument('--seed', dest = 'seed', 554 action = 'store', type = int, default = None, 555 help = 'The random seed for the game (will be randomly generated if not set).') 556 557 parser.add_argument('--max-moves', dest = 'max_moves', 558 action = 'store', type = int, default = DEFAULT_MAX_MOVES, 559 help = 'The maximum number of moves (total for all agents) allowed in this game (-1 for unlimited) (default: %(default)s).') 560 561 parser.add_argument('--agent-start-timeout', dest = 'agent_start_timeout', 562 action = 'store', type = float, default = DEFAULT_AGENT_START_TIMEOUT, 563 help = ('The maximum number of seconds each agent is allowed when starting a game (<= 0 for unlimited time) (default: %(default)s).' 564 + ' Note that the "none" isolation level cannot enforce timeouts.')) 565 566 parser.add_argument('--agent-end-timeout', dest = 'agent_end_timeout', 567 action = 'store', type = float, default = DEFAULT_AGENT_END_TIMEOUT, 568 help = ('The maximum number of seconds each agent is allowed when ending a game (<= 0 for unlimited time) (default: %(default)s).' 569 + ' Note that the "none" isolation level cannot enforce timeouts.')) 570 571 parser.add_argument('--agent-action-timeout', dest = 'agent_action_timeout', 572 action = 'store', type = float, default = DEFAULT_AGENT_ACTION_TIMEOUT, 573 help = ('The maximum number of seconds each agent is allowed when getting an action (<= 0 for unlimited time) (default: %(default)s).' 574 + ' Note that the "none" isolation level cannot enforce timeouts.')) 575 576 parser.add_argument('--isolation', dest = 'isolation_level', metavar = 'LEVEL', 577 action = 'store', type = str, default = chessai.core.isolation.level.Level.NONE.value, 578 choices = chessai.core.isolation.level.LEVELS, 579 help = ('Set the agent isolation level for this game (default: %(default)s).' 580 + ' Choose one of:' 581 + ' `none` -- Do not make any attempt to isolate the agent code from the game (fastest and least secure),' 582 + ' `process` -- Run the agent code in a separate process' 583 + ' (offers some protection, but still vulnerable to disk or execution exploits),' 584 + ' `tcp` -- Open TCP listeners to communicate with agents (most secure, requires additional work to set up agents).')) 585 586 parser.add_argument('--agent-arg', dest = 'raw_agent_args', metavar = 'ARG', 587 action = 'append', type = str, default = [], 588 help = ('Specify arguments directly to agents (may be used multiple times).' 589 + ' The value for this argument must be formatted as "player::key=value",' 590 + ' for example to set `foo = 9` for white and `bar = a` for black, we can use:' 591 + ' `--agent-arg white::foo=9 --agent-arg black::bar=a`.')) 592 593 parser.add_argument('--save-path', dest = 'save_path', 594 action = 'store', type = str, default = None, 595 help = ('If specified, write the result of this game to the specified location.' 596 + ' This file can be replayed with `--replay-path`.')) 597 598 parser.add_argument('--replay-path', dest = 'replay_path', 599 action = 'store', type = str, default = None, 600 help = ('If specified, replay the game whose result was saved at the specified path with `--save-path`.' 601 + ' This may be a PGN of the game.' 602 + ' It may also be the full path to a game, or just a filename.' 603 + ' If just a filename, than the `chessai/resources/games` directory will be checked (using a ".pgn" extension.')) 604 605 return parser
Set common CLI arguments. This is a sibling to init_from_args(), as the arguments set here can be interpreted there.
607def init_from_args( 608 args: argparse.Namespace, 609 game_class: typing.Type[Game], 610 state_class: typing.Type[chessai.core.gamestate.GameState], 611 base_agent_infos: dict[chessai.core.types.Color, chessai.core.agentinfo.AgentInfo] | None = None, 612 **kwargs: typing.Any, 613 ) -> argparse.Namespace: 614 """ 615 Take in args from a parser that was passed to set_cli_args(), 616 and initialize the proper components. 617 This will create a number of games (and related resources) 618 based on `--num-games` + `--num-training`. 619 Each of these resources will be placed in their respective list at 620 `args._fens`, `args._agent_infos`, or `args._games`. 621 """ 622 623 if (base_agent_infos is None): 624 base_agent_infos = {} 625 626 if (args.num_games <= 0): 627 raise ValueError(f"At least one game must be played (--num-games), {args.num_games} was specified.") 628 629 # Establish an RNG to generate seeds for each game using the given seed. 630 seed = args.seed 631 if (seed is None): 632 seed = random.randint(0, 2**64) 633 634 logging.debug("Using source seed for games: %d.", seed) 635 rng = random.Random(seed) 636 637 agents = [chessai.core.types.Color.WHITE, chessai.core.types.Color.BLACK] 638 agent_infos = _parse_agent_infos(agents, args.raw_agent_args, base_agent_infos) 639 640 base_save_path = args.save_path 641 642 all_fens: list[str] = [] 643 all_agent_infos = [] 644 all_games = [] 645 646 for i in range(args.num_games): 647 game_seed = rng.randint(0, 2**64) 648 649 all_agent_infos.append(copy.deepcopy(agent_infos)) 650 all_fens.append(args.board) 651 652 white_agent_info = all_agent_infos[-1].get(chessai.core.types.Color.WHITE, None) 653 if (white_agent_info is not None): 654 white_reference = white_agent_info.name 655 if (isinstance(white_reference, chessai.util.reflection.Reference)): 656 white_player = str(white_reference) 657 else: 658 white_player = white_reference 659 else: 660 white_player = '?' 661 662 black_agent_info = all_agent_infos[-1].get(chessai.core.types.Color.BLACK, None) 663 if (black_agent_info is not None): 664 black_reference = black_agent_info.name 665 if (isinstance(black_reference, chessai.util.reflection.Reference)): 666 black_player = str(black_reference) 667 else: 668 black_player = black_reference 669 else: 670 black_player = '?' 671 672 game_info = GameInfo( 673 all_agent_infos[-1], 674 start_fen = all_fens[-1], 675 isolation_level = chessai.core.isolation.level.Level(args.isolation_level), 676 max_moves = args.max_moves, 677 agent_start_timeout = args.agent_start_timeout, 678 agent_end_timeout = args.agent_end_timeout, 679 agent_action_timeout = args.agent_action_timeout, 680 seed = game_seed, 681 event = 'Casual Chessai Game', 682 site = 'https://github.com/ucsc-cse-240/chessai', 683 date = datetime.date.today().isoformat(), 684 game_round = str(i), 685 white_player = white_player, 686 black_player = black_player, 687 ) 688 689 # Suffix the save path if there is more than one game. 690 save_path = base_save_path 691 if ((save_path is not None) and (args.num_games > 1)): 692 parts = os.path.splitext(save_path) 693 save_path = f"{parts[0]}_{i:03d}{parts[1]}" 694 695 if (args.replay_path is None): 696 game_args = { 697 'game_info': game_info, 698 'save_path': save_path, 699 } 700 701 game = game_class(**game_args) 702 else: 703 raw_game = game_class.from_pgn(args.replay_path, state_class, game_info, save_path) 704 if (raw_game is None): 705 raise ValueError(f"Failed to initialize game number {i} from the PGN: '{args.replay_path}'.") 706 707 game = raw_game 708 709 game.process_args(args) 710 711 all_games.append(game) 712 713 setattr(args, '_fens', all_fens) 714 setattr(args, '_agent_infos', all_agent_infos) 715 setattr(args, '_games', all_games) 716 717 return args
Take in args from a parser that was passed to set_cli_args(),
and initialize the proper components.
This will create a number of games (and related resources)
based on --num-games + --num-training.
Each of these resources will be placed in their respective list at
args._fens, args._agent_infos, or args._games.