##########################################################################################
########################################## INFO ##########################################
##########################################################################################
# This file is part of the PyRat library.
# It is meant to be used as a library, and not to be executed directly.
# Please import necessary elements using the following syntax:
# from pyrat import <element_name>
"""
This module provides the main game loop and manages the game state.
It initializes the game with the specified parameters, manages players, and handles the game rendering.
It is the core of the PyRat game engine, allowing players to interact with the maze and each other.
"""
##########################################################################################
######################################### IMPORTS ########################################
##########################################################################################
# External imports
import copy
import math
import multiprocessing
import multiprocessing.managers as mpmanagers
import time
import traceback
import sys
import os
import datetime
import random
# PyRat imports
from pyrat.src.mazes.maze import Maze
from pyrat.src.mazes.holes_on_side_random_maze import HolesOnSideRandomMaze
from pyrat.src.mazes.uniform_holes_random_maze import UniformHolesRandomMaze
from pyrat.src.mazes.big_holes_random_maze import BigHolesRandomMaze
from pyrat.src.mazes.maze_from_dict import MazeFromDict
from pyrat.src.mazes.maze_from_matrix import MazeFromMatrix
from pyrat.src.players.player import Player
from pyrat.src.game.game_state import GameState
from pyrat.src.rendering.rendering_engine import RenderingEngine
from pyrat.src.rendering.shell_rendering_engine import ShellRenderingEngine
from pyrat.src.rendering.pygame_rendering_engine import PygameRenderingEngine
from pyrat.src.game.enums import RenderMode, GameMode, Action, StartingLocation, PlayerSkin, RandomMazeAlgorithm
from pyrat.src.game.exceptions import PyRatException, PyRatPlayerException
from pyrat.src.game.main_guard import check_main_guard
##########################################################################################
######################################### CLASSES ########################################
##########################################################################################
[docs]
class Game ():
"""
A game is a class that allows to play a game of PyRat.
It is initialized with the parameters of the game.
Players should then be added to the game using the ``add_player()`` method.
Finally, the ``start()`` method should be called to start the game.
Once the game is over, it will provide statistics about the game.
Set your own parameters to define interesting objectives for the players.
However, make sure that parameters make sense together.
For instance, you cannot set a number of cheese that is larger than the number of cells in the maze.
To ease creating games, we provide a set of default parameters.
These are used when the parameters are not set in the constructor:
"""
DEFAULT_RANDOM_SEED = None
DEFAULT_RANDOM_SEED_MAZE = None
DEFAULT_RANDOM_SEED_CHEESE = None
DEFAULT_RANDOM_SEED_PLAYERS = None
DEFAULT_MAZE_WIDTH = 15
DEFAULT_MAZE_HEIGHT = 13
DEFAULT_CELL_PERCENTAGE = 80.0
DEFAULT_WALL_PERCENTAGE = 60.0
DEFAULT_MUD_PERCENTAGE = 20.0
DEFAULT_MUD_RANGE = (4, 9)
DEFAULT_FIXED_MAZE = None
DEFAULT_RANDOM_MAZE_ALGORITHM = RandomMazeAlgorithm.BIG_HOLES
DEFAULT_NB_CHEESE = 21
DEFAULT_FIXED_CHEESE = None
DEFAULT_RENDER_MODE = RenderMode.GUI
DEFAULT_RENDER_SIMPLIFIED = False
DEFAULT_RENDERING_SPEED = 1.0
DEFAULT_TRACE_LENGTH = 0
DEFAULT_FULLSCREEN = False
DEFAULT_CLEAR_SHELL_EACH_TURN = True
DEFAULT_SAVE_PATH = "."
DEFAULT_SAVE_GAME = False
DEFAULT_PREPROCESSING_TIME = 3.0
DEFAULT_TURN_TIME = 0.1
DEFAULT_GAME_MODE_SINGLE_TEAM = GameMode.SEQUENTIAL
DEFAULT_GAME_MODE_MULTI_TEAM = GameMode.MATCH
DEFAULT_CONTINUE_ON_ERROR = False
##################################################################################
# CONSTRUCTOR #
##################################################################################
[docs]
def __init__ ( self,
random_seed: int | None = None,
random_seed_maze: int | None = None,
random_seed_cheese: int | None = None,
random_seed_players: int | None = None,
maze_width: int | None = None,
maze_height: int | None = None,
cell_percentage: float | None = None,
wall_percentage: float | None = None,
mud_percentage: float | None = None,
mud_range: tuple[int, int] | None = None,
fixed_maze: Maze | dict[int, dict[int, int]] | object | None = None, # Object for numpy.ndarray or torch.tensor (not listed to have an optional dependency on these libraries)
nb_cheese: int | None = None,
fixed_cheese: list[int] | None = None,
random_maze_algorithm: RandomMazeAlgorithm | None = None,
render_mode: RenderMode | None = None,
render_simplified: bool | None = None,
rendering_speed: float | None = None,
trace_length: int | None = None,
fullscreen: bool | None = None,
clear_shell_each_turn: bool | None = None,
save_path: str | None = None,
save_game: bool | None = None,
preprocessing_time: float | None = None,
turn_time: float | None = None,
game_mode: GameMode | None = None,
continue_on_error: bool | None = None
) -> None:
"""
Initializes a new instance of the class.
Any unset parameter will be set to its default value defined in the class.
Args:
random_seed: Global random seed for all elements, or ``None`` for a random value.
random_seed_maze: Random seed for maze generation, or ``None`` for a random value.
random_seed_cheese: Random seed for cheese distribution, or ``None`` for a random value.
random_seed_players: Random seed for initial player locations, or ``None`` for a random value.
maze_width: Width of the maze (number of cells).
maze_height: Height of the maze (number of cells).
cell_percentage: Percentage of accessible cells in the maze (0% = useless maze, 100% = full rectangle).
wall_percentage: Percentage of walls in the maze (0% = empty, 100% = max walls while connected).
mud_percentage: Percentage of adjacent cell pairs separated by mud.
mud_range: Interval of turns needed to cross mud.
fixed_maze: Fixed maze in any PyRat-accepted representation (``Maze``, ``dict``, ``numpy.ndarray``, or ``torch.tensor``).
random_maze_algorithm: Algorithm to generate the maze.
nb_cheese: Number of pieces of cheese in the maze.
fixed_cheese: Fixed list of cheese locations.
render_mode: Method to display the game.
render_simplified: If ``True``, hides non-essential elements in rendering.
rendering_speed: Speed of the rendering, relative to the default one (``2.0`` is twice as fast). \
With the GUI, this sets how long a move is shown, independently of the pace of the game itself.
trace_length: Maximum trace length to display (GUI rendering only).
fullscreen: If ``True``, renders the game in fullscreen (GUI only).
clear_shell_each_turn: If ``True``, clears the shell each turn (shell rendering only).
save_path: Path where games are saved.
save_game: If ``True``, saves the game.
preprocessing_time: Time given to players before the game starts.
turn_time: Time after which players miss a turn.
game_mode: Indicates concurrency mode for players.
continue_on_error: If ``True``, continues the game if a player crashes.
"""
# Check validity
if not (random_seed is None or (isinstance(random_seed, int) and not isinstance(random_seed, bool))):
raise PyRatException(f"Argument 'random_seed' must be an integer or None (if so, default value 'Game.DEFAULT_RANDOM_SEED' is used), got {random_seed!r} of type {type(random_seed).__name__}")
if not (random_seed_maze is None or (isinstance(random_seed_maze, int) and not isinstance(random_seed_maze, bool))):
raise PyRatException(f"Argument 'random_seed_maze' must be an integer or None (if so, default value 'Game.DEFAULT_RANDOM_SEED_MAZE' is used), got {random_seed_maze!r} of type {type(random_seed_maze).__name__}")
if not (random_seed_cheese is None or (isinstance(random_seed_cheese, int) and not isinstance(random_seed_cheese, bool))):
raise PyRatException(f"Argument 'random_seed_cheese' must be an integer or None (if so, default value 'Game.DEFAULT_RANDOM_SEED_CHEESE' is used), got {random_seed_cheese!r} of type {type(random_seed_cheese).__name__}")
if not (random_seed_players is None or (isinstance(random_seed_players, int) and not isinstance(random_seed_players, bool))):
raise PyRatException(f"Argument 'random_seed_players' must be an integer or None (if so, default value 'Game.DEFAULT_RANDOM_SEED_PLAYERS' is used), got {random_seed_players!r} of type {type(random_seed_players).__name__}")
if not (random_seed is None or (random_seed is not None and 0 <= random_seed < sys.maxsize)):
raise PyRatException(f"Argument 'random_seed' should be a non-negative integer smaller than sys.maxsize ({sys.maxsize}), got {random_seed!r}")
if not (random_seed_maze is None or (random_seed_maze is not None and 0 <= random_seed_maze < sys.maxsize)):
raise PyRatException(f"Argument 'random_seed_maze' should be a non-negative integer smaller than sys.maxsize ({sys.maxsize}), got {random_seed_maze!r}")
if not (random_seed_cheese is None or (random_seed_cheese is not None and 0 <= random_seed_cheese < sys.maxsize)):
raise PyRatException(f"Argument 'random_seed_cheese' should be a non-negative integer smaller than sys.maxsize ({sys.maxsize}), got {random_seed_cheese!r}")
if not (random_seed_players is None or (random_seed_players is not None and 0 <= random_seed_players < sys.maxsize)):
raise PyRatException(f"Argument 'random_seed_players' should be a non-negative integer smaller than sys.maxsize ({sys.maxsize}), got {random_seed_players!r}")
if not (random_seed is None or (random_seed is not None and all([param is None for param in [random_seed_maze, random_seed_cheese, random_seed_players]]))):
raise PyRatException(f"Argument 'random_seed' should be given if and only if no other random seed is given, got random_seed={random_seed!r} with random_seed_maze={random_seed_maze!r}, random_seed_cheese={random_seed_cheese!r}, random_seed_players={random_seed_players!r}: use either the global seed or the specific ones")
if not isinstance(render_mode, (RenderMode, type(None))):
raise PyRatException(f"Argument 'render_mode' must be of type 'pyrat.RenderMode' or None (if so, default value 'Game.DEFAULT_RENDER_MODE' is used), got {render_mode!r} of type {type(render_mode).__name__}")
if not isinstance(turn_time, (float, type(None))):
raise PyRatException(f"Argument 'turn_time' must be a real number of type float or None (if so, default value 'Game.DEFAULT_TURN_TIME' is used), got {turn_time!r} of type {type(turn_time).__name__}")
if not (turn_time is None or turn_time >= 0):
raise PyRatException(f"Argument 'turn_time' should be non-negative, got {turn_time!r}")
if not (turn_time is None or math.isfinite(turn_time)):
raise PyRatException(f"Argument 'turn_time' should be a finite number of seconds, got {turn_time!r}")
if not isinstance(preprocessing_time, (float, type(None))):
raise PyRatException(f"Argument 'preprocessing_time' must be a real number of type float or None (if so, default value 'Game.DEFAULT_PREPROCESSING_TIME' is used), got {preprocessing_time!r} of type {type(preprocessing_time).__name__}")
if not (preprocessing_time is None or preprocessing_time >= 0):
raise PyRatException(f"Argument 'preprocessing_time' should be non-negative, got {preprocessing_time!r}")
if not (preprocessing_time is None or math.isfinite(preprocessing_time)):
raise PyRatException(f"Argument 'preprocessing_time' should be a finite number of seconds, got {preprocessing_time!r}")
if not isinstance(game_mode, (GameMode, type(None))):
raise PyRatException(f"Argument 'game_mode' must be of type 'pyrat.GameMode' or None (if so, default value 'Game.DEFAULT_GAME_MODE_SINGLE_TEAM' or 'Game.DEFAULT_GAME_MODE_MULTI_TEAM' is used), got {game_mode!r} of type {type(game_mode).__name__}")
if not isinstance(continue_on_error, (bool, type(None))):
raise PyRatException(f"Argument 'continue_on_error' must be a boolean or None (if so, default value 'Game.DEFAULT_CONTINUE_ON_ERROR' is used), got {continue_on_error!r} of type {type(continue_on_error).__name__}")
if (game_mode == GameMode.SIMULATION and render_mode == RenderMode.GUI):
raise PyRatException("Cannot render GUI in simulation mode (game_mode=GameMode.SIMULATION with render_mode=RenderMode.GUI): do not set 'render_mode' in simulation mode, as no rendering is done")
if not (fixed_maze is None or (fixed_maze is not None and all(param is None for param in [random_seed_maze, random_maze_algorithm, maze_width, maze_height, cell_percentage, wall_percentage, mud_percentage, mud_range]))):
raise PyRatException(f"Argument 'fixed_maze' should be given if and only if no other maze description is given, got 'fixed_maze' together with {[name for name, value in [('random_seed_maze', random_seed_maze), ('random_maze_algorithm', random_maze_algorithm), ('maze_width', maze_width), ('maze_height', maze_height), ('cell_percentage', cell_percentage), ('wall_percentage', wall_percentage), ('mud_percentage', mud_percentage), ('mud_range', mud_range)] if value is not None]}: remove these arguments or 'fixed_maze'")
if not (fixed_cheese is None or (fixed_cheese is not None and all(param is None for param in [random_seed_cheese, nb_cheese]))):
raise PyRatException(f"Argument 'fixed_cheese' should be given if and only if no other cheese description is given, got 'fixed_cheese' together with {[name for name, value in [('random_seed_cheese', random_seed_cheese), ('nb_cheese', nb_cheese)] if value is not None]}: remove these arguments or 'fixed_cheese'")
if not (game_mode is None or game_mode != GameMode.SIMULATION or (game_mode == GameMode.SIMULATION and all([param is None for param in [render_mode, preprocessing_time, turn_time]]))):
raise PyRatException(f"Parameters 'render_mode', 'preprocessing_time' and 'turn_time' should not be set when running in simulation mode (they are fixed automatically), got render_mode={render_mode!r}, preprocessing_time={preprocessing_time!r}, turn_time={turn_time!r}")
if (render_mode not in [None, RenderMode.GUI] and any([param is not None for param in [trace_length, fullscreen]])):
raise PyRatException(f"Parameters 'trace_length' and 'fullscreen' should be set only when rendering in GUI mode, got render_mode={render_mode!r} with trace_length={trace_length!r}, fullscreen={fullscreen!r}")
if (render_mode not in [RenderMode.ASCII, RenderMode.ANSI] and clear_shell_each_turn is not None):
raise PyRatException(f"Parameter 'clear_shell_each_turn' should be set only when rendering in shell mode (RenderMode.ASCII or RenderMode.ANSI), got render_mode={render_mode!r} (None means the default 'Game.DEFAULT_RENDER_MODE')")
if (render_mode == RenderMode.NO_RENDERING and rendering_speed is not None):
raise PyRatException(f"Parameter 'rendering_speed' should be set only when rendering in GUI or shell mode, got render_mode={render_mode!r}")
if not (maze_width is None or (isinstance(maze_width, int) and not isinstance(maze_width, bool))):
raise PyRatException(f"Argument 'maze_width' must be an integer or None (if so, default value 'Game.DEFAULT_MAZE_WIDTH' is used), got {maze_width!r} of type {type(maze_width).__name__}")
if not (maze_height is None or (isinstance(maze_height, int) and not isinstance(maze_height, bool))):
raise PyRatException(f"Argument 'maze_height' must be an integer or None (if so, default value 'Game.DEFAULT_MAZE_HEIGHT' is used), got {maze_height!r} of type {type(maze_height).__name__}")
if not (fixed_maze is None or not isinstance(fixed_maze, Maze) or fixed_maze.nb_vertices() > 1):
raise PyRatException(f"Argument 'fixed_maze' must have at least two cells, got a maze with {fixed_maze.nb_vertices()} cell(s)")
if not isinstance(render_simplified, (bool, type(None))):
raise PyRatException(f"Argument 'render_simplified' must be a boolean or None (if so, default value 'Game.DEFAULT_RENDER_SIMPLIFIED' is used), got {render_simplified!r} of type {type(render_simplified).__name__}")
if not isinstance(rendering_speed, (float, type(None))):
raise PyRatException(f"Argument 'rendering_speed' must be a real number of type float or None (if so, default value 'Game.DEFAULT_RENDERING_SPEED' is used), got {rendering_speed!r} of type {type(rendering_speed).__name__}")
if not (trace_length is None or (isinstance(trace_length, int) and not isinstance(trace_length, bool))):
raise PyRatException(f"Argument 'trace_length' must be an integer or None (if so, default value 'Game.DEFAULT_TRACE_LENGTH' is used), got {trace_length!r} of type {type(trace_length).__name__}")
if not isinstance(fullscreen, (bool, type(None))):
raise PyRatException(f"Argument 'fullscreen' must be a boolean or None (if so, default value 'Game.DEFAULT_FULLSCREEN' is used), got {fullscreen!r} of type {type(fullscreen).__name__}")
if not isinstance(random_maze_algorithm, (RandomMazeAlgorithm, type(None))):
raise PyRatException(f"Argument 'random_maze_algorithm' must be of type 'pyrat.RandomMazeAlgorithm' or None (if so, default value 'Game.DEFAULT_RANDOM_MAZE_ALGORITHM' is used), got {random_maze_algorithm!r} of type {type(random_maze_algorithm).__name__}")
if not isinstance(save_game, (bool, type(None))):
raise PyRatException(f"Argument 'save_game' must be a boolean or None (if so, default value 'Game.DEFAULT_SAVE_GAME' is used), got {save_game!r} of type {type(save_game).__name__}")
if not isinstance(save_path, (str, type(None))):
raise PyRatException(f"Argument 'save_path' must be a string or None (if so, default value 'Game.DEFAULT_SAVE_PATH' is used), got {save_path!r} of type {type(save_path).__name__}")
if not (save_path is None or not os.path.exists(save_path) or os.path.isdir(save_path)):
raise PyRatException(f"Argument 'save_path' must be a directory (created if needed), got {save_path!r}, which is an existing file")
# Store given parameters or default values
self.__random_seed = random_seed if random_seed is not None else Game.DEFAULT_RANDOM_SEED
self.__random_seed_maze = random_seed_maze if random_seed_maze is not None else Game.DEFAULT_RANDOM_SEED_MAZE
self.__random_seed_cheese = random_seed_cheese if random_seed_cheese is not None else Game.DEFAULT_RANDOM_SEED_CHEESE
self.__random_seed_players = random_seed_players if random_seed_players is not None else Game.DEFAULT_RANDOM_SEED_PLAYERS
self.__maze_width = maze_width if maze_width is not None else Game.DEFAULT_MAZE_WIDTH
self.__maze_height = maze_height if maze_height is not None else Game.DEFAULT_MAZE_HEIGHT
self.__cell_percentage = cell_percentage if cell_percentage is not None else Game.DEFAULT_CELL_PERCENTAGE
self.__wall_percentage = wall_percentage if wall_percentage is not None else Game.DEFAULT_WALL_PERCENTAGE
self.__mud_percentage = mud_percentage if mud_percentage is not None else Game.DEFAULT_MUD_PERCENTAGE
self.__mud_range = mud_range if mud_range is not None else Game.DEFAULT_MUD_RANGE
self.__fixed_maze = fixed_maze if fixed_maze is not None else Game.DEFAULT_FIXED_MAZE
self.__random_maze_algorithm = random_maze_algorithm if random_maze_algorithm is not None else Game.DEFAULT_RANDOM_MAZE_ALGORITHM
self.__nb_cheese = nb_cheese if nb_cheese is not None else Game.DEFAULT_NB_CHEESE
self.__fixed_cheese = fixed_cheese if fixed_cheese is not None else Game.DEFAULT_FIXED_CHEESE
self.__render_mode = render_mode if render_mode is not None else Game.DEFAULT_RENDER_MODE
self.__render_simplified = render_simplified if render_simplified is not None else Game.DEFAULT_RENDER_SIMPLIFIED
self.__rendering_speed = rendering_speed if rendering_speed is not None else Game.DEFAULT_RENDERING_SPEED
self.__trace_length = trace_length if trace_length is not None else Game.DEFAULT_TRACE_LENGTH
self.__fullscreen = fullscreen if fullscreen is not None else Game.DEFAULT_FULLSCREEN
self.__clear_shell_each_turn = clear_shell_each_turn if clear_shell_each_turn is not None else Game.DEFAULT_CLEAR_SHELL_EACH_TURN
self.__save_path = save_path if save_path is not None else Game.DEFAULT_SAVE_PATH
self.__save_game = save_game if save_game is not None else Game.DEFAULT_SAVE_GAME
self.__preprocessing_time = preprocessing_time if preprocessing_time is not None else Game.DEFAULT_PREPROCESSING_TIME
self.__turn_time = turn_time if turn_time is not None else Game.DEFAULT_TURN_TIME
self.__continue_on_error = continue_on_error if continue_on_error is not None else Game.DEFAULT_CONTINUE_ON_ERROR
# We will set the game mode later, as it depends on the number of players
self.__asked_game_mode = game_mode
self.__game_mode = game_mode
# If the game is in simulation mode, we enforce some parameters
if self.__asked_game_mode == GameMode.SIMULATION:
self.__preprocessing_time = 0.0
self.__turn_time = 0.0
self.__render_mode = RenderMode.NO_RENDERING
self.__game_mode = GameMode.SEQUENTIAL
# Private attributes
self.__game_random_seed_maze = None
self.__game_random_seed_cheese = None
self.__game_random_seed_players = None
self.__players_rng = None
self.__players_asked_location = []
self.__players = []
self.__initial_game_state = None
self.__actions_history = None
self.__rendering_engine = None
self.__maze = None
self.__reset_called = False
# Initialize the game
self.reset()
##################################################################################
# DUNDER METHODS #
##################################################################################
def __str__ (self) -> str:
"""
Returns a string representation of the object.
Returns:
String representation of the object.
"""
# Create the string
string = "Game object:\n"
string += "| Random seed: {}\n".format(self.__random_seed)
string += "| Random seed for maze: {}\n".format(self.__random_seed_maze)
string += "| Random seed for cheese: {}\n".format(self.__random_seed_cheese)
string += "| Random seed for players: {}\n".format(self.__random_seed_players)
string += "| Maze width: {}\n".format(self.__maze_width)
string += "| Maze height: {}\n".format(self.__maze_height)
string += "| Cell percentage: {}\n".format(self.__cell_percentage)
string += "| Wall percentage: {}\n".format(self.__wall_percentage)
string += "| Mud percentage: {}\n".format(self.__mud_percentage)
string += "| Mud range: {}\n".format(self.__mud_range)
string += "| Fixed maze: {}\n".format(self.__fixed_maze)
string += "| Random maze algorithm: {}\n".format(self.__random_maze_algorithm)
string += "| Number of cheese: {}\n".format(self.__nb_cheese)
string += "| Fixed cheese: {}\n".format(self.__fixed_cheese)
string += "| Render mode: {}\n".format(self.__render_mode)
string += "| Render simplified: {}\n".format(self.__render_simplified)
string += "| Rendering speed: {}\n".format(self.__rendering_speed)
string += "| Trace length: {}\n".format(self.__trace_length)
string += "| Fullscreen: {}\n".format(self.__fullscreen)
string += "| Clear shell each turn: {}\n".format(self.__clear_shell_each_turn)
string += "| Save path: {}\n".format(self.__save_path)
string += "| Save game: {}\n".format(self.__save_game)
string += "| Preprocessing time: {}\n".format(self.__preprocessing_time)
string += "| Turn time: {}\n".format(self.__turn_time)
string += "| Game mode: {}\n".format(self.__game_mode)
string += "| Continue on error: {}\n".format(self.__continue_on_error)
return string
##################################################################################
# PUBLIC METHODS #
##################################################################################
[docs]
def add_player ( self,
player: Player,
team: str = "",
location: StartingLocation | int = StartingLocation.CENTER
) -> None:
"""
Adds a player to the game.
In a PyRat game, players are identified by their name, which should be unique.
If a player with the same name is added, an error will be raised.
If the target location is not reachable (if there is a hole), the player will be placed at the closest reachable location.
Args:
player: Player to add.
team: Team of the player.
location: Initial location of the player (fixed index or value of the ``StartingLocation`` enumeration).
"""
# Check validity
if not isinstance(player, Player):
raise PyRatException(f"Argument 'player' must be of type 'pyrat.Player', got an object of type {type(player).__name__}")
if not isinstance(team, str):
raise PyRatException(f"Argument 'team' must be a string, got {team!r} of type {type(team).__name__}")
if not (isinstance(location, (StartingLocation, int)) and not isinstance(location, bool)):
raise PyRatException(f"Argument 'location' must be of type 'pyrat.StartingLocation' or an integer, corresponding to the index of the cell where the player should start, got {location!r} of type {type(location).__name__}")
if not (location in list(StartingLocation) or (isinstance(location, int) and self.__maze.i_exists(location))):
raise PyRatException(f"Argument 'location' must be a valid index of the maze or a value of the 'pyrat.StartingLocation' enumeration, got {location!r} for player {player.get_name()!r}, which is not a cell of the maze")
if player.get_name() in self.__initial_game_state.player_locations:
raise PyRatException(f"Player {player.get_name()!r} was already added to the game: each player must have a unique name")
if (location == StartingLocation.SAME and len(self.__players) == 0):
raise PyRatException(f"Cannot start player {player.get_name()!r} at the same location as the previous player (StartingLocation.SAME) if no player was added before: give another location for the first player")
# Set initial location
self.__players_asked_location.append(location)
corrected_location = location
if location == StartingLocation.RANDOM:
corrected_location = self.__players_rng.choice(self.__maze.get_vertices())
elif location == StartingLocation.SAME:
corrected_location = list(self.__initial_game_state.player_locations.values())[-1]
elif location == StartingLocation.CENTER:
corrected_location = self.__maze.rc_to_i(self.__maze.get_height() // 2, self.__maze.get_width() // 2)
elif location == StartingLocation.TOP_LEFT:
corrected_location = self.__maze.rc_to_i(0, 0)
elif location == StartingLocation.TOP_RIGHT:
corrected_location = self.__maze.rc_to_i(0, self.__maze.get_width() - 1)
elif location == StartingLocation.BOTTOM_LEFT:
corrected_location = self.__maze.rc_to_i(self.__maze.get_height() - 1, 0)
elif location == StartingLocation.BOTTOM_RIGHT:
corrected_location = self.__maze.rc_to_i(self.__maze.get_height() - 1, self.__maze.get_width() - 1)
# If the location is not reachable, we choose the closest reachable location
if self.__maze.i_exists(corrected_location):
self.__initial_game_state.player_locations[player.get_name()] = corrected_location
else:
valid_cells = self.__maze.get_vertices()
distances = [math.dist(self.__maze.i_to_rc(corrected_location), self.__maze.i_to_rc(cell)) for cell in valid_cells]
_, argmin_distance = min((val, idx) for (idx, val) in enumerate(distances))
self.__initial_game_state.player_locations[player.get_name()] = valid_cells[argmin_distance]
# Append to team
if team not in self.__initial_game_state.teams:
self.__initial_game_state.teams[team] = []
self.__initial_game_state.teams[team].append(player.get_name())
# Initialize other elements of game state
self.__initial_game_state.score_per_player[player.get_name()] = 0
self.__initial_game_state.muds[player.get_name()] = {"target": None, "count": 0}
# Other attributes
self.__players.append(player)
self.__actions_history[player.get_name()] = []
##################################################################################
[docs]
def reset ( self,
keep_players: bool = True,
same: bool = True
) -> None:
"""
Resets the game to its initial state.
The initial state is defined by the parameters given in the constructor.
If the game was generated with random elements, we will generate new random elements or keep the same ones, depending on the ``same`` parameter.
If ``same`` is ``True``, we will keep the same random seeds that were used to generate the previous game.
If ``same`` is ``False``, we will generate new random seeds for the game, based on what was provided in the constructor.
Note that even if you set ``same`` to ``False`` but you provided random seeds in the constructor, the game will use these random seeds.
Thus, if you declare ``same=False`` but created a game with ``random_seed=42``, this will be equivalent ``same=True``.
Another example: if you declare ``same=False`` but created a game with ``random_seed_maze=42``, the game will happen on the same maze as before, but with a different cheese distribution and random initial player locations.
Args:
keep_players: If ``True``, keeps the players in the game, otherwise removes them.
same: If ``True``, keeps the same random seeds as before, otherwise generates new random seeds.
"""
# Check validity
if not isinstance(keep_players, bool):
raise PyRatException(f"Argument 'keep_players' must be a boolean, got {keep_players!r} of type {type(keep_players).__name__}")
if not isinstance(same, bool):
raise PyRatException(f"Argument 'same' must be a boolean, got {same!r} of type {type(same).__name__}")
# Set random seeds for the game
if not same or self.__game_random_seed_maze is None:
self.__game_random_seed_maze = self.__random_seed if self.__random_seed is not None else self.__random_seed_maze if self.__random_seed_maze is not None else random.randint(0, sys.maxsize - 1)
self.__game_random_seed_cheese = self.__random_seed if self.__random_seed is not None else self.__random_seed_cheese if self.__random_seed_cheese is not None else random.randint(0, sys.maxsize - 1)
self.__game_random_seed_players = self.__random_seed if self.__random_seed is not None else self.__random_seed_players if self.__random_seed_players is not None else random.randint(0, sys.maxsize - 1)
self.__players_rng = random.Random(self.__game_random_seed_players)
# Reset game elements
self.__actions_history = {}
if not self.__asked_game_mode:
self.__game_mode = None
# Initialize the maze
if isinstance(self.__fixed_maze, Maze):
self.__maze = copy.deepcopy(self.__fixed_maze)
elif isinstance(self.__fixed_maze, dict):
self.__maze = MazeFromDict(self.__fixed_maze)
elif self.__fixed_maze is not None:
self.__maze = MazeFromMatrix(self.__fixed_maze)
elif self.__random_maze_algorithm == RandomMazeAlgorithm.UNIFORM_HOLES:
self.__maze = UniformHolesRandomMaze(self.__cell_percentage, self.__wall_percentage, self.__mud_percentage, self.__mud_range, self.__game_random_seed_maze, self.__maze_width, self.__maze_height)
elif self.__random_maze_algorithm == RandomMazeAlgorithm.HOLES_ON_SIDE:
self.__maze = HolesOnSideRandomMaze(self.__cell_percentage, self.__wall_percentage, self.__mud_percentage, self.__mud_range, self.__game_random_seed_maze, self.__maze_width, self.__maze_height)
elif self.__random_maze_algorithm == RandomMazeAlgorithm.BIG_HOLES:
self.__maze = BigHolesRandomMaze(self.__cell_percentage, self.__wall_percentage, self.__mud_percentage, self.__mud_range, self.__game_random_seed_maze, self.__maze_width, self.__maze_height)
# Initialize the rendering engine
if self.__render_mode in [RenderMode.ASCII, RenderMode.ANSI]:
use_colors = self.__render_mode == RenderMode.ANSI
self.__rendering_engine = ShellRenderingEngine(use_colors, self.__clear_shell_each_turn, self.__rendering_speed, self.__render_simplified)
elif self.__render_mode == RenderMode.GUI:
self.__rendering_engine = PygameRenderingEngine(self.__fullscreen, self.__trace_length, self.__rendering_speed, self.__render_simplified)
elif self.__render_mode == RenderMode.NO_RENDERING:
self.__rendering_engine = RenderingEngine(self.__rendering_speed, self.__render_simplified)
# Initialize the game state
previous_initial_state = copy.deepcopy(self.__initial_game_state)
self.__initial_game_state = GameState()
# Add players as they were added
for _ in range(len(self.__players)):
player = self.__players.pop(0)
player_asked_location = self.__players_asked_location.pop(0)
if keep_players:
player_team = [team for team in previous_initial_state.teams if player.get_name() in previous_initial_state.teams[team]][0]
self.add_player(player, player_team, player_asked_location)
# Indicate that the game was reset
self.__reset_called = True
##################################################################################
[docs]
def start (self) -> dict[str, object]:
"""
Starts a game, asking players for decisions until the game is over.
Returns:
Game statistics computed during the game.
Raises:
PyRatPlayerException: If a player crashes or returns an invalid action, unless the game was created with ``continue_on_error=True``.
PyRatException: If the game cannot be started, for instance if no player was added, if a game whose players run in their own processes (``MATCH`` and ``SYNCHRONOUS`` modes) is not started from a block ``if __name__ == "__main__":``, if the graphical interface cannot be opened, or if the game is interrupted by the user (Ctrl+C in the terminal).
"""
# Check validity
if not (len(self.__players) > 0):
raise PyRatException("No player was added to the game: call 'add_player' before 'start'")
if not self.__reset_called:
raise PyRatException("The game was not reset before starting: call 'reset' before starting a new game with the same 'Game' object")
# We catch exceptions that may happen during the game
try:
# Set the game mode if needed
if self.__game_mode is None:
self.__game_mode = Game.DEFAULT_GAME_MODE_SINGLE_TEAM if len(self.__initial_game_state.teams) == 1 else Game.DEFAULT_GAME_MODE_MULTI_TEAM
# In multiprocessing mode, players run in other processes, which need the game to be started from a block 'if __name__ == "__main__":'
if self.__game_mode in [GameMode.MATCH, GameMode.SYNCHRONOUS]:
check_main_guard(f"In game mode {self.__game_mode.name}, each player plays in its own process")
# Mark the game as not reset
self.__reset_called = False
# Initialize stats
stats = {"players": {}, "turns": -1}
for player in self.__players:
stats["players"][player.get_name()] = {"score": 0,
"preprocessing_duration": None,
"turn_durations": [],
"team": [team for team in self.__initial_game_state.teams if player.get_name() in self.__initial_game_state.teams[team]][0],
"actions": {Action.NOTHING.value: 0,
Action.NORTH.value: 0,
Action.EAST.value: 0,
Action.SOUTH.value: 0,
Action.WEST.value: 0,
"mud": 0,
"error": 0,
"miss": 0,
"wall" : 0}}
# In multiprocessing mode, prepare processes
maze_per_player = {player.get_name(): copy.deepcopy(self.__maze) for player in self.__players}
if self.__game_mode in [GameMode.MATCH, GameMode.SYNCHRONOUS]:
# A single manager process provides all the objects shared with the players
manager = multiprocessing.Manager()
# Create a process per player
turn_start_synchronizer = manager.Barrier(len(self.__players) + 1)
turn_timeout_lock = manager.Lock()
player_processes = {}
for player in self.__players:
player_processes[player.get_name()] = {"process": None, "input_queue": manager.Queue(), "output_queue": manager.Queue(), "turn_end_synchronizer": manager.Barrier(2)}
player_processes[player.get_name()]["process"] = multiprocessing.Process(target=_player_process_function, args=(player, maze_per_player[player.get_name()], player_processes[player.get_name()]["input_queue"], player_processes[player.get_name()]["output_queue"], turn_start_synchronizer, turn_timeout_lock, player_processes[player.get_name()]["turn_end_synchronizer"], None, None,))
player_processes[player.get_name()]["process"].start()
# If playing in match mode, we create processs to wait instead of missing players
if self.__game_mode == GameMode.MATCH:
waiter_processes = {}
for player in self.__players:
waiter_processes[player.get_name()] = {"process": None, "input_queue": manager.Queue()}
waiter_processes[player.get_name()]["process"] = multiprocessing.Process(target=_waiter_process_function, args=(waiter_processes[player.get_name()]["input_queue"], turn_start_synchronizer,))
waiter_processes[player.get_name()]["process"].start()
# Add cheese
available_cells = [i for i in self.__maze.get_vertices() if i not in self.__initial_game_state.player_locations.values()]
self.__initial_game_state.cheese.extend(self.__distribute_cheese(available_cells))
game_state = copy.deepcopy(self.__initial_game_state)
# Initial rendering of the maze
self.__rendering_engine.render(self.__players, self.__maze, game_state)
# We play until the game is over
players_ready = [player for player in self.__players]
players_running = {player.get_name(): True for player in self.__players}
all_action_names = [action.value for action in Action]
while any(players_running.values()):
# The phase starts now, so that in sequential mode, the time taken by players counts in the phase duration
phase_start = time.perf_counter()
# We communicate the state of the game to the players not in mud
game_phases = {player.get_name(): "none" for player in self.__players}
turn_actions = {player.get_name(): "miss" for player in self.__players}
durations = {player.get_name(): None for player in self.__players}
for ready_player in players_ready:
final_stats = copy.deepcopy(stats) if game_state.game_over() else {}
player_game_state = copy.deepcopy(game_state)
if self.__game_mode in [GameMode.MATCH, GameMode.SYNCHRONOUS]:
player_processes[ready_player.get_name()]["input_queue"].put((player_game_state, final_stats))
else:
turn_actions[ready_player.get_name()], game_phases[ready_player.get_name()], durations[ready_player.get_name()] = _player_process_function(ready_player, maze_per_player[ready_player.get_name()], None, None, None, None, None, player_game_state, final_stats)
# In multiprocessing mode, we wait for everybody to receive data to start, so the phase starts there
# In sequential mode, decisions are already received at this point
if self.__game_mode in [GameMode.MATCH, GameMode.SYNCHRONOUS]:
turn_start_synchronizer.wait()
phase_start = time.perf_counter()
# Wait for the rest of the phase
phase_time = self.__preprocessing_time if game_state.turn == 0 else self.__turn_time
time.sleep(max(0.0, phase_time - (time.perf_counter() - phase_start)))
# In synchronous mode, we wait for everyone
if self.__game_mode == GameMode.SYNCHRONOUS:
for player in self.__players:
player_processes[player.get_name()]["turn_end_synchronizer"].wait()
turn_actions[player.get_name()], game_phases[player.get_name()], durations[player.get_name()] = player_processes[player.get_name()]["output_queue"].get()
# In match mode, we block the possibility to return an action and check who answered in time
elif self.__game_mode == GameMode.MATCH:
# Wait at least for those in mud
for player in self.__players:
if game_state.is_in_mud(player.get_name()) and players_running[player.get_name()]:
player_processes[player.get_name()]["turn_end_synchronizer"].wait()
turn_actions[player.get_name()], game_phases[player.get_name()], durations[player.get_name()] = player_processes[player.get_name()]["output_queue"].get()
# For others, set timeout and wait for output info of those who passed just before timeout
with turn_timeout_lock:
for player in self.__players:
if not game_state.is_in_mud(player.get_name()) and players_running[player.get_name()]:
if not player_processes[player.get_name()]["output_queue"].empty():
player_processes[player.get_name()]["turn_end_synchronizer"].wait()
turn_actions[player.get_name()], game_phases[player.get_name()], durations[player.get_name()] = player_processes[player.get_name()]["output_queue"].get()
# Check which players are ready to continue
players_ready = []
for player in self.__players:
if game_phases[player.get_name()] == "postprocessing":
players_running[player.get_name()] = False
if self.__game_mode == GameMode.MATCH and (game_phases[player.get_name()] == "postprocessing" or turn_actions[player.get_name()] == "miss"):
waiter_processes[player.get_name()]["input_queue"].put(True)
else:
players_ready.append(player)
# Check for errors
crashed_players = [player.get_name() for player in self.__players if turn_actions[player.get_name()] == "error"]
if crashed_players and not self.__continue_on_error:
raise PyRatPlayerException(f"Player(s) {', '.join(repr(name) for name in crashed_players)} caused an error, exiting (see the error above, or create the game with 'continue_on_error=True' to continue despite errors)")
# We save the turn info if we are not postprocessing
if not game_state.game_over():
# Apply the actions
corrected_actions = {player.get_name(): Action(turn_actions[player.get_name()]) if turn_actions[player.get_name()] in all_action_names else Action.NOTHING for player in self.__players}
new_game_state = self.__determine_new_game_state(game_state, corrected_actions)
# Save stats
for player in self.__players:
if game_phases[player.get_name()] == "none":
stats["players"][player.get_name()]["actions"]["miss"] += 1
self.__actions_history[player.get_name()].append(Action.NOTHING)
elif game_phases[player.get_name()] != "preprocessing":
if turn_actions[player.get_name()] in all_action_names and turn_actions[player.get_name()] != Action.NOTHING.value and game_state.player_locations[player.get_name()] == new_game_state.player_locations[player.get_name()] and not new_game_state.is_in_mud(player.get_name()):
stats["players"][player.get_name()]["actions"]["wall"] += 1
else:
stats["players"][player.get_name()]["actions"][turn_actions[player.get_name()]] += 1
if turn_actions[player.get_name()] != "mud":
self.__actions_history[player.get_name()].append(corrected_actions[player.get_name()])
if durations[player.get_name()] is not None:
if game_phases[player.get_name()] == "preprocessing":
stats["players"][player.get_name()]["preprocessing_duration"] = durations[player.get_name()]
else:
stats["players"][player.get_name()]["turn_durations"].append(durations[player.get_name()])
stats["players"][player.get_name()]["score"] = new_game_state.score_per_player[player.get_name()]
stats["turns"] = game_state.turn
# Go to next turn
self.__rendering_engine.render(self.__players, self.__maze, new_game_state)
game_state = new_game_state
# If the user interrupts the game (Ctrl+C in the terminal), we apply end actions (e.g., closing the window), and report it as a PyRat error
# The KeyboardInterrupt is not chained, as it carries no useful information
except KeyboardInterrupt:
self.__end(True)
raise PyRatException("The game was interrupted by the user (Ctrl+C)") from None
# In case of another error, we apply end actions before letting the error go up to the caller
except:
self.__end(True)
raise
# Apply end actions before returning
self.__end(False)
return stats
##################################################################################
# PRIVATE METHODS #
##################################################################################
def __end ( self,
game_crashed: bool,
) -> None:
"""
Actions to perform at the end of the game if needed.
Args:
game_crashed: Indicates if the game crashed.
"""
# Check validity
if not isinstance(game_crashed, bool):
raise PyRatException(f"Argument 'game_crashed' must be a boolean, got {game_crashed!r} of type {type(game_crashed).__name__}")
# We save the game if asked
if self.__save_game and not game_crashed:
# Create the saves directory if needed
if not os.path.exists(self.__save_path):
os.makedirs(self.__save_path)
# Prepare the config dictionary
config = {"game_mode": "{GAME_MODE}",
"fixed_maze": self.__maze.as_dict(),
"fixed_cheese": self.__initial_game_state.cheese}
# Create a description of the players
player_descriptions = []
for player in self.__players:
player_descriptions.append({"name": player.get_name(),
"skin": "{SKIN_" + player.get_skin().name + "}",
"team": [team for team in self.__initial_game_state.teams if player.get_name() in self.__initial_game_state.teams[team]][0],
"location": self.__initial_game_state.player_locations[player.get_name()],
"actions": "{ACTIONS_" + player.get_name() + "}"})
# Create the players' file, forcing players to their initial locations
output_file_name = os.path.join(self.__save_path, datetime.datetime.now().strftime("%Y_%m_%d_%H_%M_%S_%f.py"))
with open(os.path.join(os.path.dirname(os.path.realpath(__file__)), "save_template.py"), "r") as save_template_file:
save_template = save_template_file.read()
save_template = save_template.replace("{PLAYERS}", str(player_descriptions).replace("}, ", "},\n "))
save_template = save_template.replace("{CONFIG}", str(config).replace(", '", ",\n '"))
save_template = save_template.replace("'{GAME_MODE}'", "GameMode.SEQUENTIAL")
for skin in PlayerSkin:
save_template = save_template.replace("'{SKIN_" + skin.name + "}'", "PlayerSkin." + skin.name)
for player in self.__players:
save_template = save_template.replace("'{ACTIONS_" + player.get_name() + "}'", "[" + ", ".join("Action." + action.name for action in self.__actions_history[player.get_name()]) + "]")
with open(output_file_name, "w") as output_file:
print(save_template, file=output_file)
# Apply ending actions of the rendering engine
self.__rendering_engine.end()
##################################################################################
def __determine_new_game_state ( self,
game_state: GameState,
actions: dict[str, Action]
) -> GameState:
"""
Updates the game state after a turn, given decisions of players.
Args:
game_state: Current game state.
actions: Action performed per player.
Returns:
New game state after the turn.
"""
# Check validity
if not isinstance(game_state, GameState):
raise PyRatException(f"Argument 'game_state' must be of type 'pyrat.GameState', got an object of type {type(game_state).__name__}")
if not isinstance(actions, dict):
raise PyRatException(f"Argument 'actions' must be a dictionary, got an object of type {type(actions).__name__}")
if not all(player_name in [player.get_name() for player in self.__players] for player_name in actions):
raise PyRatException(f"All players must be in the game, got actions for unknown players {[name for name in actions if name not in [player.get_name() for player in self.__players]]!r}")
if not all(action in Action for action in actions.values()):
raise PyRatException(f"All actions must be of type 'pyrat.Action', got invalid actions for players {[name for name, action in actions.items() if not isinstance(action, Action)]!r}")
# Initialize new game state
new_game_state = copy.deepcopy(game_state)
new_game_state.turn += 1
# Move all players accordingly
for player in self.__players:
row, col = self.__maze.i_to_rc(game_state.player_locations[player.get_name()])
target = None
if actions[player.get_name()] == Action.NORTH and row > 0:
target = self.__maze.rc_to_i(row - 1, col)
elif actions[player.get_name()] == Action.SOUTH and row < self.__maze.get_height() - 1:
target = self.__maze.rc_to_i(row + 1, col)
elif actions[player.get_name()] == Action.WEST and col > 0:
target = self.__maze.rc_to_i(row, col - 1)
elif actions[player.get_name()] == Action.EAST and col < self.__maze.get_width() - 1:
target = self.__maze.rc_to_i(row, col + 1)
if target is not None and self.__maze.i_exists(target) and self.__maze.has_edge(game_state.player_locations[player.get_name()], target):
weight = self.__maze.get_weight(game_state.player_locations[player.get_name()], target)
if weight == 1:
new_game_state.player_locations[player.get_name()] = target
elif weight > 1:
new_game_state.muds[player.get_name()]["target"] = target
new_game_state.muds[player.get_name()]["count"] = weight
# All players in mud advance a bit
for player in self.__players:
if new_game_state.is_in_mud(player.get_name()):
new_game_state.muds[player.get_name()]["count"] -= 1
if new_game_state.muds[player.get_name()]["count"] == 0:
new_game_state.player_locations[player.get_name()] = new_game_state.muds[player.get_name()]["target"]
new_game_state.muds[player.get_name()]["target"] = None
# Update cheese and scores
for c in game_state.cheese:
players_on_cheese = [player for player in self.__players if c == new_game_state.player_locations[player.get_name()]]
for player_on_cheese in players_on_cheese:
new_game_state.score_per_player[player_on_cheese.get_name()] += 1.0 / len(players_on_cheese)
if len(players_on_cheese) > 0:
new_game_state.cheese.remove(c)
# Return new game state
return new_game_state
##################################################################################
def __distribute_cheese ( self,
available_cells: list[int],
) -> list[int]:
"""
Distributes pieces of cheese in the maze, according to the provided criteria.
If a fixed list of cheese was provided, it is used.
Otherwise, the cheese is distributed randomly.
Args:
available_cells: List of indices of cells that can be used to place cheese.
Returns:
List of indices of cells containing cheese.
"""
# Check validity
if not isinstance(available_cells, list):
raise PyRatException(f"Argument 'available_cells' must be a list, got an object of type {type(available_cells).__name__}")
if not all([isinstance(cell, int) for cell in available_cells]):
raise PyRatException(f"All elements of 'available_cells' must be integers, got element {next(cell for cell in available_cells if not isinstance(cell, int))!r:.100} of type {type(next(cell for cell in available_cells if not isinstance(cell, int))).__name__}")
if not all([self.__maze.i_exists(cell) for cell in available_cells]):
raise PyRatException(f"All elements of 'available_cells' must be valid indices of the maze, got cell {next(cell for cell in available_cells if not self.__maze.i_exists(cell))!r}, which is not in the maze")
# If we ask for a fixed list of cheese, we use it
if self.__fixed_cheese is not None:
# Check validity
if not isinstance(self.__fixed_cheese, list):
raise PyRatException(f"Attribute '__fixed_cheese' (argument 'fixed_cheese' of the constructor) must be a list, got an object of type {type(self.__fixed_cheese).__name__}")
if not all([isinstance(cell, int) and not isinstance(cell, bool) for cell in self.__fixed_cheese]):
raise PyRatException(f"All elements of '__fixed_cheese' (argument 'fixed_cheese' of the constructor) must be integers, got element {next(cell for cell in self.__fixed_cheese if not (isinstance(cell, int) and not isinstance(cell, bool)))!r:.100} of type {type(next(cell for cell in self.__fixed_cheese if not (isinstance(cell, int) and not isinstance(cell, bool)))).__name__}")
if not (len(self.__fixed_cheese) > 0):
raise PyRatException("Attribute '__fixed_cheese' (argument 'fixed_cheese' of the constructor) must contain at least one cell, got an empty list")
if len(set(self.__fixed_cheese)) != len(self.__fixed_cheese):
raise PyRatException(f"All elements of '__fixed_cheese' (argument 'fixed_cheese' of the constructor) must be unique, got duplicated cells {sorted(set(cell for cell in self.__fixed_cheese if self.__fixed_cheese.count(cell) > 1))!r}")
if not (len(available_cells) >= len(self.__fixed_cheese)):
raise PyRatException(f"Not enough available cells to place the fixed cheese: {len(self.__fixed_cheese)} pieces of cheese requested, but only {len(available_cells)} cells available (cells of the maze not occupied by players)")
if not all([self.__maze.i_exists(cell) for cell in self.__fixed_cheese]):
raise PyRatException(f"All elements of '__fixed_cheese' (argument 'fixed_cheese' of the constructor) must be valid indices of the maze, got cell {next(cell for cell in self.__fixed_cheese if not self.__maze.i_exists(cell))!r}, which is not in the maze")
if not all([cell in available_cells for cell in self.__fixed_cheese]):
raise PyRatException(f"All elements of '__fixed_cheese' (argument 'fixed_cheese' of the constructor) must be in 'available_cells', got cell {next(cell for cell in self.__fixed_cheese if cell not in available_cells)!r}, which is not available (probably occupied by a player)")
# Place the cheese
cheese = copy.deepcopy(self.__fixed_cheese)
# Otherwise, we place the cheese randomly
else:
# Check validity
if not (isinstance(self.__nb_cheese, int) and not isinstance(self.__nb_cheese, bool)):
raise PyRatException(f"Attribute '__nb_cheese' (argument 'nb_cheese' of the constructor) must be an integer, got {self.__nb_cheese!r} of type {type(self.__nb_cheese).__name__}")
if not (self.__nb_cheese > 0):
raise PyRatException(f"Attribute '__nb_cheese' (argument 'nb_cheese' of the constructor) must be positive, got {self.__nb_cheese!r}")
if not (len(available_cells) >= self.__nb_cheese):
raise PyRatException(f"Not enough available cells to place the cheese: {self.__nb_cheese} pieces of cheese requested, but only {len(available_cells)} cells available (cells of the maze not occupied by players)")
# Place the cheese randomly
rng = random.Random(self.__game_random_seed_cheese)
rng.shuffle(available_cells)
cheese = available_cells[:self.__nb_cheese]
# Return the cheese
return cheese
##########################################################################################
######################################## FUNCTIONS #######################################
##########################################################################################
def _player_process_function ( player: Player,
maze: Maze,
input_queue: mpmanagers.BaseProxy | None = None,
output_queue: mpmanagers.BaseProxy | None = None,
turn_start_synchronizer: mpmanagers.BarrierProxy | None = None,
turn_timeout_lock: mpmanagers.AcquirerProxy | None = None,
turn_end_synchronizer: mpmanagers.BarrierProxy | None = None,
game_state: GameState | None = None,
final_stats: dict[str, object] | None = None,
) -> tuple[str, str, float | None] | None:
"""
This function is executed in a separate process per player.
It handles the communication with the player and calls the functions given as arguments.
It is defined outside of the class due to multiprocessing limitations.
If not using multiprocessing, the function returns the action and the duration of the turn.
Otherwise, results are sent through the output queue, and the function returns nothing when the game is over.
Args:
player: Player controlled by the process.
maze: Maze in which the player plays.
input_queue: Queue to receive the game state (set if multiprocessing).
output_queue: Queue to send the action (set if multiprocessing).
turn_start_synchronizer: Barrier to synchronize the start of the turn (set if multiprocessing).
turn_timeout_lock: Lock to synchronize the timeout of the turn (set if multiprocessing).
turn_end_synchronizer: Barrier to synchronize the end of the turn (set if multiprocessing).
game_state: Initial game state (set if sequential).
final_stats: Final stats (set if sequential).
Returns:
In sequential mode, a tuple containing the action performed by the player, the phase of the game in which the player is, and the duration of the turn.
In multiprocessing mode, ``None``.
"""
# Check validity
if not isinstance(player, Player):
raise PyRatException(f"Argument 'player' must be of type 'pyrat.Player', got an object of type {type(player).__name__}")
if not isinstance(maze, Maze):
raise PyRatException(f"Argument 'maze' must be of type 'pyrat.Maze', got an object of type {type(maze).__name__}")
if not isinstance(input_queue, (mpmanagers.BaseProxy, type(None))):
raise PyRatException(f"Argument 'input_queue' must be of type 'multiprocessing.Queue' or None, got an object of type {type(input_queue).__name__}")
if not isinstance(output_queue, (mpmanagers.BaseProxy, type(None))):
raise PyRatException(f"Argument 'output_queue' must be of type 'multiprocessing.Queue' or None, got an object of type {type(output_queue).__name__}")
if not isinstance(turn_start_synchronizer, (mpmanagers.BarrierProxy, type(None))):
raise PyRatException(f"Argument 'turn_start_synchronizer' must be of type 'multiprocessing.Barrier' or None, got an object of type {type(turn_start_synchronizer).__name__}")
if not isinstance(turn_timeout_lock, (mpmanagers.AcquirerProxy, type(None))):
raise PyRatException(f"Argument 'turn_timeout_lock' must be of type 'multiprocessing.Lock' or None, got an object of type {type(turn_timeout_lock).__name__}")
if not isinstance(turn_end_synchronizer, (mpmanagers.BarrierProxy, type(None))):
raise PyRatException(f"Argument 'turn_end_synchronizer' must be of type 'multiprocessing.Barrier' or None, got an object of type {type(turn_end_synchronizer).__name__}")
if not isinstance(game_state, (GameState, type(None))):
raise PyRatException(f"Argument 'game_state' must be of type 'pyrat.GameState' or None, got an object of type {type(game_state).__name__}")
if not isinstance(final_stats, (dict, type(None))):
raise PyRatException(f"Argument 'final_stats' must be of type 'dict' or None, got an object of type {type(final_stats).__name__}")
if not (final_stats is None or all(isinstance(key, str) for key in final_stats)):
raise PyRatException(f"Keys of 'final_stats' must be strings, got key {next(key for key in final_stats if not isinstance(key, str))!r:.100} of type {type(next(key for key in final_stats if not isinstance(key, str))).__name__}")
if not (input_queue is None and output_queue is None and turn_start_synchronizer is None and turn_timeout_lock is None and turn_end_synchronizer is None) ^ (game_state is None and final_stats is None):
raise PyRatException(f"Some arguments are for multiprocessing mode, and others for sequential mode: either set all of 'input_queue', 'output_queue', 'turn_start_synchronizer', 'turn_timeout_lock', 'turn_end_synchronizer' (multiprocessing), or 'game_state' and 'final_stats' (sequential), got non-None arguments {[name for name, value in [('input_queue', input_queue), ('output_queue', output_queue), ('turn_start_synchronizer', turn_start_synchronizer), ('turn_timeout_lock', turn_timeout_lock), ('turn_end_synchronizer', turn_end_synchronizer), ('game_state', game_state), ('final_stats', final_stats)] if value is not None]}")
# We catch exceptions that may happen during the game
use_multiprocessing = input_queue is not None
try:
# Main loop
while True:
# In multiprocessing, wait for all players ready
if use_multiprocessing:
turn_start_synchronizer.wait()
game_state, final_stats = input_queue.get()
# Call the correct function
game_phase = "turn"
duration = None
try:
# Call postprocessing once the game is over
if final_stats:
game_phase = "postprocessing"
action = "error"
player.postprocessing(maze, game_state, final_stats)
action = "ignore"
# If in mud, we return immediately (main process will wait for us in all cases)
elif game_state.is_in_mud(player.get_name()):
action = "mud"
# Otherwise, we ask for an action
else:
# Measure start time
start = time.process_time()
# Go
action = "error"
if game_state.turn == 0:
game_phase = "preprocessing"
player.preprocessing(maze, game_state)
action = "ignore"
else:
a = player.turn(maze, game_state)
if a not in list(Action):
raise PyRatPlayerException(f"Invalid action {a!r:.100} of type {type(a).__name__} returned by method 'turn' of player {player.get_name()!r}: it must return a value of the 'pyrat.Action' enumeration")
action = a.value
# Set end time
end_time = time.process_time()
duration = end_time - start
# Print error message in case of a crash
# An interruption by the user (Ctrl+C) is not a crash of the player, so it is not caught here
except Exception:
print("Player %s has crashed with the following error:" % player.get_name(), file=sys.stderr)
print(traceback.format_exc(), file=sys.stderr)
# Turn is over
if use_multiprocessing:
with turn_timeout_lock:
output_queue.put((action, game_phase, duration))
turn_end_synchronizer.wait()
if game_phase == "postprocessing":
break
else:
return action, game_phase, duration
# If the user interrupts the game (Ctrl+C in the terminal), the game must stop
# In sequential mode, we are in the process of the game, so we let the interruption go up to it
# In multiprocessing mode, the game process receives the interruption too, so this process can just end
except KeyboardInterrupt:
if not use_multiprocessing:
raise
# Ignore
except:
pass
##########################################################################################
def _waiter_process_function ( input_queue: mpmanagers.BaseProxy,
turn_start_synchronizer: mpmanagers.BarrierProxy,
) -> None:
"""
This function is executed in a separate process per player.
It handles the timeouts of the player.
It is defined outside of the class due to multiprocessing limitations.
Args:
input_queue: Queue to receive the game state.
turn_start_synchronizer: Barrier to synchronize the start of the turn.
"""
# Check validity
if not isinstance(input_queue, mpmanagers.BaseProxy):
raise PyRatException(f"Argument 'input_queue' must be of type 'multiprocessing.Queue', got an object of type {type(input_queue).__name__}")
if not isinstance(turn_start_synchronizer, mpmanagers.BarrierProxy):
raise PyRatException(f"Argument 'turn_start_synchronizer' must be of type 'multiprocessing.Barrier', got an object of type {type(turn_start_synchronizer).__name__}")
# We catch exceptions that may happen during the game
try:
# We just mark as ready
while True:
_ = input_queue.get()
turn_start_synchronizer.wait()
# Ignore
except:
pass
##########################################################################################
##########################################################################################