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