worker.collector.base_serial_collector

base_serial_collector

Please Reference ding/worker/collector/base_serial_collector.py for usage

ISerialCollector

class ding.worker.collector.base_serial_collector.ISerialCollector

Overview:

Abstract baseclass for serial collector.

Interfaces:

default_config, reset_env, reset_policy, reset, collect

Property:

envstep

abstract collect(per_collect_target: Any) → List[Any]

Overview:

Collect the corresponding data according to the specified target and return. There are different definitions in episode and sample mode.

classmethod default_config() → EasyDict

Overview:

Get collector’s default config. We merge collector’s default config with other default configs and user’s config to get the final config.

Return:

cfg: (EasyDict): collector’s default config

abstract property envstep: int

Overview:

Get the total envstep num.

abstract reset(_policy: namedtuple | None = None, _env: BaseEnvManager | None = None) → None

Overview:

Reset collector’s policy and environment. Use new policy and environment to collect data.

abstract reset_env(_env: BaseEnvManager | None = None) → None

Overview:

Reset collector’s environment. In some case, we need collector use the same policy to collect data in different environments. We can use reset_env to reset the environment.

abstract reset_policy(_policy: namedtuple | None = None) → None

Overview:

Reset collector’s policy. In some case, we need collector work in this same environment but use different policy to collect data. We can use reset_policy to reset the policy.

create_serial_collector

Overview:

Create a specific collector instance based on the config.

get_serial_collector_cls

Overview:

Get the specific collector class according to the config.

CachePool

class ding.worker.collector.base_serial_collector.CachePool(name: str, env_num: int, deepcopy: bool = False)

Overview:

CachePool is the repository of cache items.

Interfaces:

__init__, update, __getitem__, reset

__getitem__(idx: int) → Any

Overview:

Get item in cache pool.

Arguments:

• idx (int): The index of the item we need to get.

Return:

• item (Any): The item we get.

__init__(name: str, env_num: int, deepcopy: bool = False) → None

Overview:

Initialization method.

Arguments:

• name (str): name of cache

• env_num (int): number of environments

• deepcopy (bool): whether to deepcopy data

reset(idx: int) → None

Overview:

Reset the cache pool.

Arguments:

• idx (int): The index of the position we need to reset.

update(data: Dict[int, Any] | list) → None

Overview:

Update elements in cache pool.

Arguments:

• data (Dict[int, Any]): A dict containing update index-value pairs. Key is index in cache pool, and value is the new element.

TrajBuffer

class ding.worker.collector.base_serial_collector.TrajBuffer(maxlen: int, *args, deepcopy: bool = False, **kwargs)

Overview:

TrajBuffer is used to store traj_len pieces of transitions.

Interfaces:

__init__, append

__init__(maxlen: int, *args, deepcopy: bool = False, **kwargs) → None

Overview:

Initialization trajBuffer.

Arguments:

• maxlen (int): The maximum length of trajectory buffer.

• deepcopy (bool): Whether to deepcopy data when do operation.

append(data: Any) → None

Overview:

Append data to trajBuffer.

to_tensor_transitions

Overview:

Transform ths original transition return from env to tensor format.

Argument:

• data (List[Dict[str, Any]]): The data that will be transformed to tensor.

• shallow_copy_next_obs (bool): Whether to shallow copy next_obs. Default: True.

Return:

• data (List[Dict[str, Any]]): The transformed tensor-like data.

Tip

In order to save memory, If there are next_obs in the passed data, we do special treatment on next_obs so that the next_obs of each state in the data fragment is the next state’s obs and the next_obs of the last state is its own next_obsself. Besides, we set transform_scalar to False to avoid the extra .item() operation.