acds.archetypes package

Submodules

acds.archetypes.esn module

class acds.archetypes.esn.DeepReservoir(input_size: int = 1, tot_units: int = 100, n_layers: int = 1, concat: bool = False, input_scaling: float = 1.0, inter_scaling: float = 1.0, spectral_radius: float = 0.99, leaky: float = 1.0, connectivity_recurrent: int = 10, connectivity_input: int = 10, connectivity_inter: int = 10)

Bases: Module

Deep Reservoir to be used as Recurrent Neural Network.

The implementation realizes a number of stacked RNN layers using the ReservoirCell as core cell. All the reservoir layers share the same hyper-parameter values (i.e., same number of recurrent neurons, spectral radius, etc..).

forward(X: Tensor)

Forward pass.

Parameters:

X (torch.Tensor) – Input tensor, shaped as (batch, seq_len, n_inp).

Returns:

Output tensor, shaped as (batch, seq_len, n_out).

Return type:

torch.Tensor

class acds.archetypes.esn.ReservoirCell(input_size: int, units: int, input_scaling: float = 1.0, spectral_radius: float = 0.99, leaky: float = 1.0, connectivity_input: int = 10, connectivity_recurrent: int = 10)

Bases: Module

Shallow reservoir to be used as cell of a Recurrent Neural Network. The equation of the reservoir is given by:

\[h_t = (1 - \alpha) h_{t-1} + \alpha \tanh(W_{in} x_t + W_{rec} h_{t-1} + b)\]

where: - \(h_t\) is the hidden state at time t, - \(x_t\) is the input at time t, - \(W_{in}\) is the input weight matrix, - \(W_{rec}\) is the recurrent weight matrix, - \(b\) is the bias, - \(\alpha\) is the leaking rate.

The implementation is derivated from the one in https://github.com/gallicch/DeepRC-TF/blob/master/DeepRC.py

If you use this code in your work, please cite the following paper, in which the concept of Deep Reservoir Computing has been introduced:

Gallicchio, C., Micheli, A., Pedrelli, L.: Deep reservoir computing: A critical experimental analysis. Neurocomputing268, 87-99 (2017). https://doi.org/10.1016/j.neucom.2016.12.08924.

forward(xt: Tensor, h_prev: Tensor)

Computes the output of the cell given the input and previous state.

Parameters:

• xt (torch.Tensor) – input tensor shaped as (batch, time, input_dim).

• h_prev (torch.Tensor) – previous state tensor shaped as (batch, time, state_dim).

Returns:

hidden state tensor shaped as (batch, time, state_dim). torch.Tensor: hidden state tensor shaped as (batch, time, state_dim).

Return type:

torch.Tensor

class acds.archetypes.esn.ReservoirLayer(input_size: int, units: int, input_scaling: float = 1.0, spectral_radius: float = 0.99, leaky: float = 1.0, connectivity_input: int = 10, connectivity_recurrent: int = 10)

Bases: Module

Shallow reservoir to be used as Recurrent Neural Network layer.

The layer is composed by a number of ReservoirCell, each of which is used to process the input and the previous state at each time step.

forward(x: Tensor, h_prev: Tensor | None = None)

Computes the output of the cell given the input and previous state.

Parameters:

• x (torch.Tensor) – input tensor shaped as (batch, time, input_dim).

• h_prev (torch.Tensor) – previous state tensor shaped as (batch, time, state_dim). If None, the hidden state is initialized to zeros. Defaults to None.

Returns:

hidden state tensor shaped as (batch, time, state_dim). torch.Tensor: hidden state tensor shaped as (batch, time, state_dim).

Return type:

torch.Tensor

init_hidden(batch_size: int)

Initializes the hidden state to zeros.

Parameters:

batch_size (int) – size of the batch.

Returns:

hidden state tensor shaped as (batch_size, state_dim).

Return type:

torch.Tensor

acds.archetypes.hcornn module

class acds.archetypes.hcornn.hcoRNN(n_inp, n_hid, dt, gamma, epsilon, device='cpu', matrix_friction=False, train_oscillators=False)

Bases: Module

cell(x, hy, hz)

forward(x)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

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

acds.archetypes.pron module

class acds.archetypes.pron.MultistablePhysicallyImplementableRandomizedOscillatorsNetwork(n_inp, n_hid, dt, gamma, epsilon, input_scaling, device='cpu')

Bases: Module

