chessai.core.game

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

A simple container that holds common information about a game.

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)
 34    def __init__(self,
 35            agent_infos: dict[chessai.core.types.Color, chessai.core.agentinfo.AgentInfo],
 36            start_fen: str | None = None,
 37            isolation_level: chessai.core.isolation.level.Level = chessai.core.isolation.level.Level.NONE,
 38            max_moves: int = DEFAULT_MAX_MOVES,
 39            agent_start_timeout: float = DEFAULT_AGENT_START_TIMEOUT,
 40            agent_end_timeout: float = DEFAULT_AGENT_END_TIMEOUT,
 41            agent_action_timeout: float = DEFAULT_AGENT_ACTION_TIMEOUT,
 42            seed: int | None = None,
 43            event: str | None = None,
 44            site: str | None = None,
 45            date: str | None = None,
 46            game_round: str | None = None,
 47            white_player: str | None = None,
 48            black_player: str | None = None,
 49            extra_info: dict[str, typing.Any] | None = None,
 50            ) -> None:
 51        if (seed is None):
 52            seed = random.randint(0, 2**64)
 53
 54        self.seed: int = seed
 55        """ The random seed for this game's RNG. """
 56
 57        self.agent_infos: dict[chessai.core.types.Color, chessai.core.agentinfo.AgentInfo] = agent_infos
 58        """ The required information for creating the agents for this game. """
 59
 60        if (len(self.agent_infos) == 0):
 61            raise ValueError("No agents provided.")
 62
 63        if (start_fen is None):
 64            start_fen = chessai.core.gamestate.DEFAULT_FEN
 65
 66        self.start_fen: str = start_fen
 67        """ The starting fen for this game. """
 68
 69        self.isolation_level: chessai.core.isolation.level.Level = isolation_level
 70        """ The isolation level to use for this game. """
 71
 72        self.max_moves: int = max_moves
 73        """
 74        The total number of moves (between all agents) allowed for this game.
 75        If -1, unlimited moves are allowed.
 76        """
 77
 78        self.agent_start_timeout: float = agent_start_timeout
 79        """
 80        The maximum number of seconds an agent is allowed when starting a game.
 81        If <= 0, unlimited time is allowed.
 82        """
 83
 84        self.agent_end_timeout: float = agent_end_timeout
 85        """
 86        The maximum number of seconds an agent is allowed when ending a game.
 87        If <= 0, unlimited time is allowed.
 88        """
 89
 90        self.agent_action_timeout: float = agent_action_timeout
 91        """
 92        The maximum number of seconds an agent is allowed when getting an action.
 93        If <= 0, unlimited time is allowed.
 94        """
 95
 96        if (event is None):
 97            event = '?'
 98
 99        self.event: str = event
100        """ The tournament or match event name. """
101
102        if (site is None):
103            site = '?'
104
105        self.site: str = site
106        """ The location of the event. """
107
108        if (date is None):
109            date = '????.??.??'
110
111        self.date: str = date
112        """ The starting date of the game. """
113
114        if (game_round is None):
115            game_round = '?'
116
117        self.game_round: str = game_round
118        """ The playing round ordinal. """
119
120        if (white_player is None):
121            white_player = '?'
122
123        self.white_player: str = white_player
124        """ The name of the white player. """
125
126        if (black_player is None):
127            black_player = '?'
128
129        self.black_player: str = black_player
130        """ The name of the black player. """
131
132        if (extra_info is None):
133            extra_info = {}
134
135        self.extra_info: dict[str, typing.Any] = extra_info
136        """ Any additional arguments passed to the game. """
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.

