d2c.utils

Config

The general config that integrates the app_config and model_config

class ConfigBuilder(app_config: Any, model_config_path: str, work_abs_dir: str, command_args: Optional[Dict] = None, experiment_type: str = 'benchmark')

Bases: object

Builder the complete configuration with app_config and model_config, and set the parameters according to the CLI input.

The main API method is:

• build_config(): get the complete configuration.

Parameters

• app_config – the app_config;

• model_config_path (str) – the model_config file path;

• model_config_path – the absolute path of the work dir that contains the run script, data dir and models dir.

• command_args (dict) – the CLI parameters input;

• experiment_type (str) – the available options are [‘benchmark’, ‘application’].

Returns

a complete configuration that can be used by main function.

build_config() → d2c.utils.utils.Flags

The API to build the final config.

static main_hyper_params(_model_cfg: Union[Dict, easydict.EasyDict, Any]) → Dict

Get the main hyperparameters of this experiment.

Parameters

_model_cfg (dict) – the model_config.

flat_dict(x: Dict) → Generator

read_config_from_json(config_file: str, encoding: Optional[str] = None, easydict: bool = False) → Union[Dict, easydict.EasyDict]

update_config(config_file: str, hyper_params: Optional[Dict] = None, encoding: Optional[str] = None) → easydict.EasyDict

read from config_file and update it by hyper_params

update_nested_dict_by_dict(from_dict: Dict, to_dict: Dict) → Dict

update_nested_dict_by_kv(dic: Dict, keys: str, value: Any) → None

make use of the shallow copy of dict to update value.

Dataloader

Dataloader for loading dataset and generating transitions.

There are data_loaders for benchmarks and real-world applications.

class BaseBMLoader(file_path: str, state_normalize: bool = False, reward_normalize: bool = False)

Bases: d2c.utils.dataloader.BaseDataLoader

The basic class of the benchmark dataset loader. Please inherit this class to build data loaders for different benchmarks.

The main API method the user should implement is:

• _load_data(): load the transitions from the dataset file and return in requested format.

The main API method that users of this class need to know is:

• get_transitions(): process the transitions from the dataset and return a namedtuple.

Parameters

• file_path (str) – the path of the benchmark dataset;

• state_normalize (bool) – if normalize the states;

• reward_normalize (bool) – if normalize the rewards.

static get_keys(h5file: h5py._hl.files.File) → List[str]

get_transitions(split_ratio: Optional[float] = None, split_shuffle: bool = True) → collections.OrderedDict

Get the transitions from the dataset.

Parameters

• split_ratio (float) – The ratio value for splitting the data.

• split_shuffle (bool) – If choosing the splitting data randomly.

Returns

A namedtuple that contains the elements of the transitions.

static norm_reward(r: numpy.ndarray) → numpy.ndarray

Normalize the reward.

Parameters

r (np.ndarray) – reward;

Returns

normalized reward.

static norm_state(s1: numpy.ndarray, s2: numpy.ndarray) → Tuple[numpy.ndarray, numpy.ndarray, numpy.ndarray, numpy.ndarray]

Normalize the states.

Parameters

• s1 (np.ndarray) – the states;

• s2 (np.ndarray) – the next states;

Returns

normalized states

property state_shift_scale

Get the shift and scale of the state normalization.

class BaseDataLoader

Bases: abc.ABC

The base class of the dataset loader.

Inherit this class to build data loaders for benchmarks and real-world applications.

The main methods

abstract get_transitions(**kwargs)

Get the transitions from the dataset.

Returns

A namedtuple that contains the elements of the transitions.

state_shift_scale()

Get the shift and scale of the state normalization.

class D4rlDataLoader(file_path: str, state_normalize: bool = False, reward_normalize: bool = False)

Bases: d2c.utils.dataloader.BaseBMLoader

Get transitions from the D4RL dataset.

TYPE: ClassVar[str] = 'd4rl'

Logger

Tools for logging the training information.

class WandbLogger(project: Optional[str] = None, entity: Optional[str] = None, name: Optional[str] = None, run_id: Optional[str] = None, config: Optional[dict] = None, dir_: Optional[str] = None, reinit: Optional[bool] = False, mode: Optional[str] = 'online')

Bases: object

Weights and Biases logger that sends data to https://wandb.ai/.

Parameters

• project (str) – W&B project name.

• entity (str) – W&B team/organization name. Default to None.

• name (str) – W&B run name. Default to None. If None, random name is assigned.

• run_id (str) – run id of W&B run to be resumed. Default to None.

• dir (str) – An absolute path to a directory where metadata will be stored.

• reinit (bool) – Allow multiple wandb.init() calls in the same process. (default: False)

• mode (str) – Can be “online”, “offline” or “disabled”. Defaults to online.

finish() → None

static write_summary(info: Union[Dict, collections.OrderedDict]) → None

