Index
Utility classes and functions for neural networks
layers
¶
Various neural network layer types
Apply (Module)
¶
A torch module for applying an arithmetic operator on an input tensor
Source code in evotorch/neuroevolution/net/layers.py
class Apply(nn.Module):
"""A torch module for applying an arithmetic operator on an input tensor"""
def __init__(self, operator: str, argument: float):
"""`__init__(...)`: Initialize the Apply module.
Args:
operator: Must be '+', '-', '*', '/', or '**'.
Indicates which operation will be done
on the input tensor.
argument: Expected as a float, represents
the right-argument of the operation
(the left-argument being the input
tensor).
"""
nn.Module.__init__(self)
self._operator = str(operator)
assert self._operator in ("+", "-", "*", "/", "**")
self._argument = float(argument)
def forward(self, x):
op = self._operator
arg = self._argument
if op == "+":
return x + arg
elif op == "-":
return x - arg
elif op == "*":
return x * arg
elif op == "/":
return x / arg
elif op == "**":
return x**arg
else:
raise ValueError("Unknown operator:" + repr(op))
def extra_repr(self):
return "operator={}, argument={}".format(repr(self._operator), self._argument)
__init__(self, operator, argument)
special
¶
__init__(...): Initialize the Apply module.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
operator |
str |
Must be '+', '-', '', '/', or '*'. Indicates which operation will be done on the input tensor. |
required |
argument |
float |
Expected as a float, represents the right-argument of the operation (the left-argument being the input tensor). |
required |
Source code in evotorch/neuroevolution/net/layers.py
def __init__(self, operator: str, argument: float):
"""`__init__(...)`: Initialize the Apply module.
Args:
operator: Must be '+', '-', '*', '/', or '**'.
Indicates which operation will be done
on the input tensor.
argument: Expected as a float, represents
the right-argument of the operation
(the left-argument being the input
tensor).
"""
nn.Module.__init__(self)
self._operator = str(operator)
assert self._operator in ("+", "-", "*", "/", "**")
self._argument = float(argument)
extra_repr(self)
¶
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(self, x)
¶
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 :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
Source code in evotorch/neuroevolution/net/layers.py
Bin (Module)
¶
A small torch module for binning the values of tensors.
In more details, considering a lower bound value lb, an upper bound value ub, and an input tensor x, each value within x closer to lb will be converted to lb and each value within x closer to ub will be converted to ub.
Source code in evotorch/neuroevolution/net/layers.py
class Bin(nn.Module):
"""A small torch module for binning the values of tensors.
In more details, considering a lower bound value lb,
an upper bound value ub, and an input tensor x,
each value within x closer to lb will be converted to lb
and each value within x closer to ub will be converted to ub.
"""
def __init__(self, lb: float, ub: float):
"""`__init__(...)`: Initialize the Clip operator.
Args:
lb: Lower bound
ub: Upper bound
"""
nn.Module.__init__(self)
self._lb = float(lb)
self._ub = float(ub)
self._interval_size = self._ub - self._lb
self._shrink_amount = self._interval_size / 2.0
self._shift_amount = (self._ub + self._lb) / 2.0
def forward(self, x: torch.Tensor):
x = x - self._shift_amount
x = x / self._shrink_amount
x = torch.sign(x)
x = x * self._shrink_amount
x = x + self._shift_amount
return x
def extra_repr(self):
return "lb={}, ub={}".format(self._lb, self._ub)
__init__(self, lb, ub)
special
¶
__init__(...): Initialize the Clip operator.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
lb |
float |
Lower bound |
required |
ub |
float |
Upper bound |
required |
Source code in evotorch/neuroevolution/net/layers.py
def __init__(self, lb: float, ub: float):
"""`__init__(...)`: Initialize the Clip operator.
Args:
lb: Lower bound
ub: Upper bound
"""
nn.Module.__init__(self)
self._lb = float(lb)
self._ub = float(ub)
self._interval_size = self._ub - self._lb
self._shrink_amount = self._interval_size / 2.0
self._shift_amount = (self._ub + self._lb) / 2.0
extra_repr(self)
¶
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(self, x)
¶
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 :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
Clip (Module)
¶
A small torch module for clipping the values of tensors
Source code in evotorch/neuroevolution/net/layers.py
class Clip(nn.Module):
"""A small torch module for clipping the values of tensors"""
def __init__(self, lb: float, ub: float):
"""`__init__(...)`: Initialize the Clip operator.
Args:
lb: Lower bound. Values less than this will be clipped.
ub: Upper bound. Values greater than this will be clipped.
"""
nn.Module.__init__(self)
self._lb = float(lb)
self._ub = float(ub)
def forward(self, x: torch.Tensor):
return x.clamp(self._lb, self._ub)
def extra_repr(self):
return "lb={}, ub={}".format(self._lb, self._ub)
__init__(self, lb, ub)
special
¶
__init__(...): Initialize the Clip operator.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
lb |
float |
Lower bound. Values less than this will be clipped. |
required |
ub |
float |
Upper bound. Values greater than this will be clipped. |
required |
Source code in evotorch/neuroevolution/net/layers.py
extra_repr(self)
¶
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(self, x)
¶
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 :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
FeedForwardNet (Module)
¶
Representation of a feed forward neural network as a torch Module.
An example initialization of a FeedForwardNet is as follows:
net = drt.FeedForwardNet(4, [(8, 'tanh'), (6, 'tanh')])
which means that we would like to have a network which expects an input vector of length 4 and passes its input through 2 tanh-activated hidden layers (with neurons count 8 and 6, respectively). The output of the last hidden layer (of length 6) is the final output vector.
The string representation of the module obtained via the example above is:
FeedForwardNet(
(layer_0): Linear(in_features=4, out_features=8, bias=True)
(actfunc_0): Tanh()
(layer_1): Linear(in_features=8, out_features=6, bias=True)
(actfunc_1): Tanh()
)
Source code in evotorch/neuroevolution/net/layers.py
class FeedForwardNet(nn.Module):
"""
Representation of a feed forward neural network as a torch Module.
An example initialization of a FeedForwardNet is as follows:
net = drt.FeedForwardNet(4, [(8, 'tanh'), (6, 'tanh')])
which means that we would like to have a network which expects an input
vector of length 4 and passes its input through 2 tanh-activated hidden
layers (with neurons count 8 and 6, respectively).
The output of the last hidden layer (of length 6) is the final
output vector.
The string representation of the module obtained via the example above
is:
FeedForwardNet(
(layer_0): Linear(in_features=4, out_features=8, bias=True)
(actfunc_0): Tanh()
(layer_1): Linear(in_features=8, out_features=6, bias=True)
(actfunc_1): Tanh()
)
"""
LengthActTuple = Tuple[int, Union[str, Callable]]
LengthActBiasTuple = Tuple[int, Union[str, Callable], Union[bool]]
def __init__(self, input_size: int, layers: List[Union[LengthActTuple, LengthActBiasTuple]]):
"""`__init__(...)`: Initialize the FeedForward network.
Args:
input_size: Input size of the network, expected as an int.
layers: Expected as a list of tuples,
where each tuple is either of the form
`(layer_size, activation_function)`
or of the form
`(layer_size, activation_function, bias)`
in which
(i) `layer_size` is an int, specifying the number of neurons;
(ii) `activation_function` is None, or a callable object,
or a string containing the name of the activation function
('relu', 'selu', 'elu', 'tanh', 'hardtanh', or 'sigmoid');
(iii) `bias` is a boolean, specifying whether the layer
is to have a bias or not.
When omitted, bias is set to True.
"""
nn.Module.__init__(self)
for i, layer in enumerate(layers):
if len(layer) == 2:
size, actfunc = layer
bias = True
elif len(layer) == 3:
size, actfunc, bias = layer
else:
assert False, "A layer tuple of invalid size is encountered"
setattr(self, "layer_" + str(i), nn.Linear(input_size, size, bias=bias))
if isinstance(actfunc, str):
if actfunc == "relu":
actfunc = nn.ReLU()
elif actfunc == "selu":
actfunc = nn.SELU()
elif actfunc == "elu":
actfunc = nn.ELU()
elif actfunc == "tanh":
actfunc = nn.Tanh()
elif actfunc == "hardtanh":
actfunc = nn.Hardtanh()
elif actfunc == "sigmoid":
actfunc = nn.Sigmoid()
elif actfunc == "round":
actfunc = Round()
else:
raise ValueError("Unknown activation function: " + repr(actfunc))
setattr(self, "actfunc_" + str(i), actfunc)
input_size = size
def forward(self, x):
i = 0
while hasattr(self, "layer_" + str(i)):
x = getattr(self, "layer_" + str(i))(x)
f = getattr(self, "actfunc_" + str(i))
if f is not None:
x = f(x)
i += 1
return x
__init__(self, input_size, layers)
special
¶
__init__(...): Initialize the FeedForward network.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input_size |
int |
Input size of the network, expected as an int. |
required |
layers |
List[Union[Tuple[int, Union[str, Callable]], Tuple[int, Union[str, Callable], bool]]] |
Expected as a list of tuples,
where each tuple is either of the form
|
required |
Source code in evotorch/neuroevolution/net/layers.py
def __init__(self, input_size: int, layers: List[Union[LengthActTuple, LengthActBiasTuple]]):
"""`__init__(...)`: Initialize the FeedForward network.
Args:
input_size: Input size of the network, expected as an int.
layers: Expected as a list of tuples,
where each tuple is either of the form
`(layer_size, activation_function)`
or of the form
`(layer_size, activation_function, bias)`
in which
(i) `layer_size` is an int, specifying the number of neurons;
(ii) `activation_function` is None, or a callable object,
or a string containing the name of the activation function
('relu', 'selu', 'elu', 'tanh', 'hardtanh', or 'sigmoid');
(iii) `bias` is a boolean, specifying whether the layer
is to have a bias or not.
When omitted, bias is set to True.
"""
nn.Module.__init__(self)
for i, layer in enumerate(layers):
if len(layer) == 2:
size, actfunc = layer
bias = True
elif len(layer) == 3:
size, actfunc, bias = layer
else:
assert False, "A layer tuple of invalid size is encountered"
setattr(self, "layer_" + str(i), nn.Linear(input_size, size, bias=bias))
if isinstance(actfunc, str):
if actfunc == "relu":
actfunc = nn.ReLU()
elif actfunc == "selu":
actfunc = nn.SELU()
elif actfunc == "elu":
actfunc = nn.ELU()
elif actfunc == "tanh":
actfunc = nn.Tanh()
elif actfunc == "hardtanh":
actfunc = nn.Hardtanh()
elif actfunc == "sigmoid":
actfunc = nn.Sigmoid()
elif actfunc == "round":
actfunc = Round()
else:
raise ValueError("Unknown activation function: " + repr(actfunc))
setattr(self, "actfunc_" + str(i), actfunc)
input_size = size
forward(self, x)
¶
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 :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
LSTMNet (StatefulModule)
¶
Representation of an LSTM layer.
Differently from torch.nn.LSTM, the forward pass function of this class does NOT expect the hidden state, nor does it return the resulting hidden state of the pass. Instead, the hidden states are stored within the module itself.
The forward pass function can take a 1-dimensional tensor of length
input_size, or it can take a 2-dimensional tensor of size
(batch_size, input_size).
Because the instances of this class are stateful, remember to reset() the internal state when needed.
Source code in evotorch/neuroevolution/net/layers.py
class LSTMNet(StatefulModule):
"""Representation of an LSTM layer.
Differently from torch.nn.LSTM, the forward pass function of this class
does NOT expect the hidden state, nor does it return
the resulting hidden state of the pass.
Instead, the hidden states are stored within the module itself.
The forward pass function can take a 1-dimensional tensor of length
input_size, or it can take a 2-dimensional tensor of size
`(batch_size, input_size)`.
Because the instances of this class are stateful,
remember to reset() the internal state when needed.
"""
def __init__(self, **kwargs):
"""
`__init__(...)`: Initialize the LSTM net.
Args:
input_size: The input size, expected as an int.
hidden_size: Number of neurons, expected as an int.
num_layers: Number of layers of the recurrent net.
"""
StatefulModule.__init__(self, nn.LSTM, **kwargs)
__init__(self, **kwargs)
special
¶
__init__(...): Initialize the LSTM net.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input_size |
The input size, expected as an int. |
required | |
hidden_size |
Number of neurons, expected as an int. |
required | |
num_layers |
Number of layers of the recurrent net. |
required |
Source code in evotorch/neuroevolution/net/layers.py
LocomotorNet (Module)
¶
This is a control network which consists of two components: one linear, and one non-linear. The non-linear component is an input-independent set of sinusoidals waves whose amplitudes, frequencies and phases are trainable. Upon execution of a forward pass, the output of the non-linear component is the sum of all these sinusoidal waves. The linear component is a linear layer (optionally with bias) whose weights (and biases) are trainable. The final output of the LocomotorNet at the end of a forward pass is the sum of the linear and the non-linear components.
Note that this is a stateful network, where the only state
is the timestep t, which starts from 0 and gets incremented by 1
at the end of each forward pass. The reset() method resets
t back to 0.
Reference
Mario Srouji, Jian Zhang, Ruslan Salakhutdinov (2018). Structured Control Nets for Deep Reinforcement Learning.
Source code in evotorch/neuroevolution/net/layers.py
class LocomotorNet(nn.Module):
"""LocomotorNet: A locomotion-specific structured control net.
This is a control network which consists of two components:
one linear, and one non-linear. The non-linear component
is an input-independent set of sinusoidals waves whose
amplitudes, frequencies and phases are trainable.
Upon execution of a forward pass, the output of the non-linear
component is the sum of all these sinusoidal waves.
The linear component is a linear layer (optionally with bias)
whose weights (and biases) are trainable.
The final output of the LocomotorNet at the end of a forward pass
is the sum of the linear and the non-linear components.
Note that this is a stateful network, where the only state
is the timestep t, which starts from 0 and gets incremented by 1
at the end of each forward pass. The `reset()` method resets
t back to 0.
Reference:
Mario Srouji, Jian Zhang, Ruslan Salakhutdinov (2018).
Structured Control Nets for Deep Reinforcement Learning.
"""
def __init__(self, *, in_features: int, out_features: int, bias: bool = True, num_sinusoids=16):
"""`__init__(...)`: Initialize the LocomotorNet.
Args:
in_features: Length of the input vector
out_features: Length of the output vector
bias: Whether or not the linear component is to have a bias
num_sinusoids: Number of sinusoidal waves
"""
nn.Module.__init__(self)
self._in_features = in_features
self._out_features = out_features
self._bias = bias
self._num_sinusoids = num_sinusoids
self._linear_component = nn.Linear(
in_features=self._in_features, out_features=self._out_features, bias=self._bias
)
self._amplitudes = nn.ParameterList()
self._frequencies = nn.ParameterList()
self._phases = nn.ParameterList()
for _ in range(self._num_sinusoids):
for paramlist in (self._amplitudes, self._frequencies, self._phases):
paramlist.append(nn.Parameter(torch.randn(self._out_features, dtype=torch.float32)))
self.reset()
def reset(self):
"""Set the timestep t to 0"""
self._t = 0
@property
def t(self) -> int:
"""The current timestep t"""
return self._t
@property
def in_features(self) -> int:
"""Get the length of the input vector"""
return self._in_features
@property
def out_features(self) -> int:
"""Get the length of the output vector"""
return self._out_features
@property
def num_sinusoids(self) -> int:
"""Get the number of sinusoidal waves of the non-linear component"""
return self._num_sinusoids
@property
def bias(self) -> bool:
"""Get whether or not the linear component has bias"""
return self._bias
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""Execute a forward pass"""
u_linear = self._linear_component(x)
t = self._t
u_nonlinear = torch.zeros(self._out_features)
for i in range(self._num_sinusoids):
A = self._amplitudes[i]
w = self._frequencies[i]
phi = self._phases[i]
u_nonlinear = u_nonlinear + (A * torch.sin(w * t + phi))
self._t += 1
return u_linear + u_nonlinear
bias: bool
property
readonly
¶
Get whether or not the linear component has bias
in_features: int
property
readonly
¶
Get the length of the input vector
num_sinusoids: int
property
readonly
¶
Get the number of sinusoidal waves of the non-linear component
out_features: int
property
readonly
¶
Get the length of the output vector
t: int
property
readonly
¶
The current timestep t
__init__(self, *, in_features, out_features, bias=True, num_sinusoids=16)
special
¶
__init__(...): Initialize the LocomotorNet.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
in_features |
int |
Length of the input vector |
required |
out_features |
int |
Length of the output vector |
required |
bias |
bool |
Whether or not the linear component is to have a bias |
True |
num_sinusoids |
Number of sinusoidal waves |
16 |
Source code in evotorch/neuroevolution/net/layers.py
def __init__(self, *, in_features: int, out_features: int, bias: bool = True, num_sinusoids=16):
"""`__init__(...)`: Initialize the LocomotorNet.
Args:
in_features: Length of the input vector
out_features: Length of the output vector
bias: Whether or not the linear component is to have a bias
num_sinusoids: Number of sinusoidal waves
"""
nn.Module.__init__(self)
self._in_features = in_features
self._out_features = out_features
self._bias = bias
self._num_sinusoids = num_sinusoids
self._linear_component = nn.Linear(
in_features=self._in_features, out_features=self._out_features, bias=self._bias
)
self._amplitudes = nn.ParameterList()
self._frequencies = nn.ParameterList()
self._phases = nn.ParameterList()
for _ in range(self._num_sinusoids):
for paramlist in (self._amplitudes, self._frequencies, self._phases):
paramlist.append(nn.Parameter(torch.randn(self._out_features, dtype=torch.float32)))
self.reset()
forward(self, x)
¶
Execute a forward pass
Source code in evotorch/neuroevolution/net/layers.py
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""Execute a forward pass"""
u_linear = self._linear_component(x)
t = self._t
u_nonlinear = torch.zeros(self._out_features)
for i in range(self._num_sinusoids):
A = self._amplitudes[i]
w = self._frequencies[i]
phi = self._phases[i]
u_nonlinear = u_nonlinear + (A * torch.sin(w * t + phi))
self._t += 1
return u_linear + u_nonlinear
reset(self)
¶
RecurrentNet (StatefulModule)
¶
Representation of a fully connected recurrent net as a torch Module.
Differently from torch.nn.RNN, the forward pass function of this class does NOT expect the hidden state, nor does it return the resulting hidden state of the pass. Instead, the hidden states are stored within the module itself.
The forward pass function can take a 1-dimensional tensor of length input_size, or it can take a 2-dimensional tensor of size (batch_size, input_size).
Because the instances of this class are stateful, remember to reset() the internal state when needed.
Source code in evotorch/neuroevolution/net/layers.py
class RecurrentNet(StatefulModule):
"""Representation of a fully connected recurrent net as a torch Module.
Differently from torch.nn.RNN, the forward pass function of this class
does NOT expect the hidden state, nor does it return
the resulting hidden state of the pass.
Instead, the hidden states are stored within the module itself.
The forward pass function can take a 1-dimensional tensor of length
input_size, or it can take a 2-dimensional tensor of size
(batch_size, input_size).
Because the instances of this class are stateful,
remember to reset() the internal state when needed.
"""
def __init__(self, **kwargs):
"""
`__init__(...)`: Initialize the recurrent net.
Args:
input_size: The input size, expected as an int.
hidden_size: Number of neurons, expected as an int.
nonlinearity: The activation function,
expected as 'tanh' or 'relu'.
num_layers: Number of layers of the recurrent net.
"""
StatefulModule.__init__(self, nn.RNN, **kwargs)
__init__(self, **kwargs)
special
¶
__init__(...): Initialize the recurrent net.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input_size |
The input size, expected as an int. |
required | |
hidden_size |
Number of neurons, expected as an int. |
required | |
nonlinearity |
The activation function, expected as 'tanh' or 'relu'. |
required | |
num_layers |
Number of layers of the recurrent net. |
required |
Source code in evotorch/neuroevolution/net/layers.py
def __init__(self, **kwargs):
"""
`__init__(...)`: Initialize the recurrent net.
Args:
input_size: The input size, expected as an int.
hidden_size: Number of neurons, expected as an int.
nonlinearity: The activation function,
expected as 'tanh' or 'relu'.
num_layers: Number of layers of the recurrent net.
"""
StatefulModule.__init__(self, nn.RNN, **kwargs)
Round (Module)
¶
A small torch module for rounding the values of an input tensor
Source code in evotorch/neuroevolution/net/layers.py
class Round(nn.Module):
"""A small torch module for rounding the values of an input tensor"""
def __init__(self, ndigits: int = 0):
nn.Module.__init__(self)
self._ndigits = int(ndigits)
self._q = 10.0**self._ndigits
def forward(self, x):
x = x * self._q
x = torch.round(x)
x = x / self._q
return x
def extra_repr(self):
return "ndigits=" + str(self._ndigits)
extra_repr(self)
¶
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(self, x)
¶
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 :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
Slice (Module)
¶
A small torch module for getting the slice of an input tensor
Source code in evotorch/neuroevolution/net/layers.py
class Slice(nn.Module):
"""A small torch module for getting the slice of an input tensor"""
def __init__(self, from_index: int, to_index: int):
"""`__init__(...)`: Initialize the Slice operator.
Args:
from_index: The index from which the slice begins.
to_index: The exclusive index at which the slice ends.
"""
nn.Module.__init__(self)
self._from_index = from_index
self._to_index = to_index
def forward(self, x):
return x[self._from_index : self._to_index]
def extra_repr(self):
return "from_index={}, to_index={}".format(self._from_index, self._to_index)
__init__(self, from_index, to_index)
special
¶
__init__(...): Initialize the Slice operator.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
from_index |
int |
The index from which the slice begins. |
required |
to_index |
int |
The exclusive index at which the slice ends. |
required |
Source code in evotorch/neuroevolution/net/layers.py
extra_repr(self)
¶
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(self, x)
¶
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 :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
StatefulModule (Module)
¶
Base class for stateful modules. Not to be instantiated directly.
Source code in evotorch/neuroevolution/net/layers.py
class StatefulModule(nn.Module):
"""Base class for stateful modules.
Not to be instantiated directly.
"""
def __init__(self, module_class, **kwargs):
nn.Module.__init__(self)
assert "batch_first" not in kwargs, "The `batch_first` option is not supported"
self._layer = module_class(**kwargs)
self.reset()
@property
def state(self):
"""Get the tensor of the internal state.
If the recurrent network is just initialized or reset,
then there is no state, so, a None is given.
Not having a state means that an initial internal state tensor of
compatible size with the input will be created at the
first usage of this network.
Each element of this initial internal state tensor is 0.
"""
return self._state
def reset(self):
"""Reset the internal state"""
self._state = None
def forward(self, x):
if len(x.shape) == 1:
input_size = x.shape[0]
x = x.view(1, 1, input_size)
batch_size = 1
orgdim = 1
elif len(x.shape) == 2:
batch_size, input_size = x.shape
x = x.view(1, batch_size, input_size)
orgdim = 2
else:
assert False, (
"expected a tensor with 1 or 2 dimensions, " + "but received a tensor of shape " + str(x.shape)
)
if self._state is None:
x, self._state = self._layer(x)
else:
x, self._state = self._layer(x, self._state)
if orgdim == 1:
x = x.view(-1)
elif orgdim == 2:
x = x.view(batch_size, -1)
else:
assert False, "unknown value for orgdim"
return x
@property
def batch_first(self):
"""Return True if the module expects the batch dimension first.
Otherwise, return False.
"""
return self._layer.batch_first
batch_first
property
readonly
¶
Return True if the module expects the batch dimension first. Otherwise, return False.
state
property
readonly
¶
Get the tensor of the internal state. If the recurrent network is just initialized or reset, then there is no state, so, a None is given. Not having a state means that an initial internal state tensor of compatible size with the input will be created at the first usage of this network. Each element of this initial internal state tensor is 0.
forward(self, x)
¶
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 :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
Source code in evotorch/neuroevolution/net/layers.py
def forward(self, x):
if len(x.shape) == 1:
input_size = x.shape[0]
x = x.view(1, 1, input_size)
batch_size = 1
orgdim = 1
elif len(x.shape) == 2:
batch_size, input_size = x.shape
x = x.view(1, batch_size, input_size)
orgdim = 2
else:
assert False, (
"expected a tensor with 1 or 2 dimensions, " + "but received a tensor of shape " + str(x.shape)
)
if self._state is None:
x, self._state = self._layer(x)
else:
x, self._state = self._layer(x, self._state)
if orgdim == 1:
x = x.view(-1)
elif orgdim == 2:
x = x.view(batch_size, -1)
else:
assert False, "unknown value for orgdim"
return x
reset(self)
¶
StructuredControlNet (Module)
¶
Structured Control Net.
This is a control network consisting of two components: (i) a non-linear component, which is a feed-forward network; and (ii) a linear component, which is a linear layer. Both components take the input vector provided to the structured control network. The final output is the sum of the outputs of both components.
Reference
Mario Srouji, Jian Zhang, Ruslan Salakhutdinov (2018). Structured Control Nets for Deep Reinforcement Learning.
Source code in evotorch/neuroevolution/net/layers.py
class StructuredControlNet(nn.Module):
"""Structured Control Net.
This is a control network consisting of two components:
(i) a non-linear component, which is a feed-forward network; and
(ii) a linear component, which is a linear layer.
Both components take the input vector provided to the
structured control network.
The final output is the sum of the outputs of both components.
Reference:
Mario Srouji, Jian Zhang, Ruslan Salakhutdinov (2018).
Structured Control Nets for Deep Reinforcement Learning.
"""
def __init__(
self,
*,
in_features: int,
out_features: int,
num_layers: int,
hidden_size: int,
bias: bool = True,
nonlinearity: Union[str, Callable] = "tanh",
):
"""`__init__(...)`: Initialize the structured control net.
Args:
in_features: Length of the input vector
out_features: Length of the output vector
num_layers: Number of hidden layers for the non-linear component
hidden_size: Number of neurons in a hidden layer of the
non-linear component
bias: Whether or not the linear component is to have bias
nonlinearity: Activation function
"""
nn.Module.__init__(self)
self._in_features = in_features
self._out_features = out_features
self._num_layers = num_layers
self._hidden_size = hidden_size
self._bias = bias
self._nonlinearity = nonlinearity
self._linear_component = nn.Linear(
in_features=self._in_features, out_features=self._out_features, bias=self._bias
)
self._nonlinear_component = FeedForwardNet(
input_size=self._in_features,
layers=(
list((self._hidden_size, self._nonlinearity) for _ in range(self._num_layers))
+ [(self._out_features, self._nonlinearity)]
),
)
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""TODO: documentation"""
return self._linear_component(x) + self._nonlinear_component(x)
@property
def in_features(self):
"""TODO: documentation"""
return self._in_features
@property
def out_features(self):
"""TODO: documentation"""
return self._out_features
@property
def num_layers(self):
"""TODO: documentation"""
return self._num_layers
@property
def hidden_size(self):
"""TODO: documentation"""
return self._hidden_size
@property
def bias(self):
"""TODO: documentation"""
return self._bias
@property
def nonlinearity(self):
"""TODO: documentation"""
return self._nonlinearity
bias
property
readonly
¶
hidden_size
property
readonly
¶
in_features
property
readonly
¶
nonlinearity
property
readonly
¶
num_layers
property
readonly
¶
out_features
property
readonly
¶
__init__(self, *, in_features, out_features, num_layers, hidden_size, bias=True, nonlinearity='tanh')
special
¶
__init__(...): Initialize the structured control net.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
in_features |
int |
Length of the input vector |
required |
out_features |
int |
Length of the output vector |
required |
num_layers |
int |
Number of hidden layers for the non-linear component |
required |
hidden_size |
int |
Number of neurons in a hidden layer of the non-linear component |
required |
bias |
bool |
Whether or not the linear component is to have bias |
True |
nonlinearity |
Union[str, Callable] |
Activation function |
'tanh' |
Source code in evotorch/neuroevolution/net/layers.py
def __init__(
self,
*,
in_features: int,
out_features: int,
num_layers: int,
hidden_size: int,
bias: bool = True,
nonlinearity: Union[str, Callable] = "tanh",
):
"""`__init__(...)`: Initialize the structured control net.
Args:
in_features: Length of the input vector
out_features: Length of the output vector
num_layers: Number of hidden layers for the non-linear component
hidden_size: Number of neurons in a hidden layer of the
non-linear component
bias: Whether or not the linear component is to have bias
nonlinearity: Activation function
"""
nn.Module.__init__(self)
self._in_features = in_features
self._out_features = out_features
self._num_layers = num_layers
self._hidden_size = hidden_size
self._bias = bias
self._nonlinearity = nonlinearity
self._linear_component = nn.Linear(
in_features=self._in_features, out_features=self._out_features, bias=self._bias
)
self._nonlinear_component = FeedForwardNet(
input_size=self._in_features,
layers=(
list((self._hidden_size, self._nonlinearity) for _ in range(self._num_layers))
+ [(self._out_features, self._nonlinearity)]
),
)
forward(self, x)
¶
reset_module_state(net)
¶
Reset a torch module's state by calling its reset() method.
If the module is a torch.nn.Sequential, then the function applies itself recursively to the submodules of the Sequential net. If the module does not have a reset() method, nothing happens.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
net |
Module |
The torch module whose state will be reset. |
required |
Source code in evotorch/neuroevolution/net/layers.py
def reset_module_state(net: nn.Module):
"""
Reset a torch module's state by calling its reset() method.
If the module is a torch.nn.Sequential, then the function
applies itself recursively to the submodules of the Sequential net.
If the module does not have a reset() method, nothing happens.
Args:
net: The torch module whose state will be reset.
"""
if hasattr(net, "reset"):
net.reset()
elif isinstance(net, nn.Sequential):
for i_module in range(len(net)):
reset_module_state(net[i_module])
misc
¶
Utilities for reading and for writing neural network parameters
count_parameters(net)
¶
Get the number of parameters the network.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
net |
Module |
The torch module whose parameters will be counted. |
required |
Returns:
| Type | Description |
|---|---|
int |
The number of parameters, as an integer. |
Source code in evotorch/neuroevolution/net/misc.py
fill_parameters(net, vector)
¶
Fill the parameters of a torch module (net) from a vector.
No gradient information is kept.
The vector's length must be exactly the same with the number of parameters of the PyTorch module.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
net |
Module |
The torch module whose parameter values will be filled. |
required |
vector |
Tensor |
A 1-D torch tensor which stores the parameter values. |
required |
Source code in evotorch/neuroevolution/net/misc.py
@torch.no_grad()
def fill_parameters(net: nn.Module, vector: torch.Tensor):
"""Fill the parameters of a torch module (net) from a vector.
No gradient information is kept.
The vector's length must be exactly the same with the number
of parameters of the PyTorch module.
Args:
net: The torch module whose parameter values will be filled.
vector: A 1-D torch tensor which stores the parameter values.
"""
address = 0
for p in net.parameters():
d = p.data.view(-1)
n = len(d)
d[:] = torch.as_tensor(vector[address : address + n], device=d.device)
address += n
if address != len(vector):
raise IndexError("The parameter vector is larger than expected")
parameter_vector(net, *, device=None)
¶
Get all the parameters of a torch module (net) into a vector
No gradient information is kept.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
net |
Module |
The torch module whose parameters will be extracted. |
required |
device |
Union[str, torch.device] |
The device in which the parameter vector will be constructed. If the network has parameter across multiple devices, you can specify this argument so that concatenation of all the parameters will be successful. |
None |
Returns:
| Type | Description |
|---|---|
Tensor |
The parameters of the module in a 1-D tensor. |
Source code in evotorch/neuroevolution/net/misc.py
@torch.no_grad()
def parameter_vector(net: nn.Module, *, device: Optional[Device] = None) -> torch.Tensor:
"""Get all the parameters of a torch module (net) into a vector
No gradient information is kept.
Args:
net: The torch module whose parameters will be extracted.
device: The device in which the parameter vector will be constructed.
If the network has parameter across multiple devices,
you can specify this argument so that concatenation of all the
parameters will be successful.
Returns:
The parameters of the module in a 1-D tensor.
"""
dev_kwarg = {} if device is None else {"device": device}
all_vectors = []
for p in net.parameters():
all_vectors.append(torch.as_tensor(p.data.view(-1), **dev_kwarg))
return torch.cat(all_vectors)
parser
¶
Utilities for parsing string representations of neural net policies
NetParsingError (Exception)
¶
Representation of a parsing error
Source code in evotorch/neuroevolution/net/parser.py
class NetParsingError(Exception):
"""
Representation of a parsing error
"""
def __init__(
self,
message: str,
lineno: Optional[int] = None,
col_offset: Optional[int] = None,
original_error: Optional[Exception] = None,
):
"""
`__init__(...)`: Initialize the NetParsingError.
Args:
message: Error message, as string.
lineno: Erroneous line number in the string representation of the
neural network structure.
col_offset: Erroneous column number in the string representation
of the neural network structure.
original_error: If another error caused this parsing error,
that original error can be attached to this `NetParsingError`
instance via this argument.
"""
super().__init__()
self.message = message
self.lineno = lineno
self.col_offset = col_offset
self.original_error = original_error
def _to_string(self) -> str:
parts = []
parts.append(type(self).__name__)
if self.lineno is not None:
parts.append(" at line(")
parts.append(str(self.lineno - 1))
parts.append(")")
if self.col_offset is not None:
parts.append(" at column(")
parts.append(str(self.col_offset + 1))
parts.append(")")
parts.append(": ")
parts.append(self.message)
return "".join(parts)
def __str__(self) -> str:
return self._to_string()
def __repr__(self) -> str:
return self._to_string()
__init__(self, message, lineno=None, col_offset=None, original_error=None)
special
¶
__init__(...): Initialize the NetParsingError.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message |
str |
Error message, as string. |
required |
lineno |
Optional[int] |
Erroneous line number in the string representation of the neural network structure. |
None |
col_offset |
Optional[int] |
Erroneous column number in the string representation of the neural network structure. |
None |
original_error |
Optional[Exception] |
If another error caused this parsing error,
that original error can be attached to this |
None |
Source code in evotorch/neuroevolution/net/parser.py
def __init__(
self,
message: str,
lineno: Optional[int] = None,
col_offset: Optional[int] = None,
original_error: Optional[Exception] = None,
):
"""
`__init__(...)`: Initialize the NetParsingError.
Args:
message: Error message, as string.
lineno: Erroneous line number in the string representation of the
neural network structure.
col_offset: Erroneous column number in the string representation
of the neural network structure.
original_error: If another error caused this parsing error,
that original error can be attached to this `NetParsingError`
instance via this argument.
"""
super().__init__()
self.message = message
self.lineno = lineno
self.col_offset = col_offset
self.original_error = original_error
str_to_net(s, **constants)
¶
Read a string representation of a neural net structure,
and return a torch.nn.Module instance out of it.
Let us imagine that one wants to describe the following neural network structure:
from torch import nn
net = nn.Sequential(
nn.Linear(8, 16),
nn.Tanh(),
nn.Linear(16, 4, bias=False),
nn.ReLU()
)
By using str_to_net(...) one can construct the same
module via:
from evotorch.neuroevolution.net import str_to_net
net = str_to_net(
'Linear(8, 16) >> Tanh() >> Linear(16, 4, bias=False) >> ReLU()'
)
The string can also be multi-line:
One can also define constants for using them in strings:
net = str_to_net(
'''
Linear(input_size, hidden_size)
>> Tanh()
>> Linear(hidden_size, output_size, bias=False)
>> ReLU()
''',
input_size=8,
hidden_size=16,
output_size=4
)
In the neural net structure string, when one refers to a module type,
say, Linear, first the name Linear is searched for in the namespace
evotorch.neuroevolution.net.layers, and then in the namespace torch.nn.
In the case of Linear, the searched name exists in torch.nn,
and therefore, the layer type to be instantiated is accepted as
torch.nn.Linear.
Instead of Linear, if one had used the name, say,
StructuredControlNet, then, the layer type to be instantiated
would be evotorch.neuroevolution.net.layers.StructuredControlNet.
Notes regarding usage with evotorch.neuroevolution.GymNE:
While instantiating a GymNE, one can specify a neural net
structure string as the policy. Therefore, while filling the policy
string for a GymNE, all these rules mentioned above apply.
Additionally, while using str_to_net(...) internally,
GymNE defines these extra constants:
obs_length (length of the observation vector),
act_length (length of the action vector for continuous-action
environments, or number of actions for discrete-action
environments), and obs_shape (shape of the observation as a tuple,
assuming that the observation space is of type gym.spaces.Box,
usable within the string like obs_shape[0], obs_shape[1], etc.,
or simply obs_shape to refer to the entire tuple).
Therefore, while using with GymNE, one can define a
single-hidden-layered policy via this string:
(where one might choose to omit the last Tanh() as GymNE
will clip the output of the final layer to conform with the
action boundaries of the environment, which one might think as a
type of hard-tanh anyway).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
s |
str |
The string which expresses the neural net structure. |
required |
Returns:
| Type | Description |
|---|---|
Module |
The PyTorch module of the specified structure. |
Source code in evotorch/neuroevolution/net/parser.py
def str_to_net(s: str, **constants) -> nn.Module:
"""
Read a string representation of a neural net structure,
and return a `torch.nn.Module` instance out of it.
Let us imagine that one wants to describe the following
neural network structure:
```python
from torch import nn
net = nn.Sequential(
nn.Linear(8, 16),
nn.Tanh(),
nn.Linear(16, 4, bias=False),
nn.ReLU()
)
```
By using `str_to_net(...)` one can construct the same
module via:
```python
from evotorch.neuroevolution.net import str_to_net
net = str_to_net(
'Linear(8, 16) >> Tanh() >> Linear(16, 4, bias=False) >> ReLU()'
)
```
The string can also be multi-line:
```python
net = str_to_net(
'''
Linear(8, 16)
>> Tanh()
>> Linear(16, 4, bias=False)
>> ReLU()
'''
)
```
One can also define constants for using them in strings:
```python
net = str_to_net(
'''
Linear(input_size, hidden_size)
>> Tanh()
>> Linear(hidden_size, output_size, bias=False)
>> ReLU()
''',
input_size=8,
hidden_size=16,
output_size=4
)
```
In the neural net structure string, when one refers to a module type,
say, `Linear`, first the name `Linear` is searched for in the namespace
`evotorch.neuroevolution.net.layers`, and then in the namespace `torch.nn`.
In the case of `Linear`, the searched name exists in `torch.nn`,
and therefore, the layer type to be instantiated is accepted as
`torch.nn.Linear`.
Instead of `Linear`, if one had used the name, say,
`StructuredControlNet`, then, the layer type to be instantiated
would be `evotorch.neuroevolution.net.layers.StructuredControlNet`.
**Notes regarding usage with `evotorch.neuroevolution.GymNE`:**
While instantiating a `GymNE`, one can specify a neural net
structure string as the policy. Therefore, while filling the policy
string for a `GymNE`, all these rules mentioned above apply.
Additionally, while using `str_to_net(...)` internally,
`GymNE` defines these extra constants:
`obs_length` (length of the observation vector),
`act_length` (length of the action vector for continuous-action
environments, or number of actions for discrete-action
environments), and `obs_shape` (shape of the observation as a tuple,
assuming that the observation space is of type `gym.spaces.Box`,
usable within the string like `obs_shape[0]`, `obs_shape[1]`, etc.,
or simply `obs_shape` to refer to the entire tuple).
Therefore, while using with `GymNE`, one can define a
single-hidden-layered policy via this string:
```python
'Linear(obs_length, 16) >> Tanh() >> Linear(16, act_length) >> Tanh()'
```
(where one might choose to omit the last `Tanh()` as `GymNE`
will clip the output of the final layer to conform with the
action boundaries of the environment, which one might think as a
type of hard-tanh anyway).
Args:
s: The string which expresses the neural net structure.
Returns:
The PyTorch module of the specified structure.
"""
s = f"(\n{s}\n)"
return _process_expr(ast.parse(s, mode="eval").body, constants=constants)
rl
¶
This namespace various RL-specific utilities.
ActClipLayer (Module)
¶
Source code in evotorch/neuroevolution/net/rl.py
forward(self, x)
¶
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 :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
ObsNormLayer (Module)
¶
Observation normalization layer for a policy network
Source code in evotorch/neuroevolution/net/rl.py
class ObsNormLayer(nn.Module):
"""Observation normalization layer for a policy network"""
def __init__(self, stats: RunningStat, trainable_stats: bool):
"""`__init__(...)`: Initialize the observation normalization layer
Args:
stats: The RunninStat object storing the mean and stdev of
all of the observations.
trainable_stats: Whether or not the normalization data
are to be stored as trainable parameters.
"""
nn.Module.__init__(self)
mean = torch.tensor(stats.mean, dtype=torch.float32)
stdev = torch.tensor(stats.stdev, dtype=torch.float32)
if trainable_stats:
self.obs_mean = nn.Parameter(mean)
self.obs_stdev = nn.Parameter(stdev)
else:
self.obs_mean = mean
self.obs_stdev = stdev
def forward(self, x):
x = x - self.obs_mean
x = x / self.obs_stdev
return x
__init__(self, stats, trainable_stats)
special
¶
__init__(...): Initialize the observation normalization layer
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
stats |
RunningStat |
The RunninStat object storing the mean and stdev of all of the observations. |
required |
trainable_stats |
bool |
Whether or not the normalization data are to be stored as trainable parameters. |
required |
Source code in evotorch/neuroevolution/net/rl.py
def __init__(self, stats: RunningStat, trainable_stats: bool):
"""`__init__(...)`: Initialize the observation normalization layer
Args:
stats: The RunninStat object storing the mean and stdev of
all of the observations.
trainable_stats: Whether or not the normalization data
are to be stored as trainable parameters.
"""
nn.Module.__init__(self)
mean = torch.tensor(stats.mean, dtype=torch.float32)
stdev = torch.tensor(stats.stdev, dtype=torch.float32)
if trainable_stats:
self.obs_mean = nn.Parameter(mean)
self.obs_stdev = nn.Parameter(stdev)
else:
self.obs_mean = mean
self.obs_stdev = stdev
forward(self, x)
¶
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 :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
reset_env(env)
¶
Reset a gym environment.
For gym 1.0, the plan is to have a reset(...) method which returns
a two-element tuple (observation, info) where info is an object
providing any additional information regarding the initial state of
the agent. However, the old (pre 1.0) gym API (and some environments
which were written with old gym compatibility in mind) has (or have)
a reset(...) method which returns a single object that is the
initial observation.
With the assumption that the observation space of the environment
is NOT tuple, this function can work with both pre-1.0 and (hopefully)
after-1.0 versions of gym, and always returns the initial observation.
Please do not use this function on environments whose observation
spaces or tuples, because then this function cannot distinguish between
environments whose reset(...) methods return a tuple and environments
whose reset(...) methods return a single observation object but that
observation object is a tuple.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
env |
Env |
The gym environment which will be reset. |
required |
Returns:
| Type | Description |
|---|---|
Iterable |
The initial observation |
Source code in evotorch/neuroevolution/net/rl.py
def reset_env(env: gym.Env) -> Iterable:
"""
Reset a gym environment.
For gym 1.0, the plan is to have a `reset(...)` method which returns
a two-element tuple `(observation, info)` where `info` is an object
providing any additional information regarding the initial state of
the agent. However, the old (pre 1.0) gym API (and some environments
which were written with old gym compatibility in mind) has (or have)
a `reset(...)` method which returns a single object that is the
initial observation.
With the assumption that the observation space of the environment
is NOT tuple, this function can work with both pre-1.0 and (hopefully)
after-1.0 versions of gym, and always returns the initial observation.
Please do not use this function on environments whose observation
spaces or tuples, because then this function cannot distinguish between
environments whose `reset(...)` methods return a tuple and environments
whose `reset(...)` methods return a single observation object but that
observation object is a tuple.
Args:
env: The gym environment which will be reset.
Returns:
The initial observation
"""
result = env.reset()
if isinstance(result, tuple) and (len(result) == 2):
result = result[0]
return result
take_step_in_env(env, action)
¶
Take a step in the gym environment. Taking a step means performing the action provided via the arguments.
For gym 1.0, the plan is to have a step(...) method which returns a
5-elements tuple containing observation, reward, terminated,
truncated, info where terminated is a boolean indicating whether
or not the episode is terminated because of the actions taken within the
environment, and truncated is a boolean indicating whether or not the
episode is finished because the time limit is reached.
However, the old (pre 1.0) gym API (and some environments which were
written with old gym compatibility in mind) has (or have) a step(...)
method which returns 4 elements: observation, reward, done, info
where done is a boolean indicating whether or not the episode is
"done", either because of termination or because of truncation.
This function can work with both pre-1.0 and (hopefully) after-1.0
versions of gym, and always returns the 4-element tuple as its result.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
env |
Env |
The gym environment in which the given action will be performed. |
required |
Returns:
| Type | Description |
|---|---|
tuple |
A tuple in the form |
Source code in evotorch/neuroevolution/net/rl.py
def take_step_in_env(env: gym.Env, action: Iterable) -> tuple:
"""
Take a step in the gym environment.
Taking a step means performing the action provided via the arguments.
For gym 1.0, the plan is to have a `step(...)` method which returns a
5-elements tuple containing `observation`, `reward`, `terminated`,
`truncated`, `info` where `terminated` is a boolean indicating whether
or not the episode is terminated because of the actions taken within the
environment, and `truncated` is a boolean indicating whether or not the
episode is finished because the time limit is reached.
However, the old (pre 1.0) gym API (and some environments which were
written with old gym compatibility in mind) has (or have) a `step(...)`
method which returns 4 elements: `observation`, `reward`, `done`, `info`
where `done` is a boolean indicating whether or not the episode is
"done", either because of termination or because of truncation.
This function can work with both pre-1.0 and (hopefully) after-1.0
versions of gym, and always returns the 4-element tuple as its result.
Args:
env: The gym environment in which the given action will be performed.
Returns:
A tuple in the form `(observation, reward, done, info)` where
`observation` is the observation received after performing the action,
`reward` is the amount of reward gained,
`done` is a boolean value indicating whether or not the episode has
ended, and
`info` is additional information (usually as a dictionary).
"""
result = env.step(action)
if isinstance(result, tuple):
n = len(result)
if n == 4:
observation, reward, done, info = result
elif n == 5:
observation, reward, terminated, truncated, info = result
done = terminated or truncated
else:
raise ValueError(
f"The result of the `step(...)` method of the gym environment"
f" was expected as a tuple of length 4 or 5."
f" However, the received result is {repr(result)}, which is"
f" of length {len(result)}."
)
else:
raise TypeError(
f"The result of the `step(...)` method of the gym environment"
f" was expected as a tuple of length 4 or 5."
f" However, the received result is {repr(result)}, which is"
f" of type {type(result)}."
)
return observation, reward, done, info
runningstat
¶
RunningStat
¶
Tool for efficiently computing the mean and stdev of arrays. The arrays themselves are not stored separately, instead, they are accumulated.
Source code in evotorch/neuroevolution/net/runningstat.py
class RunningStat:
"""
Tool for efficiently computing the mean and stdev of arrays.
The arrays themselves are not stored separately,
instead, they are accumulated.
"""
def __init__(self):
"""
``__init__(...)``: Initialize the RunningStat.
In the beginning, the number of arrays is 0,
and the sum and the sum of squares are set as NaN.
"""
# self.sum = np.zeros(shape, dtype='float32')
# self.sumsq = np.full(shape, eps, dtype='float32')
# self.count = eps
self.reset()
def reset(self):
"""
Reset the RunningStat to its initial state.
"""
self._sum = float("nan")
self._sumsq = float("nan")
self._count = 0
def _increment(self, s, ssq, c):
# self.sum += s
# self.sumsq += ssq
# self.count += c
if self._count == 0:
self._sum = np.array(s, dtype="float32")
self._sumsq = np.array(ssq, dtype="float32")
else:
self._sum += s
self._sumsq += ssq
self._count += c
@property
def count(self) -> int:
"""
Get the number of arrays accumulated.
"""
return self._count
@property
def sum(self) -> np.ndarray:
"""
Get the sum of all accumulated arrays.
"""
return self._sum
@property
def sum_of_squares(self) -> np.ndarray:
"""
Get the sum of squares of all accumulated arrays.
"""
return self._sumsq
@property
def mean(self) -> np.ndarray:
"""
Get the mean of all accumulated arrays.
"""
return self._sum / self._count
@property
def stdev(self) -> np.ndarray:
"""
Get the standard deviation of all accumulated arrays.
"""
return np.sqrt(np.maximum(self._sumsq / self._count - np.square(self.mean), 1e-2))
# def _set_from_init(self, init_mean, init_std, init_count):
# init_mean = np.asarray(init_mean, dtype='float32')
# init_std = np.asarray(init_std, dtype='float32')
# self._sum = init_mean * init_count
# self._sumsq = (np.square(init_mean) + np.square(init_std)) * init_count
# self._count = init_count
def update(self, x: Union[np.ndarray, "RunningStat"]):
"""
Accumulate more data into the RunningStat object.
If the argument is an array, that array is added
as one more data element.
If the argument is another RunningStat instance,
all the stats accumulated by that RunningStat object
are added into this RunningStat object.
"""
if isinstance(x, RunningStat):
if x.count > 0:
self._increment(x.sum, x.sum_of_squares, x.count)
else:
self._increment(x, np.square(x), 1)
def normalize(self, x: Union[np.ndarray, list]) -> np.ndarray:
"""
Normalize the array x according to the accumulated stats.
"""
x = np.array(x, dtype="float32")
x -= self.mean
x /= self.stdev
return x
def __copy__(self):
return deepcopy(self)
def __get_repr(self):
return "<RunningStat, count: " + str(self._count) + ">"
def __str__(self):
return self.__get_repr()
def __repr__(self):
return self.__get_repr()
count: int
property
readonly
¶
Get the number of arrays accumulated.
mean: ndarray
property
readonly
¶
Get the mean of all accumulated arrays.
stdev: ndarray
property
readonly
¶
Get the standard deviation of all accumulated arrays.
sum: ndarray
property
readonly
¶
Get the sum of all accumulated arrays.
sum_of_squares: ndarray
property
readonly
¶
Get the sum of squares of all accumulated arrays.
__init__(self)
special
¶
__init__(...): Initialize the RunningStat.
In the beginning, the number of arrays is 0, and the sum and the sum of squares are set as NaN.
Source code in evotorch/neuroevolution/net/runningstat.py
def __init__(self):
"""
``__init__(...)``: Initialize the RunningStat.
In the beginning, the number of arrays is 0,
and the sum and the sum of squares are set as NaN.
"""
# self.sum = np.zeros(shape, dtype='float32')
# self.sumsq = np.full(shape, eps, dtype='float32')
# self.count = eps
self.reset()
normalize(self, x)
¶
Normalize the array x according to the accumulated stats.
reset(self)
¶
update(self, x)
¶
Accumulate more data into the RunningStat object. If the argument is an array, that array is added as one more data element. If the argument is another RunningStat instance, all the stats accumulated by that RunningStat object are added into this RunningStat object.
Source code in evotorch/neuroevolution/net/runningstat.py
def update(self, x: Union[np.ndarray, "RunningStat"]):
"""
Accumulate more data into the RunningStat object.
If the argument is an array, that array is added
as one more data element.
If the argument is another RunningStat instance,
all the stats accumulated by that RunningStat object
are added into this RunningStat object.
"""
if isinstance(x, RunningStat):
if x.count > 0:
self._increment(x.sum, x.sum_of_squares, x.count)
else:
self._increment(x, np.square(x), 1)