mate.agents package
Submodules
mate.agents.base module
Base classes for agents.
- class mate.agents.base.AgentBase(seed: int | None = None)[source]
Bases:
ABC
Base class for all agents.
- STATE_CLASS: Type[CameraStatePrivate | TargetStatePrivate]
- TEAMMATE_STATE_CLASS: Type[CameraStatePublic | TargetStatePublic]
- OPPONENT_STATE_CLASS: Type[CameraStatePublic | TargetStatePublic]
- __init__(seed: int | None = None) None [source]
Initialize the agent. This function will be called only once on initialization.
Note
Agents can obtain the number of teammates and opponents on agent.reset(observation), but not here. You are responsible for writing scalable policies and code to handle this.
- action_space: Space = None
- observation_space: Space = None
- abstract property num_teammates: int
Number of agents in the same team, including the current agent itself.
- clone() AgentBase | CameraAgentBase | TargetAgentBase [source]
Clone an independent copy of the agent.
- spawn(num_agents: int) List[AgentBase | CameraAgentBase | TargetAgentBase] [source]
Spawn new agents.
- property np_random: RandomState
The main random number generator of the agent.
- seed(seed: int | None = None) List[int] [source]
Set seed for the agent’s random number generator. This function will be called before the first call of reset().
- reset(observation: ndarray) None [source]
Reset the agent. This function will be called immediately after env.reset().
Note
observation is a 1D array, not a 2D array with an additional dimension for agent indices.
- observe(observation: ndarray, info: dict | None = None) None [source]
The agent observe the environment before sending messages. This function will be called before send_requests().
env.step() --> agent.observe() --> agent.send_requests() --> agent.receive_requests() --> agent.send_responses() --> agent.receive_responses() --> agent.act() --> env.step() --> ... | time step | - env ----------------------------------- env --------------------------------------------- env ----------------------------------- env - \ / \ / \ / observe send_requests receive_requests send_responses receive_responses act \ / \ / \ / ----------------- agent ------------------------------------------ agent -------------------------------------------- agent -------------
Note
observation is a 1D array, not a 2D array with an additional dimension for agent indices.
- abstract act(observation: ndarray, info: dict | None = None, deterministic: bool | None = None) int | ndarray [source]
Get the agent action by the observation. This function will be called before every env.step().
env.step() --> agent.observe() --> agent.send_requests() --> agent.receive_requests() --> agent.send_responses() --> agent.receive_responses() --> agent.act() --> env.step() --> ... | time step | - env ----------------------------------- env --------------------------------------------- env ----------------------------------- env - \ / \ / \ / observe send_requests receive_requests send_responses receive_responses act \ / \ / \ / ----------------- agent ------------------------------------------ agent -------------------------------------------- agent -------------
Note
observation is a 1D array, not a 2D array with an additional dimension for agent indices.
- predict(observation: ndarray, info: dict | None = None, deterministic: bool | None = None) int | ndarray [source]
Get the agent action by the observation. Shortcut method for act().
Note
You should implement method act() instead.
- __call__(observation: ndarray, info: dict | None = None, deterministic: bool | None = None) int | ndarray [source]
Shortcut method for act().
- send_requests() Iterable[Message] [source]
Prepare messages to communicate with other agents in the same team. This function will be called after observe() but before receive_requests().
env.step() --> agent.observe() --> agent.send_requests() --> agent.receive_requests() --> agent.send_responses() --> agent.receive_responses() --> agent.act() --> env.step() --> ... | time step | - env ----------------------------------- env --------------------------------------------- env ----------------------------------- env - \ / \ / \ / observe send_requests receive_requests send_responses receive_responses act \ / \ / \ / ----------------- agent ------------------------------------------ agent -------------------------------------------- agent -------------
- receive_requests(messages: Tuple[Message, ...]) None [source]
Receive messages from other agents in the same team. This function will be called after send_requests().
env.step() --> agent.observe() --> agent.send_requests() --> agent.receive_requests() --> agent.send_responses() --> agent.receive_responses() --> agent.act() --> env.step() --> ... | time step | - env ----------------------------------- env --------------------------------------------- env ----------------------------------- env - \ / \ / \ / observe send_requests receive_requests send_responses receive_responses act \ / \ / \ / ----------------- agent ------------------------------------------ agent -------------------------------------------- agent -------------
- send_responses() Iterable[Message] [source]
Prepare messages to communicate with other agents in the same team. This function will be called after receive_requests().
env.step() --> agent.observe() --> agent.send_requests() --> agent.receive_requests() --> agent.send_responses() --> agent.receive_responses() --> agent.act() --> env.step() --> ... | time step | - env ----------------------------------- env --------------------------------------------- env ----------------------------------- env - \ / \ / \ / observe send_requests receive_requests send_responses receive_responses act \ / \ / \ / ----------------- agent ------------------------------------------ agent -------------------------------------------- agent -------------
- receive_responses(messages: Tuple[Message, ...]) None [source]
Receive messages from other agents in the same team. This function will be called after send_responses() but before act().
env.step() --> agent.observe() --> agent.send_requests() --> agent.receive_requests() --> agent.send_responses() --> agent.receive_responses() --> agent.act() --> env.step() --> ... | time step | - env ----------------------------------- env --------------------------------------------- env ----------------------------------- env - \ / \ / \ / observe send_requests receive_requests send_responses receive_responses act \ / \ / \ / ----------------- agent ------------------------------------------ agent -------------------------------------------- agent -------------
- check_inputs(observation: ndarray, info: dict | None = None) Tuple[CameraStatePrivate | TargetStatePrivate, ndarray, dict, List[Message]] [source]
Preprocess the inputs for observe() and act().
- pack_message(content: Any, recipient: int | None = None) Message [source]
Pack the content into a Message object.
- get_teammate_state(observation: ndarray, index: int) Tuple[TargetStatePublic, bool] [source]
Get the teammate’s public state from observation by index.
- get_teammate_states(observation: ndarray) Tuple[Tuple[TargetStatePublic, ...], Tuple[bool, ...]] [source]
Get all teammates’ states from observation.
- get_opponent_state(observation: ndarray, index: int) Tuple[CameraStatePublic | TargetStatePublic, bool] [source]
Get the opponent agent state from observation by index.
- get_all_opponent_states(observation: ndarray) Tuple[Tuple[CameraStatePublic | TargetStatePublic, ...], Tuple[bool, ...]] [source]
Get all opponents’ states from observation.
- class mate.agents.base.CameraAgentBase(seed: int | None = None)[source]
Bases:
AgentBase
Base class for camera agents.
- STATE_CLASS
alias of
CameraStatePrivate
- TEAMMATE_STATE_CLASS
alias of
CameraStatePublic
- OPPONENT_STATE_CLASS
alias of
TargetStatePublic
- class mate.agents.base.TargetAgentBase(seed: int | None = None)[source]
Bases:
AgentBase
Base class for target agents.
- STATE_CLASS
alias of
TargetStatePrivate
- TEAMMATE_STATE_CLASS
alias of
TargetStatePublic
- OPPONENT_STATE_CLASS
alias of
CameraStatePublic
mate.agents.random module
Random agents.
- class mate.agents.random.RandomCameraAgent(seed=None, frame_skip=20)[source]
Bases:
CameraAgentBase
Random Camera Agent
Random action.
- __init__(seed=None, frame_skip=20)[source]
Initialize the agent. This function will be called only once on initialization.
Note
Agents can obtain the number of teammates and opponents on reset, but not here. You are responsible for writing scalable policies and code to handle this.
- class mate.agents.random.RandomTargetAgent(seed=None, frame_skip=20)[source]
Bases:
TargetAgentBase
Random Target Agent
Random action.
- __init__(seed=None, frame_skip=20)[source]
Initialize the agent. This function will be called only once on initialization.
Note
Agents can obtain the number of teammates and opponents on reset, but not here. You are responsible for writing scalable policies and code to handle this.
mate.agents.naive module
Built-in naive rule-based agents.
- class mate.agents.naive.NaiveCameraAgent(seed: int | None = None)[source]
Bases:
CameraAgentBase
Naive Camera Agent
Rotates anti-clockwise with the maximum viewing angle.
- class mate.agents.naive.NaiveTargetAgent(seed=None)[source]
Bases:
TargetAgentBase
Naive Target Agent
Visits all warehouses in turn.
- __init__(seed=None)[source]
Initialize the agent. This function will be called only once on initialization.
- property goal_location
Location of the current warehouse.
mate.agents.greedy module
Built-in greedy rule-based agents.
- class mate.agents.greedy.GreedyCameraAgent(seed=None, memory_period=25, filterout_unloaded=False, filterout_beyond_range=True)[source]
Bases:
CameraAgentBase
Greedy Camera Agent
Arbitrarily tracks the nearest target. If no target found, use previous action or generate a new random action.
- __init__(seed=None, memory_period=25, filterout_unloaded=False, filterout_beyond_range=True)[source]
Initialize the agent. This function will be called only once on initialization.
- reset(observation)[source]
Reset the agent. This function will be called immediately after env.reset().
- observe(observation, info=None)[source]
The agent observe the environment before sending messages. This function will be called before send_responses().
- act(observation, info=None, deterministic=None)[source]
Get the agent action by the observation. This function will be called before every env.step().
Arbitrarily track the nearest target. If no target found, use previous action or generate a new random action.
- process_messages(observation, messages)[source]
Process observation and prepare messages to teammates.
- act_from_target_states(target_states)[source]
Place the selected target at the center of the field of view.
- class mate.agents.greedy.GreedyTargetAgent(seed=None, noise_scale=0.5)[source]
Bases:
TargetAgentBase
Greedy Target Agent
Arbitrarily runs towards the destination (desired warehouse) with some noise.
- __init__(seed=None, noise_scale=0.5)[source]
Initialize the agent. This function will be called only once on initialization.
- property goal
Index of the current warehouse.
- property goal_location
Location of the current warehouse.
- reset(observation)[source]
Reset the agent. This function will be called immediately after env.reset().
- observe(observation, info=None)[source]
The agent observe the environment before sending messages. This function will be called before send_responses().
- act(observation, info=None, deterministic=None)[source]
Get the agent action by the observation. This function will be called before every env.step().
Arbitrarily run towards the warehouse with some noise.
- process_messages(observation, messages)[source]
Process observation and prepare messages to teammates.
mate.agents.heuristic module
Built-in heuristic rule-based agents.
- class mate.agents.heuristic.HeuristicCameraAgent(seed=None)[source]
Bases:
CameraAgentBase
Heuristic Camera Agent
Greedily maximizes the heuristic scores as much as possible. All agents send their observations to the centralized controller (agent 0). Then the central controller sends the goal state (camera pose) to each agent.
- __init__(seed=None)[source]
Initialize the agent. This function will be called only once on initialization.
- reset(observation)[source]
Reset the agent. This function will be called immediately after env.reset().
- act(observation, info=None, deterministic=None)[source]
Get the agent action by the observation. This function will be called before every env.step().
Arbitrarily track the nearest target. If no target found, use previous action or generate a new random action.
- send_requests()[source]
Prepare messages to communicate with other agents in the same team. This function will be called after observe() but before receive_requests().
- receive_requests(messages)[source]
Receive messages from other agents in the same team. This function will be called after send_requests().
- send_responses()[source]
Prepare messages to communicate with other agents in the same team. This function will be called after receive_requests().
- receive_responses(messages)[source]
Receive messages from other agents in the same team. This function will be called after send_responses() but before act().
- class mate.agents.heuristic.HeuristicTargetAgent(seed=None, noise_scale=0.5)[source]
Bases:
GreedyTargetAgent
Heuristic Target Agent
Arbitrarily runs towards the destination (desired warehouse) with some noise. In addition to the greedy target agent, a drift speed is added, which escapes away from the center of cameras’ field of view.
mate.agents.mixture module
Agent wrapper to built a mixture of agents.
- class mate.agents.mixture.MixtureCameraAgent(candidates, weights=None, mixture_seed=None, seed=None)[source]
Bases:
MixtureAgentMixIn
,CameraAgentBase
Mixture Camera Agent
Randomly choose a underlying camera agent in candidates at episode start.
- class mate.agents.mixture.MixtureTargetAgent(candidates, weights=None, mixture_seed=None, seed=None)[source]
Bases:
MixtureAgentMixIn
,TargetAgentBase
Mixture Target Agent
Randomly choose a underlying target agent in candidates at episode start.
mate.agents.utils module
Utility functions and classes for agents.
- mate.agents.utils.convert_coordinates(observation: ndarray, team: Team, num_cameras: int, num_targets: int, num_obstacles: int) ndarray [source]
Convert all locations of other entities in the observation to relative coordinates (exclude the current agent itself).
- mate.agents.utils.normalize_observation(observation: ndarray, observation_space: Box, additional_mask: ndarray | None = None) ndarray [source]
Rescale all entity states in the observation to [-1., +1.].
- mate.agents.utils.rescale_observation(observation: ndarray, team: Team, num_cameras: int, num_targets: int, num_obstacles: int) ndarray [source]
Rescale all entity states in the observation to [-1., +1.].
- mate.agents.utils.split_observation(observation: ndarray, team: Team, num_cameras: int, num_targets: int, num_obstacles: int) Tuple[ndarray, ndarray, ndarray, ndarray, ndarray] [source]
Splits the concatenated observations into parts.
- class mate.agents.utils.CameraStatePublic(state: ndarray, index: int)[source]
Bases:
StateBase
- DIM = 6
- class mate.agents.utils.CameraStatePrivate(state: ndarray, index: int)[source]
Bases:
CameraStatePublic
- DIM = 9
- property action_space: Box
- class mate.agents.utils.TargetStatePublic(state: ndarray, index: int)[source]
Bases:
StateBase
- DIM = 4
Module contents
Built-in classes for agents.
- class mate.agents.CameraAgentBase(seed: int | None = None)[source]
Bases:
AgentBase
Base class for camera agents.
- STATE_CLASS
alias of
CameraStatePrivate
- TEAMMATE_STATE_CLASS
alias of
CameraStatePublic
- OPPONENT_STATE_CLASS
alias of
TargetStatePublic
- class mate.agents.TargetAgentBase(seed: int | None = None)[source]
Bases:
AgentBase
Base class for target agents.
- STATE_CLASS
alias of
TargetStatePrivate
- TEAMMATE_STATE_CLASS
alias of
TargetStatePublic
- OPPONENT_STATE_CLASS
alias of
CameraStatePublic
- class mate.agents.RandomCameraAgent(seed=None, frame_skip=20)[source]
Bases:
CameraAgentBase
Random Camera Agent
Random action.
- __init__(seed=None, frame_skip=20)[source]
Initialize the agent. This function will be called only once on initialization.
Note
Agents can obtain the number of teammates and opponents on reset, but not here. You are responsible for writing scalable policies and code to handle this.
- class mate.agents.RandomTargetAgent(seed=None, frame_skip=20)[source]
Bases:
TargetAgentBase
Random Target Agent
Random action.
- __init__(seed=None, frame_skip=20)[source]
Initialize the agent. This function will be called only once on initialization.
Note
Agents can obtain the number of teammates and opponents on reset, but not here. You are responsible for writing scalable policies and code to handle this.
- class mate.agents.NaiveCameraAgent(seed: int | None = None)[source]
Bases:
CameraAgentBase
Naive Camera Agent
Rotates anti-clockwise with the maximum viewing angle.
- class mate.agents.NaiveTargetAgent(seed=None)[source]
Bases:
TargetAgentBase
Naive Target Agent
Visits all warehouses in turn.
- __init__(seed=None)[source]
Initialize the agent. This function will be called only once on initialization.
- property goal_location
Location of the current warehouse.
- class mate.agents.GreedyCameraAgent(seed=None, memory_period=25, filterout_unloaded=False, filterout_beyond_range=True)[source]
Bases:
CameraAgentBase
Greedy Camera Agent
Arbitrarily tracks the nearest target. If no target found, use previous action or generate a new random action.
- __init__(seed=None, memory_period=25, filterout_unloaded=False, filterout_beyond_range=True)[source]
Initialize the agent. This function will be called only once on initialization.
- reset(observation)[source]
Reset the agent. This function will be called immediately after env.reset().
- observe(observation, info=None)[source]
The agent observe the environment before sending messages. This function will be called before send_responses().
- act(observation, info=None, deterministic=None)[source]
Get the agent action by the observation. This function will be called before every env.step().
Arbitrarily track the nearest target. If no target found, use previous action or generate a new random action.
- process_messages(observation, messages)[source]
Process observation and prepare messages to teammates.
- act_from_target_states(target_states)[source]
Place the selected target at the center of the field of view.
- class mate.agents.GreedyTargetAgent(seed=None, noise_scale=0.5)[source]
Bases:
TargetAgentBase
Greedy Target Agent
Arbitrarily runs towards the destination (desired warehouse) with some noise.
- __init__(seed=None, noise_scale=0.5)[source]
Initialize the agent. This function will be called only once on initialization.
- property goal
Index of the current warehouse.
- property goal_location
Location of the current warehouse.
- reset(observation)[source]
Reset the agent. This function will be called immediately after env.reset().
- observe(observation, info=None)[source]
The agent observe the environment before sending messages. This function will be called before send_responses().
- act(observation, info=None, deterministic=None)[source]
Get the agent action by the observation. This function will be called before every env.step().
Arbitrarily run towards the warehouse with some noise.
- process_messages(observation, messages)[source]
Process observation and prepare messages to teammates.
- class mate.agents.HeuristicCameraAgent(seed=None)[source]
Bases:
CameraAgentBase
Heuristic Camera Agent
Greedily maximizes the heuristic scores as much as possible. All agents send their observations to the centralized controller (agent 0). Then the central controller sends the goal state (camera pose) to each agent.
- __init__(seed=None)[source]
Initialize the agent. This function will be called only once on initialization.
- reset(observation)[source]
Reset the agent. This function will be called immediately after env.reset().
- act(observation, info=None, deterministic=None)[source]
Get the agent action by the observation. This function will be called before every env.step().
Arbitrarily track the nearest target. If no target found, use previous action or generate a new random action.
- send_requests()[source]
Prepare messages to communicate with other agents in the same team. This function will be called after observe() but before receive_requests().
- receive_requests(messages)[source]
Receive messages from other agents in the same team. This function will be called after send_requests().
- send_responses()[source]
Prepare messages to communicate with other agents in the same team. This function will be called after receive_requests().
- receive_responses(messages)[source]
Receive messages from other agents in the same team. This function will be called after send_responses() but before act().
- class mate.agents.HeuristicTargetAgent(seed=None, noise_scale=0.5)[source]
Bases:
GreedyTargetAgent
Heuristic Target Agent
Arbitrarily runs towards the destination (desired warehouse) with some noise. In addition to the greedy target agent, a drift speed is added, which escapes away from the center of cameras’ field of view.
- class mate.agents.MixtureCameraAgent(candidates, weights=None, mixture_seed=None, seed=None)[source]
Bases:
MixtureAgentMixIn
,CameraAgentBase
Mixture Camera Agent
Randomly choose a underlying camera agent in candidates at episode start.
- class mate.agents.MixtureTargetAgent(candidates, weights=None, mixture_seed=None, seed=None)[source]
Bases:
MixtureAgentMixIn
,TargetAgentBase
Mixture Target Agent
Randomly choose a underlying target agent in candidates at episode start.
- mate.agents.convert_coordinates(observation: ndarray, team: Team, num_cameras: int, num_targets: int, num_obstacles: int) ndarray [source]
Convert all locations of other entities in the observation to relative coordinates (exclude the current agent itself).
- mate.agents.normalize_observation(observation: ndarray, observation_space: Box, additional_mask: ndarray | None = None) ndarray [source]
Rescale all entity states in the observation to [-1., +1.].
- mate.agents.rescale_observation(observation: ndarray, team: Team, num_cameras: int, num_targets: int, num_obstacles: int) ndarray [source]
Rescale all entity states in the observation to [-1., +1.].
- mate.agents.split_observation(observation: ndarray, team: Team, num_cameras: int, num_targets: int, num_obstacles: int) Tuple[ndarray, ndarray, ndarray, ndarray, ndarray] [source]
Splits the concatenated observations into parts.
- class mate.agents.CameraStatePrivate(state: ndarray, index: int)[source]
Bases:
CameraStatePublic
- DIM = 9
- property action_space: Box