ding.envs.env_wrappers

ding.envs.env_wrappers

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.

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.

Full Source Code

../ding/envs/env_wrappers/__init__.py

1from .env_wrappers import *