class GameResult(edq.util.serial.DictConverter):
138class GameResult(edq.util.serial.DictConverter):
139    """ The result of running a game. """
140
141    def __init__(self,
142            game_id: int,
143            game_info: GameInfo,
144            termination_reason: chessai.core.types.TerminationReason | None = None,
145            score: float = 0,
146            end_fen: str | None = None,
147            game_timeout: bool = False,
148            timeout_agent_teams: list[chessai.core.types.Color] | None = None,
149            crash_agent_teams: list[chessai.core.types.Color] | None = None,
150            winning_agent_teams: list[chessai.core.types.Color] | None = None,
151            start_time: edq.util.time.Timestamp | None = None,
152            end_time: edq.util.time.Timestamp | None = None,
153            history: list[chessai.core.agentaction.AgentActionRecord] | None = None,
154            agent_complete_records: dict[chessai.core.types.Color, chessai.core.agentaction.AgentActionRecord] | None = None,
155            **kwargs: typing.Any,
156            ) -> None:
157        self.game_id: int = game_id
158        """ The ID of the game result. """
159
160        self.game_info: GameInfo = game_info
161        """ The core information about this game. """
162
163        self.termination_reason: chessai.core.types.TerminationReason | None = termination_reason
164        """ The reason the game ended. """
165
166        self.score: float = score
167        """ The score of the game from white's perspective. """
168
169        self.end_fen = end_fen
170        """ The fen the board is in at the end of the game. """
171
172        if (start_time is None):
173            start_time = edq.util.time.Timestamp.now()
174
175        self.start_time: edq.util.time.Timestamp = start_time
176        """ The time the game started at. """
177
178        self.end_time: edq.util.time.Timestamp | None = end_time
179        """ The time the game ended at. """
180
181        if (history is None):
182            history = []
183
184        self.history: list[chessai.core.agentaction.AgentActionRecord] = history
185        """ The history of actions taken by each agent in this game. """
186
187        if (agent_complete_records is None):
188            agent_complete_records = {}
189
190        self.agent_complete_records: dict[chessai.core.types.Color, chessai.core.agentaction.AgentActionRecord] = agent_complete_records
191        """
192        The record recieved from an agent when the game finishes.
193        For agents that learn, this may include information that the agent learned this game.
194        """
195
196        self.game_timeout: bool = game_timeout
197        """ Indicates that the game has timed out (reached the maximum number of moves). """
198
199        if (timeout_agent_teams is None):
200            timeout_agent_teams = []
201
202        self.timeout_agent_teams: list[chessai.core.types.Color] = timeout_agent_teams
203        """ The list of agents that timed out in this game. """
204
205        if (crash_agent_teams is None):
206            crash_agent_teams = []
207
208        self.crash_agent_teams: list[chessai.core.types.Color] = crash_agent_teams
209        """ The list of agents that crashed in this game. """
210
211        if (winning_agent_teams is None):
212            winning_agent_teams = []
213
214        self.winning_agent_teams: list[chessai.core.types.Color] = winning_agent_teams
215        """
216        The list of agents that won this game.
217        An empty list indicates a tie.
218
219        Games may interpret this value in different ways.
220        """
221
222    def get_duration_secs(self) -> float:
223        """
224        Get the game's duration in seconds.
225        Will return positive infinity if the game has no end time
226        (it is still going or crashed (in very rare cases)).
227        """
228
229        if (self.end_time is None):
230            return math.inf
231
232        return self.end_time.sub(self.start_time).to_secs()

The result of running a game.

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)
141    def __init__(self,
142            game_id: int,
143            game_info: GameInfo,
144            termination_reason: chessai.core.types.TerminationReason | None = None,
145            score: float = 0,
146            end_fen: str | None = None,
147            game_timeout: bool = False,
148            timeout_agent_teams: list[chessai.core.types.Color] | None = None,
149            crash_agent_teams: list[chessai.core.types.Color] | None = None,
150            winning_agent_teams: list[chessai.core.types.Color] | None = None,
151            start_time: edq.util.time.Timestamp | None = None,
152            end_time: edq.util.time.Timestamp | None = None,
153            history: list[chessai.core.agentaction.AgentActionRecord] | None = None,
154            agent_complete_records: dict[chessai.core.types.Color, chessai.core.agentaction.AgentActionRecord] | None = None,
155            **kwargs: typing.Any,
156            ) -> None:
157        self.game_id: int = game_id
158        """ The ID of the game result. """
159
160        self.game_info: GameInfo = game_info
161        """ The core information about this game. """
162
163        self.termination_reason: chessai.core.types.TerminationReason | None = termination_reason
164        """ The reason the game ended. """
165
166        self.score: float = score
167        """ The score of the game from white's perspective. """
168
169        self.end_fen = end_fen
170        """ The fen the board is in at the end of the game. """
171
172        if (start_time is None):
173            start_time = edq.util.time.Timestamp.now()
174
175        self.start_time: edq.util.time.Timestamp = start_time
176        """ The time the game started at. """
177
178        self.end_time: edq.util.time.Timestamp | None = end_time
179        """ The time the game ended at. """
180
181        if (history is None):
182            history = []
183
184        self.history: list[chessai.core.agentaction.AgentActionRecord] = history
185        """ The history of actions taken by each agent in this game. """
186
187        if (agent_complete_records is None):
188            agent_complete_records = {}
189
190        self.agent_complete_records: dict[chessai.core.types.Color, chessai.core.agentaction.AgentActionRecord] = agent_complete_records
191        """
192        The record recieved from an agent when the game finishes.
193        For agents that learn, this may include information that the agent learned this game.
194        """
195
196        self.game_timeout: bool = game_timeout
197        """ Indicates that the game has timed out (reached the maximum number of moves). """
198
199        if (timeout_agent_teams is None):
200            timeout_agent_teams = []
201
202        self.timeout_agent_teams: list[chessai.core.types.Color] = timeout_agent_teams
203        """ The list of agents that timed out in this game. """
204
205        if (crash_agent_teams is None):
206            crash_agent_teams = []
207
208        self.crash_agent_teams: list[chessai.core.types.Color] = crash_agent_teams
209        """ The list of agents that crashed in this game. """
210
211        if (winning_agent_teams is None):
212            winning_agent_teams = []
213
214        self.winning_agent_teams: list[chessai.core.types.Color] = winning_agent_teams
215        """
216        The list of agents that won this game.
217        An empty list indicates a tie.
218
219        Games may interpret this value in different ways.
220        """
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:
222    def get_duration_secs(self) -> float:
223        """
224        Get the game's duration in seconds.
225        Will return positive infinity if the game has no end time
226        (it is still going or crashed (in very rare cases)).
227        """
228
229        if (self.end_time is None):
230            return math.inf
231
232        return self.end_time.sub(self.start_time).to_secs()

