ding.envs

Env

Please refer to ding/envs/env for more details.

BaseEnv

class ding.envs.BaseEnv(cfg: dict)

Overview:

Basic environment class, extended from gym.Env

Interface:

__init__, reset, close, step, random_action, create_collector_env_cfg, create_evaluator_env_cfg, enable_save_replay

abstract __init__(cfg: dict) → None

Overview:

Lazy init, only related arguments will be initialized in __init__ method, and the concrete env will be initialized the first time reset method is called.

Arguments:

• cfg (dict): Environment configuration in dict type.

abstract close() → None

Overview:

Close env and all the related resources, it should be called after the usage of env instance.

static create_collector_env_cfg(cfg: dict) → List[dict]

Overview:

Return a list of all of the environment from input config, used in env manager (a series of vectorized env), and this method is mainly responsible for envs collecting data.

Arguments:

• cfg (dict): Original input env config, which needs to be transformed into the type of creating env instance actually and generated the corresponding number of configurations.

Returns:

• env_cfg_list (List[dict]): List of cfg including all the config collector envs.

Note

Elements(env config) in collector_env_cfg/evaluator_env_cfg can be different, such as server ip and port.

static create_evaluator_env_cfg(cfg: dict) → List[dict]

Overview:

Return a list of all of the environment from input config, used in env manager (a series of vectorized env), and this method is mainly responsible for envs evaluating performance.

Arguments:

• cfg (dict): Original input env config, which needs to be transformed into the type of creating env instance actually and generated the corresponding number of configurations.

Returns:

• env_cfg_list (List[dict]): List of cfg including all the config evaluator envs.

enable_save_replay(replay_path: str) → None

Overview:

Save replay file in the given path, and this method need to be self-implemented by each env class.

Arguments:

• replay_path (str): The path to save replay file.

random_action() → Any

Overview:

Return random action generated from the original action space, usually it is convenient for test.

Returns:

• random_action (Any): Action generated randomly.

abstract reset() → Any

Overview:

Reset the env to an initial state and returns an initial observation.

Returns:

• obs (Any): Initial observation after reset.

abstract step(action: Any) → BaseEnv.timestep

Overview:

Run one timestep of the environment’s dynamics/simulation.

Arguments:

• action (Any): The action input to step with.

Returns:

• timestep (BaseEnv.timestep): The result timestep of env executing one step.

get_vec_env_setting

ding.envs.get_vec_env_setting(cfg: dict, collect: bool = True, eval_: bool = True) → Tuple[type, List[dict], List[dict]]

Overview:

Get vectorized env setting (env_fn, collector_env_cfg, evaluator_env_cfg).

Arguments:

• cfg (dict): Original input env config in user config, such as cfg.env.

Returns:

• env_fn (type): Callable object, call it with proper arguments and then get a new env instance.

• collector_env_cfg (List[dict]): A list contains the config of collecting data envs.

• evaluator_env_cfg (List[dict]): A list contains the config of evaluation envs.

Note

Elements (env config) in collector_env_cfg/evaluator_env_cfg can be different, such as server ip and port.

get_env_cls

ding.envs.get_env_cls(cfg: EasyDict) → type

Overview:

Get the env class by correspondng module of cfg and return the callable class.

Arguments:

• cfg (dict): Original input env config in user config, such as cfg.env.

Returns:

• env_cls_type (type): Env module as the corresponding callable class type.

DingEnvWrapper

class ding.envs.DingEnvWrapper(env: Env | Env | None = None, cfg: dict | None = None, seed_api: bool = True, caller: str = 'collector', is_gymnasium: bool = False)

Overview:

This is a wrapper for the BaseEnv class, used to provide a consistent environment interface.

Interfaces:

__init__, reset, step, close, seed, random_action, _wrap_env, __repr__, create_collector_env_cfg, create_evaluator_env_cfg, enable_save_replay, observation_space, action_space, reward_space, clone

__init__(env: Env | Env | None = None, cfg: dict | None = None, seed_api: bool = True, caller: str = 'collector', is_gymnasium: bool = False) → None

Overview:

Initialize the DingEnvWrapper. Either an environment instance or a config to create the environment instance should be passed in. For the former, i.e., an environment instance: The env parameter must not be None, but should be the instance. It does not support subprocess environment manager. Thus, it is usually used in simple environments. For the latter, i.e., a config to create an environment instance: The cfg parameter must contain env_id.

Arguments:

• env (Union[gym.Env, gymnasium.Env]): An environment instance to be wrapped.

• cfg (dict): The configuration dictionary to create an environment instance.

• seed_api (bool): Whether to use seed API. Defaults to True.

• caller (str): A string representing the caller of this method, including collector or evaluator. Different caller may need different wrappers. Default is ‘collector’.

• is_gymnasium (bool): Whether the environment is a gymnasium environment. Defaults to False, i.e., the environment is a gym environment.

property action_space: Space

Overview:

Return the action space of the wrapped environment. The action space represents the range and shape of possible actions that the agent can take in the environment.

Returns:

• action_space (gym.spaces.Space): The action space of the environment.

clone(caller: str = 'collector') → BaseEnv

Overview:

Clone the current environment wrapper, creating a new environment with the same settings.

Arguments:

• caller (str): A string representing the caller of this method, including collector or evaluator. Different caller may need different wrappers. Default is ‘collector’.

Returns:

• DingEnvWrapper: A new instance of the environment with the same settings.

close() → None

Overview:

Clean up the environment by closing and deleting it. This method should be called when the environment is no longer needed. Failing to call this method can lead to memory leaks.

static create_collector_env_cfg(cfg: dict) → List[dict]

Overview:

Create a list of environment configuration for collectors based on the input configuration.

Arguments:

• cfg (dict): The input configuration dictionary.

Returns:

• env_cfgs (List[dict]): The list of environment configurations for collectors.

static create_evaluator_env_cfg(cfg: dict) → List[dict]

Overview:

Create a list of environment configuration for evaluators based on the input configuration.

Arguments:

• cfg (dict): The input configuration dictionary.

Returns:

• env_cfgs (List[dict]): The list of environment configurations for evaluators.

enable_save_replay(replay_path: str | None = None) → None

Overview:

Enable the save replay functionality. The replay will be saved at the specified path.

Arguments:

• replay_path (Optional[str]): The path to save the replay, default is None.

property observation_space: Space

Overview:

Return the observation space of the wrapped environment. The observation space represents the range and shape of possible observations that the environment can provide to the agent.

Note:

If the data type of the observation space is float64, it’s converted to float32 for better compatibility with most machine learning libraries.

Returns:

• observation_space (gym.spaces.Space): The observation space of the environment.

random_action() → ndarray

Overview:

Return a random action from the action space of the environment.

Returns:

• action (np.ndarray): The random action.

reset() → ndarray

Overview:

Resets the state of the environment. If the environment is not initialized, it will be created first.

Returns:

• obs (Dict): The new observation after reset.

property reward_space: Space

Overview:

Return the reward space of the wrapped environment. The reward space represents the range and shape of possible rewards that the agent can receive as a result of its actions.

Returns:

• reward_space (gym.spaces.Space): The reward space of the environment.

seed(seed: int, dynamic_seed: bool = True) → None

Overview:

Set the seed for the environment.

Arguments:

• seed (int): The seed to set.

• dynamic_seed (bool): Whether to use dynamic seed, default is True.

step(action: int64 | ndarray) → BaseEnvTimestep

Overview:

Execute the given action in the environment, and return the timestep (observation, reward, done, info).

Arguments:

• action (Union[np.int64, np.ndarray]): The action to execute in the environment.

Returns:

• timestep (BaseEnvTimestep): The timestep after the action execution.

get_default_wrappers

ding.envs.get_default_wrappers(env_wrapper_name: str, env_id: str | None = None, caller: str = 'collector') → List[dict]

Overview:

Get default wrappers for different environments used in DingEnvWrapper.

Arguments:

• env_wrapper_name (str): The name of the environment wrapper.

• env_id (Optional[str]): The id of the specific environment, such as PongNoFrameskip-v4.

• caller (str): The caller of the environment, including collector or evaluator. Different caller may need different wrappers.

Returns:

• wrapper_list (List[dict]): The list of wrappers, each element is a config of the concrete wrapper.

Raises:

• NotImplementedError: env_wrapper_name is not in ['mujoco_default', 'atari_default',             'gym_hybrid_default', 'default']

Env Manager

Please refer to ding/envs/env_manager for more details.

create_env_manager

ding.envs.create_env_manager(manager_cfg: EasyDict, env_fn: List[Callable]) → BaseEnvManager

Overview:

Create an env manager according to manager_cfg and env functions.

Arguments:

• manager_cfg (EasyDict): Final merged env manager config.

• env_fn (List[Callable]): A list of functions to create env_num sub-environments.

ArgumentsKeys:

• type (str): Env manager type set in ENV_MANAGER_REGISTRY.register , such as base .

• import_names (List[str]): A list of module names (paths) to import before creating env manager, such as ding.envs.env_manager.base_env_manager .

Returns:

• env_manager (BaseEnvManager): The created env manager.

Tip

This method will not modify the manager_cfg , it will deepcopy the manager_cfg and then modify it.

get_env_manager_cls

ding.envs.get_env_manager_cls(cfg: EasyDict) → type

Overview:

Get the env manager class according to config, which is used to access related class variables/methods.

Arguments:

• manager_cfg (EasyDict): Final merged env manager config.

ArgumentsKeys:

• type (str): Env manager type set in ENV_MANAGER_REGISTRY.register , such as base .

• import_names (List[str]): A list of module names (paths) to import before creating env manager, such as ding.envs.env_manager.base_env_manager .

Returns:

• env_manager_cls (type): The corresponding env manager class.

BaseEnvManager

class ding.envs.BaseEnvManager(env_fn: List[Callable], cfg: EasyDict = {})

Overview:

The basic class of env manager to manage multiple vectorized environments. BaseEnvManager define all the necessary interfaces and derived class must extend this basic class.

The class is implemented by the pseudo-parallelism (i.e. serial) mechanism, therefore, this class is only used in some tiny environments and for debug purpose.

Interfaces:

reset, step, seed, close, enable_save_replay, launch, default_config, reward_shaping, enable_save_figure

Properties:

env_num, env_ref, ready_obs, ready_obs_id, ready_imgs, done, closed, method_name_list, observation_space, action_space, reward_space

__init__(env_fn: List[Callable], cfg: EasyDict = {}) → None

Overview:

Initialize the base env manager with callable the env function and the EasyDict-type config. Here we use env_fn to ensure the lazy initialization of sub-environments, which is benetificial to resource allocation and parallelism. cfg is the merged result between the default config of this class and user’s config. This construction function is in lazy-initialization mode, the actual initialization is in launch.

Arguments:

• env_fn (List[Callable]): A list of functions to create env_num sub-environments.

• cfg (EasyDict): Final merged config.

Note

For more details about how to merge config, please refer to the system document of DI-engine (en link1).

property action_space: gym.spaces.Space

Overview:

action_space is the action space of sub-environment, following the format of gym.spaces.

Returns:

• action_space (gym.spaces.Space): The action space of sub-environment.

close() → None

Overview:

Close the env manager and release all the environment resources.

property closed: bool

Overview:

closed is a property that returns whether the env manager is closed.

Returns:

• closed (bool): Whether the env manager is closed.

classmethod default_config() → EasyDict

Overview:

Return the deepcopyed default config of env manager.

Returns:

• cfg (EasyDict): The default config of env manager.

property done: bool

Overview:

done is a flag to indicate whether env manager is done, i.e., whether all sub-environments have executed enough episodes.

Returns:

• done (bool): Whether env manager is done.

enable_save_figure(env_id: int, figure_path: str) → None

Overview:

Enable a specific env to save figure (e.g. environment statistics or episode return curve).

Arguments:

• figure_path (str): The file directory path for all environments to save figures.

enable_save_replay(replay_path: List[str] | str) → None

Overview:

Enable all environments to save replay video after each episode terminates.

Arguments:

• replay_path (Union[List[str], str]): List of paths for each environment; Or one path for all environments.

property env_num: int

Overview:

env_num is the number of sub-environments in env manager.

Returns:

• env_num (int): The number of sub-environments.

property env_ref: BaseEnv

Overview:

env_ref is used to acquire some common attributes of env, like obs_shape and act_shape.

Returns:

• env_ref (BaseEnv): The reference of sub-environment.

launch(reset_param: Dict | None = None) → None

Overview:

Launch the env manager, instantiate the sub-environments and set up the environments and their parameters.

Arguments:

• reset_param (Optional[Dict]): A dict of reset parameters for each environment, key is the env_id, value is the corresponding reset parameter, defaults to None.

property method_name_list: list

Overview:

The public methods list of sub-environments that can be directly called from the env manager level. Other methods and attributes will be accessed with the __getattr__ method. Methods defined in this list can be regarded as the vectorized extension of methods in sub-environments. Sub-class of BaseEnvManager can override this method to add more methods.

Returns:

• method_name_list (list): The public methods list of sub-environments.

property observation_space: gym.spaces.Space

Overview:

observation_space is the observation space of sub-environment, following the format of gym.spaces.

Returns:

• observation_space (gym.spaces.Space): The observation space of sub-environment.

property ready_imgs: Dict[int, Any]

Overview:

Sometimes, we need to render the envs, this function is used to get the next ready renderd frame and corresponding env id.

Arguments:

• render_mode (Optional[str]): The render mode, can be ‘rgb_array’ or ‘depth_array’, which follows the definition in the render function of ding.utils .

Returns:

• ready_imgs (Dict[int, np.ndarray]): A dict with env_id keys and rendered frames.

property ready_obs: Dict[int, Any]

Overview:

Get the ready (next) observation, which is a special design to unify both aysnc/sync env manager. For each interaction between policy and env, the policy will input the ready_obs and output the action. Then the env_manager will step with the action and prepare the next ready_obs.

Returns:

• ready_obs (Dict[int, Any]): A dict with env_id keys and observation values.

Example:

>>> obs = env_manager.ready_obs
>>> stacked_obs = np.concatenate(list(obs.values()))
>>> action = policy(obs)  # here policy inputs np obs and outputs np action
>>> action = {env_id: a for env_id, a in zip(obs.keys(), action)}
>>> timesteps = env_manager.step(action)

property ready_obs_id: List[int]

Overview:

Get the ready (next) observation id, which is a special design to unify both aysnc/sync env manager.

Returns:

• ready_obs_id (List[int]): A list of env_ids for ready observations.

reset(reset_param: Dict | None = None) → None

Overview:

Forcely reset the sub-environments their corresponding parameters. Because in env manager all the sub-environments usually are reset automatically as soon as they are done, this method is only called when the caller must forcely reset all the sub-environments, such as in evaluation.

Arguments:

• reset_param (List): Dict of reset parameters for each environment, key is the env_id, value is the corresponding reset parameters.

reward_shaping(env_id: int, transitions: List[dict]) → List[dict]

Overview:

Execute reward shaping for a specific environment, which is often called when a episode terminates.

Arguments:

• env_id (int): The id of the environment to be shaped.

• transitions (List[dict]): The transition data list of the environment to be shaped.

Returns:

• transitions (List[dict]): The shaped transition data list.

property reward_space: gym.spaces.Space

Overview:

reward_space is the reward space of sub-environment, following the format of gym.spaces.

Returns:

• reward_space (gym.spaces.Space): The reward space of sub-environment.

seed(seed: Dict[int, int] | List[int] | int, dynamic_seed: bool | None = None) → None

Overview:

Set the random seed for each environment.

Arguments:

• seed (Union[Dict[int, int], List[int], int]): Dict or List of seeds for each environment; If only one seed is provided, it will be used in the same way for all environments.

• dynamic_seed (bool): Whether to use dynamic seed.

Note

For more details about dynamic_seed, please refer to the best practice document of DI-engine (en link2).

step(actions: Dict[int, Any]) → Dict[int, BaseEnvTimestep]

Overview:

Execute env step according to input actions. If some sub-environments are done after this execution, they will be reset automatically when self._auto_reset is True, otherwise they need to be reset when the caller use the reset method of env manager.

