Working on getting the core Pacman elements in.

Author: eriq-augustine

pacai/agents/dummy.py

@@ -0,0 +1,19 @@
+import pacai.core.action
+import pacai.core.agent
+import pacai.core.gamestate
+
+class DummyAgent(pacai.core.agent.Agent):
+    """
+    An agent that only takes the STOP action.
+    At first this may seem useless, but dummy agents can serve several purposes.
+    Like being a stand-in for a future agent, fallback for a failing agent, or a placeholder when running a replay.
+    """
+
+    def get_action(self, state: pacai.core.gamestate.GameState) -> pacai.core.action.Action:
+        return pacai.core.action.STOP
+
+    def game_start(self, agent_index: int, suggested_seed: int, initial_state: pacai.core.gamestate.GameState) -> None:
+        pass
+
+    def game_complete(self, final_state: pacai.core.gamestate.GameState) -> None:
+        pass

pacai/agents/random.py

@@ -0,0 +1,32 @@
+import random
+
+import pacai.core.action
+import pacai.core.agent
+import pacai.core.gamestate
+
+class RandomAgent(pacai.core.agent.Agent):
+    """ An agent that just takes random (legal) action. """
+
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, *kwargs)
+
+        self._rng: random.Random | None = None
+
+    def get_action(self, state: pacai.core.gamestate.GameState) -> pacai.core.action.Action:
+        """ Choose a random action. """
+
+        if (self._rng is None):
+            raise ValueError("Cannot get an action before starting the game.")
+
+        legal_actions = state.get_legal_actions()
+        return self._rng.choice(legal_actions)
+
+    def game_start(self, agent_index: int, suggested_seed: int, initial_state: pacai.core.gamestate.GameState) -> None:
+        """ Initialize the agent's random number generator. """
+
+        self._rng = random.Random(suggested_seed)
+
+    def game_complete(self, final_state: pacai.core.gamestate.GameState) -> None:
+        """ Do nothing. """
+
+        pass

pacai/core/agent.py

@@ -41,17 +41,14 @@ def __init__(self, args: AgentArguments) -> None:
         For example, an agent with a move delay of 50 will move twice as often as an agent with a move delay of 100.
         """
 
-    # TEST
     @abc.abstractmethod
     def get_action(self, state: pacai.core.gamestate.GameState) -> pacai.core.action.Action:
         pass
 
-    # TEST
     @abc.abstractmethod
     def game_start(self, agent_index: int, suggested_seed: int, initial_state: pacai.core.gamestate.GameState) -> None:
         pass
 
-    # TEST
     @abc.abstractmethod
     def game_complete(self, final_state: pacai.core.gamestate.GameState) -> None:
         pass

pacai/core/board.py

@@ -1,6 +1,8 @@
 import os
 import re
+import typing
 
+import pacai.core.action
 import pacai.util.file
 import pacai.util.json
 import pacai.util.reflection
@@ -18,11 +20,43 @@
 class Marker(str):
     """
     A marker represents something that can appear on a board.
-    These are similar to a game piece of token in a traditional board game (like the top hat or dog in Monolopy).
+    These are similar to a game piece of token in a traditional board game (like the top hat or dog in Monopoly).
     """
 
     pass
 
+class Position(typing.NamedTuple):
+    """
+    A 2-dimension location.
+    The first value represent row/y/height,
+    and the second value represents col/x/width.
+    """
+
+    row: int
+    """ The row / y / height of this position. """
+
+    col: int
+    """ The col / x / width of this position. """
+
+    def to_index(self, width: int) -> int:
+        """ Convert this position into a 1-dimension index. """
+        return (self.row * width) + self.col
+
+    @staticmethod
+    def from_index(index: int, width: int) -> 'Position':
+        """ Convert a 1-dimension index into a 2-dimension position. """
+        row = index // width
+        col = index % width
+
+        return Position(row, col)
+
+    def add(self, other: 'Position') -> 'Position':
+        """
+        Add another position (offset) to this one and return the result.
+        """
+
+        return Position(self.row + other.row, self.col + other.col)
+
 class Board:
     """
     A board represents the static (non-agent) components of a game.
@@ -44,9 +78,19 @@ def __init__(self,
             extra_markers: list[str] = [],
             strip: bool = True,
             **kwargs) -> None:
+        self._marker_empty: Marker = Marker(marker_empty)
+        """
+        The marker used for empty locations.
+        """
+
+        self._marker_wall: Marker = Marker(marker_wall)
+        """
+        The marker used for wall locations.
+        """
+
         self._markers: dict[str, Marker] = {
-            marker_empty: Marker(marker_empty),
-            marker_wall: Marker(marker_wall),
+            marker_empty: self._marker_empty,
+            marker_wall: self._marker_wall,
         }
         """ Map the text for a marker to the actual marker. """
 
@@ -104,6 +148,44 @@ def _process_text(self, board_text: str, strip: bool = True) -> tuple[int, int,
 
         return height, width, locations
 
+    def _get_index(self, position):
+        """
+        Get the internal 1-d index for this position.
+        Will raise if this position is not valid.
+        """
+
+        index = position.to_index(self.width)
+        if ((index < 0) or (index >= len(self._locations))):
+            raise ValueError("Invalid position: %s.", str(position))
+
+        return index
+
+    def is_wall(self, position):
+        return (self._locations[self._get_index(position)] == self._marker_wall)
+
+    def get_neighbors(self, position: Position) -> list[tuple[pacai.core.action.Action, Position]]:
+        """
+        Get positions that are directly touching (via cardinal directions) the given position
+        without being inside a wall,
+        and the action it would take to get there.
+        """
+
+        neighbors = []
+        for (action, offset) in CARDINAL_OFFSETS:
+            neighbor = position.add(offset)
+
+            if ((neighbor.row < 0) or (neighbor.col < 0)):
+                continue
+
+            if ((neighbor.row >= self.height) or (neighbor.col >= self.width)):
+                continue
+
+            if (self.is_wall(neighbor)):
+                continue
+
+            neighbors.append((action, neighbor))
+
+        return neighbors
 
 def load_path(path: str) -> Board:
     """ Load a board from a file. """
@@ -138,3 +220,10 @@ def load_string(text: str) -> Board:
 
     board_class = options.get('class', DEFAULT_BOARD_CLASS)
     return pacai.util.reflection.new_object(board_class, board_text, **options)
+
+CARDINAL_OFFSETS: list[tuple[pacai.core.action.Action, Position]] = [
+    (pacai.core.action.NORTH, Position(-1, 0)),
+    (pacai.core.action.EAST, Position(0, 1)),
+    (pacai.core.action.WEST, Position(0, -1)),
+    (pacai.core.action.SOUTH, Position(1, 0)),
+]

pacai/core/game.py

@@ -42,6 +42,7 @@ class Game(abc.ABC):
     def __init__(self,
             agent_args: list[pacai.core.agent.AgentArguments],
             isolation_level: pacai.core.isolation.Level = pacai.core.isolation.Level.NONE,
+            max_moves: int = -1,
             seed: int | None = None,
             ) -> None:
         """
@@ -55,12 +56,22 @@ def __init__(self,
             seed = random.randint(0, 2**64)
 
         self._seed: int = seed
+        """ The random seed for this game's RNG. """
 
         self._agent_args: list[pacai.core.agent.AgentArguments] = agent_args
+        """ The required information for creating the agents for this game. """
+
         if (len(self._agent_args) == 0):
             raise ValueError("No agents provided.")
 
         self._isolation_level: pacai.core.isolation.Level = isolation_level
+        """ The isolation level to use for this game. """
+
+        self._max_moves: int = max_moves
+        """
+        The total number of moves (between all agents) allowed for this game.
+        If -1, unlimited moves are allowed.
+        """
 
     def _setup(self) -> None:
         """ Prepare for a game. """
@@ -69,7 +80,7 @@ def _setup(self) -> None:
         pass
 
     @abc.abstractmethod
-    def get_initial_state(self, rng: random.Random) -> pacai.core.gamestate.GameState:
+    def get_initial_state(self, rng: random.Random, board: pacai.core.board.Board) -> pacai.core.gamestate.GameState:
         """ Create the initial state for this game. """
 
         pass
@@ -89,7 +100,9 @@ def check_end(self, state: pacai.core.gamestate.GameState) -> bool:
 
         pass
 
-    def run(self) -> GameResult:
+    # TODO(eriq): Validate that the board works for this game (e.g., number of agent positions).
+
+    def run(self, board: pacai.core.board.Board) -> GameResult:
         """
         The main "game loop" for all games.
         One round of the loop will:
@@ -119,17 +132,17 @@ def run(self) -> GameResult:
         result = GameResult(result_id, self._seed, self._agent_args)
 
         # Create the initial game state.
-        state = self.get_initial_state(rng)
+        state = self.get_initial_state(rng, board)
 
         # Notify agents about the start of the game.
         isolator.game_start(rng, state)
 
-        turn_count = 0
-        while (True):
+        move_count = 0
+        while ((self._max_moves < 0) or (move_count < self._max_moves)):
             # Choose the next agent to move.
             agent_index = self._get_next_agent_index(tickets)
 
-            logging.debug("Turn %d, agent %d, state: '%s'.", turn_count, agent_index, state)
+            logging.debug("Turn %d, agent %d, state: '%s'.", move_count, agent_index, state)
 
             # Get the next action from the agent.
             action_record = isolator.get_action(agent_index, state)
@@ -148,10 +161,12 @@ def run(self) -> GameResult:
             # Issue the agent a new ticket.
             tickets[agent_index] = tickets[agent_index].next(self._agent_args[agent_index].move_delay)
 
-            # Increment the turn count.
-            turn_count += 1
+            # Increment the move count.
+            move_count += 1
 
-        # TEST
+        # Check if this game ended naturally or in a timeout.
+        if (not state.game_over):
+            state.timeout = True
 
         # Notify agents about the end of this game.
         isolator.game_complete(state)

pacai/core/gamestate.py

@@ -1,5 +1,6 @@
 import abc
 
+import pacai.core.action
 import pacai.core.board
 
 class GameState(abc.ABC):
@@ -11,5 +12,28 @@ class GameState(abc.ABC):
     (since this class has been optimized for performance).
     """
 
-    def __init__(self, board: pacai.core.board.Board) -> None:
-        self._board: pacai.core.board.Board = board
+    def __init__(self,
+            board: pacai.core.board.Board,
+            agent_index: int = -1,
+            game_over: bool = False,
+            timeout: bool = False) -> None:
+        self.agent_index: int = agent_index
+        """
+        The index of the agent with the current move.
+        -1 indicates that the agent to move has not been selected yet.
+        """
+
+        self.board: pacai.core.board.Board = board
+        """ The current board. """
+
+        self.game_over: bool = game_over
+        """ Indicates that this state represents a complete game. """
+
+        self.timeout: bool = timeout
+        """ Indicates that the game ended in a timeout. """
+
+    @abc.abstractmethod
+    def get_legal_actions(self) -> list[pacai.core.action.Action]:
+        """ Get the moves that the current agent is allowed to make. """
+
+        pass

pacai/pacman/__init__.py

@@ -25,3 +25,33 @@ def __init__(self, board_text: str, extra_markers: list[str] = [], **kwargs) ->
         ]
 
         super().__init__(board_text, extra_markers = extra_markers, **kwargs)