Get the game's duration in seconds. Will return positive infinity if the game has no end time (it is still going or crashed (in very rare cases)).

class Game(abc.ABC):
234class Game(abc.ABC):
235    """
236    A game that can be run in chessai.
237    Games combine the rules, layouts, and agents to run.
238    """
239
240    def __init__(self,
241                 game_info: GameInfo,
242                 save_path: str | None = None,
243                 is_replay: bool = False,
244                 initial_actions: list[chessai.core.action.Action] | None = None,
245                 action_history: list[chessai.core.action.Action] | None = None,
246                 expected_result: chessai.core.gameparser.PGNResult = chessai.core.gameparser.PGNResult.UNKNOWN,
247                 ) -> None:
248        self.game_info: GameInfo = game_info
249        """ The core information about this game. """
250
251        self._save_path: str | None = save_path
252        """ Where to save the results of this game. """
253
254        self._is_replay: bool = is_replay
255        """
256        Indicates that this game is being loaded from a replay.
257        Some behavior, like saving the result, will be modified.
258        """
259
260        if (initial_actions is None):
261            initial_actions = []
262
263        self.initial_actions: list[chessai.core.action.Action] = initial_actions
264        """ The moves to be played at the beginning of the game. """
265
266        if (action_history is None):
267            action_history = []
268
269        self.action_history: list[chessai.core.action.Action] = action_history
270        """ The actions taken throughout the game. """
271
272        self.expected_result: chessai.core.gameparser.PGNResult = expected_result
273        """ The expected result from the PGN: '1-0', '0-1', '1/2-1/2', or '*'. """
274
275    @classmethod
276    def from_pgn(cls,
277                 pgn: str,
278                 state_class: typing.Type[chessai.core.gamestate.GameState],
279                 game_info: GameInfo,
280                 save_path: str | None = None,
281                 **kwargs: typing.Any) -> typing.Optional['Game']:
282        """
283        Create a Game from a PGN.
284
285        The game will be initialized with metadata from the PGN headers and will store the moves for replay.
286
287        Returns the game and whether the PGN parsing was successful.
288        """
289
290        parsed_pgn = chessai.core.gameparser.parse_pgn(pgn, state_class)
291        if (parsed_pgn is None):
292            return None
293
294        # Extract starting FEN from optional headers or use default.
295        start_fen = parsed_pgn.get_starting_fen()
296        if (start_fen is not None):
297            game_info.start_fen = start_fen
298
299        # Extract header information into the game info.
300        game_info.event = parsed_pgn.headers.event
301        game_info.site = parsed_pgn.headers.site
302        game_info.date = parsed_pgn.headers.date
303        game_info.game_round = parsed_pgn.headers.round
304        game_info.white_player = parsed_pgn.headers.white
305        game_info.black_player = parsed_pgn.headers.black
306        game_info.extra_info.update(parsed_pgn.optional_headers)
307
308        return cls(
309            game_info = game_info,
310            save_path = save_path,
311            is_replay = True,
312            initial_actions = parsed_pgn.initial_actions,
313            expected_result = parsed_pgn.result,
314            **kwargs
315        )
316
317    def to_pgn(self,
318            final_state: chessai.core.gamestate.GameState,
319            termination_reason: chessai.core.types.TerminationReason,
320            ) -> str:
321        """ Returns the PGN representation of the Game. """
322
323        if (termination_reason in [chessai.core.types.TerminationReason.CHECKMATE,
324                                   chessai.core.types.TerminationReason.FORFEIT]):
325            winners = final_state.get_winners()
326            if (chessai.core.types.Color.WHITE in winners):
327                result = chessai.core.gameparser.PGNResult.WHITE_WIN
328            elif (chessai.core.types.Color.BLACK in winners):
329                result = chessai.core.gameparser.PGNResult.BLACK_WIN
330            else:
331                result = chessai.core.gameparser.PGNResult.TIE
332        elif (termination_reason in [chessai.core.types.TerminationReason.STALEMATE,
333                                     chessai.core.types.TerminationReason.INSUFFICIENT_MATERIAL,
334                                     chessai.core.types.TerminationReason.ACCEPTED_DRAW_PROPOSAL]):
335            result = chessai.core.gameparser.PGNResult.TIE
336        elif (termination_reason == chessai.core.types.TerminationReason.IN_PROGRESS):
337            result = chessai.core.gameparser.PGNResult.IN_PROGRESS
338        else:
339            result = chessai.core.gameparser.PGNResult.UNKNOWN
340
341        headers = chessai.core.gameparser.StandardHeaders(
342                event = self.game_info.event,
343                site = self.game_info.site,
344                date = self.game_info.date,
345                game_round = self.game_info.game_round,
346                white = self.game_info.white_player,
347                black = self.game_info.black_player,
348                result = result)
349
350        return chessai.core.gameparser.to_pgn(headers, self.game_info.extra_info, type(final_state),
351                                              self.game_info.start_fen, self.action_history)
352
353    def process_args(self, args: argparse.Namespace) -> None:
354        """ Process any special arguments from the command-line. """
355
356    @abc.abstractmethod
357    def get_initial_state(self,
358            rng: random.Random,
359            fen: str | None = None) -> chessai.core.gamestate.GameState:
360        """ Create the initial state for this game. """
361
362    def process_turn(self,
363                     state: chessai.core.gamestate.GameState,
364                     action_record: chessai.core.agentaction.AgentActionRecord,
365                     result: GameResult,
366                     rng: random.Random) -> chessai.core.gamestate.GameState:
367        """
368        Process the given agent action and return an updated game state.
369        The returned game state may be a copy or modified version of the passed in game state.
370        """
371
372        # The agent has timed out.
373        if (action_record.timeout):
374            result.timeout_agent_teams.append(action_record.player)
375            result.termination_reason = chessai.core.types.TerminationReason.AGENT_TIMEOUT
376            state.process_agent_timeout(action_record.player)
377            return state
378
379        # The agent has crashed.
380        if (action_record.crashed):
381            result.crash_agent_teams.append(action_record.player)
382            result.termination_reason = chessai.core.types.TerminationReason.AGENT_CRASHED
383            state.process_agent_crash(action_record.player)
384            return state
385
386        action = action_record.get_action()
387        if (action not in state.get_legal_actions()):
388            raise ValueError(f"Illegal action for agent {action_record.player}: '{action.uci()}' of type '{type(action)}'.")
389
390        # Add the action to the action history.
391        self.action_history.append(action)
392
393        self._call_state_process_turn_full(state, action, rng)
394
395        return state
396
397    def _call_state_process_turn_full(self,
398            state: chessai.core.gamestate.GameState,
399            action: chessai.core.action.Action,
400            rng: random.Random,
401            **kwargs: typing.Any) -> None:
402        """ Call on the game state to process a full turn. """
403
404        state.process_turn_full(action, rng, **kwargs)
405
406    def check_end(self, state: chessai.core.gamestate.GameState) -> bool:
407        """
408        Check to see if the game is over.
409        Return True if the game is now over, False otherwise.
410
411        By default, this will check chessai.core.gamestate.GameState.game_over
412        and chessai.core.gamestate.GameState.is_game_over(),
413        but child games can override for more complex functionality.
414        """
415
416        return (state.game_over or state.is_game_over())
417
418    def game_complete(self, state: chessai.core.gamestate.GameState, result: GameResult) -> None:
419        """
420        Make any last adjustments to the game result after the game is over.
421        """
422
423    def run(self, ui: chessai.core.ui.UI) -> GameResult:
424        """
425        The main "game loop" for all games.
426        """
427
428        logging.debug("Starting a game with seed: %d.", self.game_info.seed)
429
430        # Create a new random number generator just for this game.
431        rng = random.Random(self.game_info.seed)
432
433        # Keep track of what happens during this game.
434        game_id = rng.randint(0, 2**64)
435        result = GameResult(game_id, self.game_info)
436
437        # Initialize the agent isolator.
438        isolator = self.game_info.isolation_level.get_isolator()
439        isolator.init_agents(self.game_info.agent_infos)
440
441        # Create the initial game state (and force it's seed).
442        state = self.get_initial_state(rng, self.game_info.start_fen)
443        state.seed = game_id
444        state.game_start()
445
446        # board_highlights: list[chessai.core.board.Highlight] = []
447
448        # Notify agents about the start of the game.
449        records = isolator.game_start(rng, state, self.game_info.agent_start_timeout)
450        for record in records.values():
451            if (record.timeout):
452                result.timeout_agent_teams.append(record.player)
453                state.process_agent_timeout(record.player)
454                result.termination_reason = chessai.core.types.TerminationReason.AGENT_TIMEOUT
455            elif (record.crashed):
456                result.crash_agent_teams.append(record.player)
457                state.process_agent_crash(record.player)
458                result.termination_reason = chessai.core.types.TerminationReason.AGENT_CRASHED
459            else:
460                continue
461                # board_highlights += record.get_board_highlights()
462
463        state.agents_game_start(records)
464
465        # Start the UI.
466        ui.game_start(state)
467
468        while (not self.check_end(state)):
469            logging.trace("Turn %d, agent %s.", state.fullmove_number, state.turn) # type: ignore[attr-defined]  # pylint: disable=no-member
470
471            if (len(self.initial_actions) > 0):
472                # Get the action from the pre-loaded actions.
473                action = self.initial_actions.pop(0)
474                agent_action = chessai.core.agentaction.AgentAction(action = action)
475                duration = edq.util.time.Duration(0)
476
477                action_record = chessai.core.agentaction.AgentActionRecord(state.turn, agent_action, duration)
478            else:
479                # Get the next action from the agent.
480                action_record = isolator.get_action(state, self.game_info.agent_action_timeout)
481
482            # Execute the next action and update the state.
483            state = self.process_turn(state, action_record, result, rng)
484
485            # Update the UI.
486            ui.update(state)
487
488            # Update the game result and move history.
489            result.history.append(action_record)
490
491            # Check for game ending conditions.
492            if (self.check_end(state)):
493                break
494
495            # Check if this game has ran for the maximum number of moves.
496            if ((self.game_info.max_moves > 0) and (len(result.history) >= self.game_info.max_moves)):
497                state.process_game_timeout()
498                result.game_timeout = True
499                result.termination_reason = chessai.core.types.TerminationReason.GAME_TIMEOUT
500                break
501
502        # Mark the end time of the game.
503        result.end_time = edq.util.time.Timestamp.now()
504
505        # Notify the state about the end of the game.
506        (winners, score) = state.game_complete()
507        result.winning_agent_teams += winners
508        result.score = score
509
510        if (result.termination_reason is None):
511            result.termination_reason = state.get_termination_reason()
512
513        result.end_fen = state.get_fen()
514
515        # Notify agents about the end of this game.
516        result.agent_complete_records = isolator.game_complete(state, self.game_info.agent_end_timeout)
517
518        # All the game to make final updates to the result.
519        self.game_complete(state, result)
520
521        # Update the UI.
522        ui.game_complete(state, result.termination_reason)
523
524        # Cleanup
525        isolator.close()
526        ui.close()
527
528        if ((not self._is_replay) and (self._save_path is not None)):
529            logging.info("Saving results to '%s'.", self._save_path)
530
531            pgn = self.to_pgn(state, result.termination_reason)
532            edq.util.dirent.write_file(self._save_path, pgn)
533
534        return result

