smarts.core.agent_manager module

class smarts.core.agent_manager.AgentManager(sim: SMARTS, interfaces: Dict[str, AgentInterface])

Tracks agent states and implements methods for managing agent life cycle.

XXX: It is very likely that this class will see heavy changes in the near future.

Documentation for specific attributes and methods will be supplied at a later time.

property active_agents: Set[str]

A list of all active agents in the simulation (agents that have a vehicle.)

add_and_emit_social_agent(agent_id: str, agent_spec, agent_model: SocialAgent)

Generates an entirely new social agent and emits a vehicle for it immediately.

Parameters:

• agent_id (str) – The agent id for the new agent.

• agent_spec (AgentSpec) – The agent spec of the new agent

• agent_model (SocialAgent) – The agent configuration of the new vehicle.

Returns:

If the agent is added. False if the agent id is already reserved by a pending ego agent or current social/ego agent.

Return type:

bool

add_ego_agent(agent_id: str, agent_interface: AgentInterface, for_trap: bool = True)

Adds an ego agent to the manager.

add_social_agent_observations_callback(callback: Callable[[Any], None], callback_id: str)

Subscribe a callback to observe social agents.

agent_for_vehicle(vehicle_id: str) → str

Get the controlling agent for the given vehicle.

agent_has_vehicle(agent_id: str) → bool

Test if an agent has an actor associated with it.

property agent_ids: Set[str]

A list of all agents in the simulation.

agent_interface_for_agent_id(agent_id: str) → AgentInterface

Get the agent interface of a specific agent.

property agent_interfaces: Dict[str, AgentInterface]

A list of all agent to agent interface mappings.

agent_name(agent_id: str) → str

Get the resolved agent name.

destroy()

Clean up remaining resources for deletion.

property ego_agent_ids: Set[str]

A list of only the active ego agents in the simulation.

fetch_agent_actions(ego_agent_actions: Dict[str, Any]) → Dict[str, Any]

Retrieve available social agent actions.

filter_response_for_ego(response_tuple: Tuple[Dict[str, Observation], Dict[str, float], Dict[str, float], Dict[str, bool]]) → Tuple[Dict[str, Observation], Dict[str, float], Dict[str, float], Dict[str, bool]]

Filter all observations, rewards, dones, and infos down to those related to ego agents.

init_ego_agents()

Initialize all ego agents.

is_boid_agent(agent_id: str) → bool

Check if an agent is a boid agent

is_boid_done(agent_id: str) → bool

Check if this boid agent should not disappear yet.

is_boid_keep_alive_agent(agent_id: str) → bool

Check if this is a persistent boid agent

is_ego(agent_id: str) → bool

Test if the agent is an ego agent.

observe() → Tuple[Dict[str, Dict[str, Observation] | Observation], Dict[str, Dict[str, float] | float], Dict[str, Dict[str, float] | float], Dict[str, Dict[str, bool] | bool]]

Generate observations from all vehicles associated with an active agent.

observe_from(vehicle_ids: Set[str], done_this_step: Set[str] | None = None) → Tuple[Dict[str, Observation], Dict[str, float], Dict[str, float], Dict[str, bool]]

Attempt to generate observations from the given vehicles.

property pending_agent_ids: Set[str]

The IDs of agents that are waiting to enter the simulation

property pending_social_agent_ids: Set[str]

The IDs of social agents that are waiting to enter the simulation

remove_pending_agent_ids(agent_ids: Set[str])

Remove an agent from the group of agents waiting to enter the simulation.

remove_social_agent_observations_callback(callback_id: str)

Remove a subscription to social agents.

reserve_social_agent_action(agent_id: str, action: Any)

Override a current social agent action.

reset_agents(observations: Dict[str, Observation]) → Dict[str, Observation]

Reset agents, feeding in an initial observation.

send_observations_to_social_agents(observations: Dict[str, Observation])

Forwards observations to managed social agents.

setup_agents()

Initializes all agents.

property shadowing_agent_ids: Set[str]

Get all agents that currently observe, but not control, a vehicle.

property social_agent_ids: Set[str]

A list of only the active social agents in the simulation.

start_social_agent(agent_id, social_agent, agent_model)

Starts a managed social agent.

switch_initial_agents(agent_interfaces: Dict[str, AgentInterface])

Replaces the initial agent interfaces with a new group. This comes into effect on next reset.

teardown()

Clean up resources.

teardown_ego_agents(filter_ids: Set | None = None)

Tears down all given ego agents passed through the filter. :param filter_ids: The whitelist of agent ids. If None, all ids are white-listed. :type filter_ids: Optional[Set[str]], optional

teardown_social_agents(filter_ids: Set | None = None)

Tears down all given social agents passed through the filter. :param filter_ids: The whitelist of agent ids. If None, all ids are white-listed. :type filter_ids: Optional[Set[str]], optional

vehicles_for_agent(agent_id: str) → List[str]

Get the vehicles associated with an agent.