ding.envs¶

环境¶

请参考 ding/envs/env 获取更多详细信息。

基础环境¶

class ding.envs.BaseEnv(cfg: dict)[source]¶

Overview:: 基础环境类，继承自 gym.Env
Interface:: __init__, reset, close, step, random_action, create_collector_env_cfg, create_evaluator_env_cfg, enable_save_replay

abstract __init__(cfg: dict) → None[source]¶

Overview:

懒初始化，只有在__init__方法中会初始化相关的参数，具体的环境将在第一次调用reset方法时初始化。

Arguments:

cfg (dict): 环境配置，以字典类型表示。

abstract close() → None[source]¶

Overview:: 关闭环境及其所有相关资源，应在使用环境实例后调用。

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

Overview:

返回输入配置中的所有环境列表，用于环境管理器（一系列向量化环境），此方法主要负责环境收集数据。

Arguments:

cfg (dict): 原始输入的环境配置，需要转换为实际创建环境实例的类型，并生成相应数量的配置。

Returns:

env_cfg_list (List[dict]): 包含所有配置收集器环境的 cfg 列表。

注意

collector_env_cfg/evaluator_env_cfg 中的元素（环境配置）可以不同，例如服务器 IP 和端口。

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

Overview:

返回输入配置中的所有环境列表，用于环境管理器（一系列向量化环境），此方法主要负责评估环境的性能。

Arguments:

cfg (dict): 原始输入的环境配置，需要转换为实际创建环境实例的类型，并生成相应数量的配置。

Returns:

env_cfg_list (List[dict]): 包含所有配置评估器环境的 cfg 列表。

enable_save_replay(replay_path: str) → None[source]¶

Overview:

在给定路径中保存回放文件，此方法需要由每个环境类自行实现。

Arguments:

replay_path (str): 保存回放文件的路径。

random_action() → Any[source]¶

Overview:

返回从原始动作空间生成的随机动作，通常用于测试。

Returns:

random_action (Any): 随机生成的动作。

abstract reset() → Any[source]¶

Overview:

将环境重置为初始状态并返回初始观察结果。

Returns:

obs (Any): 重置后的初始观察值。

abstract step(action: Any) → BaseEnv.timestep[source]¶

Overview:

运行环境动态/模拟的一个时间步长。

Arguments:

action (Any): 用于步骤的 action 输入。

Returns:

时间步长 (BaseEnv.timestep): 环境执行一步的结果时间步长。

获取向量环境设置¶

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

Overview:

获取向量化环境设置（env_fn, collector_env_cfg, evaluator_env_cfg）。

Arguments:

cfg (dict): 用户配置中的原始输入环境配置，例如 cfg.env。

Returns:

env_fn (type): 可调用对象，使用适当的参数调用它，然后获取一个新的环境实例。
collector_env_cfg (List[dict]): 一个包含收集数据环境配置的列表。
evaluator_env_cfg (List[dict]): 一个包含评估环境配置的列表。

注意

收集器环境配置/评估器环境配置中的元素（环境配置）可以不同，例如服务器IP和端口。

获取环境类¶

ding.envs.get_env_cls(cfg: EasyDict) → type[source]¶

Overview:

通过cfg的相应模块获取env类，并返回可调用的类。

Arguments:

cfg (dict): 用户配置中的原始输入环境配置，例如 cfg.env。

Returns:

env_cls_type (type): Env 模块作为相应的可调用类类型。

DingEnvWrapper¶

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

Overview:: 这是BaseEnv类的包装器，用于提供一致的环境接口。
Interfaces:: __init__, reset, step, close, seed, random_action, _wrap_env, __repr__, create_collector_env_cfg, create_evaluator_env_cfg, enable_save_replay, observation_space, action_space, reward_space, clone

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

Overview:

初始化DingEnvWrapper。应该传入一个环境实例或用于创建环境实例的配置。对于前者，即环境实例：env参数不能为None，而应该是实例。它不支持子进程环境管理器。因此，它通常用于简单的环境。对于后者，即用于创建环境实例的配置：cfg参数必须包含env_id。

Arguments:

env (Union[gym.Env, gymnasium.Env]): 要包装的环境实例。
cfg (dict): 用于创建环境实例的配置字典。
seed_api (bool): 是否使用种子API。默认为True。
调用者 (str): 一个字符串，表示此方法的调用者，包括 collector 或 evaluator。不同的调用者可能需要不同的包装器。默认是‘collector’。
is_gymnasium (bool): 环境是否为gymnasium环境。默认为False，即环境为gym环境。

property action_space: Space¶

Overview:

返回包装环境中的动作空间。动作空间表示代理在环境中可以采取的可能动作的范围和形状。

Returns:

action_space (gym.spaces.Space): 环境的动作空间。

clone(caller: str = 'collector') → BaseEnv[source]¶

Overview:

克隆当前环境包装器，创建一个具有相同设置的新环境。

Arguments:

调用者 (str): 一个字符串，表示此方法的调用者，包括 collector 或 evaluator。不同的调用者可能需要不同的包装器。默认是 'collector'。

Returns:

DingEnvWrapper：具有相同设置的环境的新实例。

close() → None[source]¶

Overview:: 通过关闭和删除环境来清理环境。当不再需要环境时，应调用此方法。未能调用此方法可能导致内存泄漏。

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

Overview:

根据输入配置创建收集器的环境配置列表。

Arguments:

cfg (dict): 输入配置字典。

Returns:

env_cfgs (List[dict]): 收集器的环境配置列表。

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

Overview:

根据输入配置创建评估者的环境配置列表。

Arguments:

cfg (dict): 输入配置字典。

Returns:

env_cfgs (List[dict]): 评估者的环境配置列表。

enable_save_replay(replay_path: str | None = None) → None[source]¶

Overview:

启用保存重放功能。重放将保存在指定路径。

Arguments:

replay_path (Optional[str]): 保存回放的路径，默认为None。

property observation_space: Space¶

Overview:

返回被包裹环境的观察空间。观察空间表示环境可以提供给代理的可能观察的范围和形状。

Note:

如果观测空间的数据类型是float64，则将其转换为float32，以便更好地与大多数机器学习库兼容。

Returns:

observation_space (gym.spaces.Space): 环境的观察空间。

random_action() → ndarray[source]¶

Overview:

从环境的动作空间中返回一个随机动作。

Returns:

动作 (np.ndarray): 随机动作。

reset() → ndarray[source]¶

Overview:

重置环境的状态。如果环境未初始化，将首先创建它。

Returns:

obs (Dict): 重置后的新观察值。

property reward_space: Space¶

Overview:

返回包装环境的奖励空间。奖励空间表示代理作为其行动结果可能获得的奖励的范围和形状。

Returns:

reward_space (gym.spaces.Space): 环境的奖励空间。

seed(seed: int, dynamic_seed: bool = True) → None[source]¶

Overview:

设置环境的种子。

Arguments:

seed (int): 要设置的种子。
dynamic_seed (bool): 是否使用动态种子，默认为 True。

step(action: int64 | ndarray) → BaseEnvTimestep[来源]¶

Overview:

在环境中执行给定的动作，并返回时间步（观察、奖励、完成、信息）。

Arguments:

action (Union[np.int64, np.ndarray]): 要在环境中执行的动作。

Returns:

时间步长 (BaseEnvTimestep): 执行动作后的时间步长。

get_default_wrappers¶

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

Overview:

获取用于DingEnvWrapper中不同环境的默认包装器。

Arguments:

env_wrapper_name (str): 环境包装器的名称。
env_id (Optional[str]): 特定环境的ID，例如 PongNoFrameskip-v4。
调用者 (str): 环境的调用者，包括 collector 或 evaluator。不同的调用者可能需要不同的包装器。