A game that can be run in chessai. Games combine the rules, layouts, and agents to run.

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]:
275    @classmethod
276    def from_pgn(cls,
277                 pgn: str,
278                 state_class: typing.Type[chessai.core.gamestate.GameState],
279                 game_info: GameInfo,
280                 save_path: str | None = None,
281                 **kwargs: typing.Any) -> typing.Optional['Game']:
282        """
283        Create a Game from a PGN.
284
285        The game will be initialized with metadata from the PGN headers and will store the moves for replay.
286
287        Returns the game and whether the PGN parsing was successful.
288        """
289
290        parsed_pgn = chessai.core.gameparser.parse_pgn(pgn, state_class)
291        if (parsed_pgn is None):
292            return None
293
294        # Extract starting FEN from optional headers or use default.
295        start_fen = parsed_pgn.get_starting_fen()
296        if (start_fen is not None):
297            game_info.start_fen = start_fen
298
299        # Extract header information into the game info.
300        game_info.event = parsed_pgn.headers.event
301        game_info.site = parsed_pgn.headers.site
302        game_info.date = parsed_pgn.headers.date
303        game_info.game_round = parsed_pgn.headers.round
304        game_info.white_player = parsed_pgn.headers.white
305        game_info.black_player = parsed_pgn.headers.black
306        game_info.extra_info.update(parsed_pgn.optional_headers)
307
308        return cls(
309            game_info = game_info,
310            save_path = save_path,
311            is_replay = True,
312            initial_actions = parsed_pgn.initial_actions,
313            expected_result = parsed_pgn.result,
314            **kwargs
315        )

