Index

Utility classes and functions for neural networks

ModuleExpectingFlatParameters

A wrapper which brings a functional interface around a torch module.

Similar to functorch.FunctionalModule, ModuleExpectingFlatParameters turns a torch.nn.Module instance to a function which expects a new leftmost argument representing the parameters of the network. Unlike functorch.FunctionalModule, a ModuleExpectingFlatParameters instance, as its name suggests, expects the network parameters to be given as a 1-dimensional (i.e. flattened) tensor. Also, unlike functorch.FunctionalModule, an instance of ModuleExpectingFlatParameters is NOT an instance of torch.nn.Module.

PyTorch modules with buffers can be wrapped by this class, but it is assumed that those buffers are constant. If the wrapped module changes the value(s) of its buffer(s) during its forward passes, most probably things will NOT work right.

As an example, let us consider the following linear layer.

import torch
from torch import nn

net = nn.Linear(3, 8)

The functional counterpart of net can be obtained via:

from evotorch.neuroevolution.net import ModuleExpectingFlatParameters

fnet = ModuleExpectingFlatParameters(net)

Now, fnet is a callable object which expects network parameters and network inputs. Let us call fnet with randomly generated network parameters and with a randomly generated input tensor.

param_length = fnet.parameter_length
random_parameters = torch.randn(param_length)
random_input = torch.randn(3)

result = fnet(random_parameters, random_input)

Source code in evotorch/neuroevolution/net/functional.py

class ModuleExpectingFlatParameters:
    """
    A wrapper which brings a functional interface around a torch module.

    Similar to `functorch.FunctionalModule`, `ModuleExpectingFlatParameters`
    turns a `torch.nn.Module` instance to a function which expects a new
    leftmost argument representing the parameters of the network.
    Unlike `functorch.FunctionalModule`, a `ModuleExpectingFlatParameters`
    instance, as its name suggests, expects the network parameters to be
    given as a 1-dimensional (i.e. flattened) tensor.
    Also, unlike `functorch.FunctionalModule`, an instance of
    `ModuleExpectingFlatParameters` is NOT an instance of `torch.nn.Module`.

    PyTorch modules with buffers can be wrapped by this class, but it is
    assumed that those buffers are constant. If the wrapped module changes
    the value(s) of its buffer(s) during its forward passes, most probably
    things will NOT work right.

    As an example, let us consider the following linear layer.

    ```python
    import torch
    from torch import nn

    net = nn.Linear(3, 8)
    ```

    The functional counterpart of `net` can be obtained via:

    ```python
    from evotorch.neuroevolution.net import ModuleExpectingFlatParameters

    fnet = ModuleExpectingFlatParameters(net)
    ```

    Now, `fnet` is a callable object which expects network parameters
    and network inputs. Let us call `fnet` with randomly generated network
    parameters and with a randomly generated input tensor.

    ```python
    param_length = fnet.parameter_length
    random_parameters = torch.randn(param_length)
    random_input = torch.randn(3)

    result = fnet(random_parameters, random_input)
    ```
    """

    @torch.no_grad()
    def __init__(self, net: nn.Module, *, disable_autograd_tracking: bool = False):
        """
        `__init__(...)`: Initialize the `ModuleExpectingFlatParameters` instance.

        Args:
            net: The module that is to be wrapped by a functional interface.
            disable_autograd_tracking: If given as True, all operations
                regarding the wrapped module will be performed in the context
                `torch.no_grad()`, forcefully disabling the autograd.
                If given as False, autograd will not be affected.
                The default is False.
        """

        # Declare the variables which will store information regarding the parameters of the module.
        self.__param_names = []
        self.__param_shapes = []
        self.__param_length = 0
        self.__param_slices = []
        self.__num_params = 0

        # Iterate over the parameters of the module and fill the related information.
        i = 0
        j = 0
        for pname, p in net.named_parameters():
            self.__param_names.append(pname)

            shape = p.shape
            self.__param_shapes.append(shape)

            length = _shape_length(shape)
            self.__param_length += length

            j = i + length
            self.__param_slices.append(slice(i, j))
            i = j

            self.__num_params += 1

        self.__buffer_dict = {bname: b.clone() for bname, b in net.named_buffers()}

        self.__net = deepcopy(net)
        self.__net.to("meta")
        self.__disable_autograd_tracking = bool(disable_autograd_tracking)

    def __transfer_buffers(self, x: torch.Tensor):
        """
        Transfer the buffer tensors to the device of the given tensor.

        Args:
            x: The tensor whose device will also store the buffer tensors.
        """
        for bname in self.__buffer_dict.keys():
            self.__buffer_dict[bname] = torch.as_tensor(self.__buffer_dict[bname], device=x.device)

    @property
    def buffers(self) -> tuple:
        """Get the stored buffers"""
        return tuple(self.__buffer_dict)

    @property
    def parameter_length(self) -> int:
        return self.__param_length

    def __call__(self, parameter_vector: torch.Tensor, x: torch.Tensor, h: Any = None) -> Any:
        """
        Call the wrapped module's forward pass procedure.

        Args:
            parameter_vector: A 1-dimensional tensor which represents the
                parameters of the tensor.
            x: The inputs.
            h: Hidden state(s), in case this is a recurrent network.
        Returns:
            The result of the forward pass.
        """
        if parameter_vector.ndim != 1:
            raise ValueError(
                f"Expected the parameters as 1 dimensional,"
                f" but the received parameter vector has {parameter_vector.ndim} dimensions"
            )
        if len(parameter_vector) != self.__param_length:
            raise ValueError(
                f"Expected a parameter vector of length {self.__param_length},"
                f" but the received parameter vector's length is {len(parameter_vector)}."
            )
        state_args = [] if h is None else [h]

        params_and_buffers = {}
        for i, pname in enumerate(self.__param_names):
            param_slice = self.__param_slices[i]
            param_shape = self.__param_shapes[i]
            param = parameter_vector[param_slice].reshape(param_shape)
            params_and_buffers[pname] = param

        # Make sure that the buffer tensors are in the same device with x
        self.__transfer_buffers(x)

        # Add the buffer tensors to the dictionary `params_and_buffers`
        params_and_buffers.update(self.__buffer_dict)

        # Prepare the no-gradient context if gradient tracking is disabled
        context = torch.no_grad() if self.__disable_autograd_tracking else nullcontext()

        # Run the module and return the results
        with context:
            return functional_call(self.__net, params_and_buffers, tuple([x, *state_args]))

buffers property

Get the stored buffers

__call__(parameter_vector, x, h=None)

Call the wrapped module's forward pass procedure.

Parameters:

Name	Type	Description	Default
parameter_vector	Tensor	A 1-dimensional tensor which represents the parameters of the tensor.	required
x	Tensor	The inputs.	required
h	Any	Hidden state(s), in case this is a recurrent network.	None

Source code in evotorch/neuroevolution/net/functional.py

def __call__(self, parameter_vector: torch.Tensor, x: torch.Tensor, h: Any = None) -> Any:
    """
    Call the wrapped module's forward pass procedure.

    Args:
        parameter_vector: A 1-dimensional tensor which represents the
            parameters of the tensor.
        x: The inputs.
        h: Hidden state(s), in case this is a recurrent network.
    Returns:
        The result of the forward pass.
    """
    if parameter_vector.ndim != 1:
        raise ValueError(
            f"Expected the parameters as 1 dimensional,"
            f" but the received parameter vector has {parameter_vector.ndim} dimensions"
        )
    if len(parameter_vector) != self.__param_length:
        raise ValueError(
            f"Expected a parameter vector of length {self.__param_length},"
            f" but the received parameter vector's length is {len(parameter_vector)}."
        )
    state_args = [] if h is None else [h]

    params_and_buffers = {}
    for i, pname in enumerate(self.__param_names):
        param_slice = self.__param_slices[i]
        param_shape = self.__param_shapes[i]
        param = parameter_vector[param_slice].reshape(param_shape)
        params_and_buffers[pname] = param

    # Make sure that the buffer tensors are in the same device with x
    self.__transfer_buffers(x)

    # Add the buffer tensors to the dictionary `params_and_buffers`
    params_and_buffers.update(self.__buffer_dict)

    # Prepare the no-gradient context if gradient tracking is disabled
    context = torch.no_grad() if self.__disable_autograd_tracking else nullcontext()

    # Run the module and return the results
    with context:
        return functional_call(self.__net, params_and_buffers, tuple([x, *state_args]))

__init__(net, *, disable_autograd_tracking=False)

__init__(...): Initialize the ModuleExpectingFlatParameters instance.

Parameters:

Name	Type	Description	Default
net	Module	The module that is to be wrapped by a functional interface.	required
disable_autograd_tracking	bool	If given as True, all operations regarding the wrapped module will be performed in the context torch.no_grad(), forcefully disabling the autograd. If given as False, autograd will not be affected. The default is False.	False

Source code in evotorch/neuroevolution/net/functional.py

@torch.no_grad()
def __init__(self, net: nn.Module, *, disable_autograd_tracking: bool = False):
    """
    `__init__(...)`: Initialize the `ModuleExpectingFlatParameters` instance.

    Args:
        net: The module that is to be wrapped by a functional interface.
        disable_autograd_tracking: If given as True, all operations
            regarding the wrapped module will be performed in the context
            `torch.no_grad()`, forcefully disabling the autograd.
            If given as False, autograd will not be affected.
            The default is False.
    """

    # Declare the variables which will store information regarding the parameters of the module.
    self.__param_names = []
    self.__param_shapes = []
    self.__param_length = 0
    self.__param_slices = []
    self.__num_params = 0

    # Iterate over the parameters of the module and fill the related information.
    i = 0
    j = 0
    for pname, p in net.named_parameters():
        self.__param_names.append(pname)

        shape = p.shape
        self.__param_shapes.append(shape)

        length = _shape_length(shape)
        self.__param_length += length

        j = i + length
        self.__param_slices.append(slice(i, j))
        i = j

        self.__num_params += 1

    self.__buffer_dict = {bname: b.clone() for bname, b in net.named_buffers()}

    self.__net = deepcopy(net)
    self.__net.to("meta")
    self.__disable_autograd_tracking = bool(disable_autograd_tracking)

__transfer_buffers(x)

Transfer the buffer tensors to the device of the given tensor.

Parameters:

Name	Type	Description	Default
x	Tensor	The tensor whose device will also store the buffer tensors.	required

Source code in evotorch/neuroevolution/net/functional.py

def __transfer_buffers(self, x: torch.Tensor):
    """
    Transfer the buffer tensors to the device of the given tensor.

    Args:
        x: The tensor whose device will also store the buffer tensors.
    """
    for bname in self.__buffer_dict.keys():
        self.__buffer_dict[bname] = torch.as_tensor(self.__buffer_dict[bname], device=x.device)

NetParsingError

Bases: 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__(message, lineno=None, col_offset=None, original_error=None)

__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 NetParsingError instance via this argument.	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

Policy

A Policy for deciding the actions for a reinforcement learning environment.

This can be seen as a stateful wrapper around a PyTorch module.

