pacai.core.gamestate

  1import copy
  2import random
  3import typing
  4
  5import PIL.Image
  6import edq.util.json
  7
  8import pacai.core.action
  9import pacai.core.agentaction
 10import pacai.core.agentinfo
 11import pacai.core.board
 12import pacai.core.font
 13import pacai.core.spritesheet
 14import pacai.core.ticket
 15import pacai.util.math
 16
 17class GameState(edq.util.json.DictConverter):
 18    """
 19    The base for all game states in pacai.
 20    A game state should contain all the information about the current state of the game.
 21
 22    Game states should only be interacted with via their methods and not their member variables
 23    (since this class has been optimized for performance).
 24    """
 25
 26    def __init__(self,
 27            board: pacai.core.board.Board | None = None,
 28            seed: int = -1,
 29            agent_index: int = -1,
 30            game_over: bool = False,
 31            agent_actions: dict[int, list[pacai.core.action.Action]] | None = None,
 32            score: float = 0,
 33            turn_count: int = 0,
 34            move_delays: dict[int, int] | None = None,
 35            tickets: dict[int, pacai.core.ticket.Ticket] | None = None,
 36            agent_infos: dict[int, pacai.core.agentinfo.AgentInfo] | None = None,
 37            **kwargs: typing.Any) -> None:
 38        if (board is None):
 39            raise ValueError("Cannot construct a game state without a board.")
 40
 41        self.board: pacai.core.board.Board = board
 42        """ The current board. """
 43
 44        self.seed: int = seed
 45        """ A utility seed that components using the game state may use to seed their own RNGs. """
 46
 47        self.agent_index: int = agent_index
 48        """
 49        The index of the agent with the current move.
 50        -1 indicates that the agent to move has not been selected yet.
 51        """
 52
 53        self.last_agent_index: int = -1
 54        """
 55        The index of the agent who just moved.
 56        -1 indicates that no move has been made yet.
 57        """
 58
 59        self.game_over: bool = game_over
 60        """ Indicates that this state represents a complete game. """
 61
 62        if (agent_actions is None):
 63            agent_actions = {}
 64
 65        self.agent_actions: dict[int, list[pacai.core.action.Action]] = agent_actions
 66        """ Keep track of the actions that each agent makes. """
 67
 68        self.score: float = score
 69        """ The current score of the game. """
 70
 71        self.turn_count: int = turn_count
 72        """ The number of turns (agent actions) that the game has had. """
 73
 74        if (move_delays is None):
 75            move_delays = {}
 76
 77        self.move_delays: dict[int, int] = move_delays
 78        """
 79        The current move delay for each agent.
 80        Every agent should always have a move delay.
 81        """
 82
 83        if (tickets is None):
 84            tickets = {}
 85
 86        self.tickets: dict[int, pacai.core.ticket.Ticket] = tickets
 87        """
 88        The current ticket for each agent.
 89        Every agent should always have a ticket once the game starts (even if it is not taking a move).
 90        """
 91
 92        # Initialize data from agent arguments if not enough info is provided.
 93        if ((len(self.move_delays) == 0) and (agent_infos is not None)):
 94            for (info_agent_index, agent_info) in agent_infos.items():
 95                self.move_delays[info_agent_index] = agent_info.move_delay
 96
 97    def copy(self) -> 'GameState':
 98        """
 99        Get a deep copy of this state.
100
101        Child classes are responsible for making any deep copies they need to.
102        """
103
104        new_state = copy.copy(self)
105
106        new_state.board = self.board.copy()
107        new_state.agent_actions = {agent_index: actions.copy() for (agent_index, actions) in self.agent_actions.items()}
108        new_state.move_delays = self.move_delays.copy()
109        new_state.tickets = self.tickets.copy()
110
111        return new_state
112
113    def game_start(self) -> None:
114        """
115        Indicate that the game is starting.
116        This will initialize some state like tickets.
117        """
118
119        # Issue initial tickets.
120        for agent_index in self.move_delays.keys():
121            self.tickets[agent_index] = pacai.core.ticket.Ticket(agent_index + self.compute_move_delay(agent_index), 0, 0)
122
123        # Choose the first agent to move.
124        self.last_agent_index = self.agent_index
125        self.agent_index = self.get_next_agent_index()
126
127    def agents_game_start(self, agent_responses: dict[int, pacai.core.agentaction.AgentActionRecord]) -> None:
128        """ Indicate that agents have been started. """
129
130    def game_complete(self) -> list[int]:
131        """
132        Indicate that the game has ended.
133        The state should take any final actions and return the indexes of the winning agents (if any).
134        """
135
136        return []
137
138    def get_num_agents(self) -> int:
139        """ Get the number of agents this state is tracking. """
140
141        return len(self.tickets)
142
143    def get_agent_indexes(self) -> list[int]:
144        """ Get the index of all agents tracked by this game state. """
145
146        return list(sorted(self.tickets.keys()))
147
148    def get_agent_positions(self) -> dict[int, pacai.core.board.Position | None]:
149        """
150        Get the positions of all agents.
151        If an agent is tracked by the game state but not on the board, it will have a None value.
152        """
153
154        positions: dict[int, pacai.core.board.Position | None] = {}
155        for agent_index in self.tickets:
156            positions[agent_index] = self.get_agent_position(agent_index)
157
158        return positions
159
160    def get_agent_position(self, agent_index: int | None = None) -> pacai.core.board.Position | None:
161        """ Get the position of the specified agent (or current agent if no agent is specified). """
162
163        if (agent_index is None):
164            agent_index = self.agent_index
165
166        if (agent_index < 0):
167            raise ValueError("No agent is active, cannot get position.")
168
169        return self.board.get_agent_position(agent_index)
170
171    def get_agent_actions(self, agent_index: int | None = None) -> list[pacai.core.action.Action]:
172        """ Get the previous actions of the specified agent (or current agent if no agent is specified). """
173
174        if (agent_index is None):
175            agent_index = self.agent_index
176
177        if (agent_index < 0):
178            return []
179
180        if (agent_index not in self.agent_actions):
181            self.agent_actions[agent_index] = []
182
183        return self.agent_actions[agent_index]
184
185    def get_last_agent_action(self, agent_index: int | None = None) -> pacai.core.action.Action | None:
186        """ Get the last action of the specified agent (or current agent if no agent is specified). """
187
188        actions = self.get_agent_actions(agent_index)
189        if (len(actions) == 0):
190            return None
191
192        return actions[-1]
193
194    def get_reverse_action(self, action: pacai.core.action.Action | None) -> pacai.core.action.Action | None:
195        """
196        Get the reverse of an action, or None if the action has no reverse.
197        By default, "reverse" is just defined in terms of cardinal directions.
198        However, this method exists so that child games can override this definition of "reverse" if necessary.
199        """
200
201        if (action is None):
202            return None
203
204        return pacai.core.action.get_reverse_direction(action)
205
206    def generate_successor(self,
207            action: pacai.core.action.Action,
208            rng: random.Random | None = None,
209            **kwargs: typing.Any) -> 'GameState':
210        """
211        Create a new deep copy of this state that represents the current agent taking the given action.
212        To just apply an action to the current state, use process_turn().
213        """
214
215        successor = self.copy()
216        successor.process_turn_full(action, rng, **kwargs)
217
218        return successor
219
220    def process_agent_timeout(self, agent_index: int) -> None:
221        """
222        Notify the state that the given agent has timed out.
223        The state should make any updates and set the end of game information.
224        """
225
226        self.game_over = True
227
228    def process_agent_crash(self, agent_index: int) -> None:
229        """
230        Notify the state that the given agent has crashed.
231        The state should make any updates and set the end of game information.
232        """
233
234        self.game_over = True
235
236    def process_game_timeout(self) -> None:
237        """
238        Notify the state that the game has reached the maximum number of turns without ending.
239        The state should make any updates and set the end of game information.
240        """
241
242        self.game_over = True
243
244    def process_turn(self,
245            action: pacai.core.action.Action,
246            rng: random.Random | None = None,
247            **kwargs: typing.Any) -> None:
248        """
249        Process the current agent's turn with the given action.
250        This may modify the current state.
251        To get a copy of a potential successor state, use generate_successor().
252        """
253
254    def process_turn_full(self,
255            action: pacai.core.action.Action,
256            rng: random.Random | None = None,
257            **kwargs: typing.Any) -> None:
258        """
259        Process the current agent's turn with the given action.
260        This will modify the current state.
261        First procecss_turn() will be called,
262        then any bookkeeping will be performed.
263        Child classes should prefer overriding the simpler process_turn().
264
265        To get a copy of a potential successor state, use generate_successor().
266        """
267
268        # If the game is over, don't do anyhting.
269        # This case can come up when planning agent actions and generating successors.
270        if (self.game_over):
271            return
272
273        self.process_turn(action, rng, **kwargs)
274
275        # Track this last action.
276        if (self.agent_index not in self.agent_actions):
277            self.agent_actions[self.agent_index] = []
278
279        self.agent_actions[self.agent_index].append(action)
280
281        # Issue this agent a new ticket.
282        self.tickets[self.agent_index] = self.tickets[self.agent_index].next(self.compute_move_delay(self.agent_index))
283
284        # If the game is not over, pick an agent for the next turn.
285        self.last_agent_index = self.agent_index
286        self.agent_index = -1
287        if (not self.game_over):
288            self.agent_index = self.get_next_agent_index()
289
290        # Increment the move count.
291        self.turn_count += 1
292
293    def compute_move_delay(self, agent_index: int) -> int:
294        """
295        The the current move delay for the agent.
296        By default, this just looks up the value in self.move_delays.
297        However, this method allows children to implement move complex functionality
298        (e.g., power-ups that affect an agent's speed).
299        """
300
301        return self.move_delays[agent_index]
302
303    def get_next_agent_index(self) -> int:
304        """
305        Get the agent that moves next.
306        Do this by looking at the agents' tickets and choosing the one with the lowest ticket.
307        """
308
309        next_index = -1
310        for (agent_index, ticket) in self.tickets.items():
311            if ((next_index == -1) or (ticket.is_before(self.tickets[next_index]))):
312                next_index = agent_index
313
314        return next_index
315
316    def sprite_lookup(self,
317            sprite_sheet: pacai.core.spritesheet.SpriteSheet,
318            position: pacai.core.board.Position,
319            marker: pacai.core.board.Marker | None = None,
320            action: pacai.core.action.Action | None = None,
321            adjacency: pacai.core.board.AdjacencyString | None = None,
322            animation_key: str | None = None,
323            ) -> PIL.Image.Image:
324        """
325        Lookup the proper sprite for a situation.
326        By default this just calls into the sprite sheet,
327        but children may override for more expressive functionality.
328        """
329
330        return sprite_sheet.get_sprite(marker = marker, action = action, adjacency = adjacency, animation_key = animation_key)
331
332    def skip_draw(self,
333            marker: pacai.core.board.Marker,
334            position: pacai.core.board.Position,
335            static: bool = False,
336            ) -> bool:
337        """ Return true if this marker/position combination should not be drawn on the board. """
338
339        return False
340
341    def get_static_positions(self) -> list[pacai.core.board.Position]:
342        """ Get a list of positions to draw on the board statically. """
343
344        return []
345
346    def get_static_text(self) -> list[pacai.core.font.BoardText]:
347        """ Get any static text to display on board positions. """
348
349        return []
350
351    def get_nonstatic_text(self) -> list[pacai.core.font.BoardText]:
352        """ Get any non-static text to display on board positions. """
353
354        return []
355
356    def get_footer_text(self) -> pacai.core.font.Text | None:
357        """
358        Get any text to draw in the UI's footer.
359        By default, this will return the score.
360        """
361
362        score_text = f" Score: {pacai.util.math.display_number(self.score, places = 2)}"
363        if (self.game_over):
364            score_text += " - Final"
365
366        return pacai.core.font.Text(score_text, horizontal_align = pacai.core.font.TextHorizontalAlign.LEFT)
367
368    def to_dict(self) -> dict[str, typing.Any]:
369        data = vars(self).copy()
370
371        data['board'] = self.board.to_dict()
372        data['tickets'] = {agent_index: ticket.to_dict() for (agent_index, ticket) in sorted(self.tickets.items())}
373        data['agent_actions'] = {agent_index: [str(action) for action in actions] for (agent_index, actions) in self.agent_actions.items()}
374
375        return data
376
377    @classmethod
378    def from_dict(cls, data: dict[str, typing.Any]) -> typing.Any:
379        data = data.copy()
380
381        data['board'] = pacai.core.board.Board.from_dict(data['board'])
382        data['tickets'] = {int(agent_index): pacai.core.ticket.Ticket.from_dict(ticket) for (agent_index, ticket) in data['tickets'].items()}
383        data['agent_actions'] = {int(agent_index): [pacai.core.action.Action(raw_action, safe = True) for raw_action in raw_actions]
384                for (agent_index, raw_actions) in data['agent_actions'].items()}
385
386        return cls(**data)
387
388    def get_legal_actions(self, position: pacai.core.board.Position | None = None) -> list[pacai.core.action.Action]:
389        """
390        Get the moves that the current agent is allowed to make.
391        Stopping is generally always considered a legal action (unless a game re-defines this behavior).
392
393        If a position is provided, it will override the current agent's position.
394        """
395
396        # Stopping is generally safe.
397        actions = [pacai.core.action.STOP]
398
399        if (position is None):
400            if (self.agent_index == -1):
401                raise ValueError("Cannot get legal actions when no agent is active.")
402
403            position = self.get_agent_position()
404
405            # If the agent is not on the board, it can only stop.
406            if (position is None):
407                return actions
408
409        # Get moves to adjacent positions.
410        neighbor_moves = self.board.get_neighbors(position)
411        for (action, _) in neighbor_moves:
412            actions.append(action)
413
414        return actions
415
416@typing.runtime_checkable
417class AgentStateEvaluationFunction(typing.Protocol):
418    """
419    A function that can be used to score a game state.
420    """
421
422    def __call__(self,
423            state: GameState,
424            agent: typing.Any | None = None,
425            action: pacai.core.action.Action | None = None,
426            **kwargs: typing.Any) -> float:
427        """
428        Compute a score for a state that the provided agent can use to decide actions.
429        The current state is the only required argument, the others are optional.
430        Passing the agent asking for this evaluation is a simple way to pass persistent state
431        (like pre-computed distances) from the agent to this function.
432        """
433
434def base_eval(
435        state: GameState,
436        agent: typing.Any | None = None,
437        action: pacai.core.action.Action | None = None,
438        **kwargs: typing.Any) -> float:
439    """ The most basic evaluation function, which just uses the state's current score. """
440
441    return float(state.score)