Create a Game from a PGN.

The game will be initialized with metadata from the PGN headers and will store the moves for replay.

Returns the game and whether the PGN parsing was successful.

def to_pgn( self, final_state: chessai.core.gamestate.GameState, termination_reason: chessai.core.types.TerminationReason) -> str:
317    def to_pgn(self,
318            final_state: chessai.core.gamestate.GameState,
319            termination_reason: chessai.core.types.TerminationReason,
320            ) -> str:
321        """ Returns the PGN representation of the Game. """
322
323        if (termination_reason in [chessai.core.types.TerminationReason.CHECKMATE,
324                                   chessai.core.types.TerminationReason.FORFEIT]):
325            winners = final_state.get_winners()
326            if (chessai.core.types.Color.WHITE in winners):
327                result = chessai.core.gameparser.PGNResult.WHITE_WIN
328            elif (chessai.core.types.Color.BLACK in winners):
329                result = chessai.core.gameparser.PGNResult.BLACK_WIN
330            else:
331                result = chessai.core.gameparser.PGNResult.TIE
332        elif (termination_reason in [chessai.core.types.TerminationReason.STALEMATE,
333                                     chessai.core.types.TerminationReason.INSUFFICIENT_MATERIAL,
334                                     chessai.core.types.TerminationReason.ACCEPTED_DRAW_PROPOSAL]):
335            result = chessai.core.gameparser.PGNResult.TIE
336        elif (termination_reason == chessai.core.types.TerminationReason.IN_PROGRESS):
337            result = chessai.core.gameparser.PGNResult.IN_PROGRESS
338        else:
339            result = chessai.core.gameparser.PGNResult.UNKNOWN
340
341        headers = chessai.core.gameparser.StandardHeaders(
342                event = self.game_info.event,
343                site = self.game_info.site,
344                date = self.game_info.date,
345                game_round = self.game_info.game_round,
346                white = self.game_info.white_player,
347                black = self.game_info.black_player,
348                result = result)
349
350        return chessai.core.gameparser.to_pgn(headers, self.game_info.extra_info, type(final_state),
351                                              self.game_info.start_fen, self.action_history)

Returns the PGN representation of the Game.

def process_args(self, args: argparse.Namespace) -> None:
353    def process_args(self, args: argparse.Namespace) -> None:
354        """ Process any special arguments from the command-line. """

Process any special arguments from the command-line.

@abc.abstractmethod
def get_initial_state( self, rng: random.Random, fen: str | None = None) -> chessai.core.gamestate.GameState:
356    @abc.abstractmethod
357    def get_initial_state(self,
358            rng: random.Random,
359            fen: str | None = None) -> chessai.core.gamestate.GameState:
360        """ Create the initial state for this game. """

