Fit systems

class deepSI.fit_systems.System_fittable(action_space=None, observation_space=None)

Subclass of system which introduces a .fit method which calls ._fit to fit the systems

Notes

This function will automaticly fit the normalization in self.norm if auto_fit_norm is set to True (default). Lastly it will set self.init_model_done to True which will keep the norm constant.

init_model(sys_data=None, nu=-1, ny=-1, auto_fit_norm=True)
fit(train_sys_data, auto_fit_norm=True, **kwargs)
_fit(normed_sys_data, **kwargs)
__module__ = 'deepSI.fit_systems.fit_system'
class deepSI.fit_systems.System_torch(action_space=None, observation_space=None)

For systems that utilize torch

parameters

The list of fittable network parameters returned by System_torch.init_nets(nu,ny)

Type:

list

optimizer

The main optimizer returned by System_torch.init_optimizer

Type:

torch.Optimizer

time

Current runtime after each epoch

Type:

numpy.ndarray

batch_id

Current total number of batch optimization steps is saved after each epoch

Type:

numpy.ndarray

Loss_train

Average training loss for each epoch

Type:

numpy.ndarray

Loss_val

Validation loss for each epoch

Type:

numpy.ndarray

Notes

subclasses should define three methods (i) init_nets(nu, ny) which returns the network parameters, (ii) make_training_data(sys_data, **loss_kwargs)` which converts the normed sys_data into training data (list of numpy arrays), (iii) loss(*training_data, **loss_kwargs) which returns the loss using the current training data

init_nets(nu, ny)

Defined in subclass and initializes networks and returns the parameters

Parameters:

  • nu (None, int or tuple) – The shape of the input u

  • ny (None, int or tuple) – The shape of the output y

loss(**loss_kwargs)

Defined in subclass which take the batch data and calculates the loss based on loss_kwargs

Parameters:

  • training_data_batch (list) – batch of the training data returned by make_training_data and converted to torch arrays

  • loss_kwargs (dict) – loss function settings passed into .fit

init_optimizer(parameters, **optimizer_kwargs)

Optionally defined in subclass to create the optimizer

Parameters:

  • parameters (list or list of dict) – system torch parameters

  • optimizer_kwargs (dict) – If ‘optimizer’ is defined than that optimizer will be used otherwise Adam will be used. The other parameters will be passed to the optimizer as a kwarg.

init_scheduler(**scheduler_kwargs)

Optionally defined in subclass to create the scheduler

Parameters:

scheduler_kwargs (dict) – If ‘scheduler’ is defined than that scheduler will be used otherwise no scheduler will be used.

init_model(sys_data=None, nu=-1, ny=-1, device='cpu', auto_fit_norm=True, optimizer_kwargs={}, parameters_optimizer_kwargs={}, scheduler_kwargs={})

This function set the nu and ny, inits the network, moves parameters to device, initilizes optimizer and initilizes logging parameters

property parameters
property parameters_with_names
cal_validation_error(val_sys_data, validation_measure='sim-NRMS')

possible validation_measure are ‘sim-NRMS’ ‘sim-NRMS_mean_channel’ ‘sim-NRMS_per_channels’ (and others defined in System_data) ‘sim-NRMS_sys_norm’

‘10-step-NRMS’ or ‘10-step-average-NRMS’ ‘10-step-last-NRMS’ ‘10-step-last-RMS’ ‘10-step-[w0,w1,w2,w3,w4,w5,w6,w7,w8,w9]-NRMS’ weighted mean 10-step-error ‘10-step-NMAE_sys_norm’ ‘10-step-MSE’ ‘X-step-{last/average}-{mode}’ #like this

#todo; User given callback. (overwrite this function?) ‘loss’ #todo ‘sim-inno’ #todo

fit(train_sys_data, val_sys_data, epochs=30, batch_size=256, loss_kwargs={}, auto_fit_norm=True, validation_measure='sim-NRMS', optimizer_kwargs={}, concurrent_val=False, cuda=False, timeout=None, verbose=1, sqrt_train=True, num_workers_data_loader=0, print_full_time_profile=False, scheduler_kwargs={})

The batch optimization method with parallel validation,

Parameters:

  • train_sys_data (System_data or System_data_list) – The system data to be fitted

  • val_sys_data (System_data or System_data_list) – The validation system data after each used after each epoch for early stopping. Use the keyword argument validation_measure to specify which measure should be used.

  • epochs (int) –

  • batch_size (int) –

  • loss_kwargs (dict) – The Keyword Arguments to be passed to the self.make_training_data and self.loss of the current fit_system.

  • auto_fit_norm (boole) – If true will use self.norm.fit(train_sys_data) which will fit it element wise.

  • validation_measure (str) – Specify which measure should be used for validation, e.g. ‘sim-RMS’, ‘10-step-last-RMS’, ‘sim-NRMS_sys_norm’, ect. See self.cal_validation_error for details.

  • optimizer_kwargs (dict) – The Keyword Arguments to be passed on to init_optimizer. notes; init_optimizer[‘optimizer’] is the optimization function used (default torch.Adam) and optimizer_kwargs[‘parameters_optimizer_kwargs’] the learning rates and such for the different elements of the models. see https://pytorch.org/docs/stable/optim.html

  • concurrent_val (boole) – If set to true a subprocess will be started which concurrently evaluates the validation method selected. Warning: if concurrent_val is set than “if __name__==’__main__’” or import from a file if using self defined method or networks.

  • cuda (bool) – if cuda will be used (often slower than not using it, be aware)

  • timeout (None or number) – Alternative to epochs to run until a set amount of time has past.

  • verbose (int) – Set to 0 for a silent run

  • sqrt_train (boole) – will sqrt the loss while printing

  • num_workers_data_loader (int) – see https://pytorch.org/docs/stable/data.html

  • print_full_time_profile (boole) – will print the full time profile, useful for debugging and basic process optimization.

  • scheduler_kwargs (dict) – learning rate scheduals are a work in progress.

Notes

This method implements a batch optimization method in the following way; each epoch the training data is scrambled and batched where each batch is passed to the self.loss method and utilized to optimize the parameters. After each epoch the systems is validated using the evaluation of a simulation or a validation split and a checkpoint will be crated if a new lowest validation loss has been achieved. (or concurrently if concurrent_val=True) After training (which can be stopped at any moment using a KeyboardInterrupt) the system is loaded with the lowest validation loss.

The default checkpoint location is “C:/Users/USER/AppData/Local/deepSI/checkpoints” for windows and ~/.deepSI/checkpoints/ for unix like. These can be loaded manually using sys.load_checkpoint(“_best”) or “_last”. (For this to work the sys.unique_code needs to be set to the correct string)

checkpoint_save_system(name='_best', directory=None)
checkpoint_load_system(name='_best', directory=None)
save_system(file)

Save the system using pickle provided by torch

Notes

This can be quite unstable for long term storage or switching between versions of this and other modules. Consider manually creating a save_system function for a long term solution. (maybe utilize checkpoint_save_system)

_check_and_refresh_optimizer_if_needed()
_refresh_optimizer()
cuda()
cpu()
to_device(device)
eval()
train()
remote_start(val_sys_data, validation_measure)
remote_send(Loss_acc_val, time_optimize)
remote_recv(wait=False)
remote_close()
__module__ = 'deepSI.fit_systems.fit_system'
class deepSI.fit_systems.Sklearn_io(na, nb, reg, feedthrough=False)
__init__(na, nb, reg, feedthrough=False)
_fit(sys_data)
io_step(uy)
__module__ = 'deepSI.fit_systems.sklearn_io'
class deepSI.fit_systems.Sklearn_io_linear(na, nb, feedthrough=False)
__init__(na, nb, feedthrough=False)
__module__ = 'deepSI.fit_systems.sklearn_io'
class deepSI.fit_systems.SS_encoder(nx=10, na=20, nb=20, feedthrough=False)

The basic implementation of the subspace encoder, with neural networks.

nx

order of the system

Type:

int

na

length of the past outputs (y) considered as input for the encoder

Type:

int

nb

length of the past inputs (u) considered as input for the encoder

Type:

int

k0

length of the encoder max(na,nb)

Type:

int

e_net
observation_spacegym.space or None

The input shape of output y. (None is a single unbounded float)

norminstance of System_data_norm

Used in most fittable systems to normalize the input output.

fitted : Boole unique_code : str

Some random unique 4 digit code (can be used for saving/loading)

namestr

concatenation of the the class name and the unique code

seedint

random seed

randomnp.random.RandomState

unique random generated initialized with seed (only created ones called)

__init__(nx=10, na=20, nb=20, feedthrough=False)
make_training_data(sys_data, **loss_kwargs)
init_nets(nu, ny)
loss(hist, ufuture, yfuture, **loss_kwargs)
init_state_multi(sys_data, nf=100, stride=1)
measure_act_multi(actions)
reset_state()
get_state()
__module__ = 'deepSI.fit_systems.encoders'
class deepSI.fit_systems.SS_encoder_general(nx=10, na=20, nb=20, feedthrough=False, e_net=<class 'deepSI.fit_systems.encoders.default_encoder_net'>, f_net=<class 'deepSI.fit_systems.encoders.default_state_net'>, h_net=<class 'deepSI.fit_systems.encoders.default_output_net'>, e_net_kwargs={}, f_net_kwargs={}, h_net_kwargs={}, na_right=0, nb_right=0)

docstring for SS_encoder_general

__init__(nx=10, na=20, nb=20, feedthrough=False, e_net=<class 'deepSI.fit_systems.encoders.default_encoder_net'>, f_net=<class 'deepSI.fit_systems.encoders.default_state_net'>, h_net=<class 'deepSI.fit_systems.encoders.default_output_net'>, e_net_kwargs={}, f_net_kwargs={}, h_net_kwargs={}, na_right=0, nb_right=0)
make_training_data(sys_data, **Loss_kwargs)
init_nets(nu, ny)
loss(uhist, yhist, ufuture, yfuture, loss_nf_cutoff=None, **Loss_kwargs)
init_state_multi(sys_data, nf=100, stride=1)
reset_state()
measure_act_multi(action)
get_state()
__module__ = 'deepSI.fit_systems.encoders'
class deepSI.fit_systems.SS_par_start(nx=10, feedthrough=False, optimizer_kwargs={})

docstring for SS_par_start

__init__(nx=10, feedthrough=False, optimizer_kwargs={})
make_training_data(sys_data, **Loss_kwargs)
init_nets(nu, ny)
loss(ids, ufuture, yfuture, **Loss_kwargs)
init_state_multi(sys_data, nf=100, stride=1)
__module__ = 'deepSI.fit_systems.encoders'
class deepSI.fit_systems.SS_encoder_general_koopman(nx=10, na=20, nb=20, feedthrough=False, include_u_in_g=True, e_net=<class 'deepSI.fit_systems.encoders.default_encoder_net'>, g_net=<class 'deepSI.utils.torch_nets.simple_res_net'>, h_net=<class 'deepSI.fit_systems.encoders.default_output_net'>, e_net_kwargs={}, g_net_kwargs={}, h_net_kwargs={})
The encoder setup with a linear transition function with an affine input (kinda like an LPV), in equations

x_k = e(u_kpast,y_kpast) x_k+1 = A@x_k + g(x_k,u_k)@u_k #affine input here (@=matrix multiply) y_k = h(x_k)

Where g is given by g_net which is by default a feedforward nn with residual (i.e. simple_res_net) called as

‘g_net(n_in=affine_dim,n_out=output_dim*input_dim,**g_net_kwargs)’

with affine_dim=nx + nu, output_dim = nx, input_dim=nu Hence, g_net produces a vector which is reshaped into a matrix (See deepSI.utils.torch_nets.general_koopman_forward_layer for details).

__init__(nx=10, na=20, nb=20, feedthrough=False, include_u_in_g=True, e_net=<class 'deepSI.fit_systems.encoders.default_encoder_net'>, g_net=<class 'deepSI.utils.torch_nets.simple_res_net'>, h_net=<class 'deepSI.fit_systems.encoders.default_output_net'>, e_net_kwargs={}, g_net_kwargs={}, h_net_kwargs={})
__module__ = 'deepSI.fit_systems.encoders'
class deepSI.fit_systems.SS_encoder_innovation(nx=10, na=20, nb=20, na_right=1, nb_right=0, feedthrough=False, e_net=<class 'deepSI.fit_systems.encoders.default_encoder_net'>, f_net=<class 'deepSI.fit_systems.encoders.default_ino_state_net'>, h_net=<class 'deepSI.fit_systems.encoders.default_output_net'>, e_net_kwargs={}, f_net_kwargs={}, h_net_kwargs={})
Similar to SS encoder but with the structure of

x_k+1 = f(x_k,u_k,eps_k) y_k = h(x_k) + eps_k

During optimization eps will be set to eps_k = y_k - h(x_k)

During simulation eps will be set to zero (None) eps_k = 0

Futhermore this requires a specialized f net build as

self.fn = self.f_net(nx=self.nx, nu=nu, ny=self.ny,**self.f_net_kwargs)

default structure for f is as follows: f(x_k,u_k,eps_k) = f0(x_k,u_k) + K eps_k (with K a matrix)

__init__(nx=10, na=20, nb=20, na_right=1, nb_right=0, feedthrough=False, e_net=<class 'deepSI.fit_systems.encoders.default_encoder_net'>, f_net=<class 'deepSI.fit_systems.encoders.default_ino_state_net'>, h_net=<class 'deepSI.fit_systems.encoders.default_output_net'>, e_net_kwargs={}, f_net_kwargs={}, h_net_kwargs={})
init_nets(nu, ny)
loss(uhist, yhist, ufuture, yfuture, **Loss_kwargs)
__module__ = 'deepSI.fit_systems.encoders'
class deepSI.fit_systems.SS_encoder_CNN_video(nx=10, na=20, nb=20, feedthrough=False, e_net=<class 'deepSI.utils.torch_nets.CNN_encoder'>, f_net=<class 'deepSI.fit_systems.encoders.default_state_net'>, h_net=<class 'deepSI.utils.torch_nets.CNN_chained_upscales'>, e_net_kwargs={}, f_net_kwargs={}, h_net_kwargs={})

The subspace encoder convolutonal neural network

Notes

The subspace encoder

__init__(nx=10, na=20, nb=20, feedthrough=False, e_net=<class 'deepSI.utils.torch_nets.CNN_encoder'>, f_net=<class 'deepSI.fit_systems.encoders.default_state_net'>, h_net=<class 'deepSI.utils.torch_nets.CNN_chained_upscales'>, e_net_kwargs={}, f_net_kwargs={}, h_net_kwargs={})
__module__ = 'deepSI.fit_systems.encoders'
class deepSI.fit_systems.SS_encoder_shotgun_MLP(nx=10, na=20, nb=20, e_net=<class 'deepSI.utils.torch_nets.CNN_encoder'>, f_net=<class 'deepSI.fit_systems.encoders.default_state_net'>, h_net=<class 'deepSI.utils.torch_nets.Shotgun_MLP'>, e_net_kwargs={}, f_net_kwargs={}, h_net_kwargs={})
__init__(nx=10, na=20, nb=20, e_net=<class 'deepSI.utils.torch_nets.CNN_encoder'>, f_net=<class 'deepSI.fit_systems.encoders.default_state_net'>, h_net=<class 'deepSI.utils.torch_nets.Shotgun_MLP'>, e_net_kwargs={}, f_net_kwargs={}, h_net_kwargs={})

Todo: fix cuda with all the arrays

loss(uhist, yhist, ufuture, yfuture, **Loss_kwargs)
apply_experiment(sys_data, save_state=False)

Does an experiment with for a given system data (fixed u)

Parameters:

sys_data (System_data or System_data_list (or list or tuple)) – The experiment which should be applied

Notes

This will initialize the state using self.init_state if sys_data.y (and u) is not None and skip the appropriate number of steps associated with it. If either is missing than self.reset_state() is used to initialize the state. Afterwards this state is advanced using sys_data.u and the output is saved at each step. Lastly, the number of skipped/copied steps in init_state is saved as sys_data.cheat_n such that it can be accounted for later.

init_state(sys_data)
step(action, h=None, w=None)
sys_data_sampler(sys_data, Ndots_per_image)
__module__ = 'deepSI.fit_systems.encoders'
class deepSI.fit_systems.SS_encoder_deriv_general(nx=10, na=20, nb=20, feedthrough=False, f_norm=None, tau=None, cut_off=inf, e_net=<class 'deepSI.fit_systems.encoders.default_encoder_net'>, f_net=<class 'deepSI.fit_systems.encoders.default_state_net'>, integrator_net=<class 'deepSI.utils.torch_nets.integrator_RK4'>, h_net=<class 'deepSI.fit_systems.encoders.default_output_net'>, e_net_kwargs={}, f_net_kwargs={}, integrator_net_kwargs={}, h_net_kwargs={}, na_right=0, nb_right=0)

The subspace encoder method to obtain continuous time models https://arxiv.org/abs/2204.09405

Equations x_t|t = encoder(upast, ypast) dxdt = 1/tau * derivn(x_t+k|t, u_t) = f_norm * derivn(x_t+k|t, u_t) #derivn is f_net on initalization x_t+k+1|t = fn(x_t+k|t, u_t) #integrator which uses derivn y_t+k|t = hn(x_t+k|t)

Here the networks are initialized as: encoder = e_net(nb, nu, na, ny, nx, **e_net_kwargs) derivn = f_net(nx, nu, **f_net_kwargs) fn = integrator_net(deriv, f_norm, **integrator_net_kwargs) #normalization is part of the integrator hn = h_net(nx, ny, nu=-1, **h_net_kwargs)

Furthermore you can use cut_off to stabelize the training. tip: use cut_off = 5*(normalized noise level) is a good value

__init__(nx=10, na=20, nb=20, feedthrough=False, f_norm=None, tau=None, cut_off=inf, e_net=<class 'deepSI.fit_systems.encoders.default_encoder_net'>, f_net=<class 'deepSI.fit_systems.encoders.default_state_net'>, integrator_net=<class 'deepSI.utils.torch_nets.integrator_RK4'>, h_net=<class 'deepSI.fit_systems.encoders.default_output_net'>, e_net_kwargs={}, f_net_kwargs={}, integrator_net_kwargs={}, h_net_kwargs={}, na_right=0, nb_right=0)

The subspace encoder method to obtain continuous time models https://arxiv.org/abs/2204.09405

Equations x_t|t = encoder(upast, ypast) dxdt = 1/tau * derivn(x_t+k|t, u_t) = f_norm * derivn(x_t+k|t, u_t) #derivn is f_net on initalization x_t+k+1|t = fn(x_t+k|t, u_t) #integrator which uses derivn y_t+k|t = hn(x_t+k|t)

Here the networks are initialized as: encoder = e_net(nb, nu, na, ny, nx, **e_net_kwargs) derivn = f_net(nx, nu, **f_net_kwargs) fn = integrator_net(deriv, f_norm, **integrator_net_kwargs) #normalization is part of the integrator hn = h_net(nx, ny, nu=-1, **h_net_kwargs)

Furthermore you can use cut_off to stabelize the training. tip: use cut_off = 5*(normalized noise level) is a good value

init_nets(nu, ny)
property dt
loss(uhist, yhist, ufuture, yfuture, **Loss_kwargs)
__module__ = 'deepSI.fit_systems.encoders'
class deepSI.fit_systems.SS_encoder_general_hf(nx=10, na=20, nb=20, feedthrough=False, hf_net=<class 'deepSI.fit_systems.encoders.hf_net_default'>, hf_net_kwargs={'f_net': <class 'deepSI.fit_systems.encoders.default_state_net'>, 'f_net_kwargs': {}, 'h_net': <class 'deepSI.fit_systems.encoders.default_output_net'>, 'h_net_kwargs': {}}, e_net=<class 'deepSI.fit_systems.encoders.default_encoder_net'>, e_net_kwargs={}, na_right=0, nb_right=0)

The encoder function with combined h and f functions

the hf_net_default has the arguments

hf_net_default(nx, nu, ny, feedthrough=False, **hf_net_kwargs)

and is used as

ynow, xnext = hfn(x,u)

__init__(nx=10, na=20, nb=20, feedthrough=False, hf_net=<class 'deepSI.fit_systems.encoders.hf_net_default'>, hf_net_kwargs={'f_net': <class 'deepSI.fit_systems.encoders.default_state_net'>, 'f_net_kwargs': {}, 'h_net': <class 'deepSI.fit_systems.encoders.default_output_net'>, 'h_net_kwargs': {}}, e_net=<class 'deepSI.fit_systems.encoders.default_encoder_net'>, e_net_kwargs={}, na_right=0, nb_right=0)
init_nets(nu, ny)
loss(uhist, yhist, ufuture, yfuture, **Loss_kwargs)
measure_act_multi(actions)
__module__ = 'deepSI.fit_systems.encoders'
class deepSI.fit_systems.Torch_io_siso(na, nb)
__init__(na, nb)
make_training_data(sys_data, **Loss_kwargs)
init_nets(nu, ny)
loss(hist, Y, **kwargs)
io_step(uy)
__module__ = 'deepSI.fit_systems.torch_io'
class deepSI.fit_systems.Torch_io(na=5, nb=5, feedthrough=False)
__init__(na=5, nb=5, feedthrough=False)
make_training_data(sys_data, **Loss_kwargs)
init_nets(nu, ny)
loss(uhist, yhist, ufuture, yfuture, **Loss_kwargs)
multi_io_step(uy)
io_step(uy)
__module__ = 'deepSI.fit_systems.torch_io'
class deepSI.fit_systems.SS_linear(seed=None, A=None, B=None, C=None, D=None, nx=2, feedthrough=False, k0=None)
__init__(seed=None, A=None, B=None, C=None, D=None, nx=2, feedthrough=False, k0=None)
_fit(sys_data, SS_A_stability=False, SS_f=20)
f(x, u)
h(x, u)
init_state(sys_data)
__module__ = 'deepSI.fit_systems.ss_linear'
class deepSI.fit_systems.SS_linear_CT(seed=None, A=None, B=None, C=None, D=None, nx=2, feedthrough=False, k0=5)
__init__(seed=None, A=None, B=None, C=None, D=None, nx=2, feedthrough=False, k0=5)
_fit(sys_data, SS_A_stability=False, SS_f=20)
f(x, u)
__module__ = 'deepSI.fit_systems.ss_linear'
class deepSI.fit_systems.IO_autoencoder(nz=4, na=5, nb=5)

docstring for IO_autoencoder

__init__(nz=4, na=5, nb=5)
make_training_data(sys_data, **Loss_kwargs)
init_nets(nu, ny)
loss(uhist, yhist, ufuture, yfuture, **Loss_kwargs)
init_state_and_measure_multi(sys_data, nf=100, stride=1)
act_measure_multi(action)
__module__ = 'deepSI.fit_systems.io_autoencoder'