write_summary_tensorboard(writer: torch.utils.tensorboard.writer.SummaryWriter, step: int, info: Dict) → None

Networks

Neural networks for RL models.

class ActorNetwork(observation_space: Union[gym.spaces.box.Box, gym.spaces.space.Space], action_space: Union[gym.spaces.box.Box, gym.spaces.space.Space], fc_layer_params: Sequence[int] = (), device: Union[str, int, torch.device] = 'cpu')

Bases: torch.nn.modules.module.Module

Stochastic Actor network.

Parameters

• observation_space (Box) – the observation space information. It is an instance of class: gym.spaces.Box.

• action_space (Box) – the action space information. It is an instance of class: gym.spaces.Box.

• fc_layer_params (tuple) – the network parameter. For example: (300, 300) means a 2-layer network with 300 units in each layer.

• device – which device to create this model on. Default to ‘cpu’.

property action_space

forward(state: Union[numpy.ndarray, torch.Tensor]) → Tuple[torch.Tensor, torch.Tensor, torch.Tensor]

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

get_log_density(state: torch.Tensor, action: torch.Tensor) → torch.Tensor

sample(state: Union[numpy.ndarray, torch.Tensor]) → torch.Tensor

sample_n(state: Union[numpy.ndarray, torch.Tensor], n: int = 1) → Tuple[torch.Tensor, torch.Tensor, torch.Tensor]

training: bool

class ActorNetworkDet(observation_space: Union[gym.spaces.box.Box, gym.spaces.space.Space], action_space: Union[gym.spaces.box.Box, gym.spaces.space.Space], fc_layer_params: Sequence[int] = (), device: Union[str, int, torch.device] = 'cpu')

Bases: torch.nn.modules.module.Module

Deterministic Actor network.

Parameters

• observation_space (Box) – the observation space information. It is an instance of class: gym.spaces.Box.

• action_space (Box) – the action space information. It is an instance of class: gym.spaces.Box.

• fc_layer_params (tuple) – the network parameter. For example: (300, 300) means a 2-layer network with 300 units in each layer.

• device – which device to create this model on. Default to ‘cpu’.

property action_space

forward(state: Union[numpy.ndarray, torch.Tensor]) → torch.Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

training: bool

class Classifier(input_dim: int, output_dim: int = 2, fc_layer_params: Sequence[int] = (), device: Union[str, int, torch.device] = 'cpu')

Bases: torch.nn.modules.module.Module

based on Multi-layer Perceptron. Discriminator network for H2O.

Parameters

• input_dim (int) – the dimension of the input.

• output_dim (int) – the dimension of the output.

• fc_layer_params (tuple) – the network parameter. For example: (300, 300) means a 2-layer network with 300 units in each layer.

• device – which device to create this model on. Default to ‘cpu’.

forward(inputs: Union[numpy.ndarray, torch.Tensor]) → torch.Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

training: bool

class ConcatClassifier(dim: int = 1, *args, **kwargs)

Bases: d2c.utils.networks.Classifier

Concatenate inputs along dimension and then pass through MLP.

Parameters

dim (int) – concatenate inputs in row or column (0 or 1).

forward(*inputs: Union[numpy.ndarray, torch.Tensor]) → torch.Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

training: bool

class CriticNetwork(observation_space: Union[gym.spaces.box.Box, gym.spaces.space.Space, int], action_space: Union[gym.spaces.box.Box, gym.spaces.space.Space, int], fc_layer_params: Sequence[int] = (), device: Union[str, int, torch.device] = 'cpu')

Bases: torch.nn.modules.module.Module

Critic Network.

Parameters

• or int observation_space (gym.spaces.Box) – the observation space information. It is an instance of class: gym.spaces.Box. observation_space can also be an integer which represents the dimension of the observation.

• or int action_space (gym.spaces.Box) – the action space information. It is an instance of class: gym.spaces.Box. action_space can also be an integer which represents the dimension of the action.

• fc_layer_params (tuple) – the network parameter. For example: (300, 300) means a 2-layer network with 300 units in each layer.

• device – which device to create this model on. Default to ‘cpu’.

forward(state: Union[numpy.ndarray, torch.Tensor], action: Union[numpy.ndarray, torch.Tensor]) → torch.Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

training: bool

class Discriminator(observation_space: Union[gym.spaces.box.Box, gym.spaces.space.Space], action_space: Union[gym.spaces.box.Box, gym.spaces.space.Space], fc_layer_params: Sequence[int] = (), device: Union[str, int, torch.device] = 'cpu')

Bases: torch.nn.modules.module.Module

A Discriminator Network(for DMIL).

Parameters

• observation_space (Box) – the observation space information. It is an instance of class: gym.spaces.Box.

• action_space (Box) – the action space information. It is an instance of class: gym.spaces.Box.