Create the initial state for this game.

def process_turn( self, state: chessai.core.gamestate.GameState, action_record: chessai.core.agentaction.AgentActionRecord, result: GameResult, rng: random.Random) -> chessai.core.gamestate.GameState:
362    def process_turn(self,
363                     state: chessai.core.gamestate.GameState,
364                     action_record: chessai.core.agentaction.AgentActionRecord,
365                     result: GameResult,
366                     rng: random.Random) -> chessai.core.gamestate.GameState:
367        """
368        Process the given agent action and return an updated game state.
369        The returned game state may be a copy or modified version of the passed in game state.
370        """
371
372        # The agent has timed out.
373        if (action_record.timeout):
374            result.timeout_agent_teams.append(action_record.player)
375            result.termination_reason = chessai.core.types.TerminationReason.AGENT_TIMEOUT
376            state.process_agent_timeout(action_record.player)
377            return state
378
379        # The agent has crashed.
380        if (action_record.crashed):
381            result.crash_agent_teams.append(action_record.player)
382            result.termination_reason = chessai.core.types.TerminationReason.AGENT_CRASHED
383            state.process_agent_crash(action_record.player)
384            return state
385
386        action = action_record.get_action()
387        if (action not in state.get_legal_actions()):
388            raise ValueError(f"Illegal action for agent {action_record.player}: '{action.uci()}' of type '{type(action)}'.")
389
390        # Add the action to the action history.
391        self.action_history.append(action)
392
393        self._call_state_process_turn_full(state, action, rng)
394
395        return state

Process the given agent action and return an updated game state. The returned game state may be a copy or modified version of the passed in game state.

def check_end(self, state: chessai.core.gamestate.GameState) -> bool:
406    def check_end(self, state: chessai.core.gamestate.GameState) -> bool:
407        """
408        Check to see if the game is over.
409        Return True if the game is now over, False otherwise.
410
411        By default, this will check chessai.core.gamestate.GameState.game_over
412        and chessai.core.gamestate.GameState.is_game_over(),
413        but child games can override for more complex functionality.
414        """
415
416        return (state.game_over or state.is_game_over())

Check to see if the game is over. Return True if the game is now over, False otherwise.

By default, this will check chessai.core.gamestate.GameState.game_over and chessai.core.gamestate.GameState.is_game_over(), but child games can override for more complex functionality.

def game_complete( self, state: chessai.core.gamestate.GameState, result: GameResult) -> None:
418    def game_complete(self, state: chessai.core.gamestate.GameState, result: GameResult) -> None:
419        """
420        Make any last adjustments to the game result after the game is over.
421        """

Make any last adjustments to the game result after the game is over.

def run(self, ui: chessai.core.ui.UI) -> GameResult:
423    def run(self, ui: chessai.core.ui.UI) -> GameResult:
424        """
425        The main "game loop" for all games.
426        """
427
428        logging.debug("Starting a game with seed: %d.", self.game_info.seed)
429
430        # Create a new random number generator just for this game.
431        rng = random.Random(self.game_info.seed)
432
433        # Keep track of what happens during this game.
434        game_id = rng.randint(0, 2**64)
435        result = GameResult(game_id, self.game_info)
436
437        # Initialize the agent isolator.
438        isolator = self.game_info.isolation_level.get_isolator()
439        isolator.init_agents(self.game_info.agent_infos)
440
441        # Create the initial game state (and force it's seed).
442        state = self.get_initial_state(rng, self.game_info.start_fen)
443        state.seed = game_id
444        state.game_start()
445
446        # board_highlights: list[chessai.core.board.Highlight] = []
447
448        # Notify agents about the start of the game.
449        records = isolator.game_start(rng, state, self.game_info.agent_start_timeout)
450        for record in records.values():
451            if (record.timeout):
452                result.timeout_agent_teams.append(record.player)
453                state.process_agent_timeout(record.player)
454                result.termination_reason = chessai.core.types.TerminationReason.AGENT_TIMEOUT
455            elif (record.crashed):
456                result.crash_agent_teams.append(record.player)
457                state.process_agent_crash(record.player)
458                result.termination_reason = chessai.core.types.TerminationReason.AGENT_CRASHED
459            else:
460                continue
461                # board_highlights += record.get_board_highlights()
462
463        state.agents_game_start(records)
464
465        # Start the UI.
466        ui.game_start(state)
467
468        while (not self.check_end(state)):
469            logging.trace("Turn %d, agent %s.", state.fullmove_number, state.turn) # type: ignore[attr-defined]  # pylint: disable=no-member
470
471            if (len(self.initial_actions) > 0):
472                # Get the action from the pre-loaded actions.
473                action = self.initial_actions.pop(0)
474                agent_action = chessai.core.agentaction.AgentAction(action = action)
475                duration = edq.util.time.Duration(0)
476
477                action_record = chessai.core.agentaction.AgentActionRecord(state.turn, agent_action, duration)
478            else:
479                # Get the next action from the agent.
480                action_record = isolator.get_action(state, self.game_info.agent_action_timeout)
481
482            # Execute the next action and update the state.
483            state = self.process_turn(state, action_record, result, rng)
484
485            # Update the UI.
486            ui.update(state)
487
488            # Update the game result and move history.
489            result.history.append(action_record)
490
491            # Check for game ending conditions.
492            if (self.check_end(state)):
493                break
494
495            # Check if this game has ran for the maximum number of moves.
496            if ((self.game_info.max_moves > 0) and (len(result.history) >= self.game_info.max_moves)):
497                state.process_game_timeout()
498                result.game_timeout = True
499                result.termination_reason = chessai.core.types.TerminationReason.GAME_TIMEOUT
500                break
501
502        # Mark the end time of the game.
503        result.end_time = edq.util.time.Timestamp.now()
504
505        # Notify the state about the end of the game.
506        (winners, score) = state.game_complete()
507        result.winning_agent_teams += winners
508        result.score = score
509
510        if (result.termination_reason is None):
511            result.termination_reason = state.get_termination_reason()
512
513        result.end_fen = state.get_fen()
514
515        # Notify agents about the end of this game.
516        result.agent_complete_records = isolator.game_complete(state, self.game_info.agent_end_timeout)
517
518        # All the game to make final updates to the result.
519        self.game_complete(state, result)
520
521        # Update the UI.
522        ui.game_complete(state, result.termination_reason)
523
524        # Cleanup
525        isolator.close()
526        ui.close()
527
528        if ((not self._is_replay) and (self._save_path is not None)):
529            logging.info("Saving results to '%s'.", self._save_path)
530
531            pgn = self.to_pgn(state, result.termination_reason)
532            edq.util.dirent.write_file(self._save_path, pgn)
533
534        return result

