Environment

class eagerx.core.env.BaseEnv(name, rate, graph, engine, backend=None, force_start=True, render_mode=None)[source]

The base class for all EAGERx environments that follows the OpenAI gym’s Env API.

Be sure to call super().__init__() inside the subclass’ constructor with the required arguments (name, graph, etc…).

A subclass should implement the following methods:

step(): Be sure to call _step() inside this method to perform the step.
reset(): Be sure to call _reset() inside this method to perform the reset.

A subclass can optionally overwrite the following properties:

observation_space: Per default, the observations, registered in the graph, are taken.
action_space: Per default, the actions, registered in the graph, are taken.

__init__(name, rate, graph, engine, backend=None, force_start=True, render_mode=None)[source]

Initializes an environment with EAGERx dynamics.

Parameters

name (str) – The name of the environment. Everything related to this environment (parameters, topics, nodes, etc…) will be registered under namespace: “/name”.
rate (float) – The rate (Hz) at which the environment will run.
graph (Graph) – The graph consisting of nodes and objects that describe the environment’s dynamics.
engine (EngineSpec) – The physics engine that will govern the environment’s dynamics. For every Object in the graph, the corresponding engine implementations is chosen.
backend (Optional[BackendSpec]) – The backend that will govern the communication for this environment. Per default, the SingleProcess backend is used.
force_start (bool) – If there already exists an environment with the same name, the existing environment is first shutdown by calling the BaseEnv() method before initializing this environment.
render_mode (Optional[str]) – The render mode that will be used for rendering the environment.

_reset(states)[source]

A private method that should be called within reset().

Parameters: states (Dict) – The desired states to be set before the start an episode. May also be an (empty) subset of registered states if not all states require a reset.
Return type: Dict
Returns: The initial observation.

_step(action)[source]

A private method that should be called within step().

Parameters: action (Dict) – The actions to be applied in the next timestep. Should include all registered actions.
Return type: Dict
Returns: The observation of the current timestep that comply with the graph’s observation space.

close()[source]

A method to stop rendering (i.e. close the render window).

A bool message to topic address “name /env/render/toggle”, which toggles the rendering on/off.

Note

Depending on the source node that is producing the images that are rendered, images may still be produced, even when the render window is not visible. This may add computational overhead and influence the run speed.

Optionally, users may subscribe to topic address “name /env/render/toggle” in the node that is producing the images to stop the production and output empty images instead.

gui(interactive=True, resolution=None, filename=None)

Opens a graphical user interface of the graph.

Note

Requires eagerx-gui:

pip3 install eagerx-gui

Parameters

interactive (Optional[bool]) – If True, an interactive application is launched. Otherwise, an RGB render of the GUI is returned. This could be useful when using a headless machine.
resolution (Optional[List[int]]) – Specifies the resolution of the returned render when interactive is False. If interactive is True, this argument is ignored.
filename (Optional[str]) – If provided, the GUI is rendered to an svg file with this name. If interactive is True, this argument is ignored.

Return type

Optional[ndarray]

Returns

RGB render of the GUI if interactive is False.

classmethod load(name, file, backend=None, force_start=True)[source]

Loads an environment corresponding to the graph state.

Parameters

name (str) – The name of the environment. Everything related to this environment (parameters, topics, nodes, etc…) will be registered under namespace: “/name”.
file (str) – A string giving the name (and the file if the file isn’t in the current working directory).
backend (Optional[BackendSpec]) – The backend that will govern the communication for this environment. Per default, the SingleProcess backend is used.
force_start (bool) – If there already exists an environment with the same name, the existing environment is first shutdown by calling the BaseEnv() method before initializing this environment.

render()[source]

A method to start rendering (i.e. open the render window).

A bool message to topic address “name /env/render/toggle”, which toggles the rendering on/off. :rtype: Optional[ndarray] :returns: Optionally, a rgb_array if env.mode=rgb_array.

abstract reset(seed=None, options=None)[source]

An abstract method that resets the environment to an initial state and returns an initial observation.

Note

To reset the graph, the private method _reset() must be called with the: desired initial states. The spaces of all states (of Objects and Nodes in the graph) are stored in

state_space().

Return type: Tuple[Union[Dict, ndarray], Dict]
Returns: The initial observation that is complies with the observation_space().

save(file)[source]

Saves the (engine-specific) graph state, that includes the engine & environment nodes.

The state is saved in .yaml format and contains the state of every added node, action, and observation and the connections between them.

Parameters: file (str) – A string giving the name (and the file if the file isn’t in the current working directory).
Return type: None

shutdown()[source]

A method to shutdown the environment.

Clear the parameters on the ROS parameter under the namespace /name.
Close nodes (i.e. release resources and perform close procedure).
Unregister topics that supplied the I/O communication between nodes.

abstract step(action)[source]

An abstract method that runs one timestep of the environment’s dynamics.

Note

To run one timestep of the graph dynamics (that essentially define the environment dynamics), this method must call the private method _step() with the actions that comply with _action_space.

When the end of an episode is reached, the user is responsible for calling reset() to reset this environment’s state.

Params action

Actions provided by the agent. Should comply with the action_space().

Return type

Tuple[Union[Dict, ndarray], float, bool, bool, Dict]

Returns

A tuple (observation, reward, terminated, truncated, info).

observation: Observations of the current timestep that comply with
the observation_space().
reward: amount of reward returned after previous action
terminated: whether the episode has ended due to a terminal state, in which case further step() calls will
return undefined results
truncated: whether the episode has ended due to a time limit, in which case further step() calls will
return undefined results
info: contains auxiliary diagnostic information (helpful for debugging, and sometimes learning)

property _action_space: gymnasium.spaces.dict.Dict

Infers the action space from the space of every action.

This space defines the format of valid actions.

Return type: Dict
Returns: A dictionary with key = action and value = Space.

property _observation_space: gymnasium.spaces.dict.Dict

Infers the observation space from the space of every observation.

This space defines the format of valid observations.

Note

Observations with window = 0 are excluded from the observation space. For observations with window > 1, the observation space is duplicated window times.

Return type: Dict
Returns: A dictionary with key = observation and value = Space.

property action_space: gymnasium.spaces.space.Space

The Space object corresponding to valid actions

Per default, the action space of all registered actions in the graph is used.

Return type: Space

property np_random: numpy.random._generator.Generator

Returns the environment’s internal _np_random that if not set will initialise with a random seed.

Returns:: Instances of np.random.Generator

Return type: Generator

property observation_space: gymnasium.spaces.space.Space

The Space object corresponding to valid observations.

Per default, the observation space of all registered observations in the graph is used.

Return type: Space

property state_space: gymnasium.spaces.dict.Dict

Infers the state space from the space of every state.

This space defines the format of valid states that can be set before the start of an episode.

Return type: Dict
Returns: A dictionary with key = state and value = Space.