• fc_layer_params (tuple) – the network parameter. For example: (300, 300) means a 2-layer network with 300 units in each layer.

• device – which device to create this model on. Default to ‘cpu’.

forward(state: Union[numpy.ndarray, torch.Tensor], action: Union[numpy.ndarray, torch.Tensor], logpi: Union[numpy.ndarray, torch.Tensor], lossf: Union[numpy.ndarray, torch.Tensor]) → torch.Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

training: bool

class MLP(input_dim: int, output_dim: int, fc_layer_params: Sequence[int] = (), device: Union[str, int, torch.device] = 'cpu')

Bases: torch.nn.modules.module.Module

Multi-layer Perceptron.

Parameters

• input_dim (int) – the dimension of the input.

• output_dim (int) – the dimension of the output.

• fc_layer_params (tuple) – the network parameter. For example: (300, 300) means a 2-layer network with 300 units in each layer.

• device – which device to create this model on. Default to ‘cpu’.

forward(inputs: Union[numpy.ndarray, torch.Tensor]) → torch.Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

training: bool

class ProbDynamicsNetwork(state_dim: int, action_dim: int, fc_layer_params: Sequence[int] = (), local_mode: bool = False, with_reward: bool = False, device: Union[str, int, torch.device] = 'cpu')

Bases: torch.nn.modules.module.Module

Stochastic Dynamics network(Probabilistic dynamics model).

Parameters

• state_dim (int) – the observation space dimension.

• action_dim (int) – the action space dimension.

• fc_layer_params (tuple) – the network parameter. For example: (300, 300) means a 2-layer network with 300 units in each layer.

• local_mode (bool) – local_mode means that this model predicts the difference to the current state.

• with_reward (bool) – if the output of the dynamics contains the reward or not.

• device – which device to create this model on. Default to ‘cpu’.

forward(state: Union[numpy.ndarray, torch.Tensor], action: Union[numpy.ndarray, torch.Tensor]) → Tuple[torch.Tensor, torch.Tensor, torch.distributions.distribution.Distribution]

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

get_log_density(state: torch.Tensor, action: torch.Tensor, output: torch.Tensor) → torch.Tensor

property max_logstd

property min_logstd

training: bool

class Scalar(init_value: float, device: Union[str, int, torch.device] = 'cpu')

Bases: torch.nn.modules.module.Module

Scalar network

Parameters

init_value (float) – initialized value for the scalar

forward() → torch.Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

training: bool

class ValueNetwork(observation_space: Union[gym.spaces.box.Box, gym.spaces.space.Space, int], fc_layer_params: Sequence[int] = (), device: Union[str, int, torch.device] = 'cpu')

Bases: torch.nn.modules.module.Module

Value Network.

Parameters

• or int observation_space (gym.spaces.Box) – The observation space information. It is an instance of class: gym.spaces.Box. It can also be an integer which represents the dimension of the observation.

• fc_layer_params (tuple) – the network parameter. For example: (300, 300) means a 2-layer network with 300 units in each layer.

• device – which device to create this model on. Default to ‘cpu’.

forward(inputs: Union[numpy.ndarray, torch.Tensor]) → torch.Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

training: bool

get_spec_means_mags(space: gym.spaces.box.Box, device: Optional[Union[str, int, torch.device]] = None) → Tuple[torch.Tensor, torch.Tensor]

miniblock(input_size: int, output_size: int = 0, norm_layer: Optional[Type[torch.nn.modules.module.Module]] = None, activation: Optional[Type[torch.nn.modules.module.Module]] = None, linear_layer: Type[torch.nn.modules.linear.Linear] = <class 'torch.nn.modules.linear.Linear'>) → List[torch.nn.modules.module.Module]

Construct a miniblock with given input/output-size, norm layer and activation.

Policies

Policies used by various agents.

class DeterministicPolicy(a_network: torch.nn.modules.module.Module)

Bases: torch.nn.modules.module.Module

Returns deterministic action.

forward(observation: Union[numpy.ndarray, torch.Tensor]) → numpy.ndarray

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

training: bool

class DeterministicSoftPolicy(a_network: torch.nn.modules.module.Module)

Bases: torch.nn.modules.module.Module

Returns mode of policy distribution.

forward(observation: Union[numpy.ndarray, torch.Tensor]) → numpy.ndarray

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

training: bool

Replaybuffer

The replay buffer for RL training.

class ReplayBuffer(state_dim: int, action_dim: int, max_size: int = 2000000, device: Union[str, int, torch.device] = 'cpu')

Bases: object

The base replay buffer.

Parameters

• state_dim (int) – the dimension of the state.

• action_dim (int) – the dimension of the action.

• max_size (int) – the maximum size of the buffer.

• device (str) – which device to create the data on. Default to ‘cpu’.