Returns:

wrapper_list (List[dict]): 包装器列表，每个元素是具体包装器的配置。

Raises:

NotImplementedError: env_wrapper_name 不在 ['mujoco_default', 'atari_default', 'gym_hybrid_default', 'default'] 中

环境管理器¶

请参考 ding/envs/env_manager 获取更多详细信息。

创建环境管理器¶

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

Overview:

根据manager_cfg和环境函数创建一个环境管理器。

Arguments:

manager_cfg (EasyDict): 最终合并的环境管理器配置。
env_fn (List[Callable]): 一个用于创建 env_num 子环境的函数列表。

ArgumentsKeys:

类型 (str): 在 ENV_MANAGER_REGISTRY.register 中设置的环境管理器类型，例如 base。
import_names (List[str]): 在创建环境管理器之前要导入的模块名称（路径）列表，例如 ding.envs.env_manager.base_env_manager。

Returns:

env_manager (BaseEnvManager): 创建的env管理器。

提示

此方法不会修改manager_cfg，它会深拷贝manager_cfg然后进行修改。

get_env_manager_cls¶

ding.envs.get_env_manager_cls(cfg: EasyDict) → type[source]¶

Overview:

根据配置获取环境管理器类，用于访问相关的类变量/方法。

Arguments:

manager_cfg (EasyDict): 最终合并的环境管理器配置。

ArgumentsKeys:

类型 (str): 在 ENV_MANAGER_REGISTRY.register 中设置的环境管理器类型，例如 base。
import_names (List[str]): 在创建环境管理器之前要导入的模块名称（路径）列表，例如 ding.envs.env_manager.base_env_manager。

Returns:

env_manager_cls (type): 对应的环境管理器类。

BaseEnvManager¶

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

Overview:

环境管理器的基类，用于管理多个向量化环境。BaseEnvManager 定义了所有必要的接口，派生类必须扩展这个基类。

该类通过伪并行（即串行）机制实现，因此，该类仅用于一些小型环境和调试目的。

Interfaces:

重置, 步骤, 种子, 关闭, 启用保存回放, 启动, 默认配置, 奖励塑造, 启用保存图表

Properties:

环境编号, 环境参考, 准备观察, 准备观察ID, 准备图像, 完成, 关闭, 方法名称列表, 观察空间, 动作空间, 奖励空间

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

Overview:

使用可调用的环境函数和EasyDict类型的配置初始化基础环境管理器。这里我们使用 env_fn 来确保子环境的延迟初始化，这对资源分配和并行性有益。cfg 是该类的默认配置和用户配置的合并结果。此构造函数处于延迟初始化模式，实际初始化在 launch 中进行。

Arguments:

env_fn (List[Callable]): 一个用于创建 env_num 子环境的函数列表。
cfg (EasyDict): 最终合并的配置。

注意

有关如何合并配置的更多详细信息，请参阅DI-engine的系统文档 (en link1)。

property action_space: gym.spaces.Space¶

Overview:

action_space 是子环境的动作空间，遵循 gym.spaces 的格式。

Returns:

action_space (gym.spaces.Space): 子环境的动作空间。

close() → None[source]¶

Overview:: 关闭环境管理器并释放所有环境资源。

property closed: bool¶

Overview:

closed 是一个属性，用于返回环境管理器是否已关闭。

Returns:

closed (bool): 环境管理器是否已关闭。

classmethod default_config() → EasyDict[source]¶

Overview:

返回环境管理器的深拷贝默认配置。

Returns:

cfg (EasyDict): 环境管理器的默认配置。

property done: bool¶

Overview:

done 是一个标志，用于指示环境管理器是否已完成，即所有子环境是否已经执行了足够的回合。

Returns:

完成 (bool): 环境管理器是否已完成。

enable_save_figure(env_id: int, figure_path: str) → None[source]¶

Overview:

启用特定环境以保存图形（例如环境统计或回合回报曲线）。

Arguments:

figure_path (str): 所有环境保存图形的文件目录路径。

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

Overview:

启用所有环境在每次情节结束后保存回放视频。

Arguments:

replay_path (Union[List[str], str]): 每个环境的路径列表；或者所有环境的一个路径。

property env_num: int¶

Overview:

env_num 是环境管理器中子环境的数量。

Returns:

env_num (int): 子环境的数量。

property env_ref: BaseEnv¶

Overview:

env_ref 用于获取环境的一些常见属性，如 obs_shape 和 act_shape。

Returns:

env_ref (BaseEnv): 子环境的引用。

launch(reset_param: Dict | None = None) → None[source]¶

Overview:

启动环境管理器，实例化子环境并设置环境及其参数。

Arguments:

reset_param (Optional[Dict]): 每个环境的重置参数字典，键是env_id，值是相应的重置参数，默认为None。

property method_name_list: list¶

Overview:

可以从环境管理器级别直接调用的子环境的公共方法列表。其他方法和属性将通过__getattr__方法访问。在此列表中定义的方法可以视为子环境中方法的向量化扩展。 BaseEnvManager的子类可以重写此方法以添加更多方法。

Returns:

method_name_list (list): 子环境的公共方法列表。

property observation_space: gym.spaces.Space¶

Overview:

observation_space 是子环境的观察空间，遵循 gym.spaces 的格式。

Returns:

observation_space (gym.spaces.Space): 子环境的观察空间。

property ready_imgs: Dict[int, Any]¶

Overview:

有时，我们需要渲染环境，此函数用于获取下一个准备好的渲染帧和相应的环境ID。

Arguments:

render_mode (Optional[str]): 渲染模式，可以是‘rgb_array’或‘depth_array’，遵循ding.utils中render函数的定义。

Returns:

ready_imgs (Dict[int, np.ndarray]): 一个包含env_id键和渲染帧的字典。

property ready_obs: Dict[int, Any]¶

Overview:

获取准备（下一个）观察结果，这是一个特殊设计，用于统一异步/同步环境管理器。对于策略和环境之间的每次交互，策略将输入ready_obs并输出动作。然后环境管理器将使用动作step并准备下一个ready_obs。

Returns:

ready_obs (Dict[int, Any]): 一个包含env_id键和观察值的字典。

Example:

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

property ready_obs_id: List[int]¶

Overview:

获取准备（下一个）观察ID，这是一个特殊设计，用于统一异步/同步环境管理器。

Returns:

ready_obs_id (List[int]): 准备观察的环境ID列表。

reset(reset_param: Dict | None = None) → None[source]¶

Overview:

强制重置子环境及其对应的参数。因为在环境管理器中，所有子环境通常会在完成后自动重置，所以此方法仅在调用者必须强制重置所有子环境时调用，例如在评估时。

Arguments:

reset_param (List): 每个环境的重置参数字典，键是env_id，值是相应的重置参数。

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

Overview:

为特定环境执行奖励塑形，通常在情节终止时调用。

Arguments:

env_id (int): 要塑造的环境的ID。
transitions (List[dict]): 要塑造的环境的转换数据列表。

Returns:

transitions (List[dict]): 格式化后的过渡数据列表。

property reward_space: gym.spaces.Space¶

Overview:

reward_space 是子环境的奖励空间，遵循 gym.spaces 的格式。

Returns:

reward_space (gym.spaces.Space): 子环境的奖励空间。

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

Overview:

为每个环境设置随机种子。

Arguments:

种子 (Union[Dict[int, int], List[int], int]): 每个环境的种子字典或列表；如果只提供一个种子，它将以相同的方式用于所有环境。
dynamic_seed (bool): 是否使用动态种子。

注意

有关dynamic_seed的更多详细信息，请参阅DI-engine的最佳实践文档 (en link2)。

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

Overview:

根据输入的动作执行环境步骤。如果在此次执行后某些子环境已完成，当self._auto_reset为True时，它们将自动重置，否则当调用者使用环境管理器的reset方法时需要手动重置。

Arguments:

actions (Dict[int, Any]): 一个动作的字典，键是env_id，值是对应的动作。动作可以是任何类型，这取决于环境，环境会处理它。通常，动作是一个numpy数组的字典，值是由外部调用者如policy生成的。

Returns:

时间步长 (Dict[int, BaseEnvTimestep]): 每个时间步长是一个 BaseEnvTimestep 对象，通常包括观察、奖励、完成、信息。一些特殊定制的环境会有特殊的时间步长定义。时间步长的长度与同步环境管理器中的动作长度相同。

Example:

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

BaseEnvManagerV2¶

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

Overview:

环境管理器的基类，用于管理多个向量化环境。BaseEnvManager 定义了所有必要的接口，派生类必须扩展这个基类。

该类通过伪并行（即串行）机制实现，因此，该类仅用于一些小型环境和调试目的。

V2 表示此环境管理器是为与 treetensor 耦合的新任务管道和接口设计的。

注意

有关新任务管道的更多详细信息，请参阅DI-engine的系统文档（system en link3）。

Interfaces:: 重置, 步骤, 种子, 关闭, 启用保存回放, 启动, 默认配置, 奖励塑造, 启用保存图表
Properties:: 环境编号, 环境参考, 准备观察, 准备观察ID, 准备图像, 完成, 关闭, 方法名称列表, 观察空间, 动作空间, 奖励空间

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

Overview:

使用可调用的环境函数和EasyDict类型的配置初始化基础环境管理器。这里我们使用 env_fn 来确保子环境的延迟初始化，这对资源分配和并行性有益。cfg 是该类的默认配置和用户配置的合并结果。此构造函数处于延迟初始化模式，实际初始化在 launch 中进行。

Arguments:

env_fn (List[Callable]): 一个用于创建 env_num 子环境的函数列表。
cfg (EasyDict): 最终合并的配置。

注意

有关如何合并配置的更多详细信息，请参阅DI-engine的系统文档 (en link1)。

property action_space: gym.spaces.Space¶

Overview:

action_space 是子环境的动作空间，遵循 gym.spaces 的格式。

Returns:

action_space (gym.spaces.Space): 子环境的动作空间。

close() → None¶

Overview:: 关闭环境管理器并释放所有环境资源。

property closed: bool¶

Overview:

closed 是一个属性，用于返回环境管理器是否已关闭。

Returns:

closed (bool): 环境管理器是否已关闭。

classmethod default_config() → EasyDict¶

Overview:

返回环境管理器的深拷贝默认配置。

Returns:

cfg (EasyDict): 环境管理器的默认配置。

property done: bool¶

Overview:

done 是一个标志，用于指示环境管理器是否已完成，即所有子环境是否已经执行了足够的回合。

Returns:

完成 (bool): 环境管理器是否已完成。

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

Overview:

启用特定环境以保存图形（例如环境统计或回合回报曲线）。

Arguments:

figure_path (str): 所有环境保存图形的文件目录路径。

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

Overview:

启用所有环境在每次情节结束后保存回放视频。

Arguments:

replay_path (Union[List[str], str]): 每个环境的路径列表；或者所有环境的一个路径。

property env_num: int¶

Overview:

env_num 是环境管理器中子环境的数量。

Returns:

env_num (int): 子环境的数量。

property env_ref: BaseEnv¶

Overview:

env_ref 用于获取环境的一些常见属性，如 obs_shape 和 act_shape。

Returns:

env_ref (BaseEnv): 子环境的引用。

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

Overview:

启动环境管理器，实例化子环境并设置环境及其参数。

Arguments:

reset_param (Optional[Dict]): 每个环境的重置参数字典，键是env_id，值是相应的重置参数，默认为None。

property method_name_list: list¶

Overview:

可以从环境管理器级别直接调用的子环境的公共方法列表。其他方法和属性将通过__getattr__方法访问。在此列表中定义的方法可以视为子环境中方法的向量化扩展。 BaseEnvManager的子类可以重写此方法以添加更多方法。

Returns:

method_name_list (list): 子环境的公共方法列表。

property observation_space: gym.spaces.Space¶

Overview:

observation_space 是子环境的观察空间，遵循 gym.spaces 的格式。

Returns:

observation_space (gym.spaces.Space): 子环境的观察空间。

property ready_imgs: Dict[int, Any]¶

Overview:

有时，我们需要渲染环境，此函数用于获取下一个准备好的渲染帧和相应的环境ID。

Arguments:

render_mode (Optional[str]): 渲染模式，可以是‘rgb_array’或‘depth_array’，遵循ding.utils中render函数的定义。

Returns:

ready_imgs (Dict[int, np.ndarray]): 一个包含env_id键和渲染帧的字典。

property ready_obs: array¶

Overview:

获取准备（下一个）观察值，这是一种特殊设计，用于统一异步/同步环境管理器。对于策略和环境之间的每次交互，策略将输入ready_obs并输出动作。然后环境管理器将使用动作step并准备下一个ready_obs。对于V2版本，观察值被转换并打包成tnp.array类型，这使得操作更加方便。

Return:

ready_obs (tnp.array): 一个堆叠的treenumpy类型的观测数据。

Example:

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

property ready_obs_id: List[int]¶

Overview:

获取准备（下一个）观察ID，这是一个特殊设计，用于统一异步/同步环境管理器。

Returns:

ready_obs_id (List[int]): 准备观察的环境ID列表。

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

Overview:

强制重置子环境及其对应的参数。因为在环境管理器中，所有子环境通常会在完成后自动重置，所以此方法仅在调用者必须强制重置所有子环境时调用，例如在评估时。

Arguments:

reset_param (List): 每个环境的重置参数字典，键是env_id，值是相应的重置参数。

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

Overview:

为特定环境执行奖励塑形，通常在情节终止时调用。

Arguments:

env_id (int): 要塑造的环境的ID。
transitions (List[dict]): 要塑造的环境的转换数据列表。

Returns:

transitions (List[dict]): 形状化的过渡数据列表。

property reward_space: gym.spaces.Space¶

Overview:

reward_space 是子环境的奖励空间，遵循 gym.spaces 的格式。

Returns:

reward_space (gym.spaces.Space): 子环境的奖励空间。

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

Overview:

为每个环境设置随机种子。

Arguments:

种子 (Union[Dict[int, int], List[int], int]): 每个环境的种子字典或列表；如果只提供一个种子，它将以相同的方式用于所有环境。
dynamic_seed (bool): 是否使用动态种子。

注意

有关dynamic_seed的更多详细信息，请参阅DI-engine的最佳实践文档 (en link2)。

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

Overview:

根据输入的动作执行环境步骤。如果在此次执行后某些子环境已完成，默认情况下它们将自动重置。

Arguments:

actions (List[tnp.ndarray]): 一个treenumpy类型的动作列表，该值由外部调用者如policy生成。

Returns:

时间步长 (List[tnp.ndarray]): 一个时间步长的列表，每个时间步长是一个 tnp.ndarray 对象，通常包括观察、奖励、完成、信息、环境ID。一些特殊的环境会有特殊的时间步长定义。时间步长的长度与同步环境管理器中的动作长度相同。为了兼容treenumpy，这里我们使用 make_key_as_identifier 和 remove_illegal_item 函数来修改原始的时间步长。

Example:

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

SyncSubprocessEnvManager¶

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

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

Overview:

初始化 AsyncSubprocessEnvManager。

Arguments:

env_fn (List[Callable]): 用于创建环境的函数
cfg (EasyDict): 配置

注意

