杂项封装器¶
常用封装器¶
- 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 - 随着步进 API 的更新,终止和截断信号是分开返回的。
- 参数:
env – 应用封装器的环境
max_episode_steps – 剧集被截断的环境步数(
elapsed >= max_episode_steps
)
- class gymnasium.wrappers.RecordVideo(env: 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, gc_trigger: Callable[[int], bool] | None = lambda episode: ...)[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 日志记录器,默认为禁用
gc_trigger – 接受整数并返回
True
(当且仅当应在此剧集之后执行垃圾回收时)的函数
- 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 – 剧集统计信息的信息键
- class gymnasium.wrappers.AtariPreprocessing(env: 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 环境
空操作重置 (Noop Reset):通过在重置时执行随机次数的空操作来获取初始状态,默认最多 30 次空操作。
帧跳过 (Frame skipping):步进之间跳过的帧数,默认为 4。
最大池化 (Max-pooling):对帧跳过中最近的两个观测进行池化。
- 生命损失时的终止信号:当智能体在环境中失去一条生命时,环境终止。
默认关闭。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 Wrapper 的内存优化优势。
- 引发:
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。此封装器在创建时自动应用,可以通过 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) – 如果为 true,则在调用
meth:render
后清除收集的帧。默认值为True
。reset_clean (bool) – 如果为 true,则在调用
meth:reset
时清除收集的帧。默认值为True
。
数据转换封装器¶
- class gymnasium.wrappers.ArrayConversion(env: Env, env_xp: ModuleType, target_xp: ModuleType, env_device: Any | None = None, target_device: Any | None = None)[source]¶
封装一个与 Array API 兼容的环境,使其可以与另一个 Array API 框架进行交互。
流行的 Array API 框架包括
numpy
、torch
、jax.numpy
、cupy
等。使用此封装器,您可以将环境的输出转换为其中任何一个框架。反之,如果可能且无需移动数据或设备传输,动作将自动映射回环境框架。存在该封装器的向量版本,
gymnasium.wrappers.vector.ArrayConversion
。示例
>>> import torch >>> import jax.numpy as jnp >>> import gymnasium as gym >>> env = gym.make("JaxEnv-vx") >>> env = ArrayConversion(env, env_xp=jnp, target_xp=torch) >>> 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.2.0 - 初次添加
- 参数:
env – 要封装的 Array API 兼容环境
env_xp – 环境所基于的 Array API 框架
target_xp – 要转换到的 Array API 框架
env_device – 环境所在的设备
target_device – 数组应返回的设备
- 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: Env, device: str | 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 – PyTorch 张量应移动到的设备
- class gymnasium.wrappers.NumpyToTorch(env: Env, device: str | 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 – 要封装的基于 NumPy 的环境
device – PyTorch 张量应移动到的设备