The main "game loop" for all games.

def set_cli_args( parser: argparse.ArgumentParser, default_board: str | None = None) -> argparse.ArgumentParser:
536def set_cli_args(parser: argparse.ArgumentParser, default_board: str | None = None) -> argparse.ArgumentParser:
537    """
538    Set common CLI arguments.
539    This is a sibling to init_from_args(), as the arguments set here can be interpreted there.
540    """
541
542    parser.add_argument('--board', dest = 'board',
543            action = 'store', type = str, default = default_board,
544            help = ('Play on this board (default: %(default)s).'
545                    + ' This may be a FEN of the board.'
546                    + ' It may also be the full path to a board, or just a filename.'
547                    + ' If just a filename, than the `chessai/resources/boards` directory will be checked (using a ".fen" extension.'))
548
549    parser.add_argument('--num-games', dest = 'num_games',
550            action = 'store', type = int, default = 1,
551            help = 'The number of games to play (default: %(default)s).')
552
553    parser.add_argument('--seed', dest = 'seed',
554            action = 'store', type = int, default = None,
555            help = 'The random seed for the game (will be randomly generated if not set).')
556
557    parser.add_argument('--max-moves', dest = 'max_moves',
558            action = 'store', type = int, default = DEFAULT_MAX_MOVES,
559            help = 'The maximum number of moves (total for all agents) allowed in this game (-1 for unlimited) (default: %(default)s).')
560
561    parser.add_argument('--agent-start-timeout', dest = 'agent_start_timeout',
562            action = 'store', type = float, default = DEFAULT_AGENT_START_TIMEOUT,
563            help = ('The maximum number of seconds each agent is allowed when starting a game (<= 0 for unlimited time) (default: %(default)s).'
564                    + ' Note that the "none" isolation level cannot enforce timeouts.'))
565
566    parser.add_argument('--agent-end-timeout', dest = 'agent_end_timeout',
567            action = 'store', type = float, default = DEFAULT_AGENT_END_TIMEOUT,
568            help = ('The maximum number of seconds each agent is allowed when ending a game (<= 0 for unlimited time) (default: %(default)s).'
569                    + ' Note that the "none" isolation level cannot enforce timeouts.'))
570
571    parser.add_argument('--agent-action-timeout', dest = 'agent_action_timeout',
572            action = 'store', type = float, default = DEFAULT_AGENT_ACTION_TIMEOUT,
573            help = ('The maximum number of seconds each agent is allowed when getting an action (<= 0 for unlimited time) (default: %(default)s).'
574                    + ' Note that the "none" isolation level cannot enforce timeouts.'))
575
576    parser.add_argument('--isolation', dest = 'isolation_level', metavar = 'LEVEL',
577            action = 'store', type = str, default = chessai.core.isolation.level.Level.NONE.value,
578            choices = chessai.core.isolation.level.LEVELS,
579            help = ('Set the agent isolation level for this game (default: %(default)s).'
580                    + ' Choose one of:'
581                    + ' `none` -- Do not make any attempt to isolate the agent code from the game (fastest and least secure),'
582                    + ' `process` -- Run the agent code in a separate process'
583                    + ' (offers some protection, but still vulnerable to disk or execution exploits),'
584                    + ' `tcp` -- Open TCP listeners to communicate with agents (most secure, requires additional work to set up agents).'))
585
586    parser.add_argument('--agent-arg', dest = 'raw_agent_args', metavar = 'ARG',
587            action = 'append', type = str, default = [],
588            help = ('Specify arguments directly to agents (may be used multiple times).'
589                    + ' The value for this argument must be formatted as "player::key=value",'
590                    + ' for example to set `foo = 9` for white and `bar = a` for black, we can use:'
591                    + ' `--agent-arg white::foo=9 --agent-arg black::bar=a`.'))
592
593    parser.add_argument('--save-path', dest = 'save_path',
594            action = 'store', type = str, default = None,
595            help = ('If specified, write the result of this game to the specified location.'
596                    + ' This file can be replayed with `--replay-path`.'))
597
598    parser.add_argument('--replay-path', dest = 'replay_path',
599            action = 'store', type = str, default = None,
600            help = ('If specified, replay the game whose result was saved at the specified path with `--save-path`.'
601                    + ' This may be a PGN of the game.'
602                    + ' It may also be the full path to a game, or just a filename.'
603                    + ' If just a filename, than the `chessai/resources/games` directory will be checked (using a ".pgn" extension.'))
604
605    return parser

