utils.default_helper

default_helper

Please Reference ding/ding/utils/default_helper.py for usage.

lists_to_dicts

Overview:

Transform a list of dicts to a dict of lists.

Arguments:

• data (Union[List[Union[dict, NamedTuple]], Tuple[Union[dict, NamedTuple]]]):

  A dict of lists need to be transformed

• recursive (bool): whether recursively deals with dict element

Returns:

• newdata (Union[Mapping[object, object], NamedTuple]): A list of dicts as a result

Example:

>>> from ding.utils import *
>>> lists_to_dicts([{1: 1, 10: 3}, {1: 2, 10: 4}])
{1: [1, 2], 10: [3, 4]}

dicts_to_lists

Overview:

Transform a dict of lists to a list of dicts.

Arguments:

• data (Mapping[object, list]): A list of dicts need to be transformed

Returns:

• newdata (List[Mapping[object, object]]): A dict of lists as a result

Example:

>>> from ding.utils import *
>>> dicts_to_lists({1: [1, 2], 10: [3, 4]})
[{1: 1, 10: 3}, {1: 2, 10: 4}]

override

Overview:

Annotation for documenting method overrides.

Arguments:

• cls (type): The superclass that provides the overridden method. If this

  cls does not actually have the method, an error is raised.

squeeze

Overview:

Squeeze data from tuple, list or dict to single object

Example:

>>> a = (4, )
>>> a = squeeze(a)
>>> print(a)
>>> 4

default_get

Overview:

Getting the value by input, checks generically on the inputs with at least data and name. If name exists in data, get the value at name; else, add name to default_get_setwith value generated by default_fn (or directly as default_value) that is checked by `` judge_fn`` to be legal.

Arguments:

• data(dict): Data input dictionary

• name(str): Key name

• default_value(Optional[Any]) = None,

• default_fn(Optional[Callable]) = Value

• judge_fn(Optional[Callable]) = None

Returns:

• ret(list): Splitted data

• residual(list): Residule list

list_split

Overview:

Split list of data by step.

Arguments:

• data(list): List of data for spliting

• step(int): Number of step for spliting

Returns:

• ret(list): List of splitted data.

• residual(list): Residule list. This value is None when data divides steps.

Example:

>>> list_split([1,2,3,4],2)
([[1, 2], [3, 4]], None)
>>> list_split([1,2,3,4],3)
([[1, 2, 3]], [4])

error_wrapper

Overview:

wrap the function, so that any Exception in the function will be catched and return the default_ret

Arguments:

• fn (Callable): the function to be wraped

• default_ret (obj): the default return when an Exception occurred in the function

Returns:

• wrapper (Callable): the wrapped function

Examples:

>>> # Used to checkfor Fakelink (Refer to utils.linklink_dist_helper.py)
>>> def get_rank():  # Get the rank of linklink model, return 0 if use FakeLink.
>>>    if is_fake_link:
>>>        return 0
>>>    return error_wrapper(link.get_rank, 0)()

LimitedSpaceContainer

class ding.utils.default_helper.LimitedSpaceContainer(min_val: int, max_val: int)

Overview:

A space simulator.

Interface:

__init__, get_residual_space, release_space

__init__(min_val: int, max_val: int) → None

Overview:

Set min_val and max_val of the container, also set cur to min_val for initialization.

Arguments:

• min_val (int): Min volume of the container, usually 0.

• max_val (int): Max volume of the container.

acquire_space() → bool

Overview:

Try to get one pice of space. If there is one, return True; Otherwise return False.

Returns:

• flag (bool): Whether there is any piece of residual space.

decrease_space() → None

Overview:

Decrease one piece in space. Decrement max_val.

get_residual_space() → int

Overview:

Get all residual pieces of space. Set cur to max_val

Arguments:

• ret (int): Residual space, calculated by max_val - cur.

increase_space() → None

Overview:

Increase one piece in space. Increment max_val.

release_space() → None

Overview:

Release only one piece of space. Decrement cur, but ensure it won’t be negative.

deep_merge_dicts

Overview:

Merge two dicts by calling deep_update

Arguments:

• original (dict): Dict 1.

• new_dict (dict): Dict 2.

Returns:

• merged_dict (dict): A new dict that is d1 and d2 deeply merged.

deep_update

Overview:

Update original dict with values from new_dict recursively.

Arguments:

• original (dict): Dictionary with default values.

• new_dict (dict): Dictionary with values to be updated

• new_keys_allowed (bool): Whether new keys are allowed.

• whitelist (Optional[List[str]]):

  List of keys that correspond to dict values where new subkeys can be introduced. This is only at the top level.

• override_all_if_type_changes(Optional[List[str]]):

  List of top level keys with value=dict, for which we always simply override the entire value (dict), if the “type” key in that value dict changes.

Note

If new key is introduced in new_dict, then if new_keys_allowed is not True, an error will be thrown. Further, for sub-dicts, if the key is in the whitelist, then new subkeys can be introduced.

flatten_dict

Overview:

Flatten the dict, see example

Arguments:

• data (dict): Original nested dict

• delimiter (str): Delimiter of the keys of the new dict

Returns:

• data (dict): Flattened nested dict

Example:

>>> a
{'a': {'b': 100}}
>>> flatten_dict(a)
{'a/b': 100}

set_pkg_seed

Overview:

Side effect function to set seed for random, numpy random, and torch's manual seed. This is usaually used in entry scipt in the section of setting random seed for all package and instance

Argument:

• seed(int): Set seed

• use_cuda(bool) Whether use cude

Examples:

>>> # ../entry/xxxenv_xxxpolicy_main.py
>>> ...
# Set random seed for all package and instance
>>> collector_env.seed(seed)
>>> evaluator_env.seed(seed, dynamic_seed=False)
>>> set_pkg_seed(seed, use_cuda=cfg.policy.cuda)
>>> ...
# Set up RL Policy, etc.
>>> ...