cell(x, hy, hz)

forward(x)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

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

class acds.archetypes.pron.PhysicallyImplementableRandomizedOscillatorsNetwork(n_inp, n_hid, dt, gamma, epsilon, input_scaling, device='cpu', input_function='linear', matrix_friction=False)

Bases: Module

Batch-first (BW1, L, I)

cell(x, hy, hz)

forward(x)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

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

acds.archetypes.pron.get_input_fn(fn_type, n_inp, n_hid, input_scaling, device)

acds.archetypes.rnn module

class acds.archetypes.rnn.GRU_DFA(input_size, hidden_size, output_size, grad_clip=5, device='cpu', truncation=None)

Bases: RNN_DFA

compute_update(last_hidden, dW, dV, db, error)

forward(x, y=None)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

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

class acds.archetypes.rnn.LSTM(n_inp: int, n_hid: int, n_out: int, gru: bool = False, rnn: bool = False)

Bases: Module

LSTM model with a readout layer.

forward(x: Tensor) → Tensor

Forward pass.

Parameters:

x (torch.Tensor) – Input tensor, shaped as (batch, seq_len, n_inp).

Returns:

Output tensor, shaped as (batch, n_out).

Return type:

torch.Tensor

class acds.archetypes.rnn.RNN_DFA(input_size, hidden_size, output_size, grad_clip=5, device='cpu', truncation=None)

Bases: Module

compute_update(last_hidden, dW, dV, db, error)

forward(x, y=None)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

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

initHidden(batch_size)

acds.archetypes.ron module

class acds.archetypes.ron.RandomizedOscillatorsNetwork(n_inp: int, n_hid: int, dt: float, gamma: float | Tuple[float, float], epsilon: float | Tuple[float, float], diffusive_gamma=0.0, rho: float = 0.99, input_scaling: float = 1.0, topology: Literal['full', 'lower', 'orthogonal', 'band', 'ring', 'toeplitz', 'antisymmetric'] = 'full', reservoir_scaler=0.0, sparsity=0.0, device='cpu')

Bases: Module

Randomized Oscillators Network. A recurrent neural network model with oscillatory dynamics. The model is defined by the following ordinary differential equation:

\[\dot{h} = -\gamma h - \epsilon \dot{h} + \tanh(W_{in} x + W_{rec} h + b)\]

where: - \(h\) is the hidden state, - \(\dot{h}\) is the derivative of the hidden state, - \(\gamma\) is the damping factor, - \(\epsilon\) is the stiffness factor, - \(W_{in}\) is the input-to-hidden weight matrix, - \(W_{rec}\) is the hidden-to-hidden weight matrix, - \(b\) is the bias vector.

The model is trained by minimizing the mean squared error between the output of the model and the target time-series.

cell(x: Tensor, hy: Tensor, hz: Tensor) → Tuple[Tensor, Tensor]

Compute the next hidden state and its derivative.

Parameters:

• x (torch.Tensor) – Input tensor.

• hy (torch.Tensor) – Current hidden state.

• hz (torch.Tensor) – Current hidden state derivative.

forward(x: Tensor, hs: Tuple[Tensor, Tensor] | None = None) → Tuple[Tensor, List[Tensor]]

Forward pass on a given input time-series.

Parameters:

• x (torch.Tensor) – Input time-series shaped as (batch, time, input_dim).

• hs (tuple, optional) – Tuple containing the initial hidden states. If None, both are initialized to zero. Defaults to None.

Returns:

Hidden states of the network shaped as (batch, time, n_hid). list: List containing the last hidden state of the network.

Return type:

torch.Tensor

acds.archetypes.run module

class acds.archetypes.run.UnicycleNetwork(n_inp, n_units, dt, lin_stiff_min=0.1, lin_stiff_max=0.5, ang_stiff_min=0.1, ang_stiff_max=0.3, lin_damping_min=0.1, lin_damping_max=0.2, ang_damping_min=0.1, ang_damping_max=0.2, eq_dist_min=0.5, eq_dist_max=1.0, eq_dist_min_ang=0.0, eq_dist_max_ang=3.141592653589793, lin_input_map=None, ang_input_map=None, n_connections=None, n_connections_anchor=2, n_connections_ang=None, n_connections_anchor_ang=2)

Bases: Module

angle_to_unit_vector(angle)

forward(u_lin, u_ang, x, z, theta, s, omega)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

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

pairwise_differences(arr)