Let us assume that we have the following PyTorch module:

from torch import nn

net = nn.Linear(5, 8)

which has 48 parameters (when all the parameters are flattened). Let us randomly generate a parameter vector for our module net:

parameters = torch.randn(48)

We can now prepare a policy:

policy = Policy(net)
policy.set_parameters(parameters)

If we generate a random observation:

observation = torch.randn(5)

We can receive our action as follows:

action = policy(observation)

If the PyTorch module that we wish to wrap is a recurrent network (i.e. a network which expects an optional second argument for the hidden state, and returns a second value which represents the updated hidden state), then, the hidden state is automatically managed by the Policy instance.

Let us assume that we have a recurrent network named recnet.

policy = Policy(recnet)
policy.set_parameters(parameters_of_recnet)

In this case, because the hidden state of the network is internally managed, the usage is still the same with our previous non-recurrent example:

action = policy(observation)

When using a recurrent module on multiple episodes, it is important to reset the hidden state of the network. This is achieved by the reset method:

policy.reset()
action1 = policy(observation1)

# action2 will be computed with the hidden state generated by the
# previous forward-pass.
action2 = policy(observation2)

policy.reset()

# action3 will be computed according to the renewed hidden state.
action3 = policy(observation3)

Both for non-recurrent and recurrent networks, it is possible to perform vectorized operations. For now, let us return to our first non-recurrent example:

net = nn.Linear(5, 8)

Instead of generating only one parameter vector, we now generate a batch of parameter vectors. Let us say that our batch size is 10:

batch_of_parameters = torch.randn(10, 48)

Like we did in the non-batched examples, we can do:

policy = Policy(net)
policy.set_parameters(batch_of_parameters)

Because we are now in the batched mode, policy now expects a batch of observations and will return a batch of actions:

batch_of_observations = torch.randn(10, 5)
batch_of_actions = policy(batch_of_observations)

When doing vectorized reinforcement learning with a recurrent module, it can be the case that only some of the environments are finished, and therefore it is necessary to reset the hidden states associated with those environments only. The reset(...) method of Policy has a second argument to specify which of the recurrent network instances are to be reset. For example, if the episodes of the environments with indices 2 and 5 are about to restart (and therefore we wish to reset the states of the networks with indices 2 and 5), then, we can do:

policy.reset(torch.tensor([2, 5]))

Source code in evotorch/neuroevolution/net/vecrl.py

class Policy:
    """
    A Policy for deciding the actions for a reinforcement learning environment.

    This can be seen as a stateful wrapper around a PyTorch module.

    Let us assume that we have the following PyTorch module:

    ```python
    from torch import nn

    net = nn.Linear(5, 8)
    ```

    which has 48 parameters (when all the parameters are flattened).
    Let us randomly generate a parameter vector for our module `net`:

    ```python
    parameters = torch.randn(48)
    ```

    We can now prepare a policy:

    ```python
    policy = Policy(net)
    policy.set_parameters(parameters)
    ```

    If we generate a random observation:

    ```python
    observation = torch.randn(5)
    ```

    We can receive our action as follows:

    ```python
    action = policy(observation)
    ```

    If the PyTorch module that we wish to wrap is a recurrent network (i.e.
    a network which expects an optional second argument for the hidden state,
    and returns a second value which represents the updated hidden state),
    then, the hidden state is automatically managed by the Policy instance.

    Let us assume that we have a recurrent network named `recnet`.

    ```python
    policy = Policy(recnet)
    policy.set_parameters(parameters_of_recnet)
    ```

    In this case, because the hidden state of the network is internally
    managed, the usage is still the same with our previous non-recurrent
    example:

    ```python
    action = policy(observation)
    ```

    When using a recurrent module on multiple episodes, it is important
    to reset the hidden state of the network. This is achieved by the
    reset method:

    ```python
    policy.reset()
    action1 = policy(observation1)

    # action2 will be computed with the hidden state generated by the
    # previous forward-pass.
    action2 = policy(observation2)

    policy.reset()

    # action3 will be computed according to the renewed hidden state.
    action3 = policy(observation3)
    ```

    Both for non-recurrent and recurrent networks, it is possible to
    perform vectorized operations. For now, let us return to our
    first non-recurrent example:

    ```python
    net = nn.Linear(5, 8)
    ```

    Instead of generating only one parameter vector, we now generate
    a batch of parameter vectors. Let us say that our batch size is 10:

    ```python
    batch_of_parameters = torch.randn(10, 48)
    ```

    Like we did in the non-batched examples, we can do:

    ```python
    policy = Policy(net)
    policy.set_parameters(batch_of_parameters)
    ```

    Because we are now in the batched mode, `policy` now expects a batch
    of observations and will return a batch of actions:

    ```python
    batch_of_observations = torch.randn(10, 5)
    batch_of_actions = policy(batch_of_observations)
    ```

    When doing vectorized reinforcement learning with a recurrent module,
    it can be the case that only some of the environments are finished,
    and therefore it is necessary to reset the hidden states associated
    with those environments only. The `reset(...)` method of Policy
    has a second argument to specify which of the recurrent network
    instances are to be reset. For example, if the episodes of the
    environments with indices 2 and 5 are about to restart (and therefore
    we wish to reset the states of the networks with indices 2 and 5),
    then, we can do:

    ```python
    policy.reset(torch.tensor([2, 5]))
    ```
    """

    def __init__(self, net: Union[str, Callable, nn.Module], **kwargs):
        """
        `__init__(...)`: Initialize the Policy.

        Args:
            net: The network to be wrapped by the Policy object.
                This can be a string, a Callable (e.g. a `torch.nn.Module`
                subclass), or a `torch.nn.Module` instance.
                When this argument is a string, the network will be
                created with the help of the function
                `evotorch.neuroevolution.net.str_to_net(...)` and then
                wrapped. Please see the `str_to_net(...)` function's
                documentation for details regarding how a network structure
                can be expressed via strings.
            kwargs: Expected in the form of additional keyword arguments,
                these keyword arguments will be passed to the provided
                Callable object (if the argument `net` is a Callable)
                or to `str_to_net(...)` (if the argument `net` is a string)
                at the moment of generating the network.
                If the argument `net` is a `torch.nn.Module` instance,
                having any additional keyword arguments will trigger an
                error, because the network is already instantiated and
                therefore, it is not possible to pass these keyword arguments.
        """
        from ..net import str_to_net
        from ..net.functional import ModuleExpectingFlatParameters, make_functional_module

        if isinstance(net, str):
            self.__module = str_to_net(net, **kwargs)
        elif isinstance(net, nn.Module):
            if len(kwargs) > 0:
                raise ValueError(
                    f"When the network is given as an `nn.Module` instance, extra network arguments cannot be used"
                    f" (because the network is already instantiated)."
                    f" However, these extra keyword arguments were received: {kwargs}."
                )
            self.__module = net
        elif isinstance(net, Callable):
            self.__module = net(**kwargs)
        else:
            raise TypeError(
                f"The class `Policy` expected a string or an `nn.Module` instance, or a Callable, but received {net}"
                f" (whose type is {type(net)})."
            )

        self.__fmodule: ModuleExpectingFlatParameters = make_functional_module(self.__module)
        self.__state: Any = None
        self.__parameters: Optional[torch.Tensor] = None

    def set_parameters(self, parameters: torch.Tensor, indices: Optional[MaskOrIndices] = None, *, reset: bool = True):
        """
        Set the parameters of the policy.

        Args:
            parameters: A 1-dimensional or a 2-dimensional tensor containing
                the flattened parameters to be used with the neural network.
                If the given parameters are two-dimensional, then, given that
                the leftmost size of the parameter tensor is `n`, the
                observations will be expected in a batch with leftmost size
                `n`, and the returned actions will also be in a batch,
                again with the leftmost size `n`.
            indices: For when the parameters were previously given via a
                2-dimensional tensor, provide this argument if you would like
                to change only some rows of the previously given parameters.
                For example, if `indices` is given as `torch.tensor([2, 4])`
                and the argument `parameters` is given as a 2-dimensional
                tensor with leftmost size 2, then the rows with indices
                2 and 4 will be replaced by these new parameters provided
                via the argument `parameters`.
            reset: If given as True, the hidden states of the networks whose
                parameters just changed will be reset. If `indices` was not
                provided at all, then this means that the parameters of all
                networks are modified, in which case, all the hidden states
                will be reset.
                If given as False, no such resetting will be done.
        """
        if self.__parameters is None:
            if indices is not None:
                raise ValueError(
                    "The argument `indices` can be used only if network parameters were previously specified."
                    " However, it seems that the method `set_parameters(...)` was not called before."
                )
            self.__parameters = parameters
        else:
            if indices is None:
                self.__parameters = parameters
            else:
                self.__parameters[indices] = parameters

        if reset:
            self.reset(indices)

    def __call__(self, x: torch.Tensor) -> torch.Tensor:
        """
        Pass the given observations through the network.

        Args:
            x: The observations, as a PyTorch tensor.
                If the parameters were given (via the method
                `set_parameters(...)`) as a 1-dimensional tensor, then this
                argument is expected to store a single observation.
                If the parameters were given as a 2-dimensional tensor,
                then, this argument is expected to store a batch of
                observations, and the leftmost size of this observation
                tensor must match with the leftmost size of the parameter
                tensor.
        Returns:
            The output tensor, which represents the action to take.
        """
        if self.__parameters is None:
            raise ValueError("Please use the method `set_parameters(...)` before calling the policy.")

        if self.__state is None:
            further_args = (x,)
        else:
            further_args = (x, self.__state)

        parameters = self.__parameters
        ndim = parameters.ndim
        if ndim == 1:
            result = self.__fmodule(parameters, *further_args)
        elif ndim == 2:
            vmapped = vmap(self.__fmodule)
            result = vmapped(parameters, *further_args)
        else:
            raise ValueError(
                f"Expected the parameters as a 1 or 2 dimensional tensor."
                f" However, the received parameters tensor has {ndim} dimensions."
            )

        if isinstance(result, torch.Tensor):
            return result
        elif isinstance(result, tuple):
            result, state = result
            self.__state = state
            return result
        else:
            raise TypeError(f"The torch module used by the Policy returned an unexpected object: {result}")

    def reset(self, indices: Optional[MaskOrIndices] = None, *, copy: bool = True):
        """
        Reset the hidden states, if the contained module is a recurrent network.

        Args:
            indices: Optionally a sequence of integers or a sequence of
                booleans, specifying which networks' states will be
                reset. If left as None, then the states of all the networks
                will be reset.
            copy: When `indices` is given as something other than None,
                if `copy` is given as True, then the resetting will NOT
                be done in-place. Instead, a new copy of the hidden state
                will first be created, and then the specified regions
                of this new copy will be cleared, and then finally this
                modified copy will be declared as the new hidden state.
                It is a common practice for recurrent neural network
                implementations to return the same tensor both as its
                output and as (part of) its hidden state. With `copy=False`,
                the resetting would be done in-place, and the action
                tensor could be involuntarily reset as well.
                This in-place modification could cause silent bugs
                if the unintended modification on the action tensor
                happens BEFORE the action is sent to the reinforcement
                learning environment.
                To prevent such situations, the default value for the argument
                `copy` is True.
        """
        if indices is None:
            self.__state = None
        else:
            if self.__state is not None:
                with torch.no_grad():
                    if copy:
                        self.__state = deepcopy(self.__state)
                    reset_tensors(self.__state, indices)

    @property
    def parameters(self) -> torch.Tensor:
        """
        The currently used parameters.
        """
        return self.__parameters

    @property
    def h(self) -> Optional[torch.Tensor]:
        """
        The hidden state of the contained recurrent network, if any.

        If the contained recurrent network did not generate a hidden state
        yet, or if the contained network is not recurrent, then the result
        will be None.
        """
        return self.__state

    @property
    def parameter_length(self) -> int:
        """
        Length of the parameter tensor.
        """
        return self.__fmodule.parameter_length

    @property
    def wrapped_module(self) -> nn.Module:
        """
        The wrapped `torch.nn.Module` instance.
        """
        return self.__module

    def to_torch_module(self, parameter_vector: torch.Tensor) -> nn.Module:
        """
        Get a copy of the contained network, parameterized as specified.

        Args:
            parameter_vector: The parameters to be used by the new network.
        Returns:
            Copy of the contained network, as a `torch.nn.Module` instance.
        """
        with torch.no_grad():
            net = deepcopy(self.__module).to(parameter_vector.device)
            nnu.vector_to_parameters(parameter_vector, net.parameters())
        return net