wait_num: 每次环境返回收集的最小数量
step_wait_timeout: 每次环境返回收集的最小数量

close() → None¶

Overview:: 关闭环境管理器并释放所有相关资源。

classmethod default_config() → EasyDict¶

Overview:

返回环境管理器的深拷贝默认配置。

Returns:

cfg (EasyDict): 环境管理器的默认配置。

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

Overview:

设置每个环境的回放保存路径。

Arguments:

replay_path (Union[List[str], str]): 每个环境的路径列表；或者所有环境的一个路径。

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

Overview:

设置环境及其参数。

Arguments:

reset_param (Optional[Dict]): 每个环境的重置参数字典，键是env_id，值是对应的重置参数。

property ready_imgs: Dict[int, Any]¶

Overview:: 获取下一个渲染帧。
Return:: 一个包含渲染帧及其环境ID的字典。
Note:: 渲染的帧以 np.ndarray 形式返回。

property ready_obs: Dict[int, Any]¶

Overview:

获取下一个观测值。

Return:

一个包含观测值及其环境ID的字典。

Note:

观测值以np.ndarray的形式返回。

Example:

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

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

Overview:

重置环境及其参数。

Arguments:

reset_param (List): 每个环境的重置参数字典，键是env_id，值是对应的重置参数。

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

Overview:

为每个环境设置随机种子。

Arguments:

种子 (Union[Dict[int, int], List[int], int]): 每个环境的种子字典或列表；如果只提供一个种子，它将以相同的方式用于所有环境。
dynamic_seed (bool): 是否使用动态种子。

注意

有关dynamic_seed的更多详细信息，请参阅DI-engine的最佳实践文档 (en link2)。

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

Overview:

逐步执行所有环境。如果完成，则重置环境。

Arguments:

动作 (Dict[int, Any]): {环境ID: 动作}

Returns:

时间步长 (Dict[int, namedtuple]): {环境ID: 时间步长}. 时间步长是一个包含观察、奖励、完成状态、环境信息的 BaseEnvTimestep 元组。

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

注意

出现在actions中的env_id也将在timesteps中返回。
每个环境都由一个子进程单独运行。一旦一个环境完成，它会立即被重置。

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

Overview:: 子进程的目标函数运行。

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

Overview:: 子进程目标函数的更健壮版本。默认使用。

SubprocessEnvManagerV2¶

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

Overview:: SyncSubprocessEnvManager 用于新的任务管道和与 treetensor 耦合的接口。

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

Overview:

初始化 AsyncSubprocessEnvManager。

Arguments:

env_fn (List[Callable]): 用于创建环境的函数
cfg (EasyDict): 配置

注意

wait_num: 每次环境返回收集的最小数量
step_wait_timeout: 每次环境返回收集的最小数量

close() → None¶

Overview:: 关闭环境管理器并释放所有相关资源。

classmethod default_config() → EasyDict¶

Overview:

返回环境管理器的深拷贝默认配置。

Returns:

cfg (EasyDict): 环境管理器的默认配置。

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

Overview:

设置每个环境的回放保存路径。

Arguments:

replay_path (Union[List[str], str]): 每个环境的路径列表；或者所有环境的一个路径。

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

Overview:

设置环境及其参数。

Arguments:

reset_param (Optional[Dict]): 每个环境的重置参数字典，键是env_id，值是对应的重置参数。

property ready_imgs: Dict[int, Any]¶

Overview:: 获取下一个渲染帧。
Return:: 一个包含渲染帧及其环境ID的字典。
Note:: 渲染的帧以 np.ndarray 形式返回。

property ready_obs: array¶

Overview:

获取tnp.array类型的准备（下一个）观察结果，这对于异步/同步场景都是统一的。

Return:

ready_obs (tnp.array): 一个堆叠的treenumpy类型的观测数据。

Example:

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

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

Overview:

重置环境及其参数。

Arguments:

reset_param (List): 每个环境的重置参数字典，键是env_id，值是对应的重置参数。

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

Overview:

为每个环境设置随机种子。

Arguments:

种子 (Union[Dict[int, int], List[int], int]): 每个环境的种子字典或列表；如果只提供一个种子，它将以相同的方式用于所有环境。
dynamic_seed (bool): 是否使用动态种子。

注意

有关dynamic_seed的更多详细信息，请参阅DI-engine的最佳实践文档 (en link2)。

step(actions: List[ndarray] | ndarray) → List[ndarray][来源]¶

Overview:

根据输入的动作执行环境步骤。如果完成，则重置环境。

Arguments:

actions (Union[List[tnp.ndarray], tnp.ndarray]): 动作来自外部调用者，如策略。

Returns:

时间步长 (List[tnp.ndarray]): 每个时间步长是一个包含观察、奖励、完成、信息和环境ID的tnp.array。

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

Overview:: 子进程的目标函数运行。

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

Overview:: 子进程目标函数的更健壮版本。默认使用。

异步子进程环境管理器¶

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

Overview:: 创建一个AsyncSubprocessEnvManager来管理多个环境。每个环境由各自的子进程运行。
Interfaces:: 种子, 启动, 准备观察, 步骤, 重置, 活跃环境

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

Overview:

初始化 AsyncSubprocessEnvManager。

Arguments:

env_fn (List[Callable]): 用于创建环境的函数
cfg (EasyDict): 配置

注意

wait_num: 每次环境返回收集的最小数量
step_wait_timeout: 每次环境返回收集的最小数量

close() → None[source]¶

Overview:: 关闭环境管理器并释放所有相关资源。

classmethod default_config() → EasyDict¶

Overview:

返回环境管理器的深拷贝默认配置。

Returns:

cfg (EasyDict): 环境管理器的默认配置。

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

Overview:

设置每个环境的回放保存路径。

Arguments:

replay_path (Union[List[str], str]): 每个环境的路径列表；或者所有环境的一个路径。

launch(reset_param: Dict | None = None) → None[source]¶

Overview:

设置环境及其参数。

Arguments:

reset_param (Optional[Dict]): 每个环境的重置参数字典，键是env_id，值是对应的重置参数。

property ready_imgs: Dict[int, Any]¶

Overview:: 获取下一个渲染帧。
Return:: 一个包含渲染帧及其环境ID的字典。
Note:: 渲染的帧以 np.ndarray 形式返回。

property ready_obs: Dict[int, Any]¶

Overview:

获取下一个观测值。

Return:

一个包含观测值及其环境ID的字典。

Note:

观测值以np.ndarray的形式返回。

Example:

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

reset(reset_param: Dict | None = None) → None[source]¶

Overview:

重置环境及其参数。

Arguments:

reset_param (List): 每个环境的重置参数字典，键是env_id，值是对应的重置参数。

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

Overview:

为每个环境设置随机种子。

Arguments:

种子 (Union[Dict[int, int], List[int], int]): 每个环境的种子字典或列表；如果只提供一个种子，它将以相同的方式用于所有环境。
dynamic_seed (bool): 是否使用动态种子。

注意

有关dynamic_seed的更多详细信息，请参阅DI-engine的最佳实践文档 (en link2)。

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

Overview:

逐步执行所有环境。如果完成，则重置环境。

Arguments:

动作 (Dict[int, Any]): {环境ID: 动作}

Returns:

时间步长 (Dict[int, namedtuple]): {环境ID: 时间步长}. 时间步长是一个包含观察、奖励、完成状态、环境信息的 BaseEnvTimestep 元组。

Example:

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

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

Overview:: 子进程的目标函数运行。

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

Overview:: 子进程目标函数的更健壮版本。默认使用。

GymVectorEnvManager¶

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

Overview:: 创建一个GymVectorEnvManager来管理多个环境。每个环境由各自的子进程运行。
Interfaces:: 种子, 准备观察, 步骤, 重置, 关闭

