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
DEFAULT_MAX_MOVES: int = -1
DEFAULT_AGENT_START_TIMEOUT: float = 0.0
DEFAULT_AGENT_END_TIMEOUT: float = 0.0
DEFAULT_AGENT_ACTION_TIMEOUT: float = 0.0
DEFAULT_AGENT: str = 'agent-random'
class GameInfo(edq.util.serial.DictConverter):
 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.

GameInfo( agent_infos: dict[chessai.core.types.Color, chessai.core.agentinfo.AgentInfo], start_fen: str | None = None, isolation_level: chessai.core.isolation.level.Level = <Level.NONE: 'none'>, max_moves: int = -1, agent_start_timeout: float = 0.0, agent_end_timeout: float = 0.0, agent_action_timeout: float = 0.0, seed: int | None = None, event: str | None = None, site: str | None = None, date: str | None = None, game_round: str | None = None, white_player: str | None = None, black_player: str | None = None, extra_info: dict[str, typing.Any] | None = None, wait_for_user: bool = False)
 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. """
seed: int

The random seed for this game's RNG.

The required information for creating the agents for this game.

start_fen: str

The starting fen for this game.

The isolation level to use for this game.

max_moves: int

The total number of moves (between all agents) allowed for this game. If -1, unlimited moves are allowed.

agent_start_timeout: float

The maximum number of seconds an agent is allowed when starting a game. If <= 0, unlimited time is allowed.

agent_end_timeout: float

The maximum number of seconds an agent is allowed when ending a game. If <= 0, unlimited time is allowed.

agent_action_timeout: float

The maximum number of seconds an agent is allowed when getting an action. If <= 0, unlimited time is allowed.

event: str

The tournament or match event name.

site: str

The location of the event.

date: str

The starting date of the game.

game_round: str

The playing round ordinal.

white_player: str

The name of the white player.

black_player: str

The name of the black player.

extra_info: dict[str, typing.Any]

Any additional arguments passed to the game.

wait_for_user: bool

Pause the game between each move until the user is ready.

class GameResult(edq.util.serial.DictConverter):
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.

GameResult( game_id: int, game_info: GameInfo, termination_reason: chessai.core.types.TerminationReason | None = None, score: float = 0, end_fen: str | None = None, game_timeout: bool = False, timeout_agent_teams: list[chessai.core.types.Color] | None = None, crash_agent_teams: list[chessai.core.types.Color] | None = None, winning_agent_teams: list[chessai.core.types.Color] | None = None, start_time: edq.util.time.Timestamp | None = None, end_time: edq.util.time.Timestamp | None = None, history: list[chessai.core.agentaction.AgentActionRecord] | None = None, agent_complete_records: dict[chessai.core.types.Color, chessai.core.agentaction.AgentActionRecord] | None = None, **kwargs: Any)
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        """
game_id: int

The ID of the game result.

game_info: GameInfo

The core information about this game.

termination_reason: chessai.core.types.TerminationReason | None

The reason the game ended.

score: float

The score of the game from white's perspective.

end_fen

The fen the board is in at the end of the game.

The time the game started at.

end_time: edq.util.time.Timestamp | None

The time the game ended at.

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.

game_timeout: bool

Indicates that the game has timed out (reached the maximum number of moves).

timeout_agent_teams: list[chessai.core.types.Color]

The list of agents that timed out in this game.

crash_agent_teams: list[chessai.core.types.Color]

The list of agents that crashed in this game.

winning_agent_teams: list[chessai.core.types.Color]

The list of agents that won this game. An empty list indicates a tie.

Games may interpret this value in different ways.

def get_duration_secs(self) -> float:
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)).

class Game(abc.ABC):
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.

game_info: GameInfo

The core information about this game.

initial_actions: list[chessai.core.action.Action]

The moves to be played at the beginning of the game.

action_history: list[chessai.core.action.Action]

The actions taken throughout the game.

The expected result from the PGN: '1-0', '0-1', '1/2-1/2', or '*'.

@classmethod
def from_pgn( cls, pgn: str, state_class: Type[chessai.core.gamestate.GameState], game_info: GameInfo, save_path: str | None = None, **kwargs: Any) -> Optional[Game]:
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.

def to_pgn( self, final_state: chessai.core.gamestate.GameState, termination_reason: chessai.core.types.TerminationReason) -> str:
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.

def process_args(self, args: argparse.Namespace) -> None:
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.

@abc.abstractmethod
def get_initial_state( self, rng: random.Random, fen: str | None = None) -> chessai.core.gamestate.GameState:
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.

def process_turn( self, state: chessai.core.gamestate.GameState, action_record: chessai.core.agentaction.AgentActionRecord, result: GameResult, rng: random.Random) -> chessai.core.gamestate.GameState:
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.

def check_end(self, state: chessai.core.gamestate.GameState) -> bool:
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.

def game_complete( self, state: chessai.core.gamestate.GameState, result: GameResult) -> None:
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.

def run(self, ui: chessai.core.ui.UI) -> GameResult:
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.

def set_cli_args( parser: argparse.ArgumentParser, default_board: str | None = None) -> argparse.ArgumentParser:
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.

def init_from_args( args: argparse.Namespace, game_class: Type[Game], state_class: Type[chessai.core.gamestate.GameState], base_agent_infos: dict[chessai.core.types.Color, chessai.core.agentinfo.AgentInfo] | None = None, **kwargs: Any) -> argparse.Namespace:
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.