Set common CLI arguments. This is a sibling to init_from_args(), as the arguments set here can be interpreted there.

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:
607def init_from_args(
608        args: argparse.Namespace,
609        game_class: typing.Type[Game],
610        state_class: typing.Type[chessai.core.gamestate.GameState],
611        base_agent_infos: dict[chessai.core.types.Color, chessai.core.agentinfo.AgentInfo] | None = None,
612        **kwargs: typing.Any,
613        ) -> argparse.Namespace:
614    """
615    Take in args from a parser that was passed to set_cli_args(),
616    and initialize the proper components.
617    This will create a number of games (and related resources)
618    based on `--num-games` + `--num-training`.
619    Each of these resources will be placed in their respective list at
620    `args._fens`, `args._agent_infos`, or `args._games`.
621    """
622
623    if (base_agent_infos is None):
624        base_agent_infos = {}
625
626    if (args.num_games <= 0):
627        raise ValueError(f"At least one game must be played (--num-games), {args.num_games} was specified.")
628
629    # Establish an RNG to generate seeds for each game using the given seed.
630    seed = args.seed
631    if (seed is None):
632        seed = random.randint(0, 2**64)
633
634    logging.debug("Using source seed for games: %d.", seed)
635    rng = random.Random(seed)
636
637    agents = [chessai.core.types.Color.WHITE, chessai.core.types.Color.BLACK]
638    agent_infos = _parse_agent_infos(agents, args.raw_agent_args, base_agent_infos)
639
640    base_save_path = args.save_path
641
642    all_fens: list[str] = []
643    all_agent_infos = []
644    all_games = []
645
646    for i in range(args.num_games):
647        game_seed = rng.randint(0, 2**64)
648
649        all_agent_infos.append(copy.deepcopy(agent_infos))
650        all_fens.append(args.board)
651
652        white_agent_info = all_agent_infos[-1].get(chessai.core.types.Color.WHITE, None)
653        if (white_agent_info is not None):
654            white_reference = white_agent_info.name
655            if (isinstance(white_reference, chessai.util.reflection.Reference)):
656                white_player = str(white_reference)
657            else:
658                white_player = white_reference
659        else:
660            white_player = '?'
661
662        black_agent_info = all_agent_infos[-1].get(chessai.core.types.Color.BLACK, None)
663        if (black_agent_info is not None):
664            black_reference = black_agent_info.name
665            if (isinstance(black_reference, chessai.util.reflection.Reference)):
666                black_player = str(black_reference)
667            else:
668                black_player = black_reference
669        else:
670            black_player = '?'
671
672        game_info = GameInfo(
673            all_agent_infos[-1],
674            start_fen = all_fens[-1],
675            isolation_level = chessai.core.isolation.level.Level(args.isolation_level),
676            max_moves = args.max_moves,
677            agent_start_timeout = args.agent_start_timeout,
678            agent_end_timeout = args.agent_end_timeout,
679            agent_action_timeout = args.agent_action_timeout,
680            seed = game_seed,
681            event = 'Casual Chessai Game',
682            site = 'https://github.com/ucsc-cse-240/chessai',
683            date = datetime.date.today().isoformat(),
684            game_round = str(i),
685            white_player = white_player,
686            black_player = black_player,
687        )
688
689        # Suffix the save path if there is more than one game.
690        save_path = base_save_path
691        if ((save_path is not None) and (args.num_games > 1)):
692            parts = os.path.splitext(save_path)
693            save_path = f"{parts[0]}_{i:03d}{parts[1]}"
694
695        if (args.replay_path is None):
696            game_args = {
697                'game_info': game_info,
698                'save_path': save_path,
699            }
700
701            game = game_class(**game_args)
702        else:
703            raw_game = game_class.from_pgn(args.replay_path, state_class, game_info, save_path)
704            if (raw_game is None):
705                raise ValueError(f"Failed to initialize game number {i} from the PGN: '{args.replay_path}'.")
706
707            game = raw_game
708
709        game.process_args(args)
710
711        all_games.append(game)
712
713    setattr(args, '_fens', all_fens)
714    setattr(args, '_agent_infos', all_agent_infos)
715    setattr(args, '_games', all_games)
716
717    return args

Take in args from a parser that was passed to set_cli_args(), and initialize the proper components. This will create a number of games (and related resources) based on --num-games + --num-training. Each of these resources will be placed in their respective list at args._fens, args._agent_infos, or args._games.