h property

The hidden state of the contained recurrent network, if any.

If the contained recurrent network did not generate a hidden state yet, or if the contained network is not recurrent, then the result will be None.

parameter_length property

Length of the parameter tensor.

parameters property

The currently used parameters.

wrapped_module property

The wrapped torch.nn.Module instance.

__call__(x)

Pass the given observations through the network.

Parameters:

Name	Type	Description	Default
x	Tensor	The observations, as a PyTorch tensor. If the parameters were given (via the method set_parameters(...)) as a 1-dimensional tensor, then this argument is expected to store a single observation. If the parameters were given as a 2-dimensional tensor, then, this argument is expected to store a batch of observations, and the leftmost size of this observation tensor must match with the leftmost size of the parameter tensor.	required

Source code in evotorch/neuroevolution/net/vecrl.py

def __call__(self, x: torch.Tensor) -> torch.Tensor:
    """
    Pass the given observations through the network.

    Args:
        x: The observations, as a PyTorch tensor.
            If the parameters were given (via the method
            `set_parameters(...)`) as a 1-dimensional tensor, then this
            argument is expected to store a single observation.
            If the parameters were given as a 2-dimensional tensor,
            then, this argument is expected to store a batch of
            observations, and the leftmost size of this observation
            tensor must match with the leftmost size of the parameter
            tensor.
    Returns:
        The output tensor, which represents the action to take.
    """
    if self.__parameters is None:
        raise ValueError("Please use the method `set_parameters(...)` before calling the policy.")

    if self.__state is None:
        further_args = (x,)
    else:
        further_args = (x, self.__state)

    parameters = self.__parameters
    ndim = parameters.ndim
    if ndim == 1:
        result = self.__fmodule(parameters, *further_args)
    elif ndim == 2:
        vmapped = vmap(self.__fmodule)
        result = vmapped(parameters, *further_args)
    else:
        raise ValueError(
            f"Expected the parameters as a 1 or 2 dimensional tensor."
            f" However, the received parameters tensor has {ndim} dimensions."
        )

    if isinstance(result, torch.Tensor):
        return result
    elif isinstance(result, tuple):
        result, state = result
        self.__state = state
        return result
    else:
        raise TypeError(f"The torch module used by the Policy returned an unexpected object: {result}")

__init__(net, **kwargs)

__init__(...): Initialize the Policy.

Parameters:

Name	Type	Description	Default
net	Union[str, Callable, Module]	The network to be wrapped by the Policy object. This can be a string, a Callable (e.g. a torch.nn.Module subclass), or a torch.nn.Module instance. When this argument is a string, the network will be created with the help of the function evotorch.neuroevolution.net.str_to_net(...) and then wrapped. Please see the str_to_net(...) function's documentation for details regarding how a network structure can be expressed via strings.	required
kwargs		Expected in the form of additional keyword arguments, these keyword arguments will be passed to the provided Callable object (if the argument net is a Callable) or to str_to_net(...) (if the argument net is a string) at the moment of generating the network. If the argument net is a torch.nn.Module instance, having any additional keyword arguments will trigger an error, because the network is already instantiated and therefore, it is not possible to pass these keyword arguments.	{}

Source code in evotorch/neuroevolution/net/vecrl.py

def __init__(self, net: Union[str, Callable, nn.Module], **kwargs):
    """
    `__init__(...)`: Initialize the Policy.

    Args:
        net: The network to be wrapped by the Policy object.
            This can be a string, a Callable (e.g. a `torch.nn.Module`
            subclass), or a `torch.nn.Module` instance.
            When this argument is a string, the network will be
            created with the help of the function
            `evotorch.neuroevolution.net.str_to_net(...)` and then
            wrapped. Please see the `str_to_net(...)` function's
            documentation for details regarding how a network structure
            can be expressed via strings.
        kwargs: Expected in the form of additional keyword arguments,
            these keyword arguments will be passed to the provided
            Callable object (if the argument `net` is a Callable)
            or to `str_to_net(...)` (if the argument `net` is a string)
            at the moment of generating the network.
            If the argument `net` is a `torch.nn.Module` instance,
            having any additional keyword arguments will trigger an
            error, because the network is already instantiated and
            therefore, it is not possible to pass these keyword arguments.
    """
    from ..net import str_to_net
    from ..net.functional import ModuleExpectingFlatParameters, make_functional_module

    if isinstance(net, str):
        self.__module = str_to_net(net, **kwargs)
    elif isinstance(net, nn.Module):
        if len(kwargs) > 0:
            raise ValueError(
                f"When the network is given as an `nn.Module` instance, extra network arguments cannot be used"
                f" (because the network is already instantiated)."
                f" However, these extra keyword arguments were received: {kwargs}."
            )
        self.__module = net
    elif isinstance(net, Callable):
        self.__module = net(**kwargs)
    else:
        raise TypeError(
            f"The class `Policy` expected a string or an `nn.Module` instance, or a Callable, but received {net}"
            f" (whose type is {type(net)})."
        )

    self.__fmodule: ModuleExpectingFlatParameters = make_functional_module(self.__module)
    self.__state: Any = None
    self.__parameters: Optional[torch.Tensor] = None

reset(indices=None, *, copy=True)

Reset the hidden states, if the contained module is a recurrent network.

Parameters:

Name	Type	Description	Default
indices	Optional[MaskOrIndices]	Optionally a sequence of integers or a sequence of booleans, specifying which networks' states will be reset. If left as None, then the states of all the networks will be reset.	None
copy	bool	When indices is given as something other than None, if copy is given as True, then the resetting will NOT be done in-place. Instead, a new copy of the hidden state will first be created, and then the specified regions of this new copy will be cleared, and then finally this modified copy will be declared as the new hidden state. It is a common practice for recurrent neural network implementations to return the same tensor both as its output and as (part of) its hidden state. With copy=False, the resetting would be done in-place, and the action tensor could be involuntarily reset as well. This in-place modification could cause silent bugs if the unintended modification on the action tensor happens BEFORE the action is sent to the reinforcement learning environment. To prevent such situations, the default value for the argument copy is True.	True

Source code in evotorch/neuroevolution/net/vecrl.py

def reset(self, indices: Optional[MaskOrIndices] = None, *, copy: bool = True):
    """
    Reset the hidden states, if the contained module is a recurrent network.

    Args:
        indices: Optionally a sequence of integers or a sequence of
            booleans, specifying which networks' states will be
            reset. If left as None, then the states of all the networks
            will be reset.
        copy: When `indices` is given as something other than None,
            if `copy` is given as True, then the resetting will NOT
            be done in-place. Instead, a new copy of the hidden state
            will first be created, and then the specified regions
            of this new copy will be cleared, and then finally this
            modified copy will be declared as the new hidden state.
            It is a common practice for recurrent neural network
            implementations to return the same tensor both as its
            output and as (part of) its hidden state. With `copy=False`,
            the resetting would be done in-place, and the action
            tensor could be involuntarily reset as well.
            This in-place modification could cause silent bugs
            if the unintended modification on the action tensor
            happens BEFORE the action is sent to the reinforcement
            learning environment.
            To prevent such situations, the default value for the argument
            `copy` is True.
    """
    if indices is None:
        self.__state = None
    else:
        if self.__state is not None:
            with torch.no_grad():
                if copy:
                    self.__state = deepcopy(self.__state)
                reset_tensors(self.__state, indices)

set_parameters(parameters, indices=None, *, reset=True)

Set the parameters of the policy.

Parameters:

Name	Type	Description	Default
parameters	Tensor	A 1-dimensional or a 2-dimensional tensor containing the flattened parameters to be used with the neural network. If the given parameters are two-dimensional, then, given that the leftmost size of the parameter tensor is n, the observations will be expected in a batch with leftmost size n, and the returned actions will also be in a batch, again with the leftmost size n.	required
indices	Optional[MaskOrIndices]	For when the parameters were previously given via a 2-dimensional tensor, provide this argument if you would like to change only some rows of the previously given parameters. For example, if indices is given as torch.tensor([2, 4]) and the argument parameters is given as a 2-dimensional tensor with leftmost size 2, then the rows with indices 2 and 4 will be replaced by these new parameters provided via the argument parameters.	None
reset	bool	If given as True, the hidden states of the networks whose parameters just changed will be reset. If indices was not provided at all, then this means that the parameters of all networks are modified, in which case, all the hidden states will be reset. If given as False, no such resetting will be done.	True

Source code in evotorch/neuroevolution/net/vecrl.py

def set_parameters(self, parameters: torch.Tensor, indices: Optional[MaskOrIndices] = None, *, reset: bool = True):
    """
    Set the parameters of the policy.

    Args:
        parameters: A 1-dimensional or a 2-dimensional tensor containing
            the flattened parameters to be used with the neural network.
            If the given parameters are two-dimensional, then, given that
            the leftmost size of the parameter tensor is `n`, the
            observations will be expected in a batch with leftmost size
            `n`, and the returned actions will also be in a batch,
            again with the leftmost size `n`.
        indices: For when the parameters were previously given via a
            2-dimensional tensor, provide this argument if you would like
            to change only some rows of the previously given parameters.
            For example, if `indices` is given as `torch.tensor([2, 4])`
            and the argument `parameters` is given as a 2-dimensional
            tensor with leftmost size 2, then the rows with indices
            2 and 4 will be replaced by these new parameters provided
            via the argument `parameters`.
        reset: If given as True, the hidden states of the networks whose
            parameters just changed will be reset. If `indices` was not
            provided at all, then this means that the parameters of all
            networks are modified, in which case, all the hidden states
            will be reset.
            If given as False, no such resetting will be done.
    """
    if self.__parameters is None:
        if indices is not None:
            raise ValueError(
                "The argument `indices` can be used only if network parameters were previously specified."
                " However, it seems that the method `set_parameters(...)` was not called before."
            )
        self.__parameters = parameters
    else:
        if indices is None:
            self.__parameters = parameters
        else:
            self.__parameters[indices] = parameters

    if reset:
        self.reset(indices)

