utils

argparser

class MockArgs(dir, policy_name, vfcn_name)

Bases: tuple

Create new instance of MockArgs(dir, policy_name, vfcn_name)

property dir: Alias for field number 0

property policy_name: Alias for field number 1

property vfcn_name: Alias for field number 2

get_argparser() → ArgumentParser[source]: Return Pyrado’s default argument parser.

averaging

Only tested for 1-dim inputs.

class RunningExpDecayingAverage(alpha: float)[source]

Bases: object

Implementation of an exponentially decaying average estimator

Constructor

Parameters:: alpha – decay factor

reset(alpha: Optional[float] = None)[source]: Reset internal variables.

class RunningMemoryAverage(capacity: int)[source]

Bases: object

Implementation of an estimator that computes the average for a memorized buffer

Constructor

Parameters:: capacity – memory size

property memory: [<class 'numpy.ndarray'>, <class 'torch.Tensor'>, None]

reset(capacity: Optional[float] = None)[source]: Reset internal variables.

bijective_transformation

class BijectiveTransformation[source]

Bases: ABC

Base class for bijective transformations to be used for e.g. transforming the domain parameter space of an env.

These transformations are useful for avoiding infeasible values such as negative masses. The reasoning behind this is that some learning methods work on the set of real numbers, thus we make them learn in the transformed space, here the log-space, without telling them.

abstract ensures_non_negativity() → bool[source]: Whether this transformations ensures non-negativity, i.e., whether inverse() only returns non-negative values.

abstract forward(value: Union[int, float, ndarray, Tensor]) → Union[int, float, ndarray, Tensor][source]

Map a value from the space to the transformed space.

Parameters:: value – value in the original space
Returns:: value in the transformed space

abstract inverse(value: Union[int, float, ndarray, Tensor]) → Union[int, float, ndarray, Tensor][source]

Map a value from the transformed space to the actual space.

Parameters:: value – value in the transformed space
Returns:: value in the original space

class IdentityTransformation[source]

Bases: BijectiveTransformation

Transformation that does nothing.

ensures_non_negativity() → bool[source]: Whether this transformations ensures non-negativity, i.e., whether inverse() only returns non-negative values.

forward(value: Union[int, float, ndarray, Tensor]) → Union[int, float, ndarray, Tensor][source]

Map a value from the space to the transformed space.

Parameters:: value – value in the original space
Returns:: value in the transformed space

inverse(value: Union[int, float, ndarray, Tensor]) → Union[int, float, ndarray, Tensor][source]

Map a value from the transformed space to the actual space.

Parameters:: value – value in the transformed space
Returns:: value in the original space

class LogTransformation[source]

Bases: BijectiveTransformation

Transformation to make the values look like they are in log-space.

ensures_non_negativity() → bool[source]: Whether this transformations ensures non-negativity, i.e., whether inverse() only returns non-negative values.

forward(value: Union[int, float, ndarray, Tensor]) → Union[int, float, ndarray, Tensor][source]

Map a value from the space to the transformed space.

Parameters:: value – value in the original space
Returns:: value in the transformed space

inverse(value: Union[int, float, ndarray, Tensor]) → Union[int, float, ndarray, Tensor][source]

Map a value from the transformed space to the actual space.

Parameters:: value – value in the transformed space
Returns:: value in the original space

class SqrtTransformation[source]

Bases: BijectiveTransformation

Transformation to make the values look like they are in sqrt-space. This is not actually bijective!

ensures_non_negativity() → bool[source]: Whether this transformations ensures non-negativity, i.e., whether inverse() only returns non-negative values.

forward(value: Union[int, float, ndarray, Tensor]) → Union[int, float, ndarray, Tensor][source]

Map a value from the space to the transformed space.

Parameters:: value – value in the original space
Returns:: value in the transformed space

inverse(value: Union[int, float, ndarray, Tensor]) → Union[int, float, ndarray, Tensor][source]

Map a value from the transformed space to the actual space.

Parameters:: value – value in the transformed space
Returns:: value in the original space

checks

check_all_equal(iterable) → bool[source]

Check if all elements of an iterable are equal.

Parameters:: iterable – iterable to check
Returns:: bool saying if all elements are equal

check_all_lengths_equal(iterable) → bool[source]

Check if the length of all elements of an iterable are equal.

Parameters:: iterable – iterable to check
Returns:: bool saying if all elements are equal

check_all_shapes_equal(iterable) → bool[source]

Check if the shape of all elements of an iterable are equal.

Parameters:: iterable – iterable to check
Returns:: bool saying if all elements are equal

check_all_types_equal(iterable) → bool[source]

Check if all elements of an iterable are if the same type.

Parameters:: iterable – iterable to check
Returns:: bool saying if all elements are of equal type

is_iterable(obj) → bool[source]

Check if the input is iterable by trying to create an iterator from the input.

Parameters:: obj – any object
Returns:: True if input is iterable, else False

is_iterator(obj) → bool[source]

Check if the input is an iterator by trying to call next() on it.

Parameters:: obj – any object
Returns:: True if input is an iterator, else False

is_sequence(obj) → bool[source]

Check if the input is a sequence.

Note

Checking if obj is an instance of collections.Sequence will return the wrong result for objects that implement the sequence protocol but do not inherit from collections.Sequence

Parameters:: obj – any object
Returns:: True if input is iterable, else False

data_processing

Only tested for 1-dim inputs, e.g. time series of rewards.

class MinMaxScaler(bound_lo: Union[int, float, ndarray, Tensor], bound_up: Union[int, float, ndarray, Tensor])[source]

Bases: object

A stateful min-max scaler that remembers the lower and upper bound for later un-unscaling

Constructor

Parameters:

bound_lo – lower bound for the transformed data
bound_up – upper bound for the transformed data

scale_back(data: Union[ndarray, Tensor]) → Union[ndarray, Tensor][source]

Rescale the input data back to its original value range

Parameters:: data – input ndarray or Tensor scaled to be in $[a, b]$
Returns:: unscaled ndarray or Tensor

scale_to(data: Union[ndarray, Tensor]) → Union[ndarray, Tensor][source]

Transform the input data to be in $[a, b]$, where $a$ and $b$ are defined during construction.

Parameters:: data – unscaled input ndarray or Tensor
Returns:: ndarray or Tensor scaled to be in $[a, b]$

class RunningNormalizer[source]

Bases: object

Normalizes given data based on the history of observed data, such that all outputs are in range [-1, 1]

Constructor

reset()[source]: Reset internal variables.

class RunningStandardizer[source]

Bases: object

Implementation of Welford’s online algorithm

Constructor

reset()[source]: Reset internal variables.

class Standardizer[source]

Bases: object

A stateful standardizer that remembers the mean and standard deviation for later un-standardization

standardize(data: Union[ndarray, Tensor], eps: float = 1e-08) → Union[ndarray, Tensor][source]

Standardize the input data to make it $~ N(0, 1)$ and store the input’s mean and standard deviation.

Parameters:

data – input ndarray or Tensor
eps – factor for numerical stability

Returns:

standardized ndarray or Tensor

unstandardize(data: Union[ndarray, Tensor]) → Union[ndarray, Tensor][source]

Revert the previous standardization of the input data to make it $~ N(\mu, \sigma)$.

Parameters:: data – input ndarray or Tensor
Returns:: un-standardized ndarray or Tensor

correct_atleast_2d(x: Union[ndarray, Tensor]) → Union[ndarray, Tensor][source]

Compensate for numpy’s or PyTorch’s atleast_2d() in case the input was 1-dimensional.

Parameters:: x – input
Returns:: potentially transposed input

normalize(x: Union[ndarray, Tensor], axis: int = -1, order: int = 1, eps: float = 1e-08) → Union[ndarray, Tensor][source]

Normalize a numpy array or a PyTroch Tensor without changing the input. Choosing axis=1 and norm_order=1 makes all columns of sum to 1.

Parameters:

x – input to normalize
axis – axis of the array to normalize along
order – order of the norm (e.g., L1 norm: absolute values, L2 norm: quadratic values)
eps – lower bound on the norm, to avoid division by zero

Returns:

normalized array

scale_min_max(data: Union[ndarray, Tensor], bound_lo: Union[int, float, ndarray, Tensor], bound_up: Union[int, float, ndarray, Tensor]) → Union[ndarray, Tensor][source]

Transform the input data to to be in $[a, b]$.

Parameters:

data – unscaled input ndarray or Tensor
bound_lo – lower bound for the transformed data
bound_up – upper bound for the transformed data

Returns:

ndarray or Tensor scaled to be in $[a, b]$

standardize(data: Union[ndarray, Tensor], eps: float = 1e-08) → Union[ndarray, Tensor][source]

Standardize the input data to make it $~ N(0, 1)$.

Parameters:

data – input ndarray or Tensor
eps – factor for numerical stability

Returns:

standardized ndarray or Tensor

data_sets

class TimeSeriesDataSet(data: Tensor, window_size: int, ratio_train: float, standardize_data: bool = False, scale_min_max_data: bool = False, name: str = 'UnnamedDataSet')[source]

Bases: Dataset

Class for storing time series data sets

Constructor

Parameters:

data – complete raw data set, where the samples are along the first dimension
window_size – length of the sequences fed to the policy for predicting the next value
ratio_train – ratio of the training samples w.r.t. the total sample count
standardize_data – if True, the data is standardized to be $~ N(0,1)$
scale_min_max_data – if True, the data is scaled to $\in [-1, 1]$
name – descriptive name for the data set

static cut_to_window_size(data: Tensor, ws: int) → Tensor[source]

Cut off samples such that all training sequences have length window_size.

Parameters:

data – input data set with samples along the first dimension
ws – input window length as used for learning

Returns:

data of proper length

property dim_data: int: Get the data’s number of dimensions.

property num_samples_trn: int: Get the number of samples in the training subset.

property num_samples_tst: int: Get the number of samples in the testing subset.

property ratio_train: float: Get the ratio of the training samples w.r.t. the total sample count.

property window_size: int: Get the length of the sequences fed to the policy for predicting the next value.

create_sequences(data: Tensor, len_seq: int) → List[Tuple][source]

Create input sequences with associated target values from data. The targets are the next data point after an input sequence of given length.

Parameters:

data – data to be split into inputs and targets
len_seq – length of the input sequences

Returns:

list of tuples with inout and target data pairs

create_shuffled_sequences(data: Tensor, len_seq: int) → List[Tuple][source]

Create input sequences with associated target values from data using create_sequences(), and shuffle the order of the sequences afterwards.

Parameters:

data – data to be split into inputs and targets
len_seq – length of the input sequences

Returns:

list of randomly ordered tuples with inout and target data pairs

data_types

class DSSpec(function: str, goal: Union[ndarray, float, int], **kwargs)[source]

Bases: dict

The specification of a dynamical system implemented in RcsPySim and used in combination with ADN We are directly subclassing dict beacuse PropertySource in RcsPySim extracts information from dicts.

Constructor

Parameters:

function – name of the dynamics function, e.g. linear, msd, or msd_nlin
goal – desired state, e.g. a task space position or velocity
kwargs – additional attributes set by subclasses

class EnvSpec(obs_space: ~pyrado.spaces.base.Space, act_space: ~pyrado.spaces.base.Space, state_space: ~pyrado.spaces.base.Space = <pyrado.spaces.empty.EmptySpace object>)[source]

Bases: tuple

The specification of an environment’s input and output space

Create new instance of EnvSpec(obs_space, act_space, state_space)

property act_space: Alias for field number 1

property obs_space: Alias for field number 0

property state_space: Alias for field number 2

class LinDSSpec(function: str, goal: Union[ndarray, float, int], errorDynamics: Union[ndarray, float, int])[source]

Bases: DSSpec

The specification of a linear first order dynamical system implemented in RcsPySim. The dynamics are defined as $\dot{x} = A (x_{des} - x_{curr})$ with the error dynamics matrix $A$. Example: LinDSSpec(function=’lin’, goal=np.array([-1., 2.]), errorDynamics=np.eye(2))

Constructor

Parameters:

function – name of the dynamics function, e.g. msd, or msd_nlin
goal – desired state, e.g. a task space position or velocity
errorDynamics – a matrix determining the dynamics or the error difference between

class MSDDSSpec(function: str, goal: Union[ndarray, float, int], attractorStiffness: Union[float, int], damping: Union[float, int], mass: Union[float, int] = 1.0)[source]

Bases: DSSpec

The specification of a second order dynamical system i.e. a linear or nonlinear mass spring damper system implemented in RcsPySim. Example: MSDDSSpec(function=’msd_nlin’, goal=np.array([-1., 2.]), attractorStiffness=50., damping=10., mass=1.)

Constructor

Parameters:

function – name of the dynamics function, e.g. msd, or msd_nlin
goal – desired state, e.g. a task space position or velocity
attractorStiffness – spring stiffness parameter of the dynamical system
damping – damping parameter of the dynamical system
mass – mass parameter of the dynamical system

class RenderMode(text: bool = False, video: bool = False, render: bool = False)[source]

Bases: tuple

The specification of a render mode, do not print or render anything by default

Create new instance of RenderMode(text, video, render)

property render: Alias for field number 2

property text: Alias for field number 0

property video: Alias for field number 1

class TimeSeriesDataPair(inp_seq: Tensor, targ: Tensor)[source]

Bases: tuple

Pair of an input sequence and an associated target value for training time series prediction.

Create new instance of TimeSeriesDataPair(inp_seq, targ)

property inp_seq: Alias for field number 0

property targ: Alias for field number 1

dict_arraylike_to_float(d: dict)[source]

Convert every scalar array-like entry the provided dict to to a float. This function is useful when the metrics (e.g., calculated with numpy) should be saved in a YAML-file.

Parameters:: d – input dict with 1-element arrays
Return d:: output dict with float entries where conversion was possible

dict_path_access(d: dict, path: str, default: Optional[T] = None, default_for_last_layer_only: bool = False) → T[source]

Gets a dict entry using a jq-like naming system. To get the entry foo from the topmost dictionary, use the path foo. To access an entry baz in a dictionary that is stored in the top dictionary under the key bar, use the path bar.baz.

Parameters:

d – the dictionary
path – the path
default – if no entry corresponding to the key was found, this value is returned
default_for_last_layer_only – if True, only return the default value if the entry was not found in the last dictionary, but the previous dictionaries where found (e.g. for bar.baz, d[‘bar’] exists, but d[‘bar’][‘baz’] does not; if those where not found, an error is raised; if False, also return the default value of non-existent dictionaries along the path

Returns:

the entry, if found; if any part of the path except for the last does not exist, and default_for_last_layer_only is True, an error is raised; in all other cases, the default value is returned

fill_list_of_arrays(loa: Sequence[ndarray], des_len: int, fill_ele=nan) → Sequence[ndarray][source]

Fill a list of arrays with potential unequal length (first axis) will provided elements.

Parameters:

loa – list of ndarrays
des_len – desired length of the resulting arrays
fill_ele – element to fill the arrays with

Returns:

list of ndarrays with equal length

merge_dicts(dicts: Sequence[dict]) → dict[source]

Marge a list of dicts in to one dict. The resulting dict can have different value types, but the dicts later in the list override earlier ones.

Parameters:: dicts – input dicts
Returns:: merged dict

merge_dicts_same_dtype(dicts: ~typing.Sequence[dict], dtype=<class 'set'>) → dict[source]

Marge a list of dicts in to one dict. The resulting dict has only one value type for all entries, but the dicts do not override each other.

Parameters:

dicts – input dicts
dtype – data type for constructing the default dict

Returns:

merged dict

repeat_interleave(sequence: Union[list, tuple], num_reps) → Union[list, tuple][source]

Repeat every element of the input sequence, but keep the order of the elements.

Parameters:

sequence – input list or tuple
num_reps – number of repetitions

Returns:

list or tuple with repeated elements

update_matching_keys_recursively(base_dict: dict, update_dict: dict)[source]

Update a dictionary inplace with several levels of sub dictionaries with the matching entries of another dictionary.

Parameters:

base_dict – dictionary to update
update_dict – dictionary used for the update

Returns:

updated base_dict

exceptions

exception BaseErr[source]

Bases: Exception

Base class for exceptions in Pyrado

static retrieve_var_name(var, stack_level: Optional[int] = 2) → str[source]

Get the name of the given variable by searching through the locals and globals.

Parameters:

var – variable to look for
stack_level – number of frames to go up. 1 is the caller, 2 (default) is the caller’s caller.

Returns:

name of the first matching variable

exception KeyErr(*, keys: Optional[Union[str, Sequence[str]]] = None, container=None, msg: Optional[str] = None)[source]

Bases: BaseErr

Class for exceptions raised asking for a keys in an object that does not exist

Constructor

Parameters:

keys – key(s) that caused the error
container – object that should have the key
msg – offers possibility to override the error message

exception PathErr(*, given: str = '__INVALID__', msg: Optional[str] = None)[source]

Bases: BaseErr

Class for exceptions raised by passing wrong paths to folders or files

Constructor

Parameters:

given – input which caused the error
msg – offers possibility to override the error message

exception ShapeErr(*, given='__INVALID__', given_name: Optional[str] = None, expected_match=None, msg: Optional[str] = None)[source]

Bases: BaseErr

Class for exceptions raised by passing wrong values

Constructor

Parameters:

given – input which caused the error
expected_match – object which’s shape is the one the input should have
given_name – explicitly pass the name of the variable for the error message
msg – offers possibility to override the error message

static get_shape_and_name(obj, var)[source]

exception TypeErr(*, given='__INVALID__', given_name: Optional[str] = None, expected_type: Optional[Union[type, list, tuple]] = None, msg: Optional[str] = None)[source]

Bases: BaseErr

Class for exceptions raised by passing wrong data types

Constructor

Parameters:

given – input which caused the error
given_name – explicitly pass the name of the variable for the error message
expected_type – type or types that should have been passed
msg – offers possibility to override the error message

exception ValueErr(*, given='__INVALID__', given_name: Optional[str] = None, eq_constraint=None, l_constraint=None, le_constraint=None, g_constraint=None, ge_constraint=None, msg: Optional[str] = None)[source]

Bases: BaseErr

Class for exceptions raised by passing wrong values

Constructor

Parameters:

given – input which caused the error
given_name – explicitly pass the name of the variable for the error message
eq_constraint – violated equality constraint
l_constraint – violated less than constraint
le_constraint – violated less or equal than constraint
g_constraint – violated greater than constraint
ge_constraint – violated greater or equal than constraint
msg – offers possibility to override the error message

experiments

cpp_export(save_dir: PathLike, policy: Policy, env: Optional[SimEnv] = None, policy_export_name: str = 'policy_export', write_policy_node: bool = True, policy_node_name: str = 'policy')[source]

Convenience function to export the policy using PyTorch’s scripting or tracing, and the experiment’s XML configuration if the environment from RcsPySim.

Parameters:

save_dir – directory to save in
policy – (trained) policy
env – environment the policy was trained in
policy_export_name – name of the exported policy file without the file type ending
write_policy_node – if True, write the PyTorch-based control policy into the experiment’s XML configuration. This requires the experiment’s XML configuration to be exported beforehand.
policy_node_name – name of the control policies node in the XML file, e.g. ‘policy’ or ‘preStrikePolicy’

fcn_from_str(name: str) → Callable[source]

Get the matching function. This method is a workaround / utility tool to intended to work with optuna. Since we can not pass functions directly, we pass a sting.

Parameters:: name – name of the function
Returns:: the function

load_experiment(ex_dir: str, args: Optional[Any] = None) → Tuple[Optional[Union[SimEnv, EnvWrapper]], Optional[Policy], Optional[dict]][source]

Load the (training) environment and the policy. This helper function first tries to read the hyper-parameters yaml-file in the experiment’s directory to infer why entities should be loaded. If no file was found, we fall back to some heuristic and hope for the best.

Parameters:

ex_dir – experiment’s parent directory
args – arguments from the argument parser, pass None to fall back to the values from the default argparser

Returns:

environment, policy, and (optional) algorithm-specific output, e.g. value function

load_rollouts_from_dir(ex_dir: str, key: Optional[str] = 'rollout', file_exts: Tuple[str] = ('pt', 'pkl')) → Tuple[List[StepSequence], List[str]][source]

Crawl through the given directory, sort the files, and load all rollouts, i.e. all files that include the key.

Parameters:

ex_dir – directory, e.g. and experiment folder
key – word or part of a word that needs to the in the name of a file for it to be loaded
file_exts – file extensions to be considered for loading

Returns:

list of loaded rollouts, and list of file names without extension

read_csv_w_replace(path: str) → DataFrame[source]

Custom function to read a CSV file. Turns white paces into underscores for accessing the columns as exposed properties, i.e. df.prop_abc instead of df[‘prop abc’].

Parameters:: path – path to the CSV file
Returns:: Pandas DataFrame with replaced chars in columns

functions

noisy_nonlin_fcn(x: Union[Tensor, ndarray], f: float = 1.0, noise_std: float = 0.0) → Union[Tensor, ndarray][source]

A 1-dim function (sinus superposed with polynomial), representing the black box function in Bayesian optimization

Parameters:

x – function argument
noise_std – scale of the additive noise sampled from a standard normal distribution
f – frequency of the sinus wave [Hz]

Returns:

function value

rosenbrock(x: Union[Tensor, ndarray]) → Union[Tensor, ndarray][source]

The Rosenbrock function (consistent with https://docs.scipy.org/doc/scipy/reference/tutorial/optimize.html)

Parameters:: x – multi-dim column vector, or array thereof
Returns:: value of the Rosenbrock function at the input point, or array thereof

skyline(dt: Union[int, float, ndarray], t_end: Union[int, float, ndarray], t_intvl_space: BoxSpace, val_space: BoxSpace) → Tuple[ndarray, ndarray][source]

Step function that randomly samples a value from the given range, and then holds this value for a time interval which is also randomly sampled given a range of time intervals. This procedure is repeated until the sequence is long enough, i.e. dt * t_end samples.

Parameters:

dt – time step size
t_end – final time
t_intvl_space – 1-dim BoxSpace determining the range of time intervals that can be sampled
val_space – 1-dim BoxSpace determining the range of values that can be sampled

Returns:

array of time steps together with the associated array of values

input_output

color_validity(data: ndarray, valids: ndarray) → list[source]

Color the entries of an data array red or green depending on the if the entries are valid.

Parameters:

data – ndarray containing the data to print
valids – ndarray containing boolian integer values deciding gor the color (1 –> green, 0 –> red)

Returns:

list of stings

completion_context(msg: str, **kwargs)[source]

Context manager that prints a message, executes the code, and then prints a symbol on success or failure (when an exception is raised).

Parameters:

msg – message to print at the beginning, e.g. ‘Calibrating’
kwargs – keyword arguments forwarded to print_cbt()

ensure_math_mode(inp: [<class 'str'>, typing.Sequence[str]]) → [<class 'str'>, <class 'list'>][source]

Naive way to ensure that a sting is compatible with LaTeX math mode for printing.

Parameters:: inp – input string
Return s:: sting in math mode

input_timeout(prompt: str, t_timeout: [<class 'float'>, <class 'int'>] = 30, default: ~typing.Optional[str] = None) → str[source]

Ask the user for an input and quit with a runtime error if no input was given

Parameters:

prompt – text to be printed before recording the input
t_timeout – time limit to enter an input
default – default output that is returned if the timeout was raised, pass None for no default

Returns:

return the input as a string captured by the system

insert_newlines(string: str, every: int) → str[source]

Inserts multiple line breaks.

Parameters:

string – input string to be broken into multiple lines
every – gap in number of characters between every line break

Returns:

the input sting with line breaks

print_cbt(msg: str, color: str = '', bright: bool = False, tag: str = '', end='\n')[source]

Print a colored (and bright) message with a tag in the beginning.

Parameters:

msg – string to print
color – color to print in, default ‘’ is the IDE’s/system’s default
bright – flag if the message should be printed bright
tag – tag to be printed in brackets in front of the message
end – endline symbol forwarded to print()

print_cbt_once(msg: str, color: str = '', bright: bool = False, tag: str = '', end='\n')[source]: Wrapped version of print_cbt that only prints once.

select_query(items: ~typing.Sequence, max_display: int = 10, fallback: ~typing.Optional[~typing.Callable[[], ~pyrado.utils.input_output.T]] = None, item_formatter: ~typing.Callable[[~pyrado.utils.input_output.T], str] = <class 'str'>, header: str = 'Available options:', footer: str = 'Please enter the number of the option to use.') → T[source]

Ask the user to select an item out of the given list.

Parameters:

items – items for the query
max_display – maximum number of items
fallback – callable to select select an experiment from a hint if the input is not a number
item_formatter – callable to select the parts of an Experiment instance which should be printed, by default str turns the complete Experiment instance into a string.
header – string to print above the query
footer – string to print above the query

Returns:

interactive query for selecting an experiment

math

class UnitCubeProjector(bound_lo: Union[ndarray, Tensor], bound_up: Union[ndarray, Tensor])[source]

Bases: object

Project to a unit qube $[0, 1]^d$ and back using explicit bounds

Constructor

Parameters:

bound_lo – lower bound
bound_up – upper bound

project_back(data: Union[ndarray, Tensor]) → Union[ndarray, Tensor][source]

Revert the previous normalization using the stored explicit bounds

Parameters:: data – input from the uni space
Returns:: element of the original space

project_to(data: Union[ndarray, Tensor]) → Union[ndarray, Tensor][source]

Normalize every dimension individually using the stored explicit bounds and the L_1 norm.

Parameters:: data – input to project to the unit space
Returns:: element of the unit cube

clamp_symm(inp: Tensor, up_lo: Tensor) → Tensor[source]

Symmetrically clip the entries of a tensor by the entries of a tensor with only positive entries. One use case of this function is if you want to clip the parameter update by a ratio of the old parameters.

Parameters:

inp – input tensor
up_lo – upper and (negative of the) lower bound for each entry

Returns:

clipped tensor

cosine_similarity(x: Tensor, y: Tensor) → Tensor[source]

Compute the cosine similarity between two tensors $D_text{cos}(x,y) = frac{x^T y}{|x| cdot |y|}$.

Parameters:

x – input tensor
y – input tensor

Returns:

cosine similarity value

cov(x: Tensor, data_along_rows: bool = False)[source]

Compute the covariance matrix given data.

Note

Only real valued matrices are supported

Parameters:

x – matrix containing multiple observations of multiple variables
data_along_rows – if True the variables are stacked along the columns, else they are along the rows

Returns:

covariance matrix given the data

explained_var(y_mdl: ~typing.Union[~numpy.ndarray, ~torch.Tensor], y_obs: ~typing.Union[~numpy.ndarray, ~torch.Tensor]) -> (<class 'numpy.ndarray'>, <class 'torch.Tensor'>)[source]

Calculate proportion of the variance “explained” by the model (see coefficient of determination R^2) .. note:: R^2 = 0.49 implies that 49% of the variability of the dependent variable has been accounted for.

Parameters:

y_mdl – predictions from the model
y_obs – observed data (ground truth)

Returns:

proportion of the explained variance (i.e. R^2 value)

logmeanexp(x: Union[Tensor, ndarray], dim: int = 0) → Union[Tensor, ndarray][source]

Numerically stable way to compute $\log \left( \frac{1}{N} \sum_{i=1}^N \exp(x) \right)$

Parameters:

x – input tensor
dim – dimension to compute the logmeanexp` along

Returns:

natural logarithm of the mean of the exponentiated values

numerical_differentiation_coeffs(stencils: Union[list, ndarray], order: int, step_size: Union[int, float] = 1) → Tuple[ndarray, int][source]

Compute the coefficients for discrete time numerical differentiation. These can later be convolved with the signal.

Parameters:

stencils – stencil points $s$, negative means in the past, e.g. [-1, 0, 1] for a central difference or [-3, -1, 0] for a backward difference without equally space stencil points
order – order of the derivative $d$, must be lower than the number of stencil points
step_size – step size $h$

Returns:

coefficients for computing the derivative of order order and the $O(h^{N-d})$

rmse(x: Union[ndarray, Tensor], y: Union[ndarray, Tensor], dim: int = 0) → Union[ndarray, Tensor][source]

Compute the root mean squared error between two inputs.

Parameters:

x – input data, e.g. output of a neural network
y – input data, e.g. target values for the output
dim – dimension to compute the RMSE along

Returns:

soft_update(target: Tensor, source: Tensor, tau: float = 0.995)[source]

Moving average update, a.k.a. Polyak update.

Parameters:

target – PyTroch tensor with the next / target values
source – PyTroch module with the current / source values
tau – interpolation factor for averaging, between 0 and 1

soft_update_(target: Module, source: Module, tau: float = 0.995)[source]

Moving average update, a.k.a. Polyak update. Modifies the input argument target.

Parameters:

target – PyTroch module with parameters to be updated
source – PyTroch module with parameters to update to
tau – interpolation factor for averaging, between 0 and 1

nn_layers

class IndiNonlinLayer(in_features: int, nonlin: [Callable, Sequence[Callable]], bias: bool, weight: bool = True)[source]

Bases: Module

Layer add a bias to the input, multiplies the result with a scaling factor, and then applies the provided nonlinearity. If a list of nonlinearities is provided, every dimension will be processed separately. The scaling and the bias are learnable parameters.

Constructor

Parameters:

in_features – size of each input sample
nonlin – nonlinearity
bias – if True, a learnable bias is subtracted, else no bias is used
weight – if True (default), the input is multiplied with a learnable scaling factor

extra_repr() → str[source]

Set the extra representation of the module

To print customized extra information, you should re-implement this method in your own modules. Both single-line and multi-line strings are acceptable.

forward(inp: Tensor) → Tensor[source]

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 MirrConv1d(in_channels, out_channels, kernel_size, stride=1, padding=0, dilation=1, groups=1, bias=False, padding_mode='zeros')[source]

Bases: _ConvNd

Overriding Conv1d module implementation from PyTorch 1.4 to re-use parts of the convolution weights by mirroring the first half of the kernel (along the columns). This way we can save (close to) half of the parameters, under the assumption that we have a kernel that obeys this kind of symmetry. The biases are left unchanged.

Initializes internal Module state, shared by both nn.Module and ScriptModule.

bias: Optional[Tensor]

dilation: Tuple[int, ...]

forward(inp: Tensor) → Tensor[source]

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.

groups: int

kernel_size: Tuple[int, ...]

out_channels: int

output_padding: Tuple[int, ...]

padding: Tuple[int, ...]

padding_mode: str

stride: Tuple[int, ...]

transposed: bool

weight: Tensor

class PositiveScaleLayer(in_features: int, init_weight: float = 1.0)[source]

Bases: Module

Layer which scales (strictly positive) the input using a learnable scaling factor

Constructor

Parameters:

in_features – size of each input sample
init_weight – initial scaling factor

extra_repr() → str[source]

Set the extra representation of the module

To print customized extra information, you should re-implement this method in your own modules. Both single-line and multi-line strings are acceptable.

forward(inp: Tensor) → Tensor[source]

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 ScaleLayer(in_features: int, init_weight: float = 1.0)[source]

Bases: Module

Layer which scales the output of the input using a learnable scaling factor

Constructor

Parameters:

in_features – size of each input sample
init_weight – initial scaling factor

extra_repr() → str[source]

Set the extra representation of the module

To print customized extra information, you should re-implement this method in your own modules. Both single-line and multi-line strings are acceptable.

forward(inp: Tensor) → Tensor[source]

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

optimizers

class GSS(params, param_min: Tensor, param_max: Tensor)[source]

Bases: Optimizer

Golden Section Search optimizer

reset()[source]

step(closure: Callable)[source]

Performs a single optimization step.

Parameters:: closure – a closure that reevaluates the model and returns the loss
Returns:: accumulated loss for all parameter groups after the parameter update

ordering

filter_los_by_lok(strs: list, keys: list) → list[source]

Filter a list of strings by a list of keys strings.

Parameters:

strs – list of strings to filter
keys – keys-strings to filter by

Returns:

list with unique elements of the input list which contain at least one of the keys

get_immediate_subdirs(parent_dir: str)[source]

Get all 1st level subdirectories of a specified directory (i.e. a path).

Parameters:: parent_dir – directory in which to look for subdirectories
Returns:: list of names of all 1st level subdirectories

natural_sort(lst: list)[source]

Sort a list like a human would do. Normal list sort does 1, 10, 11, 2, 3. But this function yields 1, 2, 3, 10, 11.

Parameters:: lst – input list
Returns:: alpha-numerically sorted list

remove_none_from_list(lst: list)[source]

Create a new list from the given one without None entries.

Parameters:: lst – list to remove the None entries from
Returns:: list without None entries from

properties

class cached_property(func)[source]

Bases: object

Decorator that turns a function into a cached property. When the property is first accessed, the function is called to compute the value. Later calls use the cached value.

Note

Requires a __dict__ field, so it won’t work on named tuples.

saving_loading

load(name: str, load_dir: PathLike, prefix: str = '', suffix: str = '', obj=None, verbose: bool = False)[source]

Load an object object using a prefix or suffix, depending on the meta information.

Parameters:

name – name of the object for loading including the file extension, e.g. ‘policy.pt’ for PyTorch modules like a Pyrado Policy instance
load_dir – directory to load from
prefix – prefix for altering the name, e.g. ‘iter_0_…’
suffix – suffix for altering the name, e.g. ‘…_ref’
obj – PyTorch module to load into, this can be None except for the case if you want to load and save the module’s state_dict
verbose – if True, print the path of what has been loaded

save(obj, name: str, save_dir: PathLike, prefix: str = '', suffix: str = '', use_state_dict: bool = False, verbose: bool = False)[source]

Save an object object using a prefix or suffix, depending on the meta information.

Parameters:

obj – PyTorch or pickled object to save
name – name of the object for loading including the file extension, e.g. ‘policy.pt’ for PyTorch modules like a Pyrado Policy instance
save_dir – directory to save in
prefix – prefix for altering the name, e.g. ‘iter_0_…’
suffix – suffix for altering the name, e.g. ‘…_ref’
use_state_dict – if True save the state_dict, else save the entire module. This only has an effect if PyTorch modules (file_ext = ‘pt’) are saved.
verbose – if True, print the path where the object has been saved to

sbi

create_embedding(name: str, env_spec: EnvSpec, *args, **kwargs) → Embedding[source]

Create an embedding to use with sbi.

Parameters:

name – identifier of the embedding
env_spec – environment specification
args – positional arguments forwarded to the embedding’s constructor
kwargs – keyword arguments forwarded to the embedding’s constructor

Returns:

embedding instance

tensor

deepcopy_or_clone(copy_from: CopyType) → CopyType[source]

Unfortunately, deepcopy() only works for leave tensors right now. Thus, we need to iterate throught the input and clone the PyTorch tensors where possible.

Parameters:: copy_from – list or dict to copy
Returns:: copy of the input where every tensor was cloned, and every other data type was deepcopied

insert_tensor_col(x: Tensor, idx: int, col: Tensor) → Tensor[source]

Insert a column into a PyTorch tensor`.

Parameters:

x – original tensor
idx – column index where to insert the column
col – tensor to insert

Returns:

tensor with new column at index idx

stack_tensor_dict_list(tensor_dict_list: list) → dict[source]

Stack a list of dictionaries of {tensors or dict of tensors}.

Parameters:: tensor_dict_list – a list of dicts of {tensors or dict of tensors}.
Returns:: a dict of {stacked tensors or dict of stacked tensors}

stack_tensor_list(tensor_list: list) → Tensor[source]

Convenience function for stacking a list of tensors

Parameters:: tensor_list – list of tensors to stack (along a new dim)
Returns:: tensor of at least 1-dim

Module contents

get_class_name(obj) → str[source]

Get an arbitrary objects name.

Parameters:: obj – any object
Returns:: name of the class of the given object

run_once(fcn)[source]

Wrapper to ensure that a function is only called once.

Parameters:: fcn – function that should only be called once
Returns:: wrapped function