pacai.agents.searchproblem

  1import logging
  2import typing
  3
  4import edq.util.time
  5
  6import pacai.core.action
  7import pacai.core.agent
  8import pacai.core.agentaction
  9import pacai.core.agentinfo
 10import pacai.core.board
 11import pacai.core.gamestate
 12import pacai.core.search
 13import pacai.search.common
 14import pacai.search.random
 15import pacai.search.position
 16import pacai.util.alias
 17import pacai.util.reflection
 18
 19DEFAULT_PROBLEM: str = pacai.util.alias.SEARCH_PROBLEM_POSITION.long
 20DEFAULT_PROBLEM_COST: str = pacai.util.alias.COST_FUNC_UNIT.long
 21DEFAULT_SOLVER: str = pacai.util.alias.SEARCH_SOLVER_RANDOM.long
 22DEFAULT_HEURISTIC: str = pacai.util.alias.HEURISTIC_NULL.long
 23
 24class SearchProblemAgent(pacai.core.agent.Agent):
 25    """
 26    An agent that works by first solving a searh problem (pacai.core.search.Problem),
 27    and then executing the path found during the search.
 28    """
 29
 30    def __init__(self,
 31            problem: type[pacai.core.search.SearchProblem] | pacai.util.reflection.Reference | str = DEFAULT_PROBLEM,
 32            problem_cost: pacai.core.search.CostFunction | pacai.util.reflection.Reference | str = DEFAULT_PROBLEM_COST,
 33            solver: pacai.core.search.SearchProblemSolver | pacai.util.reflection.Reference | str = DEFAULT_SOLVER,
 34            heuristic: pacai.core.search.SearchHeuristic | pacai.util.reflection.Reference | str = DEFAULT_HEURISTIC,
 35            **kwargs: typing.Any) -> None:
 36        super().__init__(**kwargs)
 37
 38        claen_problem_class = pacai.util.reflection.resolve_and_fetch(type, problem)
 39        self._problem_class: type[pacai.core.search.SearchProblem] = claen_problem_class
 40        """ The search problem class this agent will use. """
 41
 42        claen_problem_cost_function = pacai.util.reflection.resolve_and_fetch(pacai.core.search.CostFunction, problem_cost)
 43        self._problem_cost_function: pacai.core.search.CostFunction = claen_problem_cost_function
 44        """ The cost function for this agent's search problem. """
 45
 46        claen_solver_function = pacai.util.reflection.resolve_and_fetch(pacai.core.search.SearchProblemSolver, solver)
 47        self._solver_function: pacai.core.search.SearchProblemSolver = claen_solver_function
 48        """ The search solver function this agent will use. """
 49
 50        claen_heuristic_function = pacai.util.reflection.resolve_and_fetch(pacai.core.search.SearchHeuristic, heuristic)
 51        self._heuristic_function: pacai.core.search.SearchHeuristic = claen_heuristic_function
 52        """ The search heuristic function this agent will use. """
 53
 54        self._actions: list[pacai.core.action.Action] = []
 55        """ The actions that the search solver came up with. """
 56
 57        logging.debug("Created a SearchProblemAgent using problem '%s', cost function '%s', solver '%s', and heuristic '%s'.",
 58                pacai.util.reflection.get_qualified_name(problem),
 59                pacai.util.reflection.get_qualified_name(problem_cost),
 60                pacai.util.reflection.get_qualified_name(solver),
 61                pacai.util.reflection.get_qualified_name(heuristic))
 62
 63    def get_action(self, state: pacai.core.gamestate.GameState) -> pacai.core.action.Action:
 64        if (len(self._actions) == 0):
 65            return pacai.core.action.STOP
 66
 67        return self._actions.pop(0)
 68
 69    def game_start_full(self,
 70            agent_index: int,
 71            suggested_seed: int,
 72            initial_state: pacai.core.gamestate.GameState,
 73            ) -> pacai.core.agentaction.AgentAction:
 74        # Do the standard game initialization steps.
 75        super().game_start_full(agent_index, suggested_seed, initial_state)
 76
 77        # This is the agent's first time seeing the game's state (which includes the board).
 78        # Create a search problem using the game's state, and solve the problem.
 79
 80        start_time = edq.util.time.Timestamp.now()
 81        (solution, position_history, expanded_node_count) = self._do_search(initial_state)
 82        end_time = edq.util.time.Timestamp.now()
 83
 84        self._actions = solution.actions
 85
 86        logging.info("Path found with %d steps and a total cost of %0.2f in %0.2f seconds. %d search nodes expanded.",
 87                len(solution.actions), solution.cost, (end_time.sub(start_time).to_secs()), expanded_node_count)
 88
 89        # Highlight visited locations in the UI to visually represent our search pattern.
 90        highlights = []
 91        for (i, position) in enumerate(position_history):
 92            # Gradually increase the highlight intensity from the start to the end.
 93            intensity = (i + 1) / len(position_history)
 94
 95            highlights.append(pacai.core.board.Highlight(position, intensity))
 96
 97        return pacai.core.agentaction.AgentAction(board_highlights = highlights)
 98
 99    def _do_search(self,
100            state: pacai.core.gamestate.GameState,
101            ) -> tuple[pacai.core.search.SearchSolution, list[pacai.core.board.Position], int]:
102        """
103        Perform the actual search operation.
104        Children may override this to change searching behavior.
105        Return: (solution, position history, expanded node count).
106        """
107
108        search_problem = self._problem_class(game_state = state, cost_function = self._problem_cost_function)
109        solution = self._solver_function(search_problem, self._heuristic_function, self.rng)
110
111        if (solution.goal_node is not None):
112            search_problem.complete(solution.goal_node)
113
114        return (solution, search_problem.position_history, search_problem.expanded_node_count)
115
116class GreedySubproblemSearchAgent(SearchProblemAgent):
117    """
118    An agent that greedily solves several search problems (instead of just one main one).
119    This agent will repeatedly create and solve search problems until the game state signals the game is over
120    (pacai.core.gamestate.GameState.game_over == True).
121    Once the goal is reached, the actions from all subproblem solutions will be concatenated to form the final list of actions.
122    """
123
124    def _do_search(self,
125            state: pacai.core.gamestate.GameState,
126            ) -> tuple[pacai.core.search.SearchSolution, list[pacai.core.board.Position], int]:
127        actions = []
128        total_cost = 0.0
129        goal_node = None
130        total_position_history = []
131        total_expanded_node_count = 0
132
133        while (not state.game_over):
134            # Solve the subproblem.
135            (solution, position_history, expanded_node_count) = super()._do_search(state)
136
137            if (solution.goal_node is None):
138                raise ValueError("Failed to solve subproblem.")
139
140            # Add all the components of the sub-solution to the total solution.
141            actions += solution.actions
142            total_cost += solution.cost
143            goal_node = solution.goal_node
144            total_position_history += position_history
145            total_expanded_node_count += expanded_node_count
146
147            # Move to the next state by applying all the actions.
148            for action in solution.actions:
149                state = state.generate_successor(action, self.rng)
150
151        solution = pacai.core.search.SearchSolution(actions, total_cost, goal_node)
152        return (solution, total_position_history, total_expanded_node_count)