杂项包装器¶
常用包装器¶
- class gymnasium.wrappers.TimeLimit(env: Env, max_episode_steps: int)[source]¶
通过在超过最大步数时截断环境来限制环境的步数。
如果环境本身未定义截断,这是发出截断信号的唯一位置。至关重要的是,这不同于源自基础环境作为 MDP 一部分的 terminated 信号。不存在矢量包装器。
- 使用 TimeLimit 包装器的示例
>>> from gymnasium.wrappers import TimeLimit >>> from gymnasium.envs.classic_control import CartPoleEnv
>>> spec = gym.spec("CartPole-v1") >>> spec.max_episode_steps 500 >>> env = gym.make("CartPole-v1") >>> env # TimeLimit is included within the environment stack <TimeLimit<OrderEnforcing<PassiveEnvChecker<CartPoleEnv<CartPole-v1>>>>> >>> env.spec EnvSpec(id='CartPole-v1', ..., max_episode_steps=500, ...) >>> env = gym.make("CartPole-v1", max_episode_steps=3) >>> env.spec EnvSpec(id='CartPole-v1', ..., max_episode_steps=3, ...) >>> env = TimeLimit(CartPoleEnv(), max_episode_steps=10) >>> env <TimeLimit<CartPoleEnv instance>>
- TimeLimit 确定剧集步骤的示例
>>> env = gym.make("CartPole-v1", max_episode_steps=3) >>> _ = env.reset(seed=123) >>> _ = env.action_space.seed(123) >>> _, _, terminated, truncated, _ = env.step(env.action_space.sample()) >>> terminated, truncated (False, False) >>> _, _, terminated, truncated, _ = env.step(env.action_space.sample()) >>> terminated, truncated (False, False) >>> _, _, terminated, truncated, _ = env.step(env.action_space.sample()) >>> terminated, truncated (False, True)
- 更改日志
v0.10.6 - 最初添加
v0.25.0 - 随着 step API 的更新,终止和截断信号将分别返回。
- 参数:
env – 要应用包装器的环境
max_episode_steps – 剧集被截断后的环境步数 (
elapsed >= max_episode_steps
)
- class gymnasium.wrappers.RecordVideo(env: gym.Env[ObsType, ActType], video_folder: str, episode_trigger: Callable[[int], bool] | None = None, step_trigger: Callable[[int], bool] | None = None, video_length: int = 0, name_prefix: str = 'rl-video', fps: int | None = None, disable_logger: bool = True)[source]¶
使用环境的渲染函数录制环境剧集的视频。
通常,你只希望间歇性地录制剧集,比如每百个剧集或每千个环境步骤录制一次。为此,你可以指定
episode_trigger
或step_trigger
。它们应该是返回布尔值的函数,表示是否应该在当前剧集或步骤分别开始录制。episode_trigger
应该在录制应该开始的剧集返回True
。step_trigger
应该在录制应该开始的第 n 个环境步骤返回True
,其中 n 对所有之前的剧集求和。如果既没有传递episode_trigger
也没有传递step_trigger
,则将使用默认的episode_trigger
,即capped_cubic_video_schedule()
。此函数将在每个 3 的幂次方的剧集开始视频,直到 1000,然后每 1000 个剧集开始视频。默认情况下,录制将在调用重置时停止。但是,你也可以通过传递video_length
的严格正值来创建固定长度的录制(可能跨越多个剧集)。不存在此包装器的矢量版本。
- 示例 - 运行环境 50 个剧集,从第 0 个剧集开始每 10 个剧集保存一次视频
>>> import os >>> import gymnasium as gym >>> env = gym.make("LunarLander-v3", render_mode="rgb_array") >>> trigger = lambda t: t % 10 == 0 >>> env = RecordVideo(env, video_folder="./save_videos1", episode_trigger=trigger, disable_logger=True) >>> for i in range(50): ... termination, truncation = False, False ... _ = env.reset(seed=123) ... while not (termination or truncation): ... obs, rew, termination, truncation, info = env.step(env.action_space.sample()) ... >>> env.close() >>> len(os.listdir("./save_videos1")) 5
- 示例 - 运行环境 5 个剧集,每 200 个步骤开始录制一次,确保每个视频有 100 帧
>>> import os >>> import gymnasium as gym >>> env = gym.make("LunarLander-v3", render_mode="rgb_array") >>> trigger = lambda t: t % 200 == 0 >>> env = RecordVideo(env, video_folder="./save_videos2", step_trigger=trigger, video_length=100, disable_logger=True) >>> for i in range(5): ... termination, truncation = False, False ... _ = env.reset(seed=123) ... _ = env.action_space.seed(123) ... while not (termination or truncation): ... obs, rew, termination, truncation, info = env.step(env.action_space.sample()) ... >>> env.close() >>> len(os.listdir("./save_videos2")) 2
- 示例 - 运行 3 个剧集,录制所有内容,但以 1000 帧为块
>>> import os >>> import gymnasium as gym >>> env = gym.make("LunarLander-v3", render_mode="rgb_array") >>> env = RecordVideo(env, video_folder="./save_videos3", video_length=1000, disable_logger=True) >>> for i in range(3): ... termination, truncation = False, False ... _ = env.reset(seed=123) ... while not (termination or truncation): ... obs, rew, termination, truncation, info = env.step(env.action_space.sample()) ... >>> env.close() >>> len(os.listdir("./save_videos3")) 2
- 更改日志
v0.25.0 - 最初添加以替换
wrappers.monitoring.VideoRecorder
- 参数:
env – 将要被包装的环境
video_folder (str) – 将存储录制的文件夹
episode_trigger – 接受整数并返回
True
的函数,当且仅当应该在此剧集开始录制时step_trigger – 接受整数并返回
True
的函数,当且仅当应该在此步骤开始录制时video_length (int) – 录制剧集的长度。如果为 0,则录制整个剧集。否则,将捕获指定长度的片段
name_prefix (str) – 将预先附加到录制的文件名
fps (int) – 视频的每秒帧数。为环境提供自定义视频 fps,如果为
None
,则使用环境元数据render_fps
键(如果存在),否则使用默认值 30。disable_logger (bool) – 是否禁用 moviepy 日志记录器,默认情况下禁用
- class gymnasium.wrappers.RecordEpisodeStatistics(env: Env[ObsType, ActType], buffer_length: int = 100, stats_key: str = 'episode')[source]¶
此包装器将跟踪累积奖励和剧集长度。
在剧集结束时,剧集的统计数据将使用键
episode
添加到info
中。如果使用向量化环境,还会使用键_episode
,它指示相应索引处的环境是否具有剧集统计数据。包装器的向量版本存在,gymnasium.wrappers.vector.RecordEpisodeStatistics
。剧集完成之后,
info
将看起来像这样>>> info = { ... "episode": { ... "r": "<cumulative reward>", ... "l": "<episode length>", ... "t": "<elapsed time since beginning of episode>" ... }, ... }
对于向量化环境,输出将采用以下形式
>>> infos = { ... "episode": { ... "r": "<array of cumulative reward>", ... "l": "<array of episode length>", ... "t": "<array of elapsed time since beginning of episode>" ... }, ... "_episode": "<boolean array of length num-envs>" ... }
此外,最新的奖励和剧集长度存储在缓冲区中,可以通过
wrapped_env.return_queue
和wrapped_env.length_queue
分别访问。- 变量:
time_queue (*) – 最后
deque_size
个剧集的时间长度return_queue (*) – 最后
deque_size
个剧集的累积奖励length_queue (*) – 最后
deque_size
个剧集的长度
- 更改日志
v0.15.4 - 最初添加
v1.0.0 - 删除了向量环境支持(参见
gymnasium.wrappers.vector.RecordEpisodeStatistics
)并添加属性time_queue
- 参数:
env (Env) – 要应用包装器的环境
buffer_length – 缓冲区
return_queue
,length_queue
和time_queue
的大小stats_key – 剧集统计数据的 info 键
- class gymnasium.wrappers.AtariPreprocessing(env: gym.Env, noop_max: int = 30, frame_skip: int = 4, screen_size: int | tuple[int, int] = 84, terminal_on_life_loss: bool = False, grayscale_obs: bool = True, grayscale_newaxis: bool = False, scale_obs: bool = False)[source]¶
实现了 Atari 环境的常用预处理技术(不包括帧堆叠)。
对于帧堆叠,请使用
gymnasium.wrappers.FrameStackObservation
。包装器没有向量版本此类遵循 Machado 等人(2018 年)的指南,“重温街机学习环境:通用智能体的评估协议和开放问题”。
具体来说,以下预处理阶段适用于 Atari 环境
无操作重置:通过在重置时执行随机数量的无操作操作来获取初始状态,默认最大 30 个无操作操作。
帧跳过:在步长之间跳过的帧数,默认值为 4。
最大池化:在最近两次来自帧跳过的观察结果上进行池化。
- 生命值损失时的终止信号:当代理在环境中损失生命时,环境将终止。
默认情况下已关闭。Machado 等人(2018 年)不建议这样做。
调整为方形图像:默认情况下将 Atari 环境的原始观察形状从 210x180 调整为 84x84。
灰度观察:使观察结果变为灰度,默认情况下已启用。
灰度新轴:扩展观察结果的最后一个通道,使得图像为 3 维,默认情况下未启用。
缩放观察:是否将观察结果缩放到 [0, 1) 或 [0, 255) 之间,默认情况下未缩放。
示例
>>> import gymnasium as gym >>> import ale_py >>> gym.register_envs(ale_py) >>> env = gym.make("ALE/Pong-v5", frameskip=1) >>> env = AtariPreprocessing( ... env, ... noop_max=10, frame_skip=4, terminal_on_life_loss=True, ... screen_size=84, grayscale_obs=False, grayscale_newaxis=False ... )
- 更改日志
在 gym v0.12.2 中添加(gym #1455)
- 参数:
env (Env) – 要应用预处理的环境
noop_max (int) – 对于无操作重置,重置时采取的最大无操作操作数,要关闭,请设置为 0。
frame_skip (int) – 影响代理体验游戏的频率,代理观察结果之间的新观察结果之间的帧数。
screen_size (int | tuple[int, int]) – 调整 Atari 帧的大小。
terminal_on_life_loss (bool) – 如果为 True,则
step()
只要生命值损失就返回terminated=True。grayscale_obs (bool) – 如果为 True,则返回灰度观察结果,否则返回 RGB 观察结果。
grayscale_newaxis (bool) – 如果为 True 且 grayscale_obs=True,则会向灰度观察结果添加一个通道轴,使其成为 3 维。
scale_obs (bool) – 如果为 True,则返回范围在 [0,1) 内的归一化观察结果。它还限制了 FrameStack 包装器的内存优化优势。
- 引发:
DependencyNotInstalled – 未安装 opencv-python 包
ValueError – 在原始环境中禁用帧跳过
不常见包装器¶
- class gymnasium.wrappers.Autoreset(env: Env)[source]¶
当达到终止或截断状态时,包装环境会自动重置。
这遵循向量自动重置 API,其中在剧集终止或截断后的步骤中,环境将重置。
- 更改日志
v0.24.0 - 最初添加为 AutoResetWrapper
v1.0.0 - 重命名为 Autoreset,并且自动重置顺序已更改为在环境终止或截断后的步骤中重置。因此,“final_observation” 和 “final_info” 已被删除。
- 参数:
env (gym.Env) – 要应用包装器的环境
- class gymnasium.wrappers.PassiveEnvChecker(env: Env[ObsType, ActType])[source]¶
一个被动包装器,它围绕着
step
,reset
和render
函数,以检查它们是否遵循 Gymnasium 的 API。此包装器在 make 期间会自动应用,可以通过 disable_env_checker 禁用。包装器没有向量版本。
示例
>>> import gymnasium as gym >>> env = gym.make("CartPole-v1") >>> env <TimeLimit<OrderEnforcing<PassiveEnvChecker<CartPoleEnv<CartPole-v1>>>>> >>> env = gym.make("CartPole-v1", disable_env_checker=True) >>> env <TimeLimit<OrderEnforcing<CartPoleEnv<CartPole-v1>>>>
- 更改日志
v0.24.1 - 最初添加,但在某些方面存在缺陷
v0.25.0 - 错误已修复
v0.29.0 - 删除了对 Box 观察空间和动作空间的无限边界以及不规则边界形状的警告
使用环境初始化包装器,运行观察空间和动作空间测试。
- class gymnasium.wrappers.HumanRendering(env: Env[ObsType, ActType])[source]¶
允许对支持“rgb_array”渲染的环境进行类似人类的渲染。
此包装器在您已实现一个可以生成 RGB 图像的环境但尚未实现任何代码将图像渲染到屏幕上时特别有用。如果您想在您的环境中使用此包装器,请记住在环境的元数据中指定
"render_fps"
。包装环境的
render_mode
必须是'rgb_array'
或'rgb_array_list'
。不存在此包装器的矢量版本。
示例
>>> import gymnasium as gym >>> from gymnasium.wrappers import HumanRendering >>> env = gym.make("LunarLander-v3", render_mode="rgb_array") >>> wrapped = HumanRendering(env) >>> obs, _ = wrapped.reset() # This will start rendering to the screen
包装器也可以在实例化环境时直接应用,只需将
render_mode="human"
传递给make
。只有在环境未原生实现人类渲染(即render_mode
不包含"human"
)时,才会应用包装器。>>> env = gym.make("phys2d/CartPole-v1", render_mode="human") # CartPoleJax-v1 doesn't implement human-rendering natively >>> obs, _ = env.reset() # This will start rendering to the screen
警告:如果基础环境使用
render_mode="rgb_array_list"
,其(即 *基础环境*)的渲染方法将始终返回一个空列表。>>> env = gym.make("LunarLander-v3", render_mode="rgb_array_list") >>> wrapped = HumanRendering(env) >>> obs, _ = wrapped.reset() >>> env.render() # env.render() will always return an empty list! []
- 更改日志
v0.25.0 - 首次添加
- 参数:
env – 被包装的环境
- class gymnasium.wrappers.OrderEnforcing(env: Env[ObsType, ActType], disable_render_order_enforcing: bool = False)[source]¶
如果在
reset
之前调用了step
或render
,将产生错误。不存在此包装器的矢量版本。
示例
>>> import gymnasium as gym >>> from gymnasium.wrappers import OrderEnforcing >>> env = gym.make("CartPole-v1", render_mode="human") >>> env = OrderEnforcing(env) >>> env.step(0) Traceback (most recent call last): ... gymnasium.error.ResetNeeded: Cannot call env.step() before calling env.reset() >>> env.render() Traceback (most recent call last): ... gymnasium.error.ResetNeeded: Cannot call `env.render()` before calling `env.reset()`, if this is an intended action, set `disable_render_order_enforcing=True` on the OrderEnforcer wrapper. >>> _ = env.reset() >>> env.render() >>> _ = env.step(0) >>> env.close()
- 更改日志
v0.22.0 - 首次添加
v0.24.0 - 为渲染函数添加了顺序强制
- 参数:
env – 要包装的环境
disable_render_order_enforcing – 是否禁用渲染顺序强制
- class gymnasium.wrappers.RenderCollection(env: Env[ObsType, ActType], pop_frames: bool = True, reset_clean: bool = True)[source]¶
收集环境的渲染帧,例如
render
返回一个list[RenderedFrame]
。不存在此包装器的矢量版本。
示例
返回
render
未被调用的步数的帧列表。>>> import gymnasium as gym >>> env = gym.make(“LunarLander-v3”, render_mode=”rgb_array”) >>> env = RenderCollection(env) >>> _ = env.reset(seed=123) >>> for _ in range(5): … _ = env.step(env.action_space.sample()) … >>> frames = env.render() >>> len(frames) 6>>> frames = env.render() >>> len(frames) 0
返回该回合运行的步数的帧列表。>>> import gymnasium as gym >>> env = gym.make(“LunarLander-v3”, render_mode=”rgb_array”) >>> env = RenderCollection(env, pop_frames=False) >>> _ = env.reset(seed=123) >>> for _ in range(5): … _ = env.step(env.action_space.sample()) … >>> frames = env.render() >>> len(frames) 6
>>> frames = env.render() >>> len(frames) 6
收集所有回合的所有帧,在调用渲染时不清除它们。>>> import gymnasium as gym >>> env = gym.make(“LunarLander-v3”, render_mode=”rgb_array”) >>> env = RenderCollection(env, pop_frames=False, reset_clean=False) >>> _ = env.reset(seed=123) >>> for _ in range(5): … _ = env.step(env.action_space.sample()) … >>> _ = env.reset(seed=123) >>> for _ in range(5): … _ = env.step(env.action_space.sample()) … >>> frames = env.render() >>> len(frames) 12
>>> frames = env.render() >>> len(frames) 12
- 更改日志
v0.26.2 - 首次添加
- 参数:
env – 被包装的环境
pop_frames (bool) – 如果为真,则在调用
meth:render
后清除集合帧。默认值为True
。reset_clean (bool) – 如果为真,则在调用
meth:reset
后清除集合帧。默认值为True
。
数据转换包装器¶
- class gymnasium.wrappers.JaxToNumpy(env: Env[ObsType, ActType])[source]¶
包装基于 Jax 的环境,使其能够与 NumPy 数组交互。
动作必须以 numpy 数组的形式提供,并且观察结果将以 numpy 数组的形式返回。包装器的向量版本存在于
gymnasium.wrappers.vector.JaxToNumpy
中。笔记
Jax 到 NumPy 和 NumPy 到 Jax 的转换不能保证往返(jax -> numpy -> jax)反之亦然。原因是 jax 不支持非数组值,因此 numpy
int_32(5) -> DeviceArray([5], dtype=jnp.int23)
示例
>>> import gymnasium as gym >>> env = gym.make("JaxEnv-vx") >>> env = JaxToNumpy(env) >>> obs, _ = env.reset(seed=123) >>> type(obs) <class 'numpy.ndarray'> >>> action = env.action_space.sample() >>> obs, reward, terminated, truncated, info = env.step(action) >>> type(obs) <class 'numpy.ndarray'> >>> type(reward) <class 'float'> >>> type(terminated) <class 'bool'> >>> type(truncated) <class 'bool'>
- 更改日志
v1.0.0 - 首次添加
- 参数:
env – 要包装的 jax 环境
- class gymnasium.wrappers.JaxToTorch(env: gym.Env, device: Device | None = None)[source]¶
包装基于 Jax 的环境,使其能够与 PyTorch 张量交互。
动作必须以 PyTorch 张量形式提供,并且观察结果将以 PyTorch 张量形式返回。包装器的向量版本存在于
gymnasium.wrappers.vector.JaxToTorch
中。注意
对于
rendered
,它将以 NumPy 数组形式返回,而不是 pytorch 张量。示例
>>> import torch >>> import gymnasium as gym >>> env = gym.make("JaxEnv-vx") >>> env = JaxtoTorch(env) >>> obs, _ = env.reset(seed=123) >>> type(obs) <class 'torch.Tensor'> >>> action = torch.tensor(env.action_space.sample()) >>> obs, reward, terminated, truncated, info = env.step(action) >>> type(obs) <class 'torch.Tensor'> >>> type(reward) <class 'float'> >>> type(terminated) <class 'bool'> >>> type(truncated) <class 'bool'>
- 更改日志
v1.0.0 - 首次添加
- 参数:
env – 要包装的基于 Jax 的环境
device – 应该将 torch 张量移动到的设备
- class gymnasium.wrappers.NumpyToTorch(env: gym.Env, device: Device | None = None)[source]¶
包装基于 NumPy 的环境,使其能够与 PyTorch 张量交互。
动作必须以 PyTorch 张量形式提供,并且观察结果将以 PyTorch 张量形式返回。包装器的向量版本存在于
gymnasium.wrappers.vector.NumpyToTorch
中。注意
对于
rendered
,它将以 NumPy 数组形式返回,而不是 pytorch 张量。示例
>>> import torch >>> import gymnasium as gym >>> env = gym.make("CartPole-v1") >>> env = NumpyToTorch(env) >>> obs, _ = env.reset(seed=123) >>> type(obs) <class 'torch.Tensor'> >>> action = torch.tensor(env.action_space.sample()) >>> obs, reward, terminated, truncated, info = env.step(action) >>> type(obs) <class 'torch.Tensor'> >>> type(reward) <class 'float'> >>> type(terminated) <class 'bool'> >>> type(truncated) <class 'bool'>
- 更改日志
v1.0.0 - 首次添加
- 参数:
env – 要包装的基于 Jax 的环境
device – 应该将 torch 张量移动到的设备