Compute pairwise differences between batched vectors.

Parameters:

arr – Tensor with shape (batch, n_vectors, n_dims).

Returns:

Tensor with shape (batch, n_vectors, n_vectors, n_dims) where [:, i, j, :] is the difference between vectors i and j.

set_eq_distances_from_positions(x, z)

Set equilibrium distances based on actual distances between connected robots. Only updates springs that have non-zero stiffness (i.e., actual connections).

Parameters:

• x – numpy array or torch tensor of x positions, shape (n_units,)

• z – numpy array or torch tensor of z positions, shape (n_units,)

class acds.archetypes.run.UnicycleReservoir(n_inp, n_units, dt, n_out, lin_stiff_min=0.1, lin_stiff_max=0.5, ang_stiff_min=0.1, ang_stiff_max=0.3, lin_damping_min=0.1, lin_damping_max=0.2, ang_damping_min=0.1, ang_damping_max=0.2, eq_dist_min=0.5, eq_dist_max=1.0, eq_dist_min_ang=0.0, eq_dist_max_ang=3.141592653589793, lin_input_map=None, ang_input_map=None, n_connections=None, inp_bias=0, n_connections_anchor=2, n_connections_ang=None, n_connections_anchor_ang=2, n_past_steps_readout=0)

Bases: Module

forward(u_lin, u_ang)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

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

acds.archetypes.trainable_pron module

class acds.archetypes.trainable_pron.TrainedPhysicallyImplementableRandomizedOscillatorsNetwork(n_inp, n_hid, dt, diffusive_gamma=0.0, gamma=1.0, epsilon=1.0, device='cpu', matrix_friction=False, train_oscillators=False, train_recurrent=True, topology: Literal['orthogonal', 'antisymmetric'] = 'orthogonal')

Bases: Module

cell(x, hy, hz)

forward(x)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

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

acds.archetypes.utils module

acds.archetypes.utils.antisymmetric_matrix(W: FloatTensor) → FloatTensor

Transforms W to have an antisymmetric matrix

Parameters:

W (torch.FloatTensor) – input matrix to be transformed

Returns:

transformed matrix

Return type:

torch.FloatTensor

acds.archetypes.utils.count_parameters(model)

Return total number of parameters and trainable parameters of a PyTorch model.

acds.archetypes.utils.get_hidden_topology(n_hid: int, topology: Literal['full', 'lower', 'orthogonal', 'band', 'ring', 'toeplitz', 'antisymmetric'], sparsity: float, scaler: float) → FloatTensor

Generates the hidden-to-hidden weight matrix according to the specified topology and sparsity.

Parameters:

• n_hid (int) – number of hidden units.

• topology (str) – topology of the hidden-to-hidden weight matrix. Options are ‘full’, ‘lower’, ‘orthogonal’, ‘band’, ‘ring’, ‘toeplitz’, ‘antisymmetric’.

• sparsity (float) – sparsity of the hidden-to-hidden weight matrix.

• scaler (float) – scaling factor for the hidden-to-hidden weight matrix.

Returns:

hidden-to-hidden weight matrix.

Return type:

torch.Tensor

acds.archetypes.utils.sparse_eye_init(M: int) → FloatTensor

Generates an M x M matrix to be used as sparse identity matrix for the re-scaling of the sparse recurrent kernel in presence of non-zero leakage. The neurons are connected according to a ring topology, where each neuron receives input only from one neuron and propagates its activation only to one other neuron. All the non-zero elements are set to 1.

Parameters:

M (int) – number of hidden units.

Returns:

MxM identity matrix.

Return type:

torch.FloatTensor

acds.archetypes.utils.sparse_recurrent_tensor_init(M: int, C: int = 1) → FloatTensor

Generates an M x M matrix to be used as sparse recurrent kernel. For each column only C elements are non-zero (i.e., each recurrent neuron take sinput from C other recurrent neurons). The non-zero elements are generated randomly from a uniform distribution in [-1,1].

Parameters:

• M (int) – number of hidden units

• C (int) – number of nonzero elements

Returns:

MxM dense matrix

Return type:

torch.FloatTensor

acds.archetypes.utils.sparse_tensor_init(M: int, N: int, C: int = 1) → FloatTensor

Generates an M x N matrix to be used as sparse (input) kernel For each row only C elements are non-zero (i.e., each input dimension is projected only to C neurons). The non-zero elements are generated randomly from a uniform distribution in [-1,1]

