smarts.core.provider module

class smarts.core.provider.ActorProviderTransition[source]

Represents a transition of an actor between providers.

actor_state: ActorState

current_provider: Provider

class smarts.core.provider.Provider[source]

A Provider manages a (sub)set of actors (e.g., vehicles) that all share the same action space(s). This is a base class (interface) from which all Providers should inherit.

property actions: Set[ActionSpaceType]: The action spaces of the provider.

property actor_ids: Iterable[str]

Indicate the agents that this provider currently manages.

Returns:: A set of agents that this provider manages.
Return type:: Iterable[str]

add_actor(provider_actor: ActorState, from_provider: Provider | None = None)[source]: Management of the actor with state is being assigned (or transferred if from_provider is not None) to this Provider. Should only be called if can_accept_actor() has returned True.

can_accept_actor(state: ActorState) → bool[source]: Whether this Provider can take control of an existing actor with state that was previously managed by another Provider. The state.role field should indicate the desired role, not the previous role.

property connected: bool: Determine if the provider is still responsive. (e.g. the case that the provider is sending provider state over the internet and has stopped responding) :returns: The connection state of the provider. :rtype: bool

manages_actor(actor_id: str) → bool[source]: Returns True if the actor referenced by actor_id is managed by this Provider.

classmethod provider_id() → str[source]: The identifying name of the provider.

recover(scenario: smarts.core.scenario.Scenario, elapsed_sim_time: float, error: Exception | None = None) → Tuple[ProviderState, bool][source]

Attempt to reconnect the provider if an error or disconnection occurred. Implementations may choose to re-raise the passed in exception. :param scenario: The scenario of the current episode. :type scenario: Scenario :param elapsed_sim_time: The current elapsed simulation time. :type elapsed_sim_time: float :param error: An exception if an exception was thrown. :type error: Optional[Exception]

Returns:: the state of the provider upon recovery bool: The success/failure of the attempt to reconnect.
Return type:: ProviderState

property recovery_flags: ProviderRecoveryFlags: Flags specifying what this provider should do if it fails. (May be overridden by child classes.)

remove_actor(actor_id: str)[source]: The actor is being removed from the simulation.

reset()[source]: Reset this provider to a pre-initialized state.

set_manager(manager: ProviderManager)[source]: Indicate the manager that this provider should inform of all actor hand-offs.

setup(scenario) → ProviderState[source]: Initialize the provider with a scenario.

property source_str: str: This property should be used to fill in the source field of all ActorState objects created/managed by this Provider.

step(actions, dt: float, elapsed_sim_time: float) → ProviderState[source]

Progress the provider to generate new actor state. :param actions: one or more valid actions from the supported action_spaces of this provider :param dt: time (in seconds) to simulate during this simulation step :type dt: float :param elapsed_sim_time: amount of time (in seconds) that’s elapsed so far in the simulation :type elapsed_sim_time: float

Returns:: State representation of all actors this manages.
Return type:: ProviderState

stop_managing(actor_id: str)[source]: Tells the Provider to stop managing the specified actor; it will be managed by another Provider now.

sync(provider_state: ProviderState)[source]: Synchronize with state managed by other Providers.

teardown()[source]: Clean up provider resources.

class smarts.core.provider.ProviderManager[source]

Interface to be implemented by a class that manages a set of Providers that jointly control a set of actors, such that they can hand these off to each other when necessary. Actors can only be passed among Providers that are managed by the same ProviderManager. Providers can call these methods on the manager to do so.

provider_for_actor(actor_id: str) → Provider | None[source]

Find the provider that currently manages the given actor.

Parameters:: actor_id (str) – The actor id to query.
Returns:: The provider that manages this actor.
Return type:: (Provider|None)

provider_releases_actor(current_provider: Provider | None, state: ActorState) → Provider | None[source]

The current provider gives up control over the specified actor. The manager finds a new Provider for the actor from among the Providers managed by this ProviderManager. If no provider accepts the actor the actor is removed from all providers.

Returns:: A suitable new provider or None if a suitable one could not be found.
Return type:: (Provider|None)

provider_relinquishing_actor(current_provider: Provider | None, state: ActorState) → Tuple[Provider | None, ActorProviderTransition][source]

Find a new Provider for an actor from among the Providers managed by this ProviderManager.

Returns:: A suitable new provider or None if a suitable one could not be found.
Return type:: (Provider|None)

provider_removing_actor(provider: Provider | None, actor_id: str)[source]: Called by a Provider when it is removing an actor from the simulation. It means that the Provider is indicating that the actor no longer exists. This was added for convenience, but it isn’t always necessary to be called.

property providers: List[Provider]: The providers that are current managed by this provider manager.

transition_to_provider(new_provider: Provider, actor_provider_transition: ActorProviderTransition) → Provider | None[source]

Passes a released actor to a new provider. This depends on provider_relinquishing_actor.

Parameters:

new_provider (Provider) – The provider to transition to.
actor_provider_transition (ActorProviderTransition) – The released actor information.

Returns:

Returns the provider if successful else will return None on failure.

Return type:

(Provider|None)

class smarts.core.provider.ProviderRecoveryFlags(value)[source]

This describes actions to be taken with a provider should it fail.

ATTEMPT_RECOVERY = 4096: Provider should attempt to recover from the exception or disconnection.

EPISODE_REQUIRED = 16: Needed for the current episode. Results in episode ending.

EXPERIMENT_REQUIRED = 256: Needed for the experiment. Results in exception if an error is thrown.

NOT_REQUIRED = 0: Not needed for the current step. Error causes skip.

RELINQUISH_ACTORS = 65536: Provider should relinquish its agents if it cannot / will not recover.

class smarts.core.provider.ProviderState(actors: ~typing.List[~smarts.core.actor.ActorState] = <factory>, dt: float | None = None)[source]

State information from a provider.

actors: List[ActorState]

dt: float | None = None

filter(actor_ids: Set[str])[source]: Filter actor states down to the given actors.

intersects(actor_ids: Set[str]) → bool[source]: Returns True if any of the actor_ids are contained in this ProviderState . Returns False for empty-set containment.

merge(other: ProviderState)[source]: Merge state with another provider’s state.

replace_actor_type(updated_actors: List[ActorState], actor_state_type: type)[source]

Replaces all actors of the given type.

Parameters:

updated_actors (List[ActorState]) – The actors to use as replacement.
actor_type (str) – The actor type to replace.