pacai.core.isolation.none

  1import logging
  2import random
  3import typing
  4
  5import edq.util.time
  6
  7import pacai.core.action
  8import pacai.core.agent
  9import pacai.core.agentaction
 10import pacai.core.agentinfo
 11import pacai.core.gamestate
 12import pacai.core.isolation.isolator
 13
 14class NoneIsolator(pacai.core.isolation.isolator.AgentIsolator):
 15    """
 16    An isolator that does not do any isolation between the engine and agents.
 17    All agents will be run in the same thread (and therefore processes space).
 18    This is the simplest and fastest of all isolators, but offers the least control and protection.
 19    Agents cannot be timed out (since they run on the same thread).
 20    Agents can also access any memory, disk, or permissions that the core engine has access to.
 21    """
 22
 23    def __init__(self) -> None:
 24        self._agents: dict[int, pacai.core.agent.Agent] = {}
 25        """
 26        The agents that this isolator manages.
 27        These agents are held and ran in this thread's memory space.
 28        """
 29
 30    def init_agents(self, agent_infos: dict[int, pacai.core.agentinfo.AgentInfo]) -> None:
 31        self._agents = {}
 32        for (agent_index, agent_info) in agent_infos.items():
 33            self._agents[agent_index] = pacai.core.agent.load(agent_info)
 34
 35    def game_start(self,
 36            rng: random.Random,
 37            initial_state: pacai.core.gamestate.GameState,
 38            timeout: float,
 39            ) -> dict[int, pacai.core.agentaction.AgentActionRecord]:
 40        results = {}
 41        for (agent_index, agent) in self._agents.items():
 42            data = {
 43                'agent_index': agent_index,
 44                'suggested_seed': rng.randint(0, 2**64),
 45                'initial_state': initial_state,
 46            }
 47
 48            results[agent_index] = _call_agent_method(agent_index, agent.game_start_full, data)
 49
 50        return results
 51
 52    def game_complete(self,
 53            final_state: pacai.core.gamestate.GameState,
 54            timeout: float,
 55            ) -> dict[int, pacai.core.agentaction.AgentActionRecord]:
 56        results = {}
 57        for (agent_index, agent) in self._agents.items():
 58            data = {
 59                'final_state': final_state,
 60            }
 61
 62            results[agent_index] = _call_agent_method(agent_index, agent.game_complete_full, data)
 63
 64        return results
 65
 66    def get_action(self,
 67            state: pacai.core.gamestate.GameState,
 68            user_inputs: list[pacai.core.action.Action],
 69            timeout: float,
 70            ) -> pacai.core.agentaction.AgentActionRecord:
 71        agent = self._agents[state.agent_index]
 72        data = {
 73            'state': state,
 74            'user_inputs': user_inputs,
 75        }
 76
 77        return _call_agent_method(state.agent_index, agent.get_action_full, data)
 78
 79    def close(self) -> None:
 80        self._agents.clear()
 81
 82def _call_agent_method(
 83        agent_index: int,
 84        agent_method: typing.Callable[..., pacai.core.agentaction.AgentAction],
 85        agent_method_kwargs: dict[str, typing.Any],
 86        ) -> pacai.core.agentaction.AgentActionRecord:
 87    """ Call a method on the agent and do all the proper bookkeeping. """
 88
 89    crashed = False
 90    agent_action: pacai.core.agentaction.AgentAction | None = None
 91
 92    start_time = edq.util.time.Timestamp.now()
 93
 94    try:
 95        agent_action = agent_method(**agent_method_kwargs)
 96    except Exception as ex:
 97        crashed = True
 98        logging.warning("Agent %d crashed.", agent_index, exc_info = ex)
 99
100    end_time = edq.util.time.Timestamp.now()
101
102    return pacai.core.agentaction.AgentActionRecord(
103            agent_index = agent_index,
104            agent_action = agent_action,
105            duration = end_time.sub(start_time),
106            crashed = crashed)