Parameters:

• M (int) – number of hidden units

• N (int) – number of input units

• C (int) – number of nonzero elements

Returns:

MxN dense matrix

Return type:

torch.FloatTensor

acds.archetypes.utils.spectral_norm_scaling(W: FloatTensor, rho_desired: float) → FloatTensor

Rescales W to have rho(W) = rho_desired .

Parameters:

• W (torch.FloatTensor) – input matrix to be rescaled

• rho_desired (float) – desired spectral radius

Returns:

rescaled matrix

Return type:

torch.FloatTensor

acds.archetypes.vdpon module

class acds.archetypes.vdpon.VanDerPolOscillatorsNetwork(n_inp: int, n_hid: int, dt: float, gamma: float | Tuple[float, float], epsilon: float | Tuple[float, float], diffusive_gamma=0.0, rho: float = 0.99, input_scaling: float = 1.0, topology: Literal['full', 'lower', 'orthogonal', 'band', 'ring', 'toeplitz', 'antisymmetric'] = 'full', reservoir_scaler=0.0, sparsity=0.0, device='cpu')

Bases: Module

Van der Pol Oscillators Network. A network model with Van der Pol’s coupled oscillators, where each node is a nonlinear system defined with the Van der Pol’s equation:

\[\ddot{x} - \mu (1 - x^2) \dot{x} + x = 0\]

The model is defined by the following ordinary differential equation:

\[\dot{h} = -\gamma h - \epsilon (1-h^2) \dot{h} + \tanh(W_{in} x + W_{rec} h + b)\]

where: - \(h\) is the hidden state, - \(\dot{h}\) is the derivative of the hidden state, - \(\gamma\) is the damping factor, - \(\epsilon\) is the stiffness factor, - \(W_{in}\) is the input-to-hidden weight matrix, - \(W_{rec}\) is the hidden-to-hidden weight matrix, - \(b\) is the bias vector.

The model is trained by minimizing the mean squared error between the output of the model and the target time-series.

cell(x: Tensor, hy: Tensor, hz: Tensor) → Tuple[Tensor, Tensor]

Compute the next hidden state and its derivative.

Parameters:

• x (torch.Tensor) – Input tensor.

• hy (torch.Tensor) – Current hidden state.

• hz (torch.Tensor) – Current hidden state derivative.

forward(x: Tensor, hs: Tuple[Tensor, Tensor] | None = None) → Tuple[Tensor, List[Tensor]]

Forward pass on a given input time-series.

Parameters:

• x (torch.Tensor) – Input time-series shaped as (batch, time, input_dim).

• hs (tuple, optional) – Tuple containing the initial hidden states. If None, both are initialized to zero. Defaults to None.

Returns:

Hidden states of the network shaped as (batch, time, n_hid). list: List containing the last hidden state of the network.

Return type:

torch.Tensor

Module contents

class acds.archetypes.DeepReservoir(input_size: int = 1, tot_units: int = 100, n_layers: int = 1, concat: bool = False, input_scaling: float = 1.0, inter_scaling: float = 1.0, spectral_radius: float = 0.99, leaky: float = 1.0, connectivity_recurrent: int = 10, connectivity_input: int = 10, connectivity_inter: int = 10)

Bases: Module

Deep Reservoir to be used as Recurrent Neural Network.

The implementation realizes a number of stacked RNN layers using the ReservoirCell as core cell. All the reservoir layers share the same hyper-parameter values (i.e., same number of recurrent neurons, spectral radius, etc..).

forward(X: Tensor)

Forward pass.

Parameters:

X (torch.Tensor) – Input tensor, shaped as (batch, seq_len, n_inp).

Returns:

Output tensor, shaped as (batch, seq_len, n_out).

Return type:

torch.Tensor

class acds.archetypes.GRU_DFA(input_size, hidden_size, output_size, grad_clip=5, device='cpu', truncation=None)

Bases: RNN_DFA

compute_update(last_hidden, dW, dV, db, error)

forward(x, y=None)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

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

class acds.archetypes.LSTM(n_inp: int, n_hid: int, n_out: int, gru: bool = False, rnn: bool = False)

Bases: Module

LSTM model with a readout layer.

forward(x: Tensor) → Tensor

Forward pass.

Parameters:

x (torch.Tensor) – Input tensor, shaped as (batch, seq_len, n_inp).

Returns:

Output tensor, shaped as (batch, n_out).

Return type:

torch.Tensor

class acds.archetypes.MultistablePhysicallyImplementableRandomizedOscillatorsNetwork(n_inp, n_hid, dt, gamma, epsilon, input_scaling, device='cpu')

Bases: Module

cell(x, hy, hz)

forward(x)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

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

class acds.archetypes.PhysicallyImplementableRandomizedOscillatorsNetwork(n_inp, n_hid, dt, gamma, epsilon, input_scaling, device='cpu', input_function='linear', matrix_friction=False)

Bases: Module

Batch-first (BW1, L, I)

cell(x, hy, hz)

forward(x)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

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

class acds.archetypes.RNN_DFA(input_size, hidden_size, output_size, grad_clip=5, device='cpu', truncation=None)

Bases: Module

compute_update(last_hidden, dW, dV, db, error)

forward(x, y=None)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

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

initHidden(batch_size)

class acds.archetypes.RandomizedOscillatorsNetwork(n_inp: int, n_hid: int, dt: float, gamma: float | Tuple[float, float], epsilon: float | Tuple[float, float], diffusive_gamma=0.0, rho: float = 0.99, input_scaling: float = 1.0, topology: Literal['full', 'lower', 'orthogonal', 'band', 'ring', 'toeplitz', 'antisymmetric'] = 'full', reservoir_scaler=0.0, sparsity=0.0, device='cpu')

Bases: Module

Randomized Oscillators Network. A recurrent neural network model with oscillatory dynamics. The model is defined by the following ordinary differential equation:

\[\dot{h} = -\gamma h - \epsilon \dot{h} + \tanh(W_{in} x + W_{rec} h + b)\]

where: - \(h\) is the hidden state, - \(\dot{h}\) is the derivative of the hidden state, - \(\gamma\) is the damping factor, - \(\epsilon\) is the stiffness factor, - \(W_{in}\) is the input-to-hidden weight matrix, - \(W_{rec}\) is the hidden-to-hidden weight matrix, - \(b\) is the bias vector.

The model is trained by minimizing the mean squared error between the output of the model and the target time-series.

cell(x: Tensor, hy: Tensor, hz: Tensor) → Tuple[Tensor, Tensor]

Compute the next hidden state and its derivative.

Parameters:

• x (torch.Tensor) – Input tensor.

• hy (torch.Tensor) – Current hidden state.

• hz (torch.Tensor) – Current hidden state derivative.

forward(x: Tensor, hs: Tuple[Tensor, Tensor] | None = None) → Tuple[Tensor, List[Tensor]]

Forward pass on a given input time-series.

Parameters:

• x (torch.Tensor) – Input time-series shaped as (batch, time, input_dim).

• hs (tuple, optional) – Tuple containing the initial hidden states. If None, both are initialized to zero. Defaults to None.

Returns:

Hidden states of the network shaped as (batch, time, n_hid). list: List containing the last hidden state of the network.

Return type:

torch.Tensor

class acds.archetypes.TrainedPhysicallyImplementableRandomizedOscillatorsNetwork(n_inp, n_hid, dt, diffusive_gamma=0.0, gamma=1.0, epsilon=1.0, device='cpu', matrix_friction=False, train_oscillators=False, train_recurrent=True, topology: Literal['orthogonal', 'antisymmetric'] = 'orthogonal')

Bases: Module

cell(x, hy, hz)

forward(x)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

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

class acds.archetypes.UnicycleReservoir(n_inp, n_units, dt, n_out, lin_stiff_min=0.1, lin_stiff_max=0.5, ang_stiff_min=0.1, ang_stiff_max=0.3, lin_damping_min=0.1, lin_damping_max=0.2, ang_damping_min=0.1, ang_damping_max=0.2, eq_dist_min=0.5, eq_dist_max=1.0, eq_dist_min_ang=0.0, eq_dist_max_ang=3.141592653589793, lin_input_map=None, ang_input_map=None, n_connections=None, inp_bias=0, n_connections_anchor=2, n_connections_ang=None, n_connections_anchor_ang=2, n_past_steps_readout=0)

Bases: Module

forward(u_lin, u_ang)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

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

class acds.archetypes.hcoRNN(n_inp, n_hid, dt, gamma, epsilon, device='cpu', matrix_friction=False, train_oscillators=False)

Bases: Module

cell(x, hy, hz)

forward(x)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

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