Module dlkoopman.traj_pred

Trajectory Predictor.

TrajPred can be used to train on given equal-length trajectories of a system, then predict unknown trajectories of the system starting from new initial states. See a specific example and tutorial here.

Classes

class TrajPredDataHandler (Xtr, Xva=None, Xte=None, cfg=None)

Trajectory predictor data handler. Used to provide data to train (and optionally validate and test) the TrajPred model.

Parameters

• Xtr (Array[float], shape=(num_training_trajectories, num_indexes, input_size)) - Input trajectories to be used as training data. Array can be any data type such as numpy.array, torch.Tensor, list etc.

• Xva (Array[float], shape=(num_validation_trajectories, num_indexes, input_size), optional) - Input trajectories to be used as validation data. Same data type requirements as Xtr.

• Xte (Array[float], shape=(num_test_trajectories, num_indexes, input_size), optional) - Input trajectories to be used as test data. Same data type requirements as Xtr.

• cfg (dlkoopan.config.Config, optional) - Leave this as None to use the default configuration options. If changing any configuration option(s) is desired, create a separate Config instance and pass that as an argument (see example below).

Example

# Provide data of a system with 2-dimensional states (i.e. input_size=3)
# containing trajectories of length 4 (i.e. num_indexes=3)
# Provide 3 trajectories for training, 1 for validation, and none for testing
# Use a custom configuration with double data types on CPU only

from dlkoopman.config import Config
cfg = Config(precision="double", use_cuda=False)

dh = TrajPredDataHandler(
    Xtr = numpy.array([
        [ # 1st trajectory
            [8.4, 6.7],
            [4.2, 4.2],
            [3.4, 4.6],
            [7.7, 1.8]
        ],
        [ # 2nd trajectory
            [5.9, 1.5],
            [8.3, 5.1],
            [9.9, 0.2],
            [1.6, 7.2]
        ],
        [ # 3rd trajectory
            [5.4, 9.8],
            [2.4, 6.9],
            [8.9, 0.5],
            [9.3, 0.7]
        ]
    ]),
    Xva = numpy.array([
        [ # 1st trajectory
            [4.7, 1.4],
            [1.1, 6.2],
            [7.8, 6.9],
            [6.1, 8. ]
        ]
    ]),
    cfg = cfg
)

class TrajPred (dh, encoded_size, encoder_hidden_layers=[100], decoder_hidden_layers=[], batch_norm=False)

Trajectory predictor. Used to train on given equal-length trajectories of a system, then predict unknown trajectories of the system starting from new initial states.

Parameters

• dh (TrajPredDataHandler) - Data handler that feeds data. Configuration options of a TrajPred instance are identical to dh.cfg.

• Parameters for AutoEncoder:

  • encoded_size (int).

  • encoder_hidden_layers (list[int], optional).

  • decoder_hidden_layers (list[int], optional).

  • batch_norm (bool, optional).

Attributes

• uuid (str) - Unique ID assigned to this instance. Results will include uuid in their filename.

• log_file (Path) - Path to log file = ./log_<uuid>.log.

• input_size (int) - Dimensionality of input states. Inferred from dh.Xtr.

• encoded_size (int) - Dimensionality of encoded states. As given in input.

• ae (nets.AutoEncoder) - AutoEncoder neural network to encode input states into a linearizable domain where the Koopman matrix can be learnt, then decode them back into original domain.

• Knet (nets.Knet) - Linear layer to approximate the Koopman matrix. This is used to evolve states in the encoded domain so as to generate their trajectories.

• Lambda (torch.Tensor), eigvecs (torch.Tensor) - Eigenvalues, and eigenvectors of the trained Koopman matrix that characterizes the discrete index system $y_{i+1} = Ky_i$. The system is discrete since specific trajectory indexes are not provided, so they are always assumed to be $[0,1,2,\cdots]$. The eigendecomposition is not used in computations since the trained Knet layer performs all predictions, but is still calculated to characterize the system.

• stats (dict[list]) - Stores different metrics from training and testing. Useful for checking performance and plotting.

• error_flag (bool) - Signals if any error has occurred in training.

Methods

def train_net(self, numepochs=10, batch_size=250, early_stopping=0, early_stopping_metric='total_loss', lr=0.001, weight_decay=0.0, decoder_loss_weight=0.01, clip_grad_norm=None, clip_grad_value=None)

Train the model using dh.Xtr, and validate it on dh.Xva.

Parameters

• numepochs (int, optional) - Number of epochs for which to train. Each epoch uses the complete training data to learn the Koopman matrix.

• batch_size (int, optional) - How many trajectories to train on per batch. Set to 0 to train on all trajectories per batch (not recommended when number of training trajectories is large).

• early_stopping (int/float, optional) - Whether to terminate training early due to no improvement in validation metric.

  • If 0, no early stopping. The model will run for the complete numepochs epochs.
  • If an integer, early stop if validation metric doesn't improve for that many epochs.
  • If a float, early stop if validation performance doesn't improve for that fraction of epochs, rounded up.

• early_stopping_metric (str, optional) - Which validation metric to use for early stopping. Ignored if early_stopping=False. Possible metrics are '<recon/lin/pred/total>_loss', and '<recon/lin/pred>_anae'.

• lr (float, optional) - Learning rate for optimizer.

• weight_decay (float, optional) - L2 coefficient for weights of the neural nets.

• decoder_loss_weight (float, optional) - Weight the losses between autoencoder decoder outputs (recon and pred) by this number. This is to account for the scaling effect of the decoder.

• cond_threshold (float, optional) - Condition number of the eigenvector matrix greater than this will be reported, and singular values smaller than this fraction of the largest will be ignored for the pseudo-inverse operation.

• clip_grad_norm (float, optional) - If not None, clip the norm of gradients to this value.

• clip_grad_value (float, optional) - If not None, clip the values of gradients to [-clip_grad_value,clip_grad_value].

Effects

• self.stats is populated.

def test_net(self)

Run the trained model on test data - dh.Xte.

Effects

• self.stats is populated further.

def predict_new(self, X0) ‑> torch.Tensor

Use the trained model to predict complete trajectories for new starting states.

This is different from testing because the ground truth values are not present, thus losses and errors are not computed.

Parameters

• 'X0' (Array[float], shape=(num_new_trajectories, input_size)) - The starting states for the new trajectories that are to be predicted. Array can be any data type such as numpy.array, torch.Tensor, list, range, etc.

Returns

Xpred (torch.Tensor, shape=(num_new_trajectories, num_indexes, input_size)) - Predicted trajectories for the new starting states.