from typing import Optional, Type, List, ClassVar from pydantic import BaseModel, Field from game_utils import * from message import Message, MessageType, AgentMessage from agent_interfaces import HumanAgentCLI, OpenAIAgentInterface, HumanAgentInterface from player import Player from data_collection import save # Abstracting the Game Class is a WIP so that future games can be added class Game(BaseModel): """Base class for all games.""" # Required game_id: str """The unique id of the game.""" players: List[Player] = Field(exclude=True) """The players in the game.""" observer: Optional[Player] """An observer who can see all public messages, but doesn't actually play.""" # Default winner_id: str | None = None """The id of the player who has won the game.""" game_state: str = Field("game_start", exclude=True) """Keeps track of the current state of the game.""" awaiting_input: bool = Field(False, exclude=True) """Whether the game is currently awaiting input from a player.""" # Class Variables number_of_players: ClassVar[int] """The number of players in the game.""" player_class: ClassVar[Type[Player]] = Player """The class of the player used in the game.""" def player_from_id(self, player_id: str) -> Player: """Returns a player from their ID.""" return next((player for player in self.players if player.player_id == player_id), None) def player_from_name(self, name: str) -> Player: """Returns a player from their name.""" return next((player for player in self.players if player.name == name), None) def game_message( self, content: str, recipient: Player | List[Player] | None = None, # If None, message is broadcast to all players exclude: bool = False, # If True, the message is broadcast to all players except the chosen player message_type: MessageType = "info" ): """ Sends a message to a player or all players. If no recipient is specified, the message is broadcast to all players. If exclude is True, the message is broadcast to all players except the recipient. Some message types are only available to player with access (e.g. verbose, debug). """ if exclude or not recipient: # These are public messages, exclude is used to exclude the sender from the recipient list. recipients = [player for player in self.players if player != recipient] if self.observer: recipients.append(self.observer) else: if isinstance(recipient, Player): recipients = [recipient] else: recipients = recipient message = Message(type=message_type, content=content) recipient_ids = [] for player in recipients: if player.can_receive_message(message_type): player.interface.add_message(message) recipient_ids.append(player.player_id) agent_message = AgentMessage.from_message(message, recipient_ids, self.game_id) save(agent_message) def verbose_message(self, content: str, **kwargs): """ Sends a verbose message to all players capable of receiving them. Verbose messages are used to communicate in real time what is happening that cannot be seen publicly. Ex: "Abby is thinking..." """ self.game_message(content, **kwargs, message_type="verbose") def debug_message(self, content: str, **kwargs): """ Sends a debug message to all players capable of receiving them. Debug messages usually contain secret information and should only be sent when it wouldn't spoil the game. Ex: "Abby is the chameleon." """ self.game_message(content, **kwargs, message_type="debug") def run_game(self): """Runs the game.""" raise NotImplementedError("The run_game method must be implemented by the subclass.") def end_game(self): """Ends the game and declares a winner.""" for player in self.players: save(player) save(self) @classmethod def from_human_name( cls, human_name: str = None, human_interface: Type[HumanAgentInterface] = HumanAgentCLI, human_message_level: str = "verbose" ): """ Instantiates a game with a human player if a name is provided. Otherwise, the game is instantiated with all AI players and an observer. """ game_id = generate_game_id() # Gather Player Names if human_name: ai_names = random_names(cls.number_of_players - 1, human_name) human_index = random_index(cls.number_of_players) else: ai_names = random_names(cls.number_of_players) human_index = None # Add Players players = [] for i in range(0, cls.number_of_players): player_dict = {"game_id": game_id} if human_index == i: player_dict["name"] = human_name player_id = f"{game_id}-human" player_dict["interface"] = human_interface(agent_id=player_id, game_id=game_id) player_dict["message_level"] = human_message_level else: player_dict["name"] = ai_names.pop() player_id = f"{game_id}-{player_dict['name']}" # all AI players use the OpenAI interface for now - this can be changed in the future player_dict["interface"] = OpenAIAgentInterface(agent_id=player_id, game_id=game_id) player_dict["message_level"] = "info" player_dict["player_id"] = player_id players.append(cls.player_class(**player_dict)) # Add Observer - an Agent who can see all the messages, but doesn't actually play if human_index is None: observer = Player.observer(game_id, interface_type=human_interface) else: observer = None return cls(game_id=game_id, players=players, observer=observer)