__init__(env_fn: List[Callable], cfg: EasyDict) → None[源代码]¶: 注意

env_fn 必须创建 gym 类型的环境实例，这可能与 DI-engine 环境不同。

close() → None[source]¶

Overview:: 释放环境资源由于没有调用super.__init__，因此不需要释放BaseEnvManager的资源

property ready_obs: Dict[int, Any]¶

Overview:

获取准备（下一个）观察结果，这是一个特殊设计，用于统一异步/同步环境管理器。对于策略和环境之间的每次交互，策略将输入ready_obs并输出动作。然后环境管理器将使用动作step并准备下一个ready_obs。

Returns:

ready_obs (Dict[int, Any]): 一个包含env_id键和观察值的字典。

Example:

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

reset(reset_param: Dict | None = None) → None[source]¶

Overview:

强制重置子环境及其对应的参数。因为在环境管理器中，所有子环境通常会在完成后自动重置，所以此方法仅在调用者必须强制重置所有子环境时调用，例如在评估时。

Arguments:

reset_param (List): 每个环境的重置参数字典，键是env_id，值是相应的重置参数。

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

Overview:

为每个环境设置随机种子。

Arguments:

种子 (Union[Dict[int, int], List[int], int]): 每个环境的种子字典或列表；如果只提供一个种子，它将以相同的方式用于所有环境。
dynamic_seed (bool): 是否使用动态种子。

注意

有关dynamic_seed的更多详细信息，请参阅DI-engine的最佳实践文档 (en link2)。

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

Overview:

根据输入的动作执行环境步骤。如果在此次执行后某些子环境已完成，当self._auto_reset为True时，它们将自动重置，否则当调用者使用环境管理器的reset方法时需要手动重置。

Arguments:

actions (Dict[int, Any]): 一个动作的字典，键是env_id，值是对应的动作。动作可以是任何类型，这取决于环境，环境会处理它。通常，动作是一个numpy数组的字典，值是由外部调用者如policy生成的。

Returns:

时间步长 (Dict[int, BaseEnvTimestep]): 每个时间步长是一个 BaseEnvTimestep 对象，通常包括观察、奖励、完成、信息。一些特殊定制的环境会有特殊的时间步长定义。时间步长的长度与同步环境管理器中的动作长度相同。

Example:

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

环境包装器¶

请参考 ding/envs/env_wrappers 获取更多详细信息。

创建环境包装器¶

ding.envs.create_env_wrapper(env: Env, env_wrapper_cfg: EasyDict) → Wrapper[source]¶

Overview:

根据环境包装器配置和环境实例创建一个环境包装器。

Arguments:

env (gym.Env): 要包装的环境实例。
env_wrapper_cfg (EasyDict): 环境包装器的配置。

Returns:

env (gym.Wrapper): 被包装的环境实例。

更新形状¶

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

Overview:

获取给定包装器的新观察、动作和奖励的形状。

Arguments:

obs_shape (Any): 观察的原始形状。
act_shape (Any): 动作的原始形状。
rew_shape (Any): 奖励的原始形状。
wrapper_names (List[str]): 包装器的名称。

Returns:

obs_shape (Any): 观察的新形状。
act_shape (Any): 动作的新形状。
rew_shape (Any): 奖励的新形状。

NoopResetWrapper¶

class ding.envs.NoopResetWrapper(env: Env, noop_max: int = 30)[source]¶

Overview:

通过在重置时随机执行无操作来采样初始状态。假设无操作是动作0。

Interfaces:

__init__, reset

Properties:

env (gym.Env): 要包装的环境。
noop_max (int): 运行无操作的最大值。

__init__(env: Env, noop_max: int = 30)[source]¶

Overview:

初始化NoopResetWrapper。

Arguments:

env (gym.Env): 要包装的环境。
noop_max (int): 运行无操作的最大值。默认为30。

reset() → ndarray[source]¶

Overview:

重置环境的状态并返回初始观察结果，在执行随机数量的无操作之后。

Returns:

观察 (Any): 无操作后的初始观察。

MaxAndSkipWrapper¶

class ding.envs.MaxAndSkipWrapper(env: Env, skip: int = 4)[source]¶

Overview:

包装环境以仅返回每skip帧（帧跳过），使用最近的原始观测值（用于跨时间步的最大池化）。

Interfaces:

__init__, step

Properties:

env (gym.Env): 要包装的环境。
跳过 (int): 跳过的帧数。默认为4。

__init__(env: Env, skip: int = 4)[源代码]¶

Overview:

初始化 MaxAndSkipWrapper。

Arguments:

env (gym.Env): 要包装的环境。
跳过 (int): 跳过的帧数。默认为4。

step(action: int | ndarray) → tuple[source]¶

Overview:

执行给定的动作并重复指定的步骤数。奖励会被累加，并返回最后观察到的最大帧。

Arguments:

action (Any): 要重复的动作。

Returns:

max_frame (np.array): 最后观察值的最大值
total_reward (Any): 前一个动作后的奖励总和。
完成 (Bool): 该集是否已结束。
info (Dict): 包含辅助诊断信息（有助于调试，有时也有助于学习）

FireResetWrapper¶

class ding.envs.FireResetWrapper(env: Env)[源代码]¶

Overview:

这个包装器在环境重置时执行一个fire动作。相关讨论: https://github.com/openai/baselines/issues/240

Interfaces:

__init__, reset

Properties:

env (gym.Env): 要包装的环境。

__init__(env: Env) → None[source]¶

Overview:

初始化FireResetWrapper。假设环境的第二个动作是‘FIRE’，并且至少有三个动作。

Arguments:

env (gym.Env): 要包装的环境。

reset() → ndarray[source]¶

Overview:

重置环境状态并执行一个点火动作，即使用动作1进行重置。

Returns:

observation (np.ndarray): 重置和触发操作后的新观察。

EpisodicLifeWrapper¶

class ding.envs.EpisodicLifeWrapper(env: Env)[source]¶

Overview:

这个包装器使得生命周期结束等同于剧集结束，但仅在真正的游戏结束时重置。这有助于更好地进行价值估计。

Interfaces:

__init__, step, reset

Properties:

env (gym.Env): 要包装的环境。
生命值 (int): 当前的生命数量。
was_real_done (bool): 上一集是否因游戏结束而终止。

__init__(env: Env) → None[源代码]¶

Overview:

初始化EpisodicLifeWrapper，将lives设置为0，并将was_real_done设置为True。

Arguments:

env (gym.Env): 要包装的环境。

step(action: Any) → Tuple[ndarray, float, bool, Dict][源代码]¶

Overview:

在环境中执行给定的动作，根据新状态更新属性，并返回新的观察结果、奖励、完成状态和信息。

Arguments:

action (Any): 在环境中执行的操作。

Returns:

观察 (np.ndarray): 动作执行后并更新 self.rms 后的归一化观察。
奖励 (float): 动作执行后返回的奖励金额。
done (bool): Whether the episode has ended, in which case further step() calls will return
未定义的结果。
info (Dict): Contains auxiliary diagnostic information (helpful for debugging, and
有时候学习）。

ClipRewardWrapper¶

class ding.envs.ClipRewardWrapper(env: Env)[source]¶

Overview:

ClipRewardWrapper 类是一个 gym 奖励包装器，它根据奖励的符号将其裁剪为 {-1, 0, +1}。这可以用于在强化学习算法中标准化奖励的规模。

Interfaces:

__init__, reward

Properties:

env (gym.Env): 要包装的环境。
reward_range (Tuple[int, int]): 裁剪后的奖励值范围。

__init__(env: Env)[源代码]¶

Overview:

初始化 ClipRewardWrapper 类。

Arguments:

env (gym.Env): 要包装的环境。

reward(reward: float) → float[source]¶

Overview:

根据奖励的符号将其裁剪为{-1, 0, +1}。注意：np.sign(0) == 0。

Arguments:

奖励 (float): 原始奖励。

Returns:

奖励 (float): 被裁剪的奖励。

FrameStackWrapper¶

class ding.envs.FrameStackWrapper(env: Env, n_frames: int = 4)[source]¶

Overview:

FrameStackWrapper 是一个 gym 环境包装器，它将最新的 n 帧（通常在 Atari 中为 4 帧）堆叠为单个观察结果。它通常用于观察结果是图像的环境，并且连续的帧为代理提供了有用的时间信息。

Interfaces:

__init__, reset, step, _get_ob

Properties:

env (gym.Env): 要包装的环境。
n_frames (int): 要堆叠的帧数。
帧 (collections.deque): 一个保存最近帧的队列。
observation_space (gym.Space): 堆叠观测的空间。

__init__(env: Env, n_frames: int = 4) → None[源代码]¶

Overview:

初始化FrameStackWrapper。此过程包括设置要包装的环境、要堆叠的帧数以及观察空间。

Arguments:

env (gym.Env): 要包装的环境。
n_frame (int): 要堆叠的帧数。

reset() → ndarray[source]¶

Overview:

重置环境并使用初始观察初始化帧。

Returns:

init_obs (np.ndarray): 堆叠的初始观测值。

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

Overview:

在环境中执行给定的动作，将返回的观察结果附加到帧中，并返回堆叠的观察结果。

Arguments:

action (Any): 用于执行步骤的操作。

Returns:

self._get_ob() (np.ndarray): 堆叠的观测值。
奖励 (float): 前一个动作后返回的奖励金额。
完成 (bool): 该回合是否已结束，如果结束，则进一步的 step() 调用将返回未定义的结果。
info (Dict[str, Any]): 包含辅助诊断信息（有助于调试，有时也有助于学习）。

ScaledFloatFrameWrapper¶

class ding.envs.ScaledFloatFrameWrapper(env: Env)[source]¶

Overview:: ScaledFloatFrameWrapper 将观测值归一化到0到1之间。
Interfaces:: __init__, observation

__init__(env: Env)[源代码]¶

Overview:

初始化ScaledFloatFrameWrapper，设置用于归一化的比例和偏差。

Arguments:

env (gym.Env): 要包装的环境。

observation(observation: ndarray) → ndarray[source]¶

Overview:

将观测值缩放到范围 [0, 1] 内。

Arguments:

观察值 (np.ndarray): 原始观察值。

Returns:

scaled_observation (np.ndarray): 缩放后的观测值。

WarpFrameWrapper¶

class ding.envs.WarpFrameWrapper(env: Env, size: int = 84)[source]¶

Overview:

WarpFrameWrapper 类是一个 gym 观察包装器，它将环境观察的帧调整为指定大小（默认为 84x84）。这通常用于强化学习中观察的预处理流程，特别是来自 Atari 环境的视觉观察。

Interfaces:

__init__, observation

Properties:

env (gym.Env): 要包装的环境。
大小 (int): 帧将被调整到的大小。
observation_space (gym.Space): 被封装环境的观察空间。

__init__(env: Env, size: int = 84)[源代码]¶

Overview:

WarpFrameWrapper类的构造函数，初始化环境和大小。

Arguments:

env (gym.Env): 要包装的环境。
大小 (int): 帧将被调整到的大小。默认值为84。

observation(frame: ndarray) → ndarray[source]¶

Overview:

将帧（观察）调整到所需的大小。

Arguments:

frame (np.ndarray): 需要调整大小的帧。

Returns:

frame (np.ndarray): 调整大小后的帧。

动作重复包装器¶

class ding.envs.ActionRepeatWrapper(env: Env, action_repeat: int = 1)[source]¶

Overview:

ActionRepeatWrapper 类是一个 gym 包装器，它会在多个步骤中重复相同的动作。这个包装器在需要通过在多个时间步骤中保持相同动作来实现预期效果的环境中特别有用。例如，一些物理环境，如运动控制任务，可能需要持续的力输入才能产生显著的状态变化。

使用这个包装器可以减少问题的时间复杂度，因为它允许代理在一个时间步内执行多个动作。这可以加快学习速度，因为代理在一个时间步内需要做出的决策更少。然而，这也可能牺牲一定程度的决策精度，因为代理无法在连续的时间步之间改变其动作。

请注意，使用ActionRepeatWrapper可能并不适合所有类型的环境。具体来说，对于需要在每个时间步骤做出新决策的环境，或者动作的时间序列对结果有重大影响的环境，它可能不是最佳选择。

Interfaces:

__init__, step

Properties:

env (gym.Env): 要包装的环境。
action_repeat (int): 重复操作的次数。

__init__(env: Env, action_repeat: int = 1)[源代码]¶

Overview:

初始化 ActionRepeatWrapper 类。

Arguments:

env (gym.Env): 要包装的环境。
action_repeat (int): 重复动作的次数。默认值为1。

step(action: int | ndarray) → tuple[source]¶

Overview:

采取给定的动作并重复执行指定的步数。奖励将被累加。

Arguments:

action (Union[int, np.ndarray]): 要重复的动作。

Returns:

obs (np.ndarray): 重复操作后的观察结果。
reward (float): 重复动作后的奖励总和。
完成 (bool): 该集是否已结束。
info (Dict): 包含辅助诊断信息。

延迟奖励包装器¶

class ding.envs.DelayRewardWrapper(env: Env, delay_reward_step: int = 0)[source]¶

Overview:

DelayRewardWrapper 类是一个 gym 包装器，用于延迟奖励。它在预定义的步数内累积奖励，并仅在此间隔结束时返回累积的奖励。在其他时间，它返回的奖励为 0。

这个包装器在动作的影响不是立即可见，而是延迟到几个步骤之后的环境中特别有用。例如，在策略游戏或规划任务中，一个动作的效果可能不会直接显现，但它会促成一系列动作，最终导致奖励。在这些情况下，将奖励延迟以匹配动作效果的延迟可以使学习过程更符合问题的本质。

然而，使用这个包装器可能会增加学习的难度，因为代理需要将其行动与延迟的结果关联起来。它还引入了一种非标准的奖励结构，这可能会限制某些强化学习算法的适用性。

请注意，使用DelayRewardWrapper可能并不适合所有类型的环境。具体来说，对于那些动作效果立即可见且奖励应相应分配的环境，它可能不是最佳选择。

Interfaces:

__init__, reset, step

Properties:

env (gym.Env): 要包装的环境。
delay_reward_step (int): 延迟和累积奖励的步数。

__init__(env: Env, delay_reward_step: int = 0)[源代码]¶

Overview:

初始化 DelayRewardWrapper 类。

Arguments:

env (gym.Env): 要包装的环境。
delay_reward_step (int): 延迟和累积奖励的步数。

reset() → ndarray[来源]¶

Overview:

重置环境状态，并重置延迟奖励持续时间和当前延迟奖励。

Returns:

obs (np.ndarray): 环境的初始观察值。

step(action: int | ndarray) → tuple[source]¶

Overview:

执行给定的动作并重复指定的步数。奖励会被累加。如果步数等于延迟奖励步数，则返回累积的奖励并重置延迟奖励持续时间和当前延迟奖励。否则，返回0的奖励。

Arguments:

action (Union[int, np.ndarray]): 在步骤中采取的行动。

Returns:

obs (np.ndarray): 步骤后的观察结果。
reward (float): 延迟奖励步骤后的累积奖励或0。
完成 (bool): 该集是否已结束。
info (Dict): 包含辅助诊断信息。

ObsTransposeWrapper¶

class ding.envs.ObsTransposeWrapper(env: Env)[source]¶

Overview:

ObsTransposeWrapper 类是一个 gym 包装器，用于将观察值转置以使通道维度位于第一位。这对于某些期望通道维度为第一维度的神经网络类型可能很有帮助。

Interfaces:

__init__, observation

Properties:

env (gym.Env): 要包装的环境。
observation_space (gym.spaces.Box): 转换后的观察空间。

__init__(env: Env)[源代码]¶

Overview:

初始化ObsTransposeWrapper类并根据环境的观察空间更新观察空间。

Arguments:

env (gym.Env): 要包装的环境。

observation(obs: tuple | ndarray) → tuple | ndarray[source]¶

Overview:

将观测值转置以将通道维度放在首位。如果观测值是一个元组，则元组中的每个元素都会独立转置。

Arguments:

obs (Union[tuple, np.ndarray]): 原始观测值。

Returns:

obs (Union[tuple, np.ndarray]): 转置后的观测值。

ObsNormWrapper¶

class ding.envs.ObsNormWrapper(env: Env)[source]¶

Overview:

ObsNormWrapper 类是一个 gym 观察包装器，它根据运行均值和标准差（std）对观察进行归一化。

Interfaces:

__init__, step, reset, observation

Properties:

env (gym.Env): 要包装的环境。
data_count (int): 到目前为止观察到的数据点的数量。
clip_range (Tuple[int, int]): 归一化观测值的裁剪范围。
rms (RunningMeanStd): 观测值的运行均值和标准差。

__init__(env: Env)[source]¶

Overview:

初始化ObsNormWrapper类。

Arguments:

env (gym.Env): 要包装的环境。

observation(observation: ndarray) → ndarray[source]¶

Overview:

使用当前的运行平均值和标准差对观测值进行归一化。如果观测到的数据点少于30个，则返回原始观测值。

Arguments:

观察值 (np.ndarray): 原始观察值。

Returns:

观察值 (np.ndarray): 归一化的观察值。

reset(**kwargs)[source]¶

Overview:

重置环境以及与运行均值和标准差相关的属性。

Arguments:

kwargs (Dict): 传递给环境的reset函数的关键字参数。

Returns:

observation (np.ndarray): 环境的初始观察值。

step(action: int | ndarray)[source]¶

Overview:

在环境中采取一个动作，更新运行的平均值和标准差，并返回归一化的观察结果。

Arguments:

action (Union[int, np.ndarray]): 在环境中采取的行动。

Returns:

obs (np.ndarray): 动作后的归一化观测值。
奖励 (float): 动作后的奖励。
完成 (bool): 该集是否已结束。
info (Dict): 包含辅助诊断信息。

StaticObsNormWrapper¶

class ding.envs.StaticObsNormWrapper(env: Env, mean: ndarray, std: ndarray)[source]¶

Overview:

StaticObsNormWrapper 类是一个 gym 观察包装器，它根据预计算的平均值和标准差（std）从固定数据集中对观察进行归一化。

Interfaces:

__init__, observation

Properties:

env (gym.Env): 要包装的环境。
mean (numpy.ndarray): 固定数据集中观测值的平均值。
std (numpy.ndarray): 固定数据集中观测值的标准差。
clip_range (Tuple[int, int]): 归一化观测值的裁剪范围。

__init__(env: Env, mean: ndarray, std: ndarray)[源代码]¶

Overview:

初始化 StaticObsNormWrapper 类。

Arguments:

env (gym.Env): 要包装的环境。
mean (numpy.ndarray): 固定数据集中观测值的平均值。
std (numpy.ndarray): 固定数据集中观测值的标准差。

observation(observation: ndarray) → ndarray[来源]¶

Overview:

使用预先计算的均值和标准差对给定的观测值进行归一化。归一化后的观测值随后在指定范围内进行裁剪。

Arguments:

观察值 (np.ndarray): 原始观察值。

Returns:

observation (np.ndarray): 归一化和裁剪后的观测值。

RewardNormWrapper¶

class ding.envs.RewardNormWrapper(env: Env, reward_discount: float)[source]¶

Overview:

这个包装类根据运行的标准差对奖励进行归一化。它扩展了gym.RewardWrapper。

Interfaces:

__init__, step, reward, reset

Properties:

env (gym.Env): 要包装的环境。
累积奖励 (numpy.ndarray): 累积的奖励，初始化为零并在step方法中更新。
reward_discount (float): 奖励的折扣因子。
data_count (int): 数据的计数器，在每次step调用时递增。
rms (RunningMeanStd): 一个RunningMeanStd的实例，用于计算奖励的运行平均值和标准差。

__init__(env: Env, reward_discount: float) → None[源代码]¶

Overview:

初始化RewardNormWrapper，根据运行均值和标准差设置属性。

Arguments:

env (gym.Env): 要包装的环境。
reward_discount (float): 奖励的折扣因子。

reset(**kwargs)[source]¶

Overview:

重置环境的状态并重置属性（NumType 重置为 0，self.rms 作为重置的 rms 包装器）

Arguments:

kwargs (Dict): 使用此关键字参数重置

reward(reward: float) → float[source]¶

Overview:

如果data_count大于30，则标准化奖励。

Arguments:

奖励 (float): 原始奖励。

Returns:

奖励 (float): 归一化的奖励。

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

Overview:

使用给定的动作步进环境，更新属性并返回新的观察结果、奖励、完成状态和信息。

Arguments:

action (Any): 在环境中执行的操作。

Returns:

观察 (np.ndarray): 执行动作并更新 self.rms 后的归一化观察。
reward (float): Amount of reward returned after the action execution (normalized) and updated
self.cum_reward.
done (bool): Whether the episode has ended, in which case further step() calls will return
未定义的结果。
info (Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes
学习).

RamWrapper¶

class ding.envs.RamWrapper(env: Env, render: bool = False)[source]¶

Overview:

这个包装类将RAM环境包装成一个类似图像的环境。它扩展了gym.Wrapper。

Interfaces:

__init__, reset, step

Properties:

env (gym.Env): 要包装的环境。
observation_space (gym.spaces.Box): 被封装环境的观察空间。

__init__(env: Env, render: bool = False) → None[source]¶

Overview:

初始化RamWrapper并设置观察空间以包装RAM环境。

Arguments:

env (gym.Env): 要包装的环境。
render (bool): 是否渲染环境，默认为 False。

reset() → ndarray[source]¶

Overview:

重置环境的状态并返回一个重新塑造的观察结果。

Returns:

observation (np.ndarray): 重置和重塑后的新观察值。

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

Overview:

在环境中使用给定的动作执行一步。重复动作，累加奖励并重塑观察结果。

Arguments:

action (Any): 在环境中采取的动作。

Returns:

观察值 (np.ndarray): 经过步骤后的重塑观察值，带有类型限制。
奖励 (Any): 前一个动作后返回的奖励金额。
完成 (bool): 该回合是否已结束，如果结束，则进一步的 step() 调用将返回未定义的结果。
info (Dict): 包含辅助诊断信息（有助于调试，有时也有助于学习）。

EvalEpisodeReturnWrapper¶

class ding.envs.EvalEpisodeReturnWrapper(env: Env)[source]¶

Overview:

一个用于gym环境的包装器，它在每个时间步累积奖励，并在info中返回该回合的总奖励。这用于评估目的。

Interfaces:

__init__, reset, step

Properties:

env (gym.Env): 要包装的环境。

__init__(env: Env)[源代码]¶

Overview:

初始化EvalEpisodeReturnWrapper。这涉及设置要包装的环境。

Arguments:

env (gym.Env): 要包装的环境。

reset() → ndarray[source]¶

Overview:

重置环境并将累积奖励初始化为零。

Returns:

obs (np.ndarray): 环境的初始观察值。

step(action: Any) → tuple[来源]¶

Overview:

使用提供的动作步进环境，累积返回的奖励，并在回合结束时将总奖励添加到info中。

Arguments:

action (Any): 在环境中采取的动作。

Returns:

obs (np.ndarray): 来自环境的下一个观察值。
奖励 (float): 执行动作后获得的奖励。
完成 (bool): 该回合是否已完成。
info (Dict[str, Any]): A dictionary of extra information, which includes ‘eval_episode_return’ if
这一集已经完成。

Examples:

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

GymHybridDictActionWrapper¶

class ding.envs.GymHybridDictActionWrapper(env: Env)[source]¶

Overview:

将Gym-Hybrid的原始gym.spaces.Tuple动作空间转换为gym.spaces.Dict。

Interfaces:

__init__, action

Properties:

env (gym.Env): 要包装的环境。
action_space (gym.spaces.Dict): 新的动作空间。

__init__(env: Env) → None[源代码]¶

Overview:

初始化GymHybridDictActionWrapper，设置新的动作空间。

Arguments:

env (gym.Env): 要包装的环境。

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

Overview:

在环境中执行给定的动作，将动作从字典转换为元组，并返回新的观察结果、奖励、完成状态和信息。

Arguments:

action (Dict): 要在环境中执行的操作，以字典形式结构化。

Returns:

observation (Dict): The wrapped observation, which includes the current observation,
之前的动作和之前的奖励。
奖励 (float): 动作执行后返回的奖励金额。
done (bool): Whether the episode has ended, in which case further step() calls will return
未定义的结果。
info (Dict): Contains auxiliary diagnostic information (helpful for debugging, and
有时候学习）。

ObsPlusPrevActRewWrapper¶

class ding.envs.ObsPlusPrevActRewWrapper(env: Env)[source]¶

Overview:

此包装器用于策略NGU。它将一个字典设置为新的包装观察，其中包括当前观察、先前动作和先前奖励。

Interfaces:

__init__, reset, step

Properties:

env (gym.Env): 要包装的环境。
prev_action (int): 上一个动作。
prev_reward_extrinsic (float): 之前的奖励。

__init__(env: Env) → None[source]¶

Overview:

初始化ObsPlusPrevActRewWrapper，设置先前的动作和奖励。

Arguments:

env (gym.Env): 要包装的环境。

reset() → Dict[source]¶

Overview:

重置环境的状态，并返回包装后的观察结果。

Returns:

observation (Dict): The wrapped observation, which includes the current observation,
之前的动作和之前的奖励。

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

Overview:

在环境中执行给定的动作，保存前一个动作和奖励以供下一次观察使用，并返回新的观察结果、奖励、完成状态和信息。

Arguments:

action (Any): 在环境中执行的操作。

Returns:

observation (Dict): The wrapped observation, which includes the current observation,
之前的动作和之前的奖励。
奖励 (float): 动作执行后返回的奖励金额。
done (bool): Whether the episode has ended, in which case further step() calls will return
未定义的结果。
info (Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes
学习).

时间限制包装器¶

class ding.envs.TimeLimitWrapper(env: Env, max_limit: int)[source]¶

Overview:: 此类用于在环境中强制执行时间限制。
Interfaces:: __init__, reset, step

__init__(env: Env, max_limit: int) → None[source]¶

Overview:

初始化TimeLimitWrapper，设置时间步长的最大限制。

Arguments:

env (gym.Env): 要包装的环境。
max_limit (int): 时间步的最大限制。

reset() → ndarray[source]¶

Overview:

重置环境状态和时间计数器。

Returns:

观察值 (np.ndarray): 重置后的新观察值。

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

Overview:

在环境中执行给定的动作，更新时间计数器，并返回新的观察结果、奖励、完成状态和信息。

Arguments:

action (Any): 在环境中执行的操作。

Returns:

观察 (np.ndarray): 动作执行后的新观察。
奖励 (float): 动作执行后返回的奖励金额。
done (bool): Whether the episode has ended, in which case further step() calls will return
未定义的结果。
info (Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes
学习).

FlatObsWrapper¶

class ding.envs.FlatObsWrapper(env: Env, maxStrLen: int = 96)[source]¶

Overview:: 此类用于展平环境的观察空间。注意：仅适用于像minigrid这样的环境。
Interfaces:: __init__, observation, reset, step

__init__(env: Env, maxStrLen: int = 96) → None[source]¶

Overview:

初始化 FlatObsWrapper，设置新的观察空间。

Arguments:

env (gym.Env): 要包装的环境。
maxStrLen (int): 任务字符串的最大长度，默认值为96。

observation(obs: ndarray | Tuple) → ndarray[source]¶

Overview:

处理观测数据，将任务转换为独热编码，并将其与图像数据连接起来。

Arguments:

obs (Union[np.ndarray, Tuple]): 要处理的原始观测数据。

Returns:

obs (np.ndarray): 处理后的观测值。

reset(*args, **kwargs) → ndarray[source]¶

Overview:

重置环境的状态并返回处理后的观察结果。

Returns:

observation (np.ndarray): 重置后处理过的观察值。

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

Overview:

在环境中执行给定的动作，并返回处理后的观察结果、奖励、完成状态和信息。

Returns:

observation (np.ndarray): 动作执行后处理过的观察结果。
奖励 (float): 动作执行后返回的奖励金额。
done (bool): Whether the episode has ended, in which case further step() calls will return
未定义的结果。
info (Dict): Contains auxiliary diagnostic information (helpful for debugging, and sometimes
学习).

GymToGymnasiumWrapper¶

class ding.envs.GymToGymnasiumWrapper(env: Env)[source]¶

Overview:: 此类用于将体育馆环境包装为健身房环境。
Interfaces:: __init__, seed, reset, step

__init__(env: Env) → None[来源]¶

Overview:

初始化GymToGymnasiumWrapper。

Arguments:

env (gymnasium.Env): 要包装的gymnasium环境。

reset() → ndarray[source]¶

Overview:

重置环境的状态并返回新的观察结果。如果设置了种子，则在重置时使用它。

Returns:

观察值 (np.ndarray): 重置后的新观察值。

seed(seed: int) → None[source]¶

Overview:

设置环境的种子。

Arguments:

seed (int): 要设置的种子。

AllinObsWrapper¶

class ding.envs.AllinObsWrapper(env: Env)[source]¶

Overview:

此包装器用于策略Decision Transformer中，该策略在论文https://arxiv.org/abs/2106.01345中提出。它将一个字典{'obs': obs, 'reward': reward}设置为新的包装观察值，其中包括当前观察值和先前的奖励。

Interfaces:

__init__, reset, step, seed

Properties:

env (gym.Env): 要包装的环境。

__init__(env: Env) → None[源代码]¶

Overview:

初始化AllinObsWrapper。

Arguments:

env (gym.Env): 要包装的环境。

reset() → Dict[source]¶

Overview:

重置环境的状态并返回新的观察结果。

Returns:

observation (Dict): 重置后的新观察值，包括当前观察值和奖励。

seed(seed: int, dynamic_seed: bool = True) → None[source]¶

Overview:

设置环境的种子。

Arguments:

seed (int): 要设置的种子。
dynamic_seed (bool): 是否使用动态种子，默认为 True。

step(action: Any)[source]¶

Overview:

在环境中执行给定的动作，并返回新的观察结果、奖励、完成状态和信息。

Arguments:

action (Any): 在环境中执行的操作。

Returns:

时间步长 (BaseEnvTimestep): 执行动作后的时间步长。