Arguments:

• actions (Dict[int, Any]): A dict of actions, key is the env_id, value is corresponding action. action can be any type, it depends on the env, and the env will handle it. Ususlly, the action is a dict of numpy array, and the value is generated by the outer caller like policy.

Returns:

• timesteps (Dict[int, BaseEnvTimestep]): Each timestep is a BaseEnvTimestep object, usually including observation, reward, done, info. Some special customized environments will have the special timestep definition. The length of timesteps is the same as the length of actions in synchronous env manager.

Example:

>>> timesteps = env_manager.step(action)
>>> for env_id, timestep in enumerate(timesteps):
>>>     if timestep.done:
>>>         print('Env {} is done'.format(env_id))

BaseEnvManagerV2

class ding.envs.BaseEnvManagerV2(env_fn: List[Callable], cfg: EasyDict = {})

Overview:

The basic class of env manager to manage multiple vectorized environments. BaseEnvManager define all the necessary interfaces and derived class must extend this basic class.

The class is implemented by the pseudo-parallelism (i.e. serial) mechanism, therefore, this class is only used in some tiny environments and for debug purpose.

V2 means this env manager is designed for new task pipeline and interfaces coupled with treetensor.`

Note

For more details about new task pipeline, please refer to the system document of DI-engine (system en link3).

Interfaces:

reset, step, seed, close, enable_save_replay, launch, default_config, reward_shaping, enable_save_figure

Properties:

env_num, env_ref, ready_obs, ready_obs_id, ready_imgs, done, closed, method_name_list, observation_space, action_space, reward_space

__init__(env_fn: List[Callable], cfg: EasyDict = {}) → None

Overview:

Initialize the base env manager with callable the env function and the EasyDict-type config. Here we use env_fn to ensure the lazy initialization of sub-environments, which is benetificial to resource allocation and parallelism. cfg is the merged result between the default config of this class and user’s config. This construction function is in lazy-initialization mode, the actual initialization is in launch.

Arguments:

• env_fn (List[Callable]): A list of functions to create env_num sub-environments.

• cfg (EasyDict): Final merged config.

Note

For more details about how to merge config, please refer to the system document of DI-engine (en link1).

property action_space: gym.spaces.Space

Overview:

action_space is the action space of sub-environment, following the format of gym.spaces.

Returns:

• action_space (gym.spaces.Space): The action space of sub-environment.

close() → None

Overview:

Close the env manager and release all the environment resources.

property closed: bool

Overview:

closed is a property that returns whether the env manager is closed.

Returns:

• closed (bool): Whether the env manager is closed.

classmethod default_config() → EasyDict

Overview:

Return the deepcopyed default config of env manager.

Returns:

• cfg (EasyDict): The default config of env manager.

property done: bool

Overview:

done is a flag to indicate whether env manager is done, i.e., whether all sub-environments have executed enough episodes.

Returns:

• done (bool): Whether env manager is done.

enable_save_figure(env_id: int, figure_path: str) → None

Overview:

Enable a specific env to save figure (e.g. environment statistics or episode return curve).

Arguments:

• figure_path (str): The file directory path for all environments to save figures.

enable_save_replay(replay_path: List[str] | str) → None

Overview:

Enable all environments to save replay video after each episode terminates.

Arguments:

• replay_path (Union[List[str], str]): List of paths for each environment; Or one path for all environments.

property env_num: int

Overview:

env_num is the number of sub-environments in env manager.

Returns:

• env_num (int): The number of sub-environments.

property env_ref: BaseEnv

Overview:

env_ref is used to acquire some common attributes of env, like obs_shape and act_shape.

Returns:

• env_ref (BaseEnv): The reference of sub-environment.

launch(reset_param: Dict | None = None) → None

Overview:

Launch the env manager, instantiate the sub-environments and set up the environments and their parameters.

Arguments:

• reset_param (Optional[Dict]): A dict of reset parameters for each environment, key is the env_id, value is the corresponding reset parameter, defaults to None.

property method_name_list: list

Overview:

The public methods list of sub-environments that can be directly called from the env manager level. Other methods and attributes will be accessed with the __getattr__ method. Methods defined in this list can be regarded as the vectorized extension of methods in sub-environments. Sub-class of BaseEnvManager can override this method to add more methods.

Returns:

• method_name_list (list): The public methods list of sub-environments.

property observation_space: gym.spaces.Space

Overview:

observation_space is the observation space of sub-environment, following the format of gym.spaces.

Returns:

• observation_space (gym.spaces.Space): The observation space of sub-environment.

property ready_imgs: Dict[int, Any]

Overview:

Sometimes, we need to render the envs, this function is used to get the next ready renderd frame and corresponding env id.

Arguments:

• render_mode (Optional[str]): The render mode, can be ‘rgb_array’ or ‘depth_array’, which follows the definition in the render function of ding.utils .

Returns:

• ready_imgs (Dict[int, np.ndarray]): A dict with env_id keys and rendered frames.

property ready_obs: array

Overview:

Get the ready (next) observation, which is a special design to unify both aysnc/sync env manager. For each interaction between policy and env, the policy will input the ready_obs and output the action. Then the env_manager will step with the action and prepare the next ready_obs. For V2 version, the observation is transformed and packed up into tnp.array type, which allows more convenient operations.

Return:

• ready_obs (tnp.array): A stacked treenumpy-type observation data.

Example:

>>> obs = env_manager.ready_obs
>>> action = policy(obs)  # here policy inputs treenp obs and output np action
>>> timesteps = env_manager.step(action)

property ready_obs_id: List[int]

Overview:

Get the ready (next) observation id, which is a special design to unify both aysnc/sync env manager.

Returns:

• ready_obs_id (List[int]): A list of env_ids for ready observations.

reset(reset_param: Dict | None = None) → None

Overview:

Forcely reset the sub-environments their corresponding parameters. Because in env manager all the sub-environments usually are reset automatically as soon as they are done, this method is only called when the caller must forcely reset all the sub-environments, such as in evaluation.

Arguments:

• reset_param (List): Dict of reset parameters for each environment, key is the env_id, value is the corresponding reset parameters.

reward_shaping(env_id: int, transitions: List[dict]) → List[dict]

Overview:

Execute reward shaping for a specific environment, which is often called when a episode terminates.

Arguments:

• env_id (int): The id of the environment to be shaped.

• transitions (List[dict]): The transition data list of the environment to be shaped.

Returns:

• transitions (List[dict]): The shaped transition data list.

property reward_space: gym.spaces.Space

Overview:

reward_space is the reward space of sub-environment, following the format of gym.spaces.

Returns:

• reward_space (gym.spaces.Space): The reward space of sub-environment.

seed(seed: Dict[int, int] | List[int] | int, dynamic_seed: bool | None = None) → None

Overview:

Set the random seed for each environment.

Arguments:

• seed (Union[Dict[int, int], List[int], int]): Dict or List of seeds for each environment; If only one seed is provided, it will be used in the same way for all environments.

• dynamic_seed (bool): Whether to use dynamic seed.

Note

For more details about dynamic_seed, please refer to the best practice document of DI-engine (en link2).

step(actions: List[ndarray]) → List[ndarray]

Overview:

Execute env step according to input actions. If some sub-environments are done after this execution, they will be reset automatically by default.

Arguments:

• actions (List[tnp.ndarray]): A list of treenumpy-type actions, the value is generated by the outer caller like policy.

Returns:

• timesteps (List[tnp.ndarray]): A list of timestep, Each timestep is a tnp.ndarray object, usually including observation, reward, done, info, env_id. Some special environments will have the special timestep definition. The length of timesteps is the same as the length of actions in synchronous env manager. For the compatibility of treenumpy, here we use make_key_as_identifier and remove_illegal_item functions to modify the original timestep.

Example:

>>> timesteps = env_manager.step(action)
>>> for timestep in timesteps:
>>>     if timestep.done:
>>>         print('Env {} is done'.format(timestep.env_id))

SyncSubprocessEnvManager

class ding.envs.SyncSubprocessEnvManager(env_fn: List[Callable], cfg: EasyDict = {})

__init__(env_fn: List[Callable], cfg: EasyDict = {}) → None

Overview:

Initialize the AsyncSubprocessEnvManager.

Arguments:

• env_fn (List[Callable]): The function to create environment

• cfg (EasyDict): Config

Note

• wait_num: for each time the minimum number of env return to gather

• step_wait_timeout: for each time the minimum number of env return to gather

close() → None

Overview:

CLose the env manager and release all related resources.

classmethod default_config() → EasyDict

Overview:

Return the deepcopyed default config of env manager.

Returns:

• cfg (EasyDict): The default config of env manager.

enable_save_replay(replay_path: List[str] | str) → None

Overview:

Set each env’s replay save path.

Arguments:

• replay_path (Union[List[str], str]): List of paths for each environment; Or one path for all environments.

launch(reset_param: Dict | None = None) → None

Overview:

Set up the environments and their parameters.

Arguments:

• reset_param (Optional[Dict]): Dict of reset parameters for each environment, key is the env_id, value is the cooresponding reset parameters.

property ready_imgs: Dict[int, Any]

Overview:

Get the next renderd frames.

Return:

A dictionary with rendered frames and their environment IDs.

Note:

The rendered frames are returned in np.ndarray.

property ready_obs: Dict[int, Any]

Overview:

Get the next observations.

Return:

A dictionary with observations and their environment IDs.

Note:

The observations are returned in np.ndarray.

Example:

>>>     obs_dict = env_manager.ready_obs
>>>     actions_dict = {env_id: model.forward(obs) for env_id, obs in obs_dict.items())}

reset(reset_param: Dict | None = None) → None

Overview:

Reset the environments their parameters.

Arguments:

• reset_param (List): Dict of reset parameters for each environment, key is the env_id, value is the cooresponding reset parameters.

seed(seed: Dict[int, int] | List[int] | int, dynamic_seed: bool | None = None) → None

Overview:

Set the random seed for each environment.

Arguments:

• seed (Union[Dict[int, int], List[int], int]): Dict or List of seeds for each environment; If only one seed is provided, it will be used in the same way for all environments.

• dynamic_seed (bool): Whether to use dynamic seed.

Note

For more details about dynamic_seed, please refer to the best practice document of DI-engine (en link2).

step(actions: Dict[int, Any]) → Dict[int, namedtuple]

Overview:

Step all environments. Reset an env if done.

Arguments:

• actions (Dict[int, Any]): {env_id: action}

Returns:

• timesteps (Dict[int, namedtuple]): {env_id: timestep}. Timestep is a BaseEnvTimestep tuple with observation, reward, done, env_info.

Example:

>>>     actions_dict = {env_id: model.forward(obs) for env_id, obs in obs_dict.items())}
>>>     timesteps = env_manager.step(actions_dict):
>>>     for env_id, timestep in timesteps.items():
>>>         pass

Note

• The env_id that appears in actions will also be returned in timesteps.

• Each environment is run by a subprocess separately. Once an environment is done, it is reset immediately.

static worker_fn(p: Connection, c: Connection, env_fn_wrapper: CloudPickleWrapper, obs_buffer: ShmBuffer, method_name_list: list, reset_inplace: bool = False) → None

Overview:

Subprocess’s target function to run.

static worker_fn_robust(parent, child, env_fn_wrapper, obs_buffer, method_name_list, reset_timeout=None, step_timeout=None, reset_inplace=False) → None

Overview:

A more robust version of subprocess’s target function to run. Used by default.

SubprocessEnvManagerV2

class ding.envs.SubprocessEnvManagerV2(env_fn: List[Callable], cfg: EasyDict = {})

Overview:

SyncSubprocessEnvManager for new task pipeline and interfaces coupled with treetensor.

__init__(env_fn: List[Callable], cfg: EasyDict = {}) → None

Overview:

Initialize the AsyncSubprocessEnvManager.

Arguments:

• env_fn (List[Callable]): The function to create environment

• cfg (EasyDict): Config

Note

• wait_num: for each time the minimum number of env return to gather

• step_wait_timeout: for each time the minimum number of env return to gather

close() → None

Overview:

CLose the env manager and release all related resources.

classmethod default_config() → EasyDict

Overview:

Return the deepcopyed default config of env manager.

Returns:

• cfg (EasyDict): The default config of env manager.

enable_save_replay(replay_path: List[str] | str) → None

Overview:

Set each env’s replay save path.

Arguments:

• replay_path (Union[List[str], str]): List of paths for each environment; Or one path for all environments.

launch(reset_param: Dict | None = None) → None

Overview:

Set up the environments and their parameters.

Arguments:

• reset_param (Optional[Dict]): Dict of reset parameters for each environment, key is the env_id, value is the cooresponding reset parameters.

property ready_imgs: Dict[int, Any]

Overview:

Get the next renderd frames.

Return:

A dictionary with rendered frames and their environment IDs.

Note:

The rendered frames are returned in np.ndarray.

property ready_obs: array

Overview:

Get the ready (next) observation in tnp.array type, which is uniform for both async/sync scenarios.

Return:

• ready_obs (tnp.array): A stacked treenumpy-type observation data.

Example:

>>> obs = env_manager.ready_obs
>>> action = model(obs)  # model input np obs and output np action
>>> timesteps = env_manager.step(action)

reset(reset_param: Dict | None = None) → None

Overview:

Reset the environments their parameters.

Arguments:

• reset_param (List): Dict of reset parameters for each environment, key is the env_id, value is the cooresponding reset parameters.

seed(seed: Dict[int, int] | List[int] | int, dynamic_seed: bool | None = None) → None

Overview:

Set the random seed for each environment.

Arguments:

• seed (Union[Dict[int, int], List[int], int]): Dict or List of seeds for each environment; If only one seed is provided, it will be used in the same way for all environments.

• dynamic_seed (bool): Whether to use dynamic seed.

Note

For more details about dynamic_seed, please refer to the best practice document of DI-engine (en link2).

step(actions: List[ndarray] | ndarray) → List[ndarray]

Overview:

Execute env step according to input actions. And reset an env if done.

Arguments:

• actions (Union[List[tnp.ndarray], tnp.ndarray]): actions came from outer caller like policy.

Returns:

• timesteps (List[tnp.ndarray]): Each timestep is a tnp.array with observation, reward, done, info, env_id.

static worker_fn(p: Connection, c: Connection, env_fn_wrapper: CloudPickleWrapper, obs_buffer: ShmBuffer, method_name_list: list, reset_inplace: bool = False) → None

Overview:

Subprocess’s target function to run.

static worker_fn_robust(parent, child, env_fn_wrapper, obs_buffer, method_name_list, reset_timeout=None, step_timeout=None, reset_inplace=False) → None

Overview:

A more robust version of subprocess’s target function to run. Used by default.

AsyncSubprocessEnvManager

class ding.envs.AsyncSubprocessEnvManager(env_fn: List[Callable], cfg: EasyDict = {})

Overview:

Create an AsyncSubprocessEnvManager to manage multiple environments. Each Environment is run by a respective subprocess.

Interfaces:

seed, launch, ready_obs, step, reset, active_env

__init__(env_fn: List[Callable], cfg: EasyDict = {}) → None

Overview:

Initialize the AsyncSubprocessEnvManager.

Arguments:

• env_fn (List[Callable]): The function to create environment

• cfg (EasyDict): Config

Note

• wait_num: for each time the minimum number of env return to gather

• step_wait_timeout: for each time the minimum number of env return to gather

close() → None

Overview:

CLose the env manager and release all related resources.

classmethod default_config() → EasyDict

Overview:

Return the deepcopyed default config of env manager.

Returns:

• cfg (EasyDict): The default config of env manager.

enable_save_replay(replay_path: List[str] | str) → None

Overview:

Set each env’s replay save path.

Arguments:

• replay_path (Union[List[str], str]): List of paths for each environment; Or one path for all environments.

launch(reset_param: Dict | None = None) → None

Overview:

Set up the environments and their parameters.

Arguments:

• reset_param (Optional[Dict]): Dict of reset parameters for each environment, key is the env_id, value is the cooresponding reset parameters.

property ready_imgs: Dict[int, Any]

Overview:

Get the next renderd frames.

Return:

A dictionary with rendered frames and their environment IDs.

Note:

The rendered frames are returned in np.ndarray.

property ready_obs: Dict[int, Any]

Overview:

Get the next observations.

Return:

A dictionary with observations and their environment IDs.

Note:

The observations are returned in np.ndarray.

Example:

>>>     obs_dict = env_manager.ready_obs
>>>     actions_dict = {env_id: model.forward(obs) for env_id, obs in obs_dict.items())}

reset(reset_param: Dict | None = None) → None

Overview:

Reset the environments their parameters.

Arguments:

• reset_param (List): Dict of reset parameters for each environment, key is the env_id, value is the cooresponding reset parameters.

seed(seed: Dict[int, int] | List[int] | int, dynamic_seed: bool | None = None) → None

Overview:

Set the random seed for each environment.

Arguments:

• seed (Union[Dict[int, int], List[int], int]): Dict or List of seeds for each environment; If only one seed is provided, it will be used in the same way for all environments.

• dynamic_seed (bool): Whether to use dynamic seed.

Note

For more details about dynamic_seed, please refer to the best practice document of DI-engine (en link2).

step(actions: Dict[int, Any]) → Dict[int, namedtuple]

Overview:

Step all environments. Reset an env if done.

Arguments:

• actions (Dict[int, Any]): {env_id: action}

Returns:

• timesteps (Dict[int, namedtuple]): {env_id: timestep}. Timestep is a BaseEnvTimestep tuple with observation, reward, done, env_info.

Example:

>>>     actions_dict = {env_id: model.forward(obs) for env_id, obs in obs_dict.items())}
>>>     timesteps = env_manager.step(actions_dict):
>>>     for env_id, timestep in timesteps.items():
>>>         pass

static worker_fn(p: Connection, c: Connection, env_fn_wrapper: CloudPickleWrapper, obs_buffer: ShmBuffer, method_name_list: list, reset_inplace: bool = False) → None

Overview:

Subprocess’s target function to run.

static worker_fn_robust(parent, child, env_fn_wrapper, obs_buffer, method_name_list, reset_timeout=None, step_timeout=None, reset_inplace=False) → None

Overview:

A more robust version of subprocess’s target function to run. Used by default.

GymVectorEnvManager

class ding.envs.GymVectorEnvManager(env_fn: List[Callable], cfg: EasyDict)

Overview:

Create an GymVectorEnvManager to manage multiple environments. Each Environment is run by a respective subprocess.

Interfaces:

seed, ready_obs, step, reset, close

__init__(env_fn: List[Callable], cfg: EasyDict) → None

Note

env_fn must create gym-type environment instance, which may different DI-engine environment.

close() → None

Overview:

Release the environment resources Since not calling super.__init__, no need to release BaseEnvManager’s resources

property ready_obs: Dict[int, Any]

Overview:

Get the ready (next) observation, which is a special design to unify both aysnc/sync env manager. For each interaction between policy and env, the policy will input the ready_obs and output the action. Then the env_manager will step with the action and prepare the next ready_obs.

Returns:

• ready_obs (Dict[int, Any]): A dict with env_id keys and observation values.

Example:

>>> obs = env_manager.ready_obs
>>> stacked_obs = np.concatenate(list(obs.values()))
>>> action = policy(obs)  # here policy inputs np obs and outputs np action
>>> action = {env_id: a for env_id, a in zip(obs.keys(), action)}
>>> timesteps = env_manager.step(action)

reset(reset_param: Dict | None = None) → None

Overview:

Forcely reset the sub-environments their corresponding parameters. Because in env manager all the sub-environments usually are reset automatically as soon as they are done, this method is only called when the caller must forcely reset all the sub-environments, such as in evaluation.

Arguments:

• reset_param (List): Dict of reset parameters for each environment, key is the env_id, value is the corresponding reset parameters.

seed(seed: Dict[int, int] | List[int] | int, dynamic_seed: bool | None = None) → None

Overview:

Set the random seed for each environment.

Arguments:

• seed (Union[Dict[int, int], List[int], int]): Dict or List of seeds for each environment; If only one seed is provided, it will be used in the same way for all environments.

• dynamic_seed (bool): Whether to use dynamic seed.

Note

For more details about dynamic_seed, please refer to the best practice document of DI-engine (en link2).

step(actions: Dict[int, Any]) → Dict[int, namedtuple]

Overview:

Execute env step according to input actions. If some sub-environments are done after this execution, they will be reset automatically when self._auto_reset is True, otherwise they need to be reset when the caller use the reset method of env manager.

Arguments:

• actions (Dict[int, Any]): A dict of actions, key is the env_id, value is corresponding action. action can be any type, it depends on the env, and the env will handle it. Ususlly, the action is a dict of numpy array, and the value is generated by the outer caller like policy.

Returns:

• timesteps (Dict[int, BaseEnvTimestep]): Each timestep is a BaseEnvTimestep object, usually including observation, reward, done, info. Some special customized environments will have the special timestep definition. The length of timesteps is the same as the length of actions in synchronous env manager.

Example:

>>> timesteps = env_manager.step(action)
>>> for env_id, timestep in enumerate(timesteps):
>>>     if timestep.done:
>>>         print('Env {} is done'.format(env_id))

Env Wrapper

Please refer to ding/envs/env_wrappers for more details.

create_env_wrapper

ding.envs.create_env_wrapper(env: Env, env_wrapper_cfg: EasyDict) → Wrapper

Overview:

Create an environment wrapper according to the environment wrapper configuration and the environment instance.

Arguments:

• env (gym.Env): The environment instance to be wrapped.

• env_wrapper_cfg (EasyDict): The configuration for the environment wrapper.

Returns:

• env (gym.Wrapper): The wrapped environment instance.

update_shape

ding.envs.update_shape(obs_shape: Any, act_shape: Any, rew_shape: Any, wrapper_names: List[str]) → Tuple[Any, Any, Any]

Overview:

Get new shapes of observation, action, and reward given the wrapper.

Arguments:

• obs_shape (Any): The original shape of observation.

• act_shape (Any): The original shape of action.

• rew_shape (Any): The original shape of reward.

• wrapper_names (List[str]): The names of the wrappers.

Returns:

• obs_shape (Any): The new shape of observation.

• act_shape (Any): The new shape of action.

• rew_shape (Any): The new shape of reward.

NoopResetWrapper

class ding.envs.NoopResetWrapper(env: Env, noop_max: int = 30)

Overview:

Sample initial states by taking random number of no-ops on reset. No-op is assumed to be action 0.

Interfaces:

__init__, reset

Properties:

• env (gym.Env): the environment to wrap.

• noop_max (int): the maximum value of no-ops to run.

__init__(env: Env, noop_max: int = 30)

Overview:

Initialize the NoopResetWrapper.

Arguments:

• env (gym.Env): the environment to wrap.

• noop_max (int): the maximum value of no-ops to run. Defaults to 30.

reset() → ndarray

Overview:

Resets the state of the environment and returns an initial observation, after taking a random number of no-ops.

Returns:

• observation (Any): The initial observation after no-ops.

MaxAndSkipWrapper

class ding.envs.MaxAndSkipWrapper(env: Env, skip: int = 4)

Overview:

Wraps the environment to return only every skip-th frame (frameskipping) using most recent raw observations (for max pooling across time steps).

Interfaces:

__init__, step

Properties:

• env (gym.Env): The environment to wrap.

• skip (int): Number of skip-th frame. Defaults to 4.

__init__(env: Env, skip: int = 4)

Overview:

Initialize the MaxAndSkipWrapper.

Arguments:

• env (gym.Env): The environment to wrap.

• skip (int): Number of skip-th frame. Defaults to 4.

step(action: int | ndarray) → tuple

Overview:

Take the given action and repeat it for a specified number of steps. The rewards are summed up and the maximum frame over the last observations is returned.

Arguments:

• action (Any): The action to repeat.

Returns:

• max_frame (np.array): Max over last observations

• total_reward (Any): Sum of rewards after previous action.

• done (Bool): Whether the episode has ended.

• info (Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes learning)

FireResetWrapper

class ding.envs.FireResetWrapper(env: Env)

Overview:

This wrapper takes a fire action at environment reset. Related discussion: https://github.com/openai/baselines/issues/240

Interfaces:

__init__, reset

Properties:

• env (gym.Env): The environment to wrap.

__init__(env: Env) → None

Overview:

Initialize the FireResetWrapper. Assume that the second action of the environment is ‘FIRE’ and there are at least three actions.

Arguments:

• env (gym.Env): The environment to wrap.

reset() → ndarray

Overview:

Resets the state of the environment and executes a fire action, i.e. reset with action 1.

Returns:

• observation (np.ndarray): New observation after reset and fire action.

EpisodicLifeWrapper

class ding.envs.EpisodicLifeWrapper(env: Env)

Overview:

This wrapper makes end-of-life equivalent to end-of-episode, but only resets on true game over. This helps in better value estimation.

Interfaces:

__init__, step, reset

Properties:

• env (gym.Env): The environment to wrap.

• lives (int): The current number of lives.

• was_real_done (bool): Whether the last episode was ended due to game over.

__init__(env: Env) → None

Overview:

Initialize the EpisodicLifeWrapper, setting lives to 0 and was_real_done to True.

Arguments:

• env (gym.Env): The environment to wrap.

step(action: Any) → Tuple[ndarray, float, bool, Dict]

Overview:

Execute the given action in the environment, update properties based on the new state and return the new observation, reward, done status and info.

Arguments:

• action (Any): The action to execute in the environment.

Returns:

• observation (np.ndarray): Normalized observation after the action execution and updated self.rms.

• reward (float): Amount of reward returned after the action execution.

• done (bool): Whether the episode has ended, in which case further step() calls will return

  undefined results.

• info (Dict): Contains auxiliary diagnostic information (helpful for debugging, and

  sometimes learning).

ClipRewardWrapper

class ding.envs.ClipRewardWrapper(env: Env)

Overview:

The ClipRewardWrapper class is a gym reward wrapper that clips the reward to {-1, 0, +1} based on its sign. This can be used to normalize the scale of the rewards in reinforcement learning algorithms.

Interfaces:

__init__, reward

Properties:

• env (gym.Env): the environment to wrap.

• reward_range (Tuple[int, int]): the range of the reward values after clipping.

__init__(env: Env)

Overview:

Initialize the ClipRewardWrapper class.

Arguments:

• env (gym.Env): the environment to wrap.

reward(reward: float) → float

Overview:

Clip the reward to {-1, 0, +1} based on its sign. Note: np.sign(0) == 0.

Arguments:

• reward (float): the original reward.

Returns:

• reward (float): the clipped reward.

FrameStackWrapper

class ding.envs.FrameStackWrapper(env: Env, n_frames: int = 4)

Overview:

FrameStackWrapper is a gym environment wrapper that stacks the latest n frames (generally 4 in Atari) as a single observation. It is commonly used in environments where the observation is an image, and consecutive frames provide useful temporal information for the agent.

Interfaces:

__init__, reset, step, _get_ob

Properties:

• env (gym.Env): The environment to wrap.

• n_frames (int): The number of frames to stack.

• frames (collections.deque): A queue that holds the most recent frames.

• observation_space (gym.Space): The space of the stacked observations.

__init__(env: Env, n_frames: int = 4) → None

Overview:

Initialize the FrameStackWrapper. This process includes setting up the environment to wrap, the number of frames to stack, and the observation space.

Arguments:

• env (gym.Env): The environment to wrap.

• n_frame (int): The number of frames to stack.

reset() → ndarray

Overview:

Reset the environment and initialize frames with the initial observation.

Returns:

• init_obs (np.ndarray): The stacked initial observations.

step(action: Any) → Tuple[ndarray, float, bool, Dict[str, Any]]

Overview:

Perform a step in the environment with the given action, append the returned observation to frames, and return the stacked observations.

Arguments:

• action (Any): The action to perform a step with.

Returns:

• self._get_ob() (np.ndarray): The stacked observations.

• reward (float): The amount of reward returned after the previous action.

• done (bool): Whether the episode has ended, in which case further step() calls will return undefined results.

• info (Dict[str, Any]): Contains auxiliary diagnostic information (helpful for debugging, and sometimes learning).

ScaledFloatFrameWrapper

class ding.envs.ScaledFloatFrameWrapper(env: Env)

Overview:

The ScaledFloatFrameWrapper normalizes observations to between 0 and 1.

Interfaces:

__init__, observation

__init__(env: Env)

Overview:

Initialize the ScaledFloatFrameWrapper, setting the scale and bias for normalization.

Arguments:

• env (gym.Env): the environment to wrap.

observation(observation: ndarray) → ndarray

Overview:

Scale the observation to be within the range [0, 1].

Arguments:

• observation (np.ndarray): the original observation.

Returns:

• scaled_observation (np.ndarray): the scaled observation.

WarpFrameWrapper

class ding.envs.WarpFrameWrapper(env: Env, size: int = 84)

Overview:

The WarpFrameWrapper class is a gym observation wrapper that resizes the frame of an environment observation to a specified size (default is 84x84). This is often used in the preprocessing pipeline of observations in reinforcement learning, especially for visual observations from Atari environments.

Interfaces:

__init__, observation

Properties:

• env (gym.Env): the environment to wrap.

• size (int): the size to which the frames are to be resized.

• observation_space (gym.Space): the observation space of the wrapped environment.

__init__(env: Env, size: int = 84)

Overview:

Constructor for WarpFrameWrapper class, initializes the environment and the size.

Arguments:

• env (gym.Env): the environment to wrap.

• size (int): the size to which the frames are to be resized. Default is 84.

observation(frame: ndarray) → ndarray

Overview:

Resize the frame (observation) to the desired size.

Arguments:

• frame (np.ndarray): the frame to be resized.

Returns:

• frame (np.ndarray): the resized frame.

ActionRepeatWrapper

class ding.envs.ActionRepeatWrapper(env: Env, action_repeat: int = 1)

Overview:

The ActionRepeatWrapper class is a gym wrapper that repeats the same action for a number of steps. This wrapper is particularly useful in environments where the desired effect is achieved by maintaining the same action across multiple time steps. For instance, some physical environments like motion control tasks might require consistent force input to produce a significant state change.

Using this wrapper can reduce the temporal complexity of the problem, as it allows the agent to perform multiple actions within a single time step. This can speed up learning, as the agent has fewer decisions to make within a time step. However, it may also sacrifice some level of decision-making precision, as the agent cannot change its action across successive time steps.

Note that the use of the ActionRepeatWrapper may not be suitable for all types of environments. Specifically, it may not be the best choice for environments where new decisions must be made at each time step, or where the time sequence of actions has a significant impact on the outcome.

Interfaces:

__init__, step

Properties:

• env (gym.Env): the environment to wrap.

• action_repeat (int): the number of times to repeat the action.

__init__(env: Env, action_repeat: int = 1)

Overview:

Initialize the ActionRepeatWrapper class.

Arguments:

• env (gym.Env): the environment to wrap.

• action_repeat (int): the number of times to repeat the action. Default is 1.

step(action: int | ndarray) → tuple

Overview:

Take the given action and repeat it for a specified number of steps. The rewards are summed up.

Arguments:

• action (Union[int, np.ndarray]): The action to repeat.

Returns:

• obs (np.ndarray): The observation after repeating the action.

• reward (float): The sum of rewards after repeating the action.

• done (bool): Whether the episode has ended.

• info (Dict): Contains auxiliary diagnostic information.

DelayRewardWrapper

class ding.envs.DelayRewardWrapper(env: Env, delay_reward_step: int = 0)

Overview:

The DelayRewardWrapper class is a gym wrapper that delays the reward. It cumulates the reward over a predefined number of steps and returns the cumulated reward only at the end of this interval. At other times, it returns a reward of 0.

This wrapper is particularly useful in environments where the impact of an action is not immediately observable, but rather delayed over several steps. For instance, in strategic games or planning tasks, the effect of an action may not be directly noticeable, but it contributes to a sequence of actions that leads to a reward. In these cases, delaying the reward to match the action-effect delay can make the learning process more consistent with the problem’s nature.

However, using this wrapper may increase the difficulty of learning, as the agent needs to associate its actions with delayed outcomes. It also introduces a non-standard reward structure, which could limit the applicability of certain reinforcement learning algorithms.

Note that the use of the DelayRewardWrapper may not be suitable for all types of environments. Specifically, it may not be the best choice for environments where the effect of actions is immediately observable and the reward should be assigned accordingly.

Interfaces:

__init__, reset, step

Properties:

• env (gym.Env): the environment to wrap.

• delay_reward_step (int): the number of steps over which to delay and cumulate the reward.

__init__(env: Env, delay_reward_step: int = 0)

Overview:

Initialize the DelayRewardWrapper class.

Arguments:

• env (gym.Env): the environment to wrap.

• delay_reward_step (int): the number of steps over which to delay and cumulate the reward.

reset() → ndarray

Overview:

Resets the state of the environment and resets the delay reward duration and current delay reward.

Returns:

• obs (np.ndarray): the initial observation of the environment.

step(action: int | ndarray) → tuple

Overview:

Take the given action and repeat it for a specified number of steps. The rewards are summed up. If the number of steps equals the delay reward step, return the cumulated reward and reset the delay reward duration and current delay reward. Otherwise, return a reward of 0.

Arguments:

• action (Union[int, np.ndarray]): the action to take in the step.

Returns:

• obs (np.ndarray): The observation after the step.

• reward (float): The cumulated reward after the delay reward step or 0.

• done (bool): Whether the episode has ended.

• info (Dict): Contains auxiliary diagnostic information.

ObsTransposeWrapper

class ding.envs.ObsTransposeWrapper(env: Env)

Overview:

The ObsTransposeWrapper class is a gym wrapper that transposes the observation to put the channel dimension first. This can be helpful for certain types of neural networks that expect the channel dimension to be the first dimension.

Interfaces:

__init__, observation

Properties:

• env (gym.Env): The environment to wrap.

• observation_space (gym.spaces.Box): The transformed observation space.

__init__(env: Env)

Overview:

Initialize the ObsTransposeWrapper class and update the observation space according to the environment’s observation space.

Arguments:

• env (gym.Env): The environment to wrap.

observation(obs: tuple | ndarray) → tuple | ndarray

Overview:

Transpose the observation to put the channel dimension first. If the observation is a tuple, each element in the tuple is transposed independently.

Arguments:

• obs (Union[tuple, np.ndarray]): The original observation.

Returns:

• obs (Union[tuple, np.ndarray]): The transposed observation.

ObsNormWrapper

class ding.envs.ObsNormWrapper(env: Env)

Overview:

The ObsNormWrapper class is a gym observation wrapper that normalizes observations according to running mean and standard deviation (std).

Interfaces:

__init__, step, reset, observation

Properties:

• env (gym.Env): the environment to wrap.

• data_count (int): the count of data points observed so far.

• clip_range (Tuple[int, int]): the range to clip the normalized observation.

• rms (RunningMeanStd): running mean and standard deviation of the observations.

__init__(env: Env)

Overview:

Initialize the ObsNormWrapper class.

Arguments:

• env (gym.Env): the environment to wrap.

observation(observation: ndarray) → ndarray

Overview:

Normalize the observation using the current running mean and std. If less than 30 data points have been observed, return the original observation.

Arguments:

• observation (np.ndarray): the original observation.

Returns:

• observation (np.ndarray): the normalized observation.

reset(**kwargs)

Overview:

Reset the environment and the properties related to the running mean and std.

Arguments:

• kwargs (Dict): keyword arguments to be passed to the environment’s reset function.

Returns:

• observation (np.ndarray): the initial observation of the environment.

step(action: int | ndarray)

Overview:

Take an action in the environment, update the running mean and std, and return the normalized observation.

Arguments:

• action (Union[int, np.ndarray]): the action to take in the environment.

Returns:

• obs (np.ndarray): the normalized observation after the action.

• reward (float): the reward after the action.

• done (bool): whether the episode has ended.

• info (Dict): contains auxiliary diagnostic information.

StaticObsNormWrapper

class ding.envs.StaticObsNormWrapper(env: Env, mean: ndarray, std: ndarray)

Overview:

The StaticObsNormWrapper class is a gym observation wrapper that normalizes observations according to a precomputed mean and standard deviation (std) from a fixed dataset.

Interfaces:

__init__, observation

Properties:

• env (gym.Env): the environment to wrap.

• mean (numpy.ndarray): the mean of the observations in the fixed dataset.

• std (numpy.ndarray): the standard deviation of the observations in the fixed dataset.

• clip_range (Tuple[int, int]): the range to clip the normalized observation.

__init__(env: Env, mean: ndarray, std: ndarray)

Overview:

Initialize the StaticObsNormWrapper class.

Arguments:

• env (gym.Env): the environment to wrap.

• mean (numpy.ndarray): the mean of the observations in the fixed dataset.

• std (numpy.ndarray): the standard deviation of the observations in the fixed dataset.

observation(observation: ndarray) → ndarray

Overview:

Normalize the given observation using the precomputed mean and std. The normalized observation is then clipped within the specified range.

Arguments:

• observation (np.ndarray): the original observation.

Returns:

• observation (np.ndarray): the normalized and clipped observation.

RewardNormWrapper

class ding.envs.RewardNormWrapper(env: Env, reward_discount: float)

Overview:

This wrapper class normalizes the reward according to running std. It extends the gym.RewardWrapper.

Interfaces:

__init__, step, reward, reset

Properties:

• env (gym.Env): The environment to wrap.

• cum_reward (numpy.ndarray): The cumulated reward, initialized as zero and updated in step method.

• reward_discount (float): The discount factor for reward.

• data_count (int): A counter for data, incremented in each step call.

• rms (RunningMeanStd): An instance of RunningMeanStd to compute the running mean and std of reward.

__init__(env: Env, reward_discount: float) → None

Overview:

Initialize the RewardNormWrapper, setup the properties according to running mean and std.

Arguments:

• env (gym.Env): The environment to wrap.

• reward_discount (float): The discount factor for reward.

reset(**kwargs)

Overview:

Resets the state of the environment and reset properties (NumType ones to 0, and self.rms as reset rms wrapper)

Arguments:

• kwargs (Dict): Reset with this key argumets

reward(reward: float) → float

Overview:

Normalize reward if data_count is more than 30.

Arguments:

• reward (float): The raw reward.

Returns:

• reward (float): Normalized reward.

step(action: Any) → Tuple[ndarray, float, bool, Dict]

Overview:

Step the environment with the given action, update properties and return the new observation, reward, done status and info.

Arguments:

• action (Any): The action to execute in the environment.

Returns:

• observation (np.ndarray): Normalized observation after executing the action and updated self.rms.

• reward (float): Amount of reward returned after the action execution (normalized) and updated

  self.cum_reward.

• done (bool): Whether the episode has ended, in which case further step() calls will return

  undefined results.

• info (Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes

  learning).

RamWrapper

class ding.envs.RamWrapper(env: Env, render: bool = False)

Overview:

This wrapper class wraps a RAM environment into an image-like environment. It extends the gym.Wrapper.

Interfaces:

__init__, reset, step

Properties:

• env (gym.Env): The environment to wrap.

• observation_space (gym.spaces.Box): The observation space of the wrapped environment.

__init__(env: Env, render: bool = False) → None

Overview:

Initialize the RamWrapper and set up the observation space to wrap the RAM environment.

Arguments:

• env (gym.Env): The environment to wrap.

• render (bool): Whether to render the environment, default is False.

reset() → ndarray

Overview:

Resets the state of the environment and returns a reshaped observation.

Returns:

• observation (np.ndarray): New observation after reset and reshaped.

step(action: Any) → Tuple[ndarray, Any, bool, Dict]

Overview:

Execute one step within the environment with the given action. Repeat action, sum reward and reshape the observation.

Arguments:

• action (Any): The action to take in the environment.

Returns:

• observation (np.ndarray): Reshaped observation after step with type restriction.

• reward (Any): Amount of reward returned after previous action.

• done (bool): Whether the episode has ended, in which case further step() calls will return undefined results.

• info (Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes learning).

EvalEpisodeReturnWrapper

class ding.envs.EvalEpisodeReturnWrapper(env: Env)

Overview:

A wrapper for a gym environment that accumulates rewards at every timestep, and returns the total reward at the end of the episode in info. This is used for evaluation purposes.

Interfaces:

__init__, reset, step

Properties:

• env (gym.Env): the environment to wrap.

__init__(env: Env)

Overview:

Initialize the EvalEpisodeReturnWrapper. This involves setting up the environment to wrap.

Arguments:

• env (gym.Env): The environment to wrap.

reset() → ndarray

Overview:

Reset the environment and initialize the accumulated reward to zero.

Returns:

• obs (np.ndarray): The initial observation from the environment.

step(action: Any) → tuple

Overview:

Step the environment with the provided action, accumulate the returned reward, and add the total reward to info if the episode is done.

Arguments:

• action (Any): The action to take in the environment.

Returns:

• obs (np.ndarray): The next observation from the environment.

• reward (float): The reward from taking the action.

• done (bool): Whether the episode is done.

• info (Dict[str, Any]): A dictionary of extra information, which includes ‘eval_episode_return’ if

  the episode is done.

Examples:

>>> env = gym.make("CartPole-v1")
>>> env = EvalEpisodeReturnWrapper(env)
>>> obs = env.reset()
>>> done = False
>>> while not done:
...     action = env.action_space.sample()  # Replace with your own policy
...     obs, reward, done, info = env.step(action)
...     if done:
...         print("Total episode reward:", info['eval_episode_return'])

GymHybridDictActionWrapper

class ding.envs.GymHybridDictActionWrapper(env: Env)

Overview:

Transform Gym-Hybrid’s original gym.spaces.Tuple action space to gym.spaces.Dict.

Interfaces:

__init__, action

Properties:

• env (gym.Env): The environment to wrap.

• action_space (gym.spaces.Dict): The new action space.

__init__(env: Env) → None

Overview:

Initialize the GymHybridDictActionWrapper, setting up the new action space.

Arguments:

• env (gym.Env): The environment to wrap.

step(action: Dict) → Tuple[Dict, float, bool, Dict]

Overview:

Execute the given action in the environment, transform the action from Dict to Tuple, and return the new observation, reward, done status and info.

Arguments:

• action (Dict): The action to execute in the environment, structured as a dictionary.

Returns:

• observation (Dict): The wrapped observation, which includes the current observation,

  previous action and previous reward.

• reward (float): Amount of reward returned after the action execution.

• done (bool): Whether the episode has ended, in which case further step() calls will return

  undefined results.

• info (Dict): Contains auxiliary diagnostic information (helpful for debugging, and

  sometimes learning).

ObsPlusPrevActRewWrapper

class ding.envs.ObsPlusPrevActRewWrapper(env: Env)

Overview:

This wrapper is used in policy NGU. It sets a dict as the new wrapped observation, which includes the current observation, previous action and previous reward.

Interfaces:

__init__, reset, step

Properties:

• env (gym.Env): The environment to wrap.

• prev_action (int): The previous action.

• prev_reward_extrinsic (float): The previous reward.

__init__(env: Env) → None

Overview:

Initialize the ObsPlusPrevActRewWrapper, setting up the previous action and reward.

Arguments:

• env (gym.Env): The environment to wrap.

reset() → Dict

Overview:

Resets the state of the environment, and returns the wrapped observation.

Returns:

• observation (Dict): The wrapped observation, which includes the current observation,

  previous action and previous reward.

step(action: Any) → Tuple[Dict, float, bool, Dict]

Overview:

Execute the given action in the environment, save the previous action and reward to be used in the next observation, and return the new observation, reward, done status and info.

Arguments:

• action (Any): The action to execute in the environment.

Returns:

• observation (Dict): The wrapped observation, which includes the current observation,

  previous action and previous reward.

• reward (float): Amount of reward returned after the action execution.

• done (bool): Whether the episode has ended, in which case further step() calls will return

  undefined results.

• info (Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes

  learning).

TimeLimitWrapper

class ding.envs.TimeLimitWrapper(env: Env, max_limit: int)

Overview:

This class is used to enforce a time limit on the environment.

Interfaces:

__init__, reset, step

__init__(env: Env, max_limit: int) → None

Overview:

Initialize the TimeLimitWrapper, setting up the maximum limit of time steps.

Arguments:

• env (gym.Env): The environment to wrap.

• max_limit (int): The maximum limit of time steps.

reset() → ndarray

Overview:

Resets the state of the environment and the time counter.

Returns:

• observation (np.ndarray): The new observation after reset.

step(action: Any) → Tuple[ndarray, float, bool, Dict]

Overview:

Execute the given action in the environment, update the time counter, and return the new observation, reward, done status and info.

Arguments:

• action (Any): The action to execute in the environment.

Returns:

• observation (np.ndarray): The new observation after the action execution.

• reward (float): Amount of reward returned after the action execution.

• done (bool): Whether the episode has ended, in which case further step() calls will return

  undefined results.

• info (Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes

  learning).

FlatObsWrapper

class ding.envs.FlatObsWrapper(env: Env, maxStrLen: int = 96)

Overview:

This class is used to flatten the observation space of the environment. Note: only suitable for environments like minigrid.

Interfaces:

__init__, observation, reset, step

__init__(env: Env, maxStrLen: int = 96) → None

Overview:

Initialize the FlatObsWrapper, setup the new observation space.

Arguments:

• env (gym.Env): The environment to wrap.

• maxStrLen (int): The maximum length of mission string, default is 96.

observation(obs: ndarray | Tuple) → ndarray

Overview:

Process the observation, convert the mission into one-hot encoding and concatenate it with the image data.

Arguments:

• obs (Union[np.ndarray, Tuple]): The raw observation to process.

Returns:

• obs (np.ndarray): The processed observation.

reset(*args, **kwargs) → ndarray

Overview:

Resets the state of the environment and returns the processed observation.

Returns:

• observation (np.ndarray): The processed observation after reset.

step(*args, **kwargs) → Tuple[ndarray, float, bool, Dict]

Overview:

Execute the given action in the environment, and return the processed observation, reward, done status, and info.

Returns:

• observation (np.ndarray): The processed observation after the action execution.

• reward (float): Amount of reward returned after the action execution.

• done (bool): Whether the episode has ended, in which case further step() calls will return

  undefined results.

• info (Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes

  learning).

GymToGymnasiumWrapper

class ding.envs.GymToGymnasiumWrapper(env: Env)

Overview:

This class is used to wrap a gymnasium environment to a gym environment.

Interfaces:

__init__, seed, reset, step

__init__(env: Env) → None

Overview:

Initialize the GymToGymnasiumWrapper.

Arguments:

• env (gymnasium.Env): The gymnasium environment to wrap.

reset() → ndarray

Overview:

Resets the state of the environment and returns the new observation. If a seed was set, use it in the reset.

Returns:

• observation (np.ndarray): The new observation after reset.

seed(seed: int) → None

Overview:

Set the seed for the environment.

Arguments:

• seed (int): The seed to set.

AllinObsWrapper

class ding.envs.AllinObsWrapper(env: Env)

Overview:

This wrapper is used in policy Decision Transformer, which is proposed in paper https://arxiv.org/abs/2106.01345. It sets a dict {‘obs’: obs, ‘reward’: reward} as the new wrapped observation, which includes the current observation and previous reward.

Interfaces:

__init__, reset, step, seed

Properties:

• env (gym.Env): The environment to wrap.

__init__(env: Env) → None

Overview:

Initialize the AllinObsWrapper.

Arguments:

• env (gym.Env): The environment to wrap.

reset() → Dict

Overview:

Resets the state of the environment and returns the new observation.

Returns:

• observation (Dict): The new observation after reset, includes the current observation and reward.

seed(seed: int, dynamic_seed: bool = True) → None

Overview:

Set the seed for the environment.

Arguments:

• seed (int): The seed to set.

• dynamic_seed (bool): Whether to use dynamic seed, default is True.

step(action: Any)

Overview:

Execute the given action in the environment, and return the new observation, reward, done status, and info.

Arguments:

• action (Any): The action to execute in the environment.

Returns:

• timestep (BaseEnvTimestep): The timestep after the action execution.