to_torch_module(parameter_vector)

Get a copy of the contained network, parameterized as specified.

Parameters:

Name	Type	Description	Default
parameter_vector	Tensor	The parameters to be used by the new network.	required

Source code in evotorch/neuroevolution/net/vecrl.py

def to_torch_module(self, parameter_vector: torch.Tensor) -> nn.Module:
    """
    Get a copy of the contained network, parameterized as specified.

    Args:
        parameter_vector: The parameters to be used by the new network.
    Returns:
        Copy of the contained network, as a `torch.nn.Module` instance.
    """
    with torch.no_grad():
        net = deepcopy(self.__module).to(parameter_vector.device)
        nnu.vector_to_parameters(parameter_vector, net.parameters())
    return net

RunningNorm

An online observation normalization tool

Source code in evotorch/neuroevolution/net/runningnorm.py

class RunningNorm:
    """
    An online observation normalization tool
    """

    def __init__(
        self,
        *,
        shape: Union[tuple, int],
        dtype: DType,
        device: Optional[Device] = None,
        min_variance: float = 1e-2,
        clip: Optional[tuple] = None,
    ) -> None:
        """
        `__init__(...)`: Initialize the RunningNorm

        Args:
            shape: Observation shape. Can be an integer or a tuple.
            dtype: The dtype of the observations.
            device: The device in which the observation stats are held.
                If left as None, the device is assumed to be "cpu".
            min_variance: A lower bound for the variance to be used in
                the normalization computations.
                In other words, if the computed variance according to the
                collected observations ends up lower than `min_variance`,
                this `min_variance` will be used instead (in an elementwise
                manner) while computing the normalized observations.
                As in Salimans et al. (2017), the default is 1e-2.
            clip: Can be left as None (which is the default), or can be
                given as a pair of real numbers.
                This is used for clipping the observations after the
                normalization operation.
                In Salimans et al. (2017), (-5.0, +5.0) was used.
        """

        # Make sure that the shape is stored as a torch.Size object.
        if isinstance(shape, Iterable):
            self._shape = torch.Size(shape)
        else:
            self._shape = torch.Size([int(shape)])

        # Store the number of dimensions
        self._ndim = len(self._shape)

        # Store the dtype and the device
        self._dtype = to_torch_dtype(dtype)
        self._device = "cpu" if device is None else device

        # Initialize the internally stored data as empty
        self._sum: Optional[torch.Tensor] = None
        self._sum_of_squares: Optional[torch.Tensor] = None
        self._count: int = 0

        # Store the minimum variance
        self._min_variance = float(min_variance)

        if clip is not None:
            # If a clip tuple was provided, store the specified lower and upper bounds
            lb, ub = clip
            self._lb = float(lb)
            self._ub = float(ub)
        else:
            # If a clip tuple was not provided the bounds are stored as None
            self._lb = None
            self._ub = None

    def to(self, device: Device) -> "RunningNorm":
        """
        If the target device is a different device, then make a copy of this
        RunningNorm instance on the target device.
        If the target device is the same with this RunningNorm's device, then
        return this RunningNorm itself.

        Args:
            device: The target device.
        Returns:
            The RunningNorm on the target device. This can be a copy, or the
            original RunningNorm instance itself.
        """
        if torch.device(device) == torch.device(self.device):
            return self
        else:
            new_running_norm = object.__new__(type(self))

            already_handled = {"_sum", "_sum_of_squares", "_device"}
            new_running_norm._sum = self._sum.to(device)
            new_running_norm._sum_of_squares = self._sum_of_squares.to(device)
            new_running_norm._device = device

            for k, v in self.__dict__.items():
                if k not in already_handled:
                    setattr(new_running_norm, k, deepcopy(v))

            return new_running_norm

    @property
    def device(self) -> Device:
        """
        The device in which the observation stats are held
        """
        return self._device

    @property
    def dtype(self) -> DType:
        """
        The dtype of the stored observation stats
        """
        return self._dtype

    @property
    def shape(self) -> tuple:
        """
        Observation shape
        """
        return self._shape

    @property
    def min_variance(self) -> float:
        """
        Minimum variance
        """
        return self._min_variance

    @property
    def low(self) -> Optional[float]:
        """
        The lower component of the bounds given in the `clip` tuple.
        If `clip` was initialized as None, this is also None.
        """
        return self._lb

    @property
    def high(self) -> Optional[float]:
        """
        The higher (upper) component of the bounds given in the `clip` tuple.
        If `clip` was initialized as None, this is also None.
        """
        return self._ub

    def _like_its_own(self, x: Iterable) -> torch.Tensor:
        return torch.as_tensor(x, dtype=self._dtype, device=self._device)

    def _verify(self, x: Iterable) -> torch.Tensor:
        x = self._like_its_own(x)
        if x.ndim == self._ndim:
            if x.shape != self._shape:
                raise ValueError(
                    f"This RunningNorm instance was initialized with shape: {self._shape}."
                    f" However, the provided tensor has an incompatible shape: {x._shape}."
                )
        elif x.ndim == (self._ndim + 1):
            if x.shape[1:] != self._shape:
                raise ValueError(
                    f"This RunningNorm instance was initialized with shape: {self._shape}."
                    f" The provided tensor is shaped {x.shape}."
                    f" Accepting the tensor's leftmost dimension as the batch size,"
                    f" the remaining shape is incompatible: {x.shape[1:]}"
                )
        else:
            raise ValueError(
                f"This RunningNorm instance was initialized with shape: {self._shape}."
                f" The provided tensor is shaped {x.shape}."
                f" The number of dimensions of the given tensor is incompatible."
            )
        return x

    def _has_no_data(self) -> bool:
        return (self._sum is None) and (self._sum_of_squares is None) and (self._count == 0)

    def _has_data(self) -> bool:
        return (self._sum is not None) and (self._sum_of_squares is not None) and (self._count > 0)

    def reset(self):
        """
        Remove all the collected observation data.
        """
        self._sum = None
        self._sum_of_squares = None
        self._count = 0

    @torch.no_grad()
    def update(self, x: Union[Iterable, "RunningNorm"], mask: Optional[Iterable] = None, *, verify: bool = True):
        """
        Update the stored stats with new observation data.

        Args:
            x: The new observation(s), as a PyTorch tensor, or any Iterable
                that can be converted to a PyTorch tensor, or another
                RunningNorm instance.
                If given as a tensor or as an Iterable, the shape of `x` can
                be the same with observation shape, or it can be augmented
                with an extra leftmost dimension.
                In the case of augmented dimension, `x` is interpreted not as
                a single observation, but as a batch of observations.
                If `x` is another RunningNorm instance, the stats stored by
                this RunningNorm instance will be updated with all the data
                stored by `x`.
            mask: Can be given as a 1-dimensional Iterable of booleans ONLY
                if `x` represents a batch of observations.
                If a `mask` is provided, the i-th observation within the
                observation batch `x` will be taken into account only if
                the i-th item of the `mask` is True.
            verify: Whether or not to verify the shape of the given Iterable
                objects. The default is True.
        """
        if isinstance(x, RunningNorm):
            # If we are to update our stats according to another RunningNorm instance

            if x._count > 0:
                # We bother only if x is non-empty

                if mask is not None:
                    # We were given another RunningNorm, not a batch of observations.
                    # So, we do not expect to receive a mask tensor.
                    # If a mask was provided, then this is an unexpected way of calling this function.
                    # We therefore raise an error.
                    raise ValueError(
                        "The `mask` argument is expected as None if the first argument is a RunningNorm."
                        " However, `mask` is found as something other than None."
                    )

                if self._shape != x._shape:
                    # If the shapes of this RunningNorm and of the other RunningNorm
                    # do not match, then we cannot use `x` for updating our stats.
                    # It might be the case that `x` was initialized for another
                    # task, with differently sized observations.
                    # We therefore raise an error.
                    raise ValueError(
                        f"The RunningNorm to be updated has the shape {self._shape}"
                        f" The other RunningNorm has the shape {self._shape}"
                        f" These shapes are incompatible."
                    )

                if self._has_no_data():
                    # If this RunningNorm has no data at all, then we clone the
                    # data of x.
                    self._sum = self._like_its_own(x._sum.clone())
                    self._sum_of_squares = self._like_its_own(x._sum_of_squares.clone())
                    self._count = x._count
                elif self._has_data():
                    # If this RunningNorm has its own data, then we update the
                    # stored data with the data stored by x.
                    self._sum += self._like_its_own(x._sum)
                    self._sum_of_squares += self._like_its_own(x._sum_of_squares)
                    self._count += x._count
                else:
                    assert False, "RunningNorm is in an invalid state! This might be a bug."
        else:
            # This is the case where the received argument x is not a
            # RunningNorm object, but an Iterable.

            if verify:
                # If we have the `verify` flag, then we make sure that
                # x is a tensor of the correct shape
                x = self._verify(x)

            if x.ndim == self._ndim:
                # If the shape of x is exactly the same with the observation shape
                # then we assume that x represents a single observation, and not a
                # batch of observations.

                if mask is not None:
                    # Since we are dealing with a single observation,
                    # we do not expect to receive a mask argument.
                    # If the mask argument was provided, then this is an unexpected
                    # usage of this function.
                    # We therefore raise an error.
                    raise ValueError(
                        "The `mask` argument is expected as None if the first argument is a single observation"
                        " (i.e. not a batch of observations, with an extra leftmost dimension)."
                        " However, `mask` is found as something other than None."
                    )

                # Since x is a single observation,
                # the sum of observations extracted from x is x itself,
                # and the sum of squared observations extracted from x is
                # the square of x itself.
                sum_of_x = x
                sum_of_x_squared = x.square()
                # We extracted a single observation from x
                n = 1
            elif x.ndim == (self._ndim + 1):
                # If the number of dimensions of x is one more than the number
                # of dimensions of this RunningNorm, then we assume that x is a batch
                # of observations.

                if mask is not None:
                    # If a mask is provided, then we first make sure that it is a tensor
                    # of dtype bool in the correct device.
                    mask = torch.as_tensor(mask, dtype=torch.bool, device=self._device)

                    if mask.ndim != 1:
                        # We expect the mask to be 1-dimensional.
                        # If not, we raise an error.
                        raise ValueError(
                            f"The `mask` tensor was expected as a 1-dimensional tensor."
                            f" However, its shape is {mask.shape}."
                        )

                    if len(mask) != x.shape[0]:
                        # If the length of the mask is not the batch size of x,
                        # then there is a mismatch.
                        # We therefore raise an error.
                        raise ValueError(
                            f"The shape of the given tensor is {x.shape}."
                            f" Therefore, the batch size of observations is {x.shape[0]}."
                            f" However, the given `mask` tensor does not has an incompatible length: {len(mask)}."
                        )

                    # We compute how many True items we have in the mask.
                    # This integer gives us how many observations we extract from x.
                    n = int(torch.sum(torch.as_tensor(mask, dtype=torch.int64, device=self._device)))

                    # We now re-cast the mask as the observation dtype (so that True items turn to 1.0
                    # and False items turn to 0.0), and then increase its number of dimensions so that
                    # it can operate directly with x.
                    mask = self._like_its_own(mask).reshape(torch.Size([x.shape[0]] + ([1] * (x.ndim - 1))))

                    # Finally, we multiply x with the mask. This means that the observations with corresponding
                    # mask values as False are zeroed out.
                    x = x * mask
                else:
                    # This is the case where we did not receive a mask.
                    # We can simply say that the number of observations to extract from x
                    # is the size of its leftmost dimension, i.e. the batch size.
                    n = x.shape[0]

                # With or without a mask, we are now ready to extract the sum and sum of squares
                # from x.
                sum_of_x = torch.sum(x, dim=0)
                sum_of_x_squared = torch.sum(x.square(), dim=0)
            else:
                # This is the case where the number of dimensions of x is unrecognized.
                # This case is actually already checked by the _verify(...) method earlier.
                # This defensive fallback case is only for when verify=False and it turned out
                # that the ndim is invalid.
                raise ValueError(f"Invalid shape: {x.shape}")

            # At this point, we handled all the valid cases regarding the Iterable x,
            # and we have our sum_of_x (sum of all observations), sum_of_squares
            # (sum of all squared observations), and n (number of observations extracted
            # from x).

            if self._has_no_data():
                # If our RunningNorm is empty, the observation data we extracted from x
                # become our RunningNorm's new data.
                self._sum = sum_of_x
                self._sum_of_squares = sum_of_x_squared
                self._count = n
            elif self._has_data():
                # If our RunningNorm is not empty, the stored data is updated with the
                # data extracted from x.
                self._sum += sum_of_x
                self._sum_of_squares += sum_of_x_squared
                self._count += n
            else:
                # This is an erroneous state where the internal data looks neither
                # existent nor completely empty.
                # This might be the result of a bug, or maybe this instance's
                # protected variables were tempered with from the outside.
                assert False, "RunningNorm is in an invalid state! This might be a bug."

    @property
    @torch.no_grad()
    def stats(self) -> CollectedStats:
        """
        The collected data's mean and standard deviation (stdev) in a tuple
        """

        # Using the internally stored sum, sum_of_squares, and count,
        # compute E[x] and E[x^2]
        E_x = self._sum / self._count
        E_x2 = self._sum_of_squares / self._count

        # The mean is E[x]
        mean = E_x

        # The variance is E[x^2] - (E[x])^2, elementwise clipped such that
        # it cannot go below min_variance
        variance = _clamp(E_x2 - E_x.square(), self._min_variance, None)

        # Standard deviation is finally computed as the square root of the variance
        stdev = torch.sqrt(variance)

        # Return the stats in a named tuple
        return CollectedStats(mean=mean, stdev=stdev)

    @property
    def mean(self) -> torch.Tensor:
        """
        The collected data's mean
        """
        return self._sum / self._count

    @property
    def stdev(self) -> torch.Tensor:
        """
        The collected data's standard deviation
        """
        return self.stats.stdev

    @property
    def sum(self) -> torch.Tensor:
        """
        The collected data's sum
        """
        return self._sum

    @property
    def sum_of_squares(self) -> torch.Tensor:
        """
        Sum of squares of the collected data
        """
        return self._sum_of_squares

    @property
    def count(self) -> int:
        """
        Number of observations encountered
        """
        return self._count

    @torch.no_grad()
    def normalize(self, x: Iterable, *, result_as_numpy: Optional[bool] = None, verify: bool = True) -> Iterable:
        """
        Normalize the given observation x.

        Args:
            x: The observation(s), as a PyTorch tensor, or any Iterable
                that is convertable to a PyTorch tensor.
                `x` can be a single observation, or it can be a batch
                of observations (with an extra leftmost dimension).
            result_as_numpy: Whether or not to return the normalized
                observation as a numpy array.
                If left as None (which is the default), then the returned
                type depends on x: a PyTorch tensor is returned if x is a
                PyTorch tensor, and a numpy array is returned otherwise.
                If True, the result is always a numpy array.
                If False, the result is always a PyTorch tensor.
            verify: Whether or not to check the type and dimensions of x.
                This is True by default.
                Note that, if `verify` is False, this function will not
                properly check the type of `x` and will assume that `x`
                is a PyTorch tensor.
        Returns:
            The normalized observation, as a PyTorch tensor or a numpy array.
        """

        if self._count == 0:
            # If this RunningNorm instance has no data yet,
            # then we do not know how to do the normalization.
            # We therefore raise an error.
            raise ValueError("Cannot do normalization because no data is collected yet.")

        if verify:
            # Here we verify the type and shape of x.

            if result_as_numpy is None:
                # If there is not an explicit request about the return type,
                # we infer the return type from the type of x:
                # if x is a tensor, we return a tensor;
                # otherwise, we assume x to be a CPU-bound iterable, and
                # therefore we return a numpy array.
                result_as_numpy = not isinstance(x, torch.Tensor)
            else:
                result_as_numpy = bool(result_as_numpy)

            # We call _verify() to make sure that x is of correct shape
            # and is properly converted to a PyTorch tensor.
            x = self._verify(x)

        # We get the mean and stdev of the collected data
        mean, stdev = self.stats

        # Now we compute the normalized observation, clipped according to the
        # lower and upper bounds expressed by the `clip` tuple, if exists.
        result = _clamp((x - mean) / stdev, self._lb, self._ub)

        if result_as_numpy:
            # If we are to return the result as a numpy array, we do the
            # necessary conversion.
            result = result.cpu().numpy()

        # Finally, return the result
        return result

    @torch.no_grad()
    def update_and_normalize(self, x: Iterable, mask: Optional[Iterable] = None) -> Iterable:
        """
        Update the observation stats according to x, then normalize x.

        Args:
            x: The observation(s), as a PyTorch tensor, or as an Iterable
                which can be converted to a PyTorch tensor.
                The shape of x can be the same with the observaiton shape,
                or it can be augmented with an extra leftmost dimension
                to express a batch of observations.
            mask: Can be given as a 1-dimensional Iterable of booleans ONLY
                if `x` represents a batch of observations.
                If a `mask` is provided, the i-th observation within the
                observation batch `x` will be taken into account only if
                the the i-th item of the `mask` is True.
        Returns:
            The normalized counterpart of the observation(s) expressed by x.
        """
        result_as_numpy = not isinstance(x, torch.Tensor)
        x = self._verify(x)

        self.update(x, mask, verify=False)
        result = self.normalize(x, verify=False)

        if result_as_numpy:
            result = result.cpu().numpy()

        return result

    def to_layer(self) -> "ObsNormLayer":
        """
        Make a PyTorch module which normalizes the its inputs.

        Returns:
            An ObsNormLayer instance.
        """
        mean, stdev = self.stats
        low = self.low
        high = self.high
        return ObsNormLayer(mean=mean, stdev=stdev, low=low, high=high)

    def __repr__(self) -> str:
        return f"<{self.__class__.__name__}, count: {self.count}>"

    def __copy__(self) -> "RunningNorm":
        return deepcopy(self)