add(*, state: Union[numpy.ndarray, torch.Tensor], action: Union[numpy.ndarray, torch.Tensor], next_state: Union[numpy.ndarray, torch.Tensor], next_action: Union[numpy.ndarray, torch.Tensor], reward: Union[numpy.ndarray, torch.Tensor, float, int], done: Union[numpy.ndarray, torch.Tensor, float, int], cost: Optional[Union[numpy.ndarray, torch.Tensor, float, int]] = None) → None

Add a transition into the buffer.

Parameters

• state (np.ndarray) – the state with shape (1, state_dim) or (state_dim,)

• action (np.ndarray) – the action with shape (1, action_dim) or (action_dim)

• next_state (np.ndarray) – the next_state with shape (1, state_dim) or (state_dim,)

• next_action (np.ndarray) – the next_action with shape (1, action_dim) or (action_dim)

• reward (np.ndarray) – the reward with shape (1,) or ()

• done (np.ndarray) – the done with shape (1,) or ()

• cost (np.ndarray) – the cost with shape (1,) or ()

add_transitions(*, state: Union[numpy.ndarray, torch.Tensor], action: Union[numpy.ndarray, torch.Tensor], next_state: Union[numpy.ndarray, torch.Tensor], next_action: Union[numpy.ndarray, torch.Tensor], reward: Optional[Union[numpy.ndarray, torch.Tensor]] = None, done: Optional[Union[numpy.ndarray, torch.Tensor]] = None, cost: Optional[Union[numpy.ndarray, torch.Tensor]] = None) → None

Add a batch of transitions into the buffer.

Parameters

• state (np.ndarray) – the state with shape (batch_size, state_dim)

• action (np.ndarray) – the action with shape (batch_size, action_dim)

• next_state (np.ndarray) – the next_state with shape (batch_size, state_dim)

• next_action (np.ndarray) – the next_action with shape (batch_size, action_dim)

• reward (np.ndarray) – the reward with shape (batch_size,)

• done (np.ndarray) – the done with shape (batch_size,)

• cost (np.ndarray) – the cost with shape (batch_size,)

property capacity

The capacity of the replay buffer.

property data

All the transitions in the buffer.

get_batch_indices(indices: numpy.ndarray) → collections.OrderedDict

Get the batch of data according to the given indices.

sample_batch(batch_size: int) → collections.OrderedDict

Sample a batch of data randomly.

Parameters

batch_size (int) – the batch size of the sample data.

property shuffle_indices

Returning the shuffled indices of the transitions in the buffer.

property size

The number of the transitions in the replay buffer.

Utils

A collection of some little utils.

class Flags(**kwargs)

Bases: object

abs_file_path(file, relative_path)

add_gaussian_noise(data: numpy.ndarray, space: Union[gym.spaces.box.Box, gym.spaces.space.Space], std: float) → numpy.ndarray

chain_gene(*args: List[Generator]) → Generator

Connect several Generator objects into one Generator object.

generate_xml_path() → str

get_optimizer(name: str) → Callable

Get an optimizer generator that returns an optimizer according to lr.

get_summary_str(step: Optional[int] = None, info: Optional[Dict] = None, prefix: str = '') → str

maybe_makedirs(log_dir: str) → None

parse_xml_name(env_name: str) → str

set_seed(seed: int) → None

to_array_as(x, y)

update_source_env_density(variety_degree: float, env_name: str) → None

update_source_env_friction(variety_degree: float, env_name: str) → None

update_source_env_gravity(variety_degree: float, env_name: str) → None

update_target_env(env_name: str) → None

update_xml(index: str, env_name: str) → None

Wrappers

A collection of gym wrappers.

class NormalizeBoxActionWrapper(env: gym.core.Env)

Bases: gym.core.Env[gym.core.ObsType, gym.core.ActType]

Rescale the action space of the environment.

action(action: numpy.ndarray) → numpy.ndarray

Returns a modified action before env.step() is called.

reverse_action(scaled_action: numpy.ndarray) → numpy.ndarray

Returns a reversed action.

class NormalizeStateWrapper(env: gym.core.Env, shift: numpy.ndarray, scale: numpy.ndarray)

Bases: gym.core.Env[gym.core.ObsType, gym.core.ActType]

Wraps an environment to shift and scale observations.

observation(observation: numpy.ndarray) → numpy.ndarray

Returns a modified observation.

check_and_normalize_box_actions(env: gym.core.Env) → gym.core.Env

Wrap env to normalize actions if [low, high] != [-1, 1].

wrapped_norm_obs_env(gym_env: gym.core.Env, shift: numpy.ndarray, scale: numpy.ndarray) → gym.core.Env

Create a gym environment with normalized observations.

Parameters

• gym_env – the original gym env.

• shift (np.ndarray) – a numpy vector to shift observations.

• scale (np.ndarray) – a numpy vector to scale observations.

Returns: An initialized gym environment.