+
+        # TODO(eriq): Verify that the ghost indexes match.
+
+        self._agents: list[int | None] = [None]
+        """
+        Keep explicit track of each agent on this board.
+        The agents are indexed by their same index in the game.
+        Pacman is always 0 and the ghosts are indexed by the order they appear.
+
+        A None position means that the agent is not currently on the board.
+
+        Note that this quick lookup requires additional maintenance when agents move.
+        """
+
+        for index in range(len(self._locations)):
+            marker = self._locations[index]
+            if (marker not in {MARKER_PACMAN, MARKER_GHOST}):
+                continue
+
+            if (marker == MARKER_PACMAN):
+                self._agents[0] = index
+            else:
+                self._agents.append(index)
+
+    def get_agent_position(self, agent_index: int) -> pacai.core.board.Position | None:
+        index = self._agents[agent_index]
+        if (index is None):
+            return None
+
+        return pacai.core.board.Position.from_index(index, self.width)

pacai/pacman/board.py

@@ -0,0 +1,26 @@
+import abc
+import random
+
+import pacai.core.agent
+import pacai.core.board
+import pacai.core.game
+import pacai.core.gamestate
+import pacai.pacman.gamestate
+
+class Game(pacai.core.game.Game):
+    """
+    A game following the standard rules of PacMan.
+    """
+
+    def get_initial_state(self, rng: random.Random, board: pacai.core.board.Board) -> pacai.core.gamestate.GameState:
+        return pacai.pacman.gamestate.GameState(board)
+
+    @abc.abstractmethod
+    def process_action(self, state: pacai.core.gamestate.GameState, action_record: pacai.core.agent.ActionRecord) -> pacai.core.gamestate.GameState:
+        # TEST
+        return state
+
+    @abc.abstractmethod
+    def check_end(self, state: pacai.core.gamestate.GameState) -> bool:
+        # TEST
+        return True