count property

Number of observations encountered

device property

The device in which the observation stats are held

dtype property

The dtype of the stored observation stats

high property

The higher (upper) component of the bounds given in the clip tuple. If clip was initialized as None, this is also None.

low property

The lower component of the bounds given in the clip tuple. If clip was initialized as None, this is also None.

mean property

The collected data's mean

min_variance property

Minimum variance

shape property

Observation shape

stats property

The collected data's mean and standard deviation (stdev) in a tuple

stdev property

The collected data's standard deviation

sum property

The collected data's sum

sum_of_squares property

Sum of squares of the collected data

__init__(*, shape, dtype, device=None, min_variance=0.01, clip=None)

__init__(...): Initialize the RunningNorm

Parameters:

Name	Type	Description	Default
shape	Union[tuple, int]	Observation shape. Can be an integer or a tuple.	required
dtype	DType	The dtype of the observations.	required
device	Optional[Device]	The device in which the observation stats are held. If left as None, the device is assumed to be "cpu".	None
min_variance	float	A lower bound for the variance to be used in the normalization computations. In other words, if the computed variance according to the collected observations ends up lower than min_variance, this min_variance will be used instead (in an elementwise manner) while computing the normalized observations. As in Salimans et al. (2017), the default is 1e-2.	0.01
clip	Optional[tuple]	Can be left as None (which is the default), or can be given as a pair of real numbers. This is used for clipping the observations after the normalization operation. In Salimans et al. (2017), (-5.0, +5.0) was used.	None

Source code in evotorch/neuroevolution/net/runningnorm.py

def __init__(
    self,
    *,
    shape: Union[tuple, int],
    dtype: DType,
    device: Optional[Device] = None,
    min_variance: float = 1e-2,
    clip: Optional[tuple] = None,
) -> None:
    """
    `__init__(...)`: Initialize the RunningNorm

    Args:
        shape: Observation shape. Can be an integer or a tuple.
        dtype: The dtype of the observations.
        device: The device in which the observation stats are held.
            If left as None, the device is assumed to be "cpu".
        min_variance: A lower bound for the variance to be used in
            the normalization computations.
            In other words, if the computed variance according to the
            collected observations ends up lower than `min_variance`,
            this `min_variance` will be used instead (in an elementwise
            manner) while computing the normalized observations.
            As in Salimans et al. (2017), the default is 1e-2.
        clip: Can be left as None (which is the default), or can be
            given as a pair of real numbers.
            This is used for clipping the observations after the
            normalization operation.
            In Salimans et al. (2017), (-5.0, +5.0) was used.
    """

    # Make sure that the shape is stored as a torch.Size object.
    if isinstance(shape, Iterable):
        self._shape = torch.Size(shape)
    else:
        self._shape = torch.Size([int(shape)])

    # Store the number of dimensions
    self._ndim = len(self._shape)

    # Store the dtype and the device
    self._dtype = to_torch_dtype(dtype)
    self._device = "cpu" if device is None else device

    # Initialize the internally stored data as empty
    self._sum: Optional[torch.Tensor] = None
    self._sum_of_squares: Optional[torch.Tensor] = None
    self._count: int = 0

    # Store the minimum variance
    self._min_variance = float(min_variance)

    if clip is not None:
        # If a clip tuple was provided, store the specified lower and upper bounds
        lb, ub = clip
        self._lb = float(lb)
        self._ub = float(ub)
    else:
        # If a clip tuple was not provided the bounds are stored as None
        self._lb = None
        self._ub = None

normalize(x, *, result_as_numpy=None, verify=True)

Normalize the given observation x.

Parameters:

