`ding.envs`¶

`ding.envs` ¶

`BaseEnv` ¶

Bases: Env, ABC

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

`init(cfg)` `abstractmethod` ¶

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 (:obj:dict): Environment configuration in dict type.

`reset()` `abstractmethod` ¶

Overview

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

Returns: - obs (:obj:Any): Initial observation after reset.

`close()` `abstractmethod` ¶

Overview

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

`step(action)` `abstractmethod` ¶

Overview

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

Arguments: - action (:obj:Any): The action input to step with. Returns: - timestep (:obj:BaseEnv.timestep): The result timestep of env executing one step.

`seed(seed)` `abstractmethod` ¶

Overview

Set the seed for this env's random number generator(s).

Arguments: - seed (:obj:Any): Random seed.

`repr()` `abstractmethod` ¶

Overview

Return the information string of this env instance.

Returns: - info (:obj:str): Information of this env instance, like type and arguments.

`create_collector_env_cfg(cfg)` `staticmethod` ¶

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 (:obj: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 (:obj: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.

`create_evaluator_env_cfg(cfg)` `staticmethod` ¶

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 (:obj: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 (:obj:List[dict]): List of cfg including all the config evaluator envs.

`enable_save_replay(replay_path)` ¶

Overview

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

Arguments: - replay_path (:obj:str): The path to save replay file.

`random_action()` ¶

Overview

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

Returns: - random_action (:obj:Any): Action generated randomly.

`DingEnvWrapper` ¶

Bases: BaseEnv

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

`observation_space` `property` ¶

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.

`action_space` `property` ¶

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.

`reward_space` `property` ¶

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.

`init(env=None, cfg=None, seed_api=True, caller='collector', is_gymnasium=False)` ¶

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 (:obj:Union[gym.Env, gymnasium.Env]): An environment instance to be wrapped. - cfg (:obj:dict): The configuration dictionary to create an environment instance. - seed_api (:obj:bool): Whether to use seed API. Defaults to True. - caller (:obj:str): A string representing the caller of this method, including collector or evaluator. Different caller may need different wrappers. Default is 'collector'. - is_gymnasium (:obj:bool): Whether the environment is a gymnasium environment. Defaults to False, i.e., the environment is a gym environment.

`reset()` ¶

Overview

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

Returns: - obs (:obj:Dict): The new observation after reset.

`close()` ¶

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.

`seed(seed, dynamic_seed=True)` ¶

Overview

Set the seed for the environment.

Arguments: - seed (:obj:int): The seed to set. - dynamic_seed (:obj:bool): Whether to use dynamic seed, default is True.

`step(action)` ¶

Overview

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

Arguments: - action (:obj:Union[np.int64, np.ndarray]): The action to execute in the environment. Returns: - timestep (:obj:BaseEnvTimestep): The timestep after the action execution.

`random_action()` ¶

Overview

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

Returns: - action (:obj:np.ndarray): The random action.

`repr()` ¶

Overview

Return the string representation of the instance.

Returns: - str (:obj:str): The string representation of the instance.

`create_collector_env_cfg(cfg)` `staticmethod` ¶

Overview

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

Arguments: - cfg (:obj:dict): The input configuration dictionary. Returns: - env_cfgs (:obj:List[dict]): The list of environment configurations for collectors.

`create_evaluator_env_cfg(cfg)` `staticmethod` ¶

Overview

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

Arguments: - cfg (:obj:dict): The input configuration dictionary. Returns: - env_cfgs (:obj:List[dict]): The list of environment configurations for evaluators.

`enable_save_replay(replay_path=None)` ¶

Overview

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

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

`clone(caller='collector')` ¶

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.

`NoopResetWrapper` ¶

Bases: Wrapper

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 (:obj:gym.Env): the environment to wrap. - noop_max (:obj:int): the maximum value of no-ops to run.

`init(env, noop_max=30)` ¶

Overview

Initialize the NoopResetWrapper.

Arguments: - env (:obj:gym.Env): the environment to wrap. - noop_max (:obj:int): the maximum value of no-ops to run. Defaults to 30.

`reset()` ¶

Overview

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

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

`MaxAndSkipWrapper` ¶

Bases: Wrapper

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 (:obj:gym.Env): The environment to wrap. - skip (:obj:int): Number of skip-th frame. Defaults to 4.

`init(env, skip=4)` ¶

Overview

Initialize the MaxAndSkipWrapper.

Arguments: - env (:obj:gym.Env): The environment to wrap. - skip (:obj:int): Number of skip-th frame. Defaults to 4.

`step(action)` ¶

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 (:obj:Any): The action to repeat. Returns: - max_frame (:obj:np.array): Max over last observations - total_reward (:obj:Any): Sum of rewards after previous action. - done (:obj:Bool): Whether the episode has ended. - info (:obj:Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes learning)

`WarpFrameWrapper` ¶

Bases: ObservationWrapper

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 (:obj:gym.Env): the environment to wrap. - size (:obj:int): the size to which the frames are to be resized. - observation_space (:obj:gym.Space): the observation space of the wrapped environment.

`init(env, size=84)` ¶

Overview

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

Arguments: - env (:obj:gym.Env): the environment to wrap. - size (:obj:int): the size to which the frames are to be resized. Default is 84.

`observation(frame)` ¶

Overview

Resize the frame (observation) to the desired size.

Arguments: - frame (:obj:np.ndarray): the frame to be resized. Returns: - frame (:obj:np.ndarray): the resized frame.

`ScaledFloatFrameWrapper` ¶

Bases: ObservationWrapper

Overview

The ScaledFloatFrameWrapper normalizes observations to between 0 and 1.

Interfaces: init, observation

`init(env)` ¶

Overview

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

Arguments: - env (:obj:gym.Env): the environment to wrap.

`observation(observation)` ¶

Overview

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

Arguments: - observation (:obj:np.ndarray): the original observation. Returns: - scaled_observation (:obj:np.ndarray): the scaled observation.

`ClipRewardWrapper` ¶

Bases: RewardWrapper

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 (:obj:gym.Env): the environment to wrap. - reward_range (:obj:Tuple[int, int]): the range of the reward values after clipping.

`init(env)` ¶

Overview

Initialize the ClipRewardWrapper class.

Arguments: - env (:obj:gym.Env): the environment to wrap.

`reward(reward)` ¶

Overview

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

Arguments: - reward (:obj:float): the original reward. Returns: - reward (:obj:float): the clipped reward.

`ActionRepeatWrapper` ¶

Bases: Wrapper

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 (:obj:gym.Env): the environment to wrap. - action_repeat (:obj:int): the number of times to repeat the action.

`init(env, action_repeat=1)` ¶

Overview

Initialize the ActionRepeatWrapper class.

Arguments: - env (:obj:gym.Env): the environment to wrap. - action_repeat (:obj:int): the number of times to repeat the action. Default is 1.

`step(action)` ¶

Overview

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

Arguments: - action (:obj:Union[int, np.ndarray]): The action to repeat. Returns: - obs (:obj:np.ndarray): The observation after repeating the action. - reward (:obj:float): The sum of rewards after repeating the action. - done (:obj:bool): Whether the episode has ended. - info (:obj:Dict): Contains auxiliary diagnostic information.

`DelayRewardWrapper` ¶

Bases: Wrapper

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 (:obj:gym.Env): the environment to wrap. - delay_reward_step (:obj:int): the number of steps over which to delay and cumulate the reward.

`init(env, delay_reward_step=0)` ¶

Overview

Initialize the DelayRewardWrapper class.

Arguments: - env (:obj:gym.Env): the environment to wrap. - delay_reward_step (:obj:int): the number of steps over which to delay and cumulate the reward.

`reset()` ¶

Overview

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

Returns: - obs (:obj:np.ndarray): the initial observation of the environment.

`step(action)` ¶

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 (:obj:Union[int, np.ndarray]): the action to take in the step. Returns: - obs (:obj:np.ndarray): The observation after the step. - reward (:obj:float): The cumulated reward after the delay reward step or 0. - done (:obj:bool): Whether the episode has ended. - info (:obj:Dict): Contains auxiliary diagnostic information.

`EvalEpisodeReturnWrapper` ¶

Bases: Wrapper

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 (:obj:gym.Env): the environment to wrap.

`init(env)` ¶

Overview

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

Arguments: - env (:obj:gym.Env): The environment to wrap.

`reset()` ¶

Overview

Reset the environment and initialize the accumulated reward to zero.

Returns: - obs (:obj:np.ndarray): The initial observation from the environment.

`step(action)` ¶

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 (:obj:Any): The action to take in the environment. Returns: - obs (:obj:np.ndarray): The next observation from the environment. - reward (:obj:float): The reward from taking the action. - done (:obj:bool): Whether the episode is done. - info (:obj: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'])

`FrameStackWrapper` ¶

Bases: Wrapper

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 (:obj:gym.Env): The environment to wrap. - n_frames (:obj:int): The number of frames to stack. - frames (:obj:collections.deque): A queue that holds the most recent frames. - observation_space (:obj:gym.Space): The space of the stacked observations.

`init(env, n_frames=4)` ¶

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 (:obj:gym.Env): The environment to wrap. - n_frame (:obj:int): The number of frames to stack.

`reset()` ¶

Overview

Reset the environment and initialize frames with the initial observation.

Returns: - init_obs (:obj:np.ndarray): The stacked initial observations.

`step(action)` ¶

Overview

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

Arguments: - action (:obj:Any): The action to perform a step with. Returns: - self._get_ob() (:obj:np.ndarray): The stacked observations. - reward (:obj:float): The amount of reward returned after the previous action. - done (:obj:bool): Whether the episode has ended, in which case further step() calls will return undefined results. - info (:obj:Dict[str, Any]): Contains auxiliary diagnostic information (helpful for debugging, and sometimes learning).

`ObsTransposeWrapper` ¶

Bases: ObservationWrapper

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 (:obj:gym.Env): The environment to wrap. - observation_space (:obj:gym.spaces.Box): The transformed observation space.

`init(env)` ¶

Overview

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

Arguments: - env (:obj:gym.Env): The environment to wrap.

`observation(obs)` ¶

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 (:obj:Union[tuple, np.ndarray]): The original observation. Returns: - obs (:obj:Union[tuple, np.ndarray]): The transposed observation.

`RunningMeanStd` ¶

Bases: object

Overview

The RunningMeanStd class is a utility that maintains a running mean and standard deviation calculation over a stream of data.

Interfaces: init, update, reset, mean, std Properties: - mean (:obj:np.ndarray): The running mean. - std (:obj:np.ndarray): The running standard deviation. - _epsilon (:obj:float): A small number to prevent division by zero when calculating standard deviation. - _shape (:obj:tuple): The shape of the data stream. - _mean (:obj:np.ndarray): The current mean of the data stream. - _var (:obj:np.ndarray): The current variance of the data stream. - _count (:obj:float): The number of data points processed.

`mean` `property` ¶

Overview

Get the current running mean.

Returns: The current running mean.

`std` `property` ¶

Overview

Get the current running standard deviation.

Returns: The current running mean.

`init(epsilon=0.0001, shape=())` ¶

Overview

Initialize the RunningMeanStd object.

Arguments: - epsilon (:obj:float, optional): A small number to prevent division by zero when calculating standard deviation. Default is 1e-4. - shape (:obj:tuple, optional): The shape of the data stream. Default is an empty tuple, which corresponds to scalars.

`update(x)` ¶

Overview

Update the running statistics with a new batch of data.

Arguments: - x (:obj:np.array): A batch of data.

`reset()` ¶

Overview

Resets the state of the environment and reset properties: _mean, _var, _count

`ObsNormWrapper` ¶

Bases: ObservationWrapper

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 (:obj:gym.Env): the environment to wrap. - data_count (:obj:int): the count of data points observed so far. - clip_range (:obj:Tuple[int, int]): the range to clip the normalized observation. - rms (:obj:RunningMeanStd): running mean and standard deviation of the observations.

`init(env)` ¶

Overview

Initialize the ObsNormWrapper class.

Arguments: - env (:obj:gym.Env): the environment to wrap.

`step(action)` ¶

Overview

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

Arguments: - action (:obj:Union[int, np.ndarray]): the action to take in the environment. Returns: - obs (:obj:np.ndarray): the normalized observation after the action. - reward (:obj:float): the reward after the action. - done (:obj:bool): whether the episode has ended. - info (:obj:Dict): contains auxiliary diagnostic information.

`observation(observation)` ¶

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 (:obj:np.ndarray): the original observation. Returns: - observation (:obj:np.ndarray): the normalized observation.

`reset(**kwargs)` ¶

Overview

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

Arguments: - kwargs (:obj:Dict): keyword arguments to be passed to the environment's reset function. Returns: - observation (:obj:np.ndarray): the initial observation of the environment.

`StaticObsNormWrapper` ¶

Bases: ObservationWrapper

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 (:obj:gym.Env): the environment to wrap. - mean (:obj:numpy.ndarray): the mean of the observations in the fixed dataset. - std (:obj:numpy.ndarray): the standard deviation of the observations in the fixed dataset. - clip_range (:obj:Tuple[int, int]): the range to clip the normalized observation.

`init(env, mean, std)` ¶

Overview

Initialize the StaticObsNormWrapper class.

Arguments: - env (:obj:gym.Env): the environment to wrap. - mean (:obj:numpy.ndarray): the mean of the observations in the fixed dataset. - std (:obj:numpy.ndarray): the standard deviation of the observations in the fixed dataset.

`observation(observation)` ¶

Overview

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

Arguments: - observation (:obj:np.ndarray): the original observation. Returns: - observation (:obj:np.ndarray): the normalized and clipped observation.

`RewardNormWrapper` ¶

Bases: RewardWrapper

Overview

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

Interfaces: init, step, reward, reset Properties: - env (:obj:gym.Env): The environment to wrap. - cum_reward (:obj:numpy.ndarray): The cumulated reward, initialized as zero and updated in step method. - reward_discount (:obj:float): The discount factor for reward. - data_count (:obj:int): A counter for data, incremented in each step call. - rms (:obj:RunningMeanStd): An instance of RunningMeanStd to compute the running mean and std of reward.

`init(env, reward_discount)` ¶

Overview

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

Arguments: - env (:obj:gym.Env): The environment to wrap. - reward_discount (:obj:float): The discount factor for reward.

`step(action)` ¶

Overview

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

Arguments: - action (:obj:Any): The action to execute in the environment. Returns: - observation (:obj:np.ndarray): Normalized observation after executing the action and updated self.rms. - reward (:obj:float): Amount of reward returned after the action execution (normalized) and updated self.cum_reward. - done (:obj:bool): Whether the episode has ended, in which case further step() calls will return undefined results. - info (:obj:Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes learning).

`reward(reward)` ¶

Overview

Normalize reward if data_count is more than 30.

Arguments: - reward (:obj:float): The raw reward. Returns: - reward (:obj:float): Normalized 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 (:obj:Dict): Reset with this key argumets

`RamWrapper` ¶

Bases: Wrapper

Overview

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

Interfaces: init, reset, step Properties: - env (:obj:gym.Env): The environment to wrap. - observation_space (:obj:gym.spaces.Box): The observation space of the wrapped environment.

`init(env, render=False)` ¶

Overview

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

Arguments: - env (:obj:gym.Env): The environment to wrap. - render (:obj:bool): Whether to render the environment, default is False.

`reset()` ¶

Overview

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

Returns: - observation (:obj:np.ndarray): New observation after reset and reshaped.

`step(action)` ¶

Overview

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

Arguments: - action (:obj:Any): The action to take in the environment. Returns: - observation (:obj:np.ndarray): Reshaped observation after step with type restriction. - reward (:obj:Any): Amount of reward returned after previous action. - done (:obj:bool): Whether the episode has ended, in which case further step() calls will return undefined results. - info (:obj:Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes learning).

`EpisodicLifeWrapper` ¶

Bases: Wrapper

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 (:obj:gym.Env): The environment to wrap. - lives (:obj:int): The current number of lives. - was_real_done (:obj:bool): Whether the last episode was ended due to game over.

`init(env)` ¶

Overview

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

Arguments: - env (:obj:gym.Env): The environment to wrap.

`step(action)` ¶

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 (:obj:Any): The action to execute in the environment. Returns: - observation (:obj:np.ndarray): Normalized observation after the action execution and updated self.rms. - reward (:obj:float): Amount of reward returned after the action execution. - done (:obj:bool): Whether the episode has ended, in which case further step() calls will return undefined results. - info (:obj:Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes learning).

`reset()` ¶

Overview

Resets the state of the environment and updates the number of lives, only when lives are exhausted. This way all states are still reachable even though lives are episodic, and the learner need not know about any of this behind-the-scenes.

Returns: - observation (:obj:np.ndarray): New observation after reset with no-op step to advance from terminal/lost life state.

`FireResetWrapper` ¶

Bases: Wrapper

Overview

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

Interfaces: init, reset Properties: - env (:obj:gym.Env): The environment to wrap.

`init(env)` ¶

Overview

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

Arguments: - env (:obj:gym.Env): The environment to wrap.

`reset()` ¶

Overview

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

Returns: - observation (:obj:np.ndarray): New observation after reset and fire action.

`GymHybridDictActionWrapper` ¶

Bases: ActionWrapper

Overview

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

Interfaces: init, action Properties: - env (:obj:gym.Env): The environment to wrap. - action_space (:obj:gym.spaces.Dict): The new action space.

`init(env)` ¶

Overview

Initialize the GymHybridDictActionWrapper, setting up the new action space.

Arguments: - env (:obj:gym.Env): The environment to wrap.

`step(action)` ¶

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 (:obj:Dict): The action to execute in the environment, structured as a dictionary. Returns: - observation (:obj:Dict): The wrapped observation, which includes the current observation, previous action and previous reward. - reward (:obj:float): Amount of reward returned after the action execution. - done (:obj:bool): Whether the episode has ended, in which case further step() calls will return undefined results. - info (:obj:Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes learning).

`ObsPlusPrevActRewWrapper` ¶

Bases: Wrapper

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 (:obj:gym.Env): The environment to wrap. - prev_action (:obj:int): The previous action. - prev_reward_extrinsic (:obj:float): The previous reward.

`init(env)` ¶

Overview

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

Arguments: - env (:obj:gym.Env): The environment to wrap.

`reset()` ¶

Overview

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

Returns: - observation (:obj:Dict): The wrapped observation, which includes the current observation, previous action and previous reward.

`step(action)` ¶

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 (:obj:Any): The action to execute in the environment. Returns: - observation (:obj:Dict): The wrapped observation, which includes the current observation, previous action and previous reward. - reward (:obj:float): Amount of reward returned after the action execution. - done (:obj:bool): Whether the episode has ended, in which case further step() calls will return undefined results. - info (:obj:Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes learning).

`TransposeWrapper` ¶

Bases: Wrapper

Overview

This class is used to transpose the observation space of the environment.

Interfaces

init, _process_obs, step, reset

`init(env)` ¶

Overview

Initialize the TransposeWrapper, setting up the new observation space.

Arguments: - env (:obj:gym.Env): The environment to wrap.

`step(action)` ¶

Overview

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

Arguments: - action (:obj:Any): The action to execute in the environment. Returns: - observation (:obj:np.ndarray): The processed observation after the action execution. - reward (:obj:float): Amount of reward returned after the action execution. - done (:obj:bool): Whether the episode has ended, in which case further step() calls will return undefined results. - info (:obj:Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes learning).

`reset()` ¶

Overview

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

Returns: - observation (:obj:np.ndarray): The processed observation after reset.

`TimeLimitWrapper` ¶

Bases: Wrapper

Overview

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

Interfaces: init, reset, step

`init(env, max_limit)` ¶

Overview

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

Arguments: - env (:obj:gym.Env): The environment to wrap. - max_limit (:obj:int): The maximum limit of time steps.

`reset()` ¶

Overview

Resets the state of the environment and the time counter.

Returns: - observation (:obj:np.ndarray): The new observation after reset.

`step(action)` ¶

Overview

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

Arguments: - action (:obj:Any): The action to execute in the environment. Returns: - observation (:obj:np.ndarray): The new observation after the action execution. - reward (:obj:float): Amount of reward returned after the action execution. - done (:obj:bool): Whether the episode has ended, in which case further step() calls will return undefined results. - info (:obj:Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes learning).

`FlatObsWrapper` ¶

Bases: Wrapper

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, maxStrLen=96)` ¶

Overview

Initialize the FlatObsWrapper, setup the new observation space.

Arguments: - env (:obj:gym.Env): The environment to wrap. - maxStrLen (:obj:int): The maximum length of mission string, default is 96.

`observation(obs)` ¶

Overview

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

Arguments: - obs (:obj:Union[np.ndarray, Tuple]): The raw observation to process. Returns: - obs (:obj:np.ndarray): The processed observation.

`reset(*args, **kwargs)` ¶

Overview

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

Returns: - observation (:obj:np.ndarray): The processed observation after reset.

`step(*args, **kwargs)` ¶

Overview

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

Returns: - observation (:obj:np.ndarray): The processed observation after the action execution. - reward (:obj:float): Amount of reward returned after the action execution. - done (:obj:bool): Whether the episode has ended, in which case further step() calls will return undefined results. - info (:obj:Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes learning).

`GymToGymnasiumWrapper` ¶

Bases: Wrapper

Overview

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

Interfaces: init, seed, reset, step

`init(env)` ¶

Overview

Initialize the GymToGymnasiumWrapper.

Arguments: - env (:obj:gymnasium.Env): The gymnasium environment to wrap.

`seed(seed)` ¶

Overview

Set the seed for the environment.

Arguments: - seed (:obj:int): The seed to set.

`reset()` ¶

Overview

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

Returns: - observation (:obj:np.ndarray): The new observation after reset.

`step(*args, **kwargs)` ¶

Overview

Execute the given action in the environment, and return the new observation, reward, done status, and info. To keep consistency with gym, the done status should be the either terminated=True or truncated=True.

`AllinObsWrapper` ¶

Bases: Wrapper

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 (:obj:gym.Env): The environment to wrap.

`init(env)` ¶

Overview

Initialize the AllinObsWrapper.

Arguments: - env (:obj:gym.Env): The environment to wrap.

`reset()` ¶

Overview

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

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

`step(action)` ¶

Overview

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

Arguments: - action (:obj:Any): The action to execute in the environment. Returns: - timestep (:obj:BaseEnvTimestep): The timestep after the action execution.

`seed(seed, dynamic_seed=True)` ¶

Overview

Set the seed for the environment.

Arguments: - seed (:obj:int): The seed to set. - dynamic_seed (:obj:bool): Whether to use dynamic seed, default is True.

`BaseEnvManager` ¶

Bases: object

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

`env_num` `property` ¶

Overview

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

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

`env_ref` `property` ¶

Overview

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

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

`observation_space` `property` ¶

Overview

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

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

`action_space` `property` ¶

Overview

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

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

`reward_space` `property` ¶

Overview

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

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

`ready_obs` `property` ¶

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 (:obj: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)

`ready_obs_id` `property` ¶

Overview

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

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

`ready_imgs` `property` ¶

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 (:obj: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 (:obj:Dict[int, np.ndarray]): A dict with env_id keys and rendered frames.

`done` `property` ¶

Overview

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

Returns: - done (:obj:bool): Whether env manager is done.

`method_name_list` `property` ¶

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 (:obj:list): The public methods list of sub-environments.

`closed` `property` ¶

Overview

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

Returns: - closed (:obj:bool): Whether the env manager is closed.

`default_config()` `classmethod` ¶

Overview

Return the deepcopyed default config of env manager.

Returns: - cfg (:obj:EasyDict): The default config of env manager.

`init(env_fn, cfg=EasyDict({}))` ¶

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 (:obj:List[Callable]): A list of functions to create env_num sub-environments. - cfg (:obj:EasyDict): Final merged config.

.. note:: For more details about how to merge config, please refer to the system document of DI-engine (en link1 <../03_system/config.html>_).

`getattr(key)` ¶

Note

If a python object doesn't have the attribute whose name is key, it will call this method. We suppose that all envs have the same attributes. If you need different envs, please implement other env managers.

`launch(reset_param=None)` ¶

Overview

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

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

`reset(reset_param=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 (:obj:List): Dict of reset parameters for each environment, key is the env_id, value is the corresponding reset parameters.

`step(actions)` ¶

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 (:obj: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 (:obj: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))

`seed(seed, dynamic_seed=None)` ¶

Overview

Set the random seed for each environment.

Arguments: - seed (:obj: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 (:obj: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 <../04_best_practice/random_seed.html>_).

`enable_save_replay(replay_path)` ¶

Overview

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

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

`enable_save_figure(env_id, figure_path)` ¶

Overview

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

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

`close()` ¶

Overview

Close the env manager and release all the environment resources.

`reward_shaping(env_id, transitions)` ¶

Overview

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

Arguments: - env_id (:obj:int): The id of the environment to be shaped. - transitions (:obj:List[dict]): The transition data list of the environment to be shaped. Returns: - transitions (:obj:List[dict]): The shaped transition data list.

`BaseEnvManagerV2` ¶

Bases: BaseEnvManager

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 <../03_system/index.html>_).

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

`ready_obs` `property` ¶

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 (:obj: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)

`step(actions)` ¶

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 (:obj:List[tnp.ndarray]): A list of treenumpy-type actions, the value is generated by the outer caller like policy. Returns: - timesteps (:obj: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))

`AsyncSubprocessEnvManager` ¶

Bases: BaseEnvManager

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

`ready_obs` `property` ¶

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())}

`ready_imgs` `property` ¶

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.

`init(env_fn, cfg=EasyDict({}))` ¶

Overview

Initialize the AsyncSubprocessEnvManager.

Arguments: - env_fn (:obj:List[Callable]): The function to create environment - cfg (:obj: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

`launch(reset_param=None)` ¶

Overview

Set up the environments and their parameters.

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

`reset(reset_param=None)` ¶

Overview

Reset the environments their parameters.

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

`step(actions)` ¶

Overview

Step all environments. Reset an env if done.

Arguments: - actions (:obj:Dict[int, Any]): {env_id: action} Returns: - timesteps (:obj: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.
- Async subprocess env manager use ``connection.wait`` to poll.

`worker_fn(p, c, env_fn_wrapper, obs_buffer, method_name_list, reset_inplace=False)` `staticmethod` ¶

Overview

Subprocess's target function to run.

`worker_fn_robust(parent, child, env_fn_wrapper, obs_buffer, method_name_list, reset_timeout=None, step_timeout=None, reset_inplace=False)` `staticmethod` ¶

Overview

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

`enable_save_replay(replay_path)` ¶

Overview

Set each env's replay save path.

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

`close()` ¶

Overview

CLose the env manager and release all related resources.

`wait(rest_conn, wait_num, timeout=None)` `staticmethod` ¶

Overview

Wait at least enough(len(ready_conn) >= wait_num) connections within timeout constraint. If timeout is None and wait_num == len(ready_conn), means sync mode; If timeout is not None, will return when len(ready_conn) >= wait_num and this method takes more than timeout seconds.

`SyncSubprocessEnvManager` ¶

Bases: AsyncSubprocessEnvManager

`step(actions)` ¶

Overview

Step all environments. Reset an env if done.

Arguments: - actions (:obj:Dict[int, Any]): {env_id: action} Returns: - timesteps (:obj: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.

`SubprocessEnvManagerV2` ¶

Bases: SyncSubprocessEnvManager

Overview

SyncSubprocessEnvManager for new task pipeline and interfaces coupled with treetensor.

`ready_obs` `property` ¶

Overview

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

Return: - ready_obs (:obj: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)

`step(actions)` ¶

Overview

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

Arguments: - actions (:obj:Union[List[tnp.ndarray], tnp.ndarray]): actions came from outer caller like policy. Returns: - timesteps (:obj:List[tnp.ndarray]): Each timestep is a tnp.array with observation, reward, done, info, env_id.

`GymVectorEnvManager` ¶

Bases: BaseEnvManager

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, cfg)` ¶

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

`close()` ¶

Overview

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

`EnvSupervisor` ¶

Bases: Supervisor

Manage multiple envs with supervisor.

New features (compared to env manager): - Consistent interface in multi-process and multi-threaded mode. - Add asynchronous features and recommend using asynchronous methods. - Reset is performed after an error is encountered in the step method.

Breaking changes (compared to env manager): - Without some states.

`ready_obs` `property` ¶

Overview

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

Return: - ready_obs (:obj: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)

`init(type_=ChildType.PROCESS, env_fn=None, retry_type=EnvRetryType.RESET, max_try=None, max_retry=None, auto_reset=True, reset_timeout=None, step_timeout=None, retry_waiting_time=None, episode_num=float('inf'), shared_memory=True, copy_on_get=True, **kwargs)` ¶

Overview

Supervisor that manage a group of envs.

Arguments: - type_ (:obj:ChildType): Type of child process. - env_fn (:obj:List[Callable]): The function to create environment - retry_type (:obj:EnvRetryType): Retry reset or renew env. - max_try (:obj:EasyDict): Max try times for reset or step action. - max_retry (:obj:Optional[int]): Alias of max_try. - auto_reset (:obj:bool): Auto reset env if reach done. - reset_timeout (:obj:Optional[int]): Timeout in seconds for reset. - step_timeout (:obj:Optional[int]): Timeout in seconds for step. - retry_waiting_time (:obj:Optional[float]): Wait time on each retry. - shared_memory (:obj:bool): Use shared memory in multiprocessing. - copy_on_get (:obj:bool): Use copy on get in multiprocessing.

`step(actions, block=True)` ¶

Overview

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

Arguments: - actions (:obj:List[tnp.ndarray]): Actions came from outer caller like policy, in structure of {env_id: actions}. - block (:obj:bool): If block, return timesteps, else return none. Returns: - timesteps (:obj:List[tnp.ndarray]): Each timestep is a tnp.array with observation, reward, done, info, env_id.

`recv(ignore_err=False)` ¶

Overview

Wait for recv payload, this function will block the thread.

Arguments: - ignore_err (:obj:bool): If ignore_err is true, payload with error object will be discarded. This option will not catch the exception. Returns: - recv_payload (:obj:RecvPayload): Recv payload.

`launch(reset_param=None, block=True)` ¶

Overview

Set up the environments and their parameters.

Arguments: - reset_param (:obj:Optional[Dict]): Dict of reset parameters for each environment, key is the env_id, value is the cooresponding reset parameters. - block (:obj:block): Whether will block the process and wait for reset states.

`reset(reset_param=None, block=True)` ¶

Overview

Reset an environment.

Arguments: - reset_param (:obj:Optional[Dict[int, List[Any]]]): Dict of reset parameters for each environment, key is the env_id, value is the cooresponding reset parameters. - block (:obj:block): Whether will block the process and wait for reset states.

`seed(seed, dynamic_seed=None)` ¶

Overview

Set the seed for each environment. The seed function will not be called until supervisor.launch was called.

Arguments: - seed (:obj:Union[Dict[int, int], List[int], int]): List of seeds for each environment; Or one seed for the first environment and other seeds are generated automatically. Note that in threading mode, no matter how many seeds are given, only the last one will take effect. Because the execution in the thread is asynchronous, the results of each experiment are different even if a fixed seed is used. - dynamic_seed (:obj:Optional[bool]): Dynamic seed is used in the training environment, trying to make the random seed of each episode different, they are all generated in the reset method by a random generator 100 * np.random.randint(1 , 1000) (but the seed of this random number generator is fixed by the environmental seed method, guranteeing the reproducibility of the experiment). You need not pass the dynamic_seed parameter in the seed method, or pass the parameter as True.

`enable_save_replay(replay_path)` ¶

Overview

Set each env's replay save path.

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

`close(timeout=None)` ¶

In order to be compatible with BaseEnvManager, the new version can use shutdown directly.

`get_vec_env_setting(cfg, collect=True, eval_=True)` ¶

Overview

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

Arguments: - cfg (:obj:dict): Original input env config in user config, such as cfg.env. Returns: - env_fn (:obj:type): Callable object, call it with proper arguments and then get a new env instance. - collector_env_cfg (:obj:List[dict]): A list contains the config of collecting data envs. - evaluator_env_cfg (:obj: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(cfg)` ¶

Overview

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

Arguments: - cfg (:obj:dict): Original input env config in user config, such as cfg.env. Returns: - env_cls_type (:obj:type): Env module as the corresponding callable class type.

`create_model_env(cfg)` ¶

Overview

Create model env, which is used in model-based RL.

`get_default_wrappers(env_wrapper_name, env_id=None, caller='collector')` ¶

Overview

Get default wrappers for different environments used in DingEnvWrapper.

Arguments: - env_wrapper_name (:obj:str): The name of the environment wrapper. - env_id (:obj:Optional[str]): The id of the specific environment, such as PongNoFrameskip-v4. - caller (:obj:str): The caller of the environment, including collector or evaluator. Different caller may need different wrappers. Returns: - wrapper_list (:obj: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']

`to_ndarray(item, dtype=None)` ¶

Overview

Convert torch.Tensor to numpy.ndarray.

Arguments: - item (:obj:Any): The torch.Tensor objects to be converted. It can be exactly a torch.Tensor object or a container (list, tuple or dict) that contains several torch.Tensor objects. - dtype (:obj:np.dtype): The type of wanted array. If set to None, its dtype will be unchanged. Returns: - item (:obj:object): The changed arrays.

Examples (ndarray): >>> t = torch.randn(3, 5) >>> tarray1 = to_ndarray(t) >>> assert tarray1.shape == (3, 5) >>> assert isinstance(tarray1, np.ndarray)

Examples (list): >>> t = [torch.randn(5, ) for i in range(3)] >>> tarray1 = to_ndarray(t, np.float32) >>> assert isinstance(tarray1, list) >>> assert tarray1[0].shape == (5, ) >>> assert isinstance(tarray1[0], np.ndarray)

.. note:

Now supports item type: :obj:`torch.Tensor`,  :obj:`dict`, :obj:`list`, :obj:`tuple` and :obj:`None`.

`import_module(modules)` ¶

Overview

Import several module as a list

Arguments: - (:obj:str list): List of module names

`update_shape(obs_shape, act_shape, rew_shape, wrapper_names)` ¶

Overview

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

Arguments: - obs_shape (:obj:Any): The original shape of observation. - act_shape (:obj:Any): The original shape of action. - rew_shape (:obj:Any): The original shape of reward. - wrapper_names (:obj:List[str]): The names of the wrappers. Returns: - obs_shape (:obj:Any): The new shape of observation. - act_shape (:obj:Any): The new shape of action. - rew_shape (:obj:Any): The new shape of reward.

`create_env_wrapper(env, env_wrapper_cfg)` ¶

Overview

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

Arguments: - env (:obj:gym.Env): The environment instance to be wrapped. - env_wrapper_cfg (:obj:EasyDict): The configuration for the environment wrapper. Returns: - env (:obj:gym.Wrapper): The wrapped environment instance.

`create_env_manager(manager_cfg, env_fn)` ¶

Overview

Create an env manager according to manager_cfg and env functions.

Arguments: - manager_cfg (:obj:EasyDict): Final merged env manager config. - env_fn (:obj:List[Callable]): A list of functions to create env_num sub-environments. ArgumentsKeys: - type (:obj:str): Env manager type set in ENV_MANAGER_REGISTRY.register , such as base . - import_names (:obj: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 (:obj: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(cfg)` ¶

Overview

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

Arguments: - manager_cfg (:obj:EasyDict): Final merged env manager config. ArgumentsKeys: - type (:obj:str): Env manager type set in ENV_MANAGER_REGISTRY.register , such as base . - import_names (:obj: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 (:obj:type): The corresponding env manager class.

Full Source Code

../ding/envs/__init__.py

from .env import *from .env_wrappers import *from .env_manager import *from .env_manager.ding_env_manager import setup_ding_env_managerfrom . import gym_env

ding.envs¶

ding.envs ¶

BaseEnv ¶

__init__(cfg) abstractmethod ¶

reset() abstractmethod ¶

close() abstractmethod ¶

step(action) abstractmethod ¶

seed(seed) abstractmethod ¶

__repr__() abstractmethod ¶

create_collector_env_cfg(cfg) staticmethod ¶

create_evaluator_env_cfg(cfg) staticmethod ¶

enable_save_replay(replay_path) ¶

random_action() ¶

DingEnvWrapper ¶

observation_space property ¶

action_space property ¶

reward_space property ¶

__init__(env=None, cfg=None, seed_api=True, caller='collector', is_gymnasium=False) ¶

reset() ¶

close() ¶

seed(seed, dynamic_seed=True) ¶

step(action) ¶

random_action() ¶

__repr__() ¶

create_collector_env_cfg(cfg) staticmethod ¶

create_evaluator_env_cfg(cfg) staticmethod ¶

enable_save_replay(replay_path=None) ¶

clone(caller='collector') ¶

NoopResetWrapper ¶

__init__(env, noop_max=30) ¶

reset() ¶

MaxAndSkipWrapper ¶

__init__(env, skip=4) ¶

step(action) ¶

WarpFrameWrapper ¶

__init__(env, size=84) ¶

observation(frame) ¶

ScaledFloatFrameWrapper ¶

__init__(env) ¶

observation(observation) ¶

ClipRewardWrapper ¶

__init__(env) ¶

reward(reward) ¶

ActionRepeatWrapper ¶

__init__(env, action_repeat=1) ¶

step(action) ¶

DelayRewardWrapper ¶

__init__(env, delay_reward_step=0) ¶

reset() ¶

step(action) ¶

EvalEpisodeReturnWrapper ¶

__init__(env) ¶

reset() ¶

step(action) ¶

FrameStackWrapper ¶

__init__(env, n_frames=4) ¶

reset() ¶

step(action) ¶

ObsTransposeWrapper ¶

__init__(env) ¶

observation(obs) ¶

RunningMeanStd ¶

mean property ¶

std property ¶

__init__(epsilon=0.0001, shape=()) ¶

update(x) ¶

reset() ¶

ObsNormWrapper ¶

__init__(env) ¶

step(action) ¶

observation(observation) ¶

reset(**kwargs) ¶

StaticObsNormWrapper ¶

__init__(env, mean, std) ¶

observation(observation) ¶

RewardNormWrapper ¶

__init__(env, reward_discount) ¶

step(action) ¶

reward(reward) ¶

reset(**kwargs) ¶

`ding.envs`¶

`ding.envs` ¶

`BaseEnv` ¶

`init(cfg)` `abstractmethod` ¶

`reset()` `abstractmethod` ¶

`close()` `abstractmethod` ¶

`step(action)` `abstractmethod` ¶

`seed(seed)` `abstractmethod` ¶

`repr()` `abstractmethod` ¶

`create_collector_env_cfg(cfg)` `staticmethod` ¶

`create_evaluator_env_cfg(cfg)` `staticmethod` ¶

`enable_save_replay(replay_path)` ¶

`random_action()` ¶

`DingEnvWrapper` ¶

`observation_space` `property` ¶

`action_space` `property` ¶

`reward_space` `property` ¶

`init(env=None, cfg=None, seed_api=True, caller='collector', is_gymnasium=False)` ¶

`reset()` ¶

`close()` ¶

`seed(seed, dynamic_seed=True)` ¶

`step(action)` ¶

`random_action()` ¶

`repr()` ¶

`create_collector_env_cfg(cfg)` `staticmethod` ¶

`create_evaluator_env_cfg(cfg)` `staticmethod` ¶

`enable_save_replay(replay_path=None)` ¶

`clone(caller='collector')` ¶

`NoopResetWrapper` ¶

`init(env, noop_max=30)` ¶

`reset()` ¶

`MaxAndSkipWrapper` ¶

`init(env, skip=4)` ¶

`step(action)` ¶

`WarpFrameWrapper` ¶

`init(env, size=84)` ¶

`observation(frame)` ¶

`ScaledFloatFrameWrapper` ¶

`init(env)` ¶

`observation(observation)` ¶

`ClipRewardWrapper` ¶

`init(env)` ¶

`reward(reward)` ¶

`ActionRepeatWrapper` ¶

`init(env, action_repeat=1)` ¶

`step(action)` ¶

`DelayRewardWrapper` ¶

`init(env, delay_reward_step=0)` ¶

`reset()` ¶

`step(action)` ¶

`EvalEpisodeReturnWrapper` ¶

`init(env)` ¶

`reset()` ¶

`step(action)` ¶

`FrameStackWrapper` ¶

`init(env, n_frames=4)` ¶

`reset()` ¶

`step(action)` ¶

`ObsTransposeWrapper` ¶

`init(env)` ¶

`observation(obs)` ¶

`RunningMeanStd` ¶

`mean` `property` ¶

`std` `property` ¶

`init(epsilon=0.0001, shape=())` ¶

`update(x)` ¶

`reset()` ¶

`ObsNormWrapper` ¶

`init(env)` ¶

`step(action)` ¶

`observation(observation)` ¶

`reset(**kwargs)` ¶

`StaticObsNormWrapper` ¶

`init(env, mean, std)` ¶

`observation(observation)` ¶

`RewardNormWrapper` ¶

`init(env, reward_discount)` ¶

`step(action)` ¶

`reward(reward)` ¶

`reset(**kwargs)` ¶