Name	Type	Description	Default
x	Iterable	The observation(s), as a PyTorch tensor, or any Iterable that is convertable to a PyTorch tensor. x can be a single observation, or it can be a batch of observations (with an extra leftmost dimension).	required
result_as_numpy	Optional[bool]	Whether or not to return the normalized observation as a numpy array. If left as None (which is the default), then the returned type depends on x: a PyTorch tensor is returned if x is a PyTorch tensor, and a numpy array is returned otherwise. If True, the result is always a numpy array. If False, the result is always a PyTorch tensor.	None
verify	bool	Whether or not to check the type and dimensions of x. This is True by default. Note that, if verify is False, this function will not properly check the type of x and will assume that x is a PyTorch tensor.	True

Source code in evotorch/neuroevolution/net/runningnorm.py

@torch.no_grad()
def normalize(self, x: Iterable, *, result_as_numpy: Optional[bool] = None, verify: bool = True) -> Iterable:
    """
    Normalize the given observation x.

    Args:
        x: The observation(s), as a PyTorch tensor, or any Iterable
            that is convertable to a PyTorch tensor.
            `x` can be a single observation, or it can be a batch
            of observations (with an extra leftmost dimension).
        result_as_numpy: Whether or not to return the normalized
            observation as a numpy array.
            If left as None (which is the default), then the returned
            type depends on x: a PyTorch tensor is returned if x is a
            PyTorch tensor, and a numpy array is returned otherwise.
            If True, the result is always a numpy array.
            If False, the result is always a PyTorch tensor.
        verify: Whether or not to check the type and dimensions of x.
            This is True by default.
            Note that, if `verify` is False, this function will not
            properly check the type of `x` and will assume that `x`
            is a PyTorch tensor.
    Returns:
        The normalized observation, as a PyTorch tensor or a numpy array.
    """

    if self._count == 0:
        # If this RunningNorm instance has no data yet,
        # then we do not know how to do the normalization.
        # We therefore raise an error.
        raise ValueError("Cannot do normalization because no data is collected yet.")

    if verify:
        # Here we verify the type and shape of x.

        if result_as_numpy is None:
            # If there is not an explicit request about the return type,
            # we infer the return type from the type of x:
            # if x is a tensor, we return a tensor;
            # otherwise, we assume x to be a CPU-bound iterable, and
            # therefore we return a numpy array.
            result_as_numpy = not isinstance(x, torch.Tensor)
        else:
            result_as_numpy = bool(result_as_numpy)

        # We call _verify() to make sure that x is of correct shape
        # and is properly converted to a PyTorch tensor.
        x = self._verify(x)

    # We get the mean and stdev of the collected data
    mean, stdev = self.stats

    # Now we compute the normalized observation, clipped according to the
    # lower and upper bounds expressed by the `clip` tuple, if exists.
    result = _clamp((x - mean) / stdev, self._lb, self._ub)

    if result_as_numpy:
        # If we are to return the result as a numpy array, we do the
        # necessary conversion.
        result = result.cpu().numpy()

    # Finally, return the result
    return result

reset()

Remove all the collected observation data.

Source code in evotorch/neuroevolution/net/runningnorm.py

def reset(self):
    """
    Remove all the collected observation data.
    """
    self._sum = None
    self._sum_of_squares = None
    self._count = 0

to(device)

If the target device is a different device, then make a copy of this RunningNorm instance on the target device. If the target device is the same with this RunningNorm's device, then return this RunningNorm itself.

Parameters:

Name	Type	Description	Default
device	Device	The target device.	required

Source code in evotorch/neuroevolution/net/runningnorm.py

def to(self, device: Device) -> "RunningNorm":
    """
    If the target device is a different device, then make a copy of this
    RunningNorm instance on the target device.
    If the target device is the same with this RunningNorm's device, then
    return this RunningNorm itself.

    Args:
        device: The target device.
    Returns:
        The RunningNorm on the target device. This can be a copy, or the
        original RunningNorm instance itself.
    """
    if torch.device(device) == torch.device(self.device):
        return self
    else:
        new_running_norm = object.__new__(type(self))

        already_handled = {"_sum", "_sum_of_squares", "_device"}
        new_running_norm._sum = self._sum.to(device)
        new_running_norm._sum_of_squares = self._sum_of_squares.to(device)
        new_running_norm._device = device

        for k, v in self.__dict__.items():
            if k not in already_handled:
                setattr(new_running_norm, k, deepcopy(v))

        return new_running_norm

to_layer()

Make a PyTorch module which normalizes the its inputs.

Returns:

Type	Description
ObsNormLayer	An ObsNormLayer instance.

Source code in evotorch/neuroevolution/net/runningnorm.py

def to_layer(self) -> "ObsNormLayer":
    """
    Make a PyTorch module which normalizes the its inputs.

    Returns:
        An ObsNormLayer instance.
    """
    mean, stdev = self.stats
    low = self.low
    high = self.high
    return ObsNormLayer(mean=mean, stdev=stdev, low=low, high=high)

update(x, mask=None, *, verify=True)

Update the stored stats with new observation data.

Parameters:

Name	Type	Description	Default
x	Union[Iterable, RunningNorm]	The new observation(s), as a PyTorch tensor, or any Iterable that can be converted to a PyTorch tensor, or another RunningNorm instance. If given as a tensor or as an Iterable, the shape of x can be the same with observation shape, or it can be augmented with an extra leftmost dimension. In the case of augmented dimension, x is interpreted not as a single observation, but as a batch of observations. If x is another RunningNorm instance, the stats stored by this RunningNorm instance will be updated with all the data stored by x.	required
mask	Optional[Iterable]	Can be given as a 1-dimensional Iterable of booleans ONLY if x represents a batch of observations. If a mask is provided, the i-th observation within the observation batch x will be taken into account only if the i-th item of the mask is True.	None
verify	bool	Whether or not to verify the shape of the given Iterable objects. The default is True.	True

Source code in evotorch/neuroevolution/net/runningnorm.py

@torch.no_grad()
def update(self, x: Union[Iterable, "RunningNorm"], mask: Optional[Iterable] = None, *, verify: bool = True):
    """
    Update the stored stats with new observation data.

    Args:
        x: The new observation(s), as a PyTorch tensor, or any Iterable
            that can be converted to a PyTorch tensor, or another
            RunningNorm instance.
            If given as a tensor or as an Iterable, the shape of `x` can
            be the same with observation shape, or it can be augmented
            with an extra leftmost dimension.
            In the case of augmented dimension, `x` is interpreted not as
            a single observation, but as a batch of observations.
            If `x` is another RunningNorm instance, the stats stored by
            this RunningNorm instance will be updated with all the data
            stored by `x`.
        mask: Can be given as a 1-dimensional Iterable of booleans ONLY
            if `x` represents a batch of observations.
            If a `mask` is provided, the i-th observation within the
            observation batch `x` will be taken into account only if
            the i-th item of the `mask` is True.
        verify: Whether or not to verify the shape of the given Iterable
            objects. The default is True.
    """
    if isinstance(x, RunningNorm):
        # If we are to update our stats according to another RunningNorm instance

        if x._count > 0:
            # We bother only if x is non-empty

            if mask is not None:
                # We were given another RunningNorm, not a batch of observations.
                # So, we do not expect to receive a mask tensor.
                # If a mask was provided, then this is an unexpected way of calling this function.
                # We therefore raise an error.
                raise ValueError(
                    "The `mask` argument is expected as None if the first argument is a RunningNorm."
                    " However, `mask` is found as something other than None."
                )

            if self._shape != x._shape:
                # If the shapes of this RunningNorm and of the other RunningNorm
                # do not match, then we cannot use `x` for updating our stats.
                # It might be the case that `x` was initialized for another
                # task, with differently sized observations.
                # We therefore raise an error.
                raise ValueError(
                    f"The RunningNorm to be updated has the shape {self._shape}"
                    f" The other RunningNorm has the shape {self._shape}"
                    f" These shapes are incompatible."
                )

            if self._has_no_data():
                # If this RunningNorm has no data at all, then we clone the
                # data of x.
                self._sum = self._like_its_own(x._sum.clone())
                self._sum_of_squares = self._like_its_own(x._sum_of_squares.clone())
                self._count = x._count
            elif self._has_data():
                # If this RunningNorm has its own data, then we update the
                # stored data with the data stored by x.
                self._sum += self._like_its_own(x._sum)
                self._sum_of_squares += self._like_its_own(x._sum_of_squares)
                self._count += x._count
            else:
                assert False, "RunningNorm is in an invalid state! This might be a bug."
    else:
        # This is the case where the received argument x is not a
        # RunningNorm object, but an Iterable.

        if verify:
            # If we have the `verify` flag, then we make sure that
            # x is a tensor of the correct shape
            x = self._verify(x)

        if x.ndim == self._ndim:
            # If the shape of x is exactly the same with the observation shape
            # then we assume that x represents a single observation, and not a
            # batch of observations.

            if mask is not None:
                # Since we are dealing with a single observation,
                # we do not expect to receive a mask argument.
                # If the mask argument was provided, then this is an unexpected
                # usage of this function.
                # We therefore raise an error.
                raise ValueError(
                    "The `mask` argument is expected as None if the first argument is a single observation"
                    " (i.e. not a batch of observations, with an extra leftmost dimension)."
                    " However, `mask` is found as something other than None."
                )

            # Since x is a single observation,
            # the sum of observations extracted from x is x itself,
            # and the sum of squared observations extracted from x is
            # the square of x itself.
            sum_of_x = x
            sum_of_x_squared = x.square()
            # We extracted a single observation from x
            n = 1
        elif x.ndim == (self._ndim + 1):
            # If the number of dimensions of x is one more than the number
            # of dimensions of this RunningNorm, then we assume that x is a batch
            # of observations.

            if mask is not None:
                # If a mask is provided, then we first make sure that it is a tensor
                # of dtype bool in the correct device.
                mask = torch.as_tensor(mask, dtype=torch.bool, device=self._device)

                if mask.ndim != 1:
                    # We expect the mask to be 1-dimensional.
                    # If not, we raise an error.
                    raise ValueError(
                        f"The `mask` tensor was expected as a 1-dimensional tensor."
                        f" However, its shape is {mask.shape}."
                    )

                if len(mask) != x.shape[0]:
                    # If the length of the mask is not the batch size of x,
                    # then there is a mismatch.
                    # We therefore raise an error.
                    raise ValueError(
                        f"The shape of the given tensor is {x.shape}."
                        f" Therefore, the batch size of observations is {x.shape[0]}."
                        f" However, the given `mask` tensor does not has an incompatible length: {len(mask)}."
                    )

                # We compute how many True items we have in the mask.
                # This integer gives us how many observations we extract from x.
                n = int(torch.sum(torch.as_tensor(mask, dtype=torch.int64, device=self._device)))

                # We now re-cast the mask as the observation dtype (so that True items turn to 1.0
                # and False items turn to 0.0), and then increase its number of dimensions so that
                # it can operate directly with x.
                mask = self._like_its_own(mask).reshape(torch.Size([x.shape[0]] + ([1] * (x.ndim - 1))))

                # Finally, we multiply x with the mask. This means that the observations with corresponding
                # mask values as False are zeroed out.
                x = x * mask
            else:
                # This is the case where we did not receive a mask.
                # We can simply say that the number of observations to extract from x
                # is the size of its leftmost dimension, i.e. the batch size.
                n = x.shape[0]

            # With or without a mask, we are now ready to extract the sum and sum of squares
            # from x.
            sum_of_x = torch.sum(x, dim=0)
            sum_of_x_squared = torch.sum(x.square(), dim=0)
        else:
            # This is the case where the number of dimensions of x is unrecognized.
            # This case is actually already checked by the _verify(...) method earlier.
            # This defensive fallback case is only for when verify=False and it turned out
            # that the ndim is invalid.
            raise ValueError(f"Invalid shape: {x.shape}")

        # At this point, we handled all the valid cases regarding the Iterable x,
        # and we have our sum_of_x (sum of all observations), sum_of_squares
        # (sum of all squared observations), and n (number of observations extracted
        # from x).

        if self._has_no_data():
            # If our RunningNorm is empty, the observation data we extracted from x
            # become our RunningNorm's new data.
            self._sum = sum_of_x
            self._sum_of_squares = sum_of_x_squared
            self._count = n
        elif self._has_data():
            # If our RunningNorm is not empty, the stored data is updated with the
            # data extracted from x.
            self._sum += sum_of_x
            self._sum_of_squares += sum_of_x_squared
            self._count += n
        else:
            # This is an erroneous state where the internal data looks neither
            # existent nor completely empty.
            # This might be the result of a bug, or maybe this instance's
            # protected variables were tempered with from the outside.
            assert False, "RunningNorm is in an invalid state! This might be a bug."

update_and_normalize(x, mask=None)

Update the observation stats according to x, then normalize x.

Parameters:

Name	Type	Description	Default
x	Iterable	The observation(s), as a PyTorch tensor, or as an Iterable which can be converted to a PyTorch tensor. The shape of x can be the same with the observaiton shape, or it can be augmented with an extra leftmost dimension to express a batch of observations.	required
mask	Optional[Iterable]	Can be given as a 1-dimensional Iterable of booleans ONLY if x represents a batch of observations. If a mask is provided, the i-th observation within the observation batch x will be taken into account only if the the i-th item of the mask is True.	None

Source code in evotorch/neuroevolution/net/runningnorm.py

@torch.no_grad()
def update_and_normalize(self, x: Iterable, mask: Optional[Iterable] = None) -> Iterable:
    """
    Update the observation stats according to x, then normalize x.

    Args:
        x: The observation(s), as a PyTorch tensor, or as an Iterable
            which can be converted to a PyTorch tensor.
            The shape of x can be the same with the observaiton shape,
            or it can be augmented with an extra leftmost dimension
            to express a batch of observations.
        mask: Can be given as a 1-dimensional Iterable of booleans ONLY
            if `x` represents a batch of observations.
            If a `mask` is provided, the i-th observation within the
            observation batch `x` will be taken into account only if
            the the i-th item of the `mask` is True.
    Returns:
        The normalized counterpart of the observation(s) expressed by x.
    """
    result_as_numpy = not isinstance(x, torch.Tensor)
    x = self._verify(x)

    self.update(x, mask, verify=False)
    result = self.normalize(x, verify=False)

    if result_as_numpy:
        result = result.cpu().numpy()

    return result

RunningStat

Tool for efficiently computing the mean and stdev of arrays. The arrays themselves are not stored separately, instead, they are accumulated.

This RunningStat is implemented as a wrapper around RunningNorm. The difference is that the interface of RunningStat is simplified to expect only numpy arrays, and expect only non-vectorized observations. With this simplified interface, RunningStat is meant to be used by GymNE, on classical non-vectorized gym tasks.

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.

    This RunningStat is implemented as a wrapper around RunningNorm.
    The difference is that the interface of RunningStat is simplified
    to expect only numpy arrays, and expect only non-vectorized
    observations.
    With this simplified interface, RunningStat is meant to be used
    by GymNE, on classical non-vectorized gym tasks.
    """

    def __init__(self):
        """
        `__init__(...)`: Initialize the RunningStat.
        """
        self._rn: Optional[RunningNorm] = None
        self.reset()

    def reset(self):
        """
        Reset the RunningStat to its initial state.
        """
        self._rn = None

    @property
    def count(self) -> int:
        """
        Get the number of arrays accumulated.
        """
        if self._rn is None:
            return 0
        else:
            return self._rn.count

    @property
    def sum(self) -> np.ndarray:
        """
        Get the sum of all accumulated arrays.
        """
        return self._rn.sum.numpy()

    @property
    def sum_of_squares(self) -> np.ndarray:
        """
        Get the sum of squares of all accumulated arrays.
        """
        return self._rn.sum_of_squares.numpy()

    @property
    def mean(self) -> np.ndarray:
        """
        Get the mean of all accumulated arrays.
        """
        return self._rn.mean.numpy()

    @property
    def stdev(self) -> np.ndarray:
        """
        Get the standard deviation of all accumulated arrays.
        """
        return self._rn.stdev.numpy()

    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:
                if self._rn is None:
                    self._rn = deepcopy(x._rn)
                else:
                    self._rn.update(x._rn)
        else:
            if self._rn is None:
                x = np.array(x, dtype="float32")
                self._rn = RunningNorm(shape=x.shape, dtype="float32", device="cpu")
            self._rn.update(x)

    def normalize(self, x: Union[np.ndarray, list]) -> np.ndarray:
        """
        Normalize the array x according to the accumulated stats.
        """
        if self._rn is None:
            return x
        else:
            x = np.array(x, dtype="float32")
            return self._rn.normalize(x)

    def __copy__(self):
        return deepcopy(self)

    def __repr__(self) -> str:
        return f"<{self.__class__.__name__}, count: {self.count}>"

    def to(self, device: Union[str, torch.device]) -> "RunningStat":
        """
        If the target device is cpu, return this RunningStat instance itself.
        A RunningStat object is meant to work with numpy arrays. Therefore,
        any device other than the cpu will trigger an error.

        Args:
            device: The target device. Only cpu is supported.
        Returns:
            The original RunningStat.
        """
        if torch.device(device) == torch.device("cpu"):
            return self
        else:
            raise ValueError(
                f"The received target device is {repr(device)}. However, RunningStat can only work on a cpu."
            )

    def to_layer(self) -> nn.Module:
        """
        Make a PyTorch module which normalizes the its inputs.

        Returns:
            An ObsNormLayer instance.
        """
        return self._rn.to_layer()

count property

Get the number of arrays accumulated.

mean property

Get the mean of all accumulated arrays.

stdev property

Get the standard deviation of all accumulated arrays.

sum property

Get the sum of all accumulated arrays.

sum_of_squares property

Get the sum of squares of all accumulated arrays.

__init__()

__init__(...): Initialize the RunningStat.

Source code in evotorch/neuroevolution/net/runningstat.py

def __init__(self):
    """
    `__init__(...)`: Initialize the RunningStat.
    """
    self._rn: Optional[RunningNorm] = None
    self.reset()

normalize(x)

Normalize the array x according to the accumulated stats.

Source code in evotorch/neuroevolution/net/runningstat.py

def normalize(self, x: Union[np.ndarray, list]) -> np.ndarray:
    """
    Normalize the array x according to the accumulated stats.
    """
    if self._rn is None:
        return x
    else:
        x = np.array(x, dtype="float32")
        return self._rn.normalize(x)

reset()

Reset the RunningStat to its initial state.

Source code in evotorch/neuroevolution/net/runningstat.py

def reset(self):
    """
    Reset the RunningStat to its initial state.
    """
    self._rn = None

to(device)

If the target device is cpu, return this RunningStat instance itself. A RunningStat object is meant to work with numpy arrays. Therefore, any device other than the cpu will trigger an error.

Parameters:

Name	Type	Description	Default
device	Union[str, device]	The target device. Only cpu is supported.	required

Source code in evotorch/neuroevolution/net/runningstat.py

def to(self, device: Union[str, torch.device]) -> "RunningStat":
    """
    If the target device is cpu, return this RunningStat instance itself.
    A RunningStat object is meant to work with numpy arrays. Therefore,
    any device other than the cpu will trigger an error.

    Args:
        device: The target device. Only cpu is supported.
    Returns:
        The original RunningStat.
    """
    if torch.device(device) == torch.device("cpu"):
        return self
    else:
        raise ValueError(
            f"The received target device is {repr(device)}. However, RunningStat can only work on a cpu."
        )

to_layer()

Make a PyTorch module which normalizes the its inputs.

Returns:

Type	Description
Module	An ObsNormLayer instance.

Source code in evotorch/neuroevolution/net/runningstat.py

def to_layer(self) -> nn.Module:
    """
    Make a PyTorch module which normalizes the its inputs.

    Returns:
        An ObsNormLayer instance.
    """
    return self._rn.to_layer()

update(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:
            if self._rn is None:
                self._rn = deepcopy(x._rn)
            else:
                self._rn.update(x._rn)
    else:
        if self._rn is None:
            x = np.array(x, dtype="float32")
            self._rn = RunningNorm(shape=x.shape, dtype="float32", device="cpu")
        self._rn.update(x)

StatefulModule

Bases: Module

A wrapper that provides a stateful interface for recurrent torch modules.

If the torch module to be wrapped is non-recurrent and its forward method has a single input (the input tensor) and a single output (the output tensor), then this wrapper module acts as a no-op wrapper.

If the torch module to be wrapped is recurrent and its forward method has two inputs (the input tensor and an optional second argument for the hidden state) and two outputs (the output tensor and the new hidden state), then this wrapper brings a new forward-passing interface. In this new interface, the forward method has a single input (the input tensor) and a single output (the output tensor). The hidden states, instead of being explicitly requested via a second argument and returned as a second result, are stored and used by the wrapper. When a new series of inputs is to be used, one has to call the reset() method of this wrapper.

Source code in evotorch/neuroevolution/net/statefulmodule.py

class StatefulModule(nn.Module):
    """
    A wrapper that provides a stateful interface for recurrent torch modules.

    If the torch module to be wrapped is non-recurrent and its forward method
    has a single input (the input tensor) and a single output (the output
    tensor), then this wrapper module acts as a no-op wrapper.

    If the torch module to be wrapped is recurrent and its forward method has
    two inputs (the input tensor and an optional second argument for the hidden
    state) and two outputs (the output tensor and the new hidden state), then
    this wrapper brings a new forward-passing interface. In this new interface,
    the forward method has a single input (the input tensor) and a single
    output (the output tensor). The hidden states, instead of being
    explicitly requested via a second argument and returned as a second
    result, are stored and used by the wrapper.
    When a new series of inputs is to be used, one has to call the `reset()`
    method of this wrapper.
    """

    def __init__(self, wrapped_module: nn.Module):
        """
        `__init__(...)`: Initialize the StatefulModule.

        Args:
            wrapped_module: The `torch.nn.Module` instance to wrap.
        """
        super().__init__()

        # Declare the variable that will store the hidden state of wrapped_module, if any.
        self._hidden: Any = None

        # Store the module that is wrapped.
        self.wrapped_module = wrapped_module

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        if self._hidden is None:
            # If there is no stored hidden state, then only pass the input tensor to the wrapped module.
            out = self.wrapped_module(x)
        else:
            # If there is a hidden state saved from the previous call to this `forward(...)` method, then pass the
            # input tensor and this stored hidden state.
            out = self.wrapped_module(x, self._hidden)

        if isinstance(out, tuple):
            # If the result of the wrapped module is a tuple, then we assume that the wrapped module returned an
            # output tensor and a hidden state. We assume the first element of this tuple as the output tensor,
            # and the second element as the new hidden state.
            # We set the variable y to the output tensor, and we store the new hidden state via the attribute
            # `_hidden`.
            y, self._hidden = out
        else:
            # If the result of the wrapped module is not a tuple, then we assume that the wrapped module returned
            # only the output tensor. We set the variable y to the output tensor, and set the attribute `_hidden`
            # as None to indicate that there was no hidden state received.
            y = out
            self._hidden = None

        # We return y, which stores the output received by the wrapped module.
        return y

    def reset(self):
        """
        Reset the hidden state, if any.
        """
        self._hidden = None

__init__(wrapped_module)

__init__(...): Initialize the StatefulModule.

Parameters:

Name	Type	Description	Default
wrapped_module	Module	The torch.nn.Module instance to wrap.	required

Source code in evotorch/neuroevolution/net/statefulmodule.py

def __init__(self, wrapped_module: nn.Module):
    """
    `__init__(...)`: Initialize the StatefulModule.

    Args:
        wrapped_module: The `torch.nn.Module` instance to wrap.
    """
    super().__init__()

    # Declare the variable that will store the hidden state of wrapped_module, if any.
    self._hidden: Any = None

    # Store the module that is wrapped.
    self.wrapped_module = wrapped_module

reset()

Reset the hidden state, if any.

Source code in evotorch/neuroevolution/net/statefulmodule.py

def reset(self):
    """
    Reset the hidden state, if any.
    """
    self._hidden = None

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

Source code in evotorch/neuroevolution/net/misc.py

def count_parameters(net: nn.Module) -> int:
    """
    Get the number of parameters the network.

    Args:
        net: The torch module whose parameters will be counted.
    Returns:
        The number of parameters, as an integer.
    """

    count = 0

    for p in net.parameters():
        count += p.numel()

    return count

device_of_module(m, default=None)

Get the device in which the module exists.

This function looks at the first parameter of the module, and returns its device. This function is not meant to be used on modules whose parameters exist on different devices.

Parameters:

Name	Type	Description	Default
m	Module	The module whose device is being queried.	required
default	Optional[Union[str, device]]	The fallback device to return if the module has no parameters. If this is left as None, the fallback device is assumed to be "cpu".	None

Source code in evotorch/neuroevolution/net/misc.py

def device_of_module(m: nn.Module, default: Optional[Union[str, torch.device]] = None) -> torch.device:
    """
    Get the device in which the module exists.

    This function looks at the first parameter of the module, and returns
    its device. This function is not meant to be used on modules whose
    parameters exist on different devices.

    Args:
        m: The module whose device is being queried.
        default: The fallback device to return if the module has no
            parameters. If this is left as None, the fallback device
            is assumed to be "cpu".
    Returns:
        The device of the module, determined from its first parameter.
    """
    if default is None:
        default = torch.device("cpu")

    device = default

    for p in m.parameters():
        device = p.device
        break

    return device

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")

make_functional_module(net, *, disable_autograd_tracking=False)

Wrap a torch module so that it has a functional interface.

Similar to functorch.make_functional(...), this function turns a torch.nn.Module instance to a function which expects a new leftmost argument representing the parameters of the network. Unlike with functorch.make_functional(...), the parameters of the network are expected in a 1-dimensional (i.e. flattened) tensor.

PyTorch modules with buffers can be wrapped by this class, but it is assumed that those buffers are constant. If the wrapped module changes the value(s) of its buffer(s) during its forward passes, most probably things will NOT work right.

As an example, let us consider the following linear layer.

import torch
from torch import nn

net = nn.Linear(3, 8)

The functional counterpart of net can be obtained via:

from evotorch.neuroevolution.net import make_functional_module

fnet = make_functional_module(net)

Now, fnet is a callable object which expects network parameters and network inputs. Let us call fnet with randomly generated network parameters and with a randomly generated input tensor.

param_length = fnet.parameter_length
random_parameters = torch.randn(param_length)
random_input = torch.randn(3)

result = fnet(random_parameters, random_input)

Parameters:

Name	Type	Description	Default
net	Module	The torch.nn.Module instance to be wrapped by a functional interface.	required
disable_autograd_tracking	bool	If given as True, all operations regarding the wrapped module will be performed in the context torch.no_grad(), forcefully disabling the autograd. If given as False, autograd will not be affected. The default is False.	False

Source code in evotorch/neuroevolution/net/functional.py

def make_functional_module(net: nn.Module, *, disable_autograd_tracking: bool = False) -> ModuleExpectingFlatParameters:
    """
    Wrap a torch module so that it has a functional interface.

    Similar to `functorch.make_functional(...)`, this function turns a
    `torch.nn.Module` instance to a function which expects a new leftmost
    argument representing the parameters of the network.
    Unlike with `functorch.make_functional(...)`, the parameters of the
    network are expected in a 1-dimensional (i.e. flattened) tensor.

    PyTorch modules with buffers can be wrapped by this class, but it is
    assumed that those buffers are constant. If the wrapped module changes
    the value(s) of its buffer(s) during its forward passes, most probably
    things will NOT work right.

    As an example, let us consider the following linear layer.

    ```python
    import torch
    from torch import nn

    net = nn.Linear(3, 8)
    ```

    The functional counterpart of `net` can be obtained via:

    ```python
    from evotorch.neuroevolution.net import make_functional_module

    fnet = make_functional_module(net)
    ```

    Now, `fnet` is a callable object which expects network parameters
    and network inputs. Let us call `fnet` with randomly generated network
    parameters and with a randomly generated input tensor.

    ```python
    param_length = fnet.parameter_length
    random_parameters = torch.randn(param_length)
    random_input = torch.randn(3)

    result = fnet(random_parameters, random_input)
    ```

    Args:
        net: The `torch.nn.Module` instance to be wrapped by a functional
            interface.
        disable_autograd_tracking: If given as True, all operations
            regarding the wrapped module will be performed in the context
            `torch.no_grad()`, forcefully disabling the autograd.
            If given as False, autograd will not be affected.
            The default is False.
    Returns:
        The functional wrapper, as an instance of
        `evotorch.neuroevolution.net.ModuleExpectingFlatParameters`.
    """
    return ModuleExpectingFlatParameters(net, disable_autograd_tracking=disable_autograd_tracking)

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	Optional[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

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)

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
from evotorch.neuroevolution.net import MultiLayered

net = MultiLayered(nn.Linear(8, 16), nn.Tanh(), nn.Linear(16, 4, bias=False), nn.ReLU())

By using str_to_net(...) one can construct an equivalent 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:

net = str_to_net(
    '''
    Linear(8, 16)
    >> Tanh()
    >> Linear(16, 4, bias=False)
    >> ReLU()
    '''
)

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.

The namespace evotorch.neuroevolution.net.layers contains its own implementations for RNN and LSTM. These recurrent layer implementations work similarly to their counterparts torch.nn.RNN and torch.nn.LSTM, except that EvoTorch's implementations do not expect the data with extra leftmost dimensions for batching and for timesteps. Instead, they expect to receive a single input and a single current hidden state, and produce a single output and a single new hidden state. These recurrent layer implementations of EvoTorch can be used within a neural net structure string. Therefore, the following examples are valid:

rnn1 = str_to_net("RNN(4, 8) >> Linear(8, 2)")

rnn2 = str_to_net(
    '''
    Linear(4, 10)
    >> Tanh()
    >> RNN(input_size=10, hidden_size=24, nonlinearity='tanh'
    >> Linear(24, 2)
    '''
)

lstm1 = str_to_net("LSTM(4, 32) >> Linear(32, 2)")

lstm2 = str_to_net("LSTM(input_size=4, hidden_size=32) >> Linear(32, 2)")

Notes regarding usage with evotorch.neuroevolution.GymNE or with evotorch.neuroevolution.VecGymNE:

While instantiating a GymNE or a VecGymNE, 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 and VecGymNE define 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 instantiating a GymNE or a VecGymNE, one can define a single-hidden-layered policy via this string:

"Linear(obs_length, 16) >> Tanh() >> Linear(16, act_length) >> Tanh()"

In the policy string above, one might choose to omit the last Tanh(), as GymNE and VecGymNE will clip the final output of the policy to conform to the action boundaries defined by the target reinforcement learning environment, and such a clipping operation might be seen as using an activation function similar to hard-tanh anyway.

Parameters:

Name	Type	Description	Default
s	str	The string which expresses the neural net structure.	required

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
    from evotorch.neuroevolution.net import MultiLayered

    net = MultiLayered(nn.Linear(8, 16), nn.Tanh(), nn.Linear(16, 4, bias=False), nn.ReLU())
    ```

    By using `str_to_net(...)` one can construct an equivalent
    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`.

    The namespace `evotorch.neuroevolution.net.layers` contains its own
    implementations for RNN and LSTM. These recurrent layer implementations
    work similarly to their counterparts `torch.nn.RNN` and `torch.nn.LSTM`,
    except that EvoTorch's implementations do not expect the data with extra
    leftmost dimensions for batching and for timesteps. Instead, they expect
    to receive a single input and a single current hidden state, and produce
    a single output and a single new hidden state. These recurrent layer
    implementations of EvoTorch can be used within a neural net structure
    string. Therefore, the following examples are valid:

    ```python
    rnn1 = str_to_net("RNN(4, 8) >> Linear(8, 2)")

    rnn2 = str_to_net(
        '''
        Linear(4, 10)
        >> Tanh()
        >> RNN(input_size=10, hidden_size=24, nonlinearity='tanh'
        >> Linear(24, 2)
        '''
    )

    lstm1 = str_to_net("LSTM(4, 32) >> Linear(32, 2)")

    lstm2 = str_to_net("LSTM(input_size=4, hidden_size=32) >> Linear(32, 2)")
    ```

    **Notes regarding usage with `evotorch.neuroevolution.GymNE`
    or with `evotorch.neuroevolution.VecGymNE`:**

    While instantiating a `GymNE` or a `VecGymNE`, 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` and `VecGymNE` define
    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 instantiating a `GymNE` or a `VecGymNE`, one can define a
    single-hidden-layered policy via this string:

    ```
    "Linear(obs_length, 16) >> Tanh() >> Linear(16, act_length) >> Tanh()"
    ```

    In the policy string above, one might choose to omit the last `Tanh()`, as
    `GymNE` and `VecGymNE` will clip the final output of the policy to conform
    to the action boundaries defined by the target reinforcement learning
    environment, and such a clipping operation might be seen as using an
    activation function similar to 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)