Skip to content

API Reference¤

Methods¤

MFVeBRNN.method.RNNTrainer ¤

RNN trainer, with supporting different RNN architectures.

Source code in src/MFVeBRNN/method/rnn_trainer.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
class RNNTrainer:
    """RNN trainer, with supporting different RNN architectures."""

    def __init__(
        self,
        net: torch.nn.Module,
        device: torch.device = torch.device("cpu"),
        seed: int = 0,
    ) -> None:
        """initialize the RNN with linear transfer layer and BNN for residual
        learning

        Parameters
        ----------
        net : torch.nn.Module
            the RNN network to be trained
        device : torch.device, optional
            device for training, by default torch.device("cpu")
        seed : int, optional
            random seed for reproducibility, by default 0
        """
        # device
        self.device = device
        # set seed for all components
        self.seed = seed
        # define the network and move it to the device
        self.net = net.to(self.device)

    def configure_optimizer_info(
        self,
        optimizer_name: str = "Adam",
        lr: float = 1e-3,
        weight_decay: float = 1e-6
    ) -> None:
        """define optimizer of the low-fidelity deterministic RNN

        Parameters
        ----------
        optimizer_name : str, optional
            name of the optimizer, by default "Adam"
        lr : float, optional
            learning rate for the optimizer, by default 1e-3
        weight_decay : float, optional
            weight decay for the optimizer, by default 1e-6
        """
        if optimizer_name == "Adam":
            self.optimizer = torch.optim.Adam(
                self.net.parameters(),
                lr=lr,
                weight_decay=weight_decay
            )
        elif optimizer_name == "SGD":
            self.lf_optimizer = torch.optim.SGD(
                self.net.parameters(),
                lr=lr,
                weight_decay=weight_decay
            )
        else:
            raise ValueError("Undefined optimizer")

    def configure_loss_function(self, loss_name: str = "MSE") -> None:
        """configure the loss function for the training process

        Parameters
        ----------
        loss_name : str, optional
            name of the loss function, by default "MSE"

        Raises
        ------
        ValueError
            Undefined loss function
        """
        if loss_name == "MSE":
            self.loss_function = torch.nn.MSELoss()
        elif loss_name == "MAE":
            self.loss_function = torch.nn.L1Loss()
        else:
            raise ValueError("Undefined loss function")

    def train(
        self,
        x_train: Tensor,
        y_train: Tensor,
        num_epochs: int,
        batch_size: int = None,
        x_val: Tensor = None,
        y_val: Tensor = None,
        verbose: bool = True,
        print_iter: int = 100,
    ) -> Tuple[float, int]:
        """train the network

        Parameters
        ----------
        x_train : Tensor
            input training data
        y_train : Tensor
            output training data
        num_epochs : int
            number of epochs for training
        batch_size : int, optional
            batch size for mini-batch training, by default None
        x_val : Tensor, optional
            input validation data, by default None
        y_val : Tensor, optional
            output validation data, by default None
        verbose : bool, optional
            print the training information, by default True
        print_iter : int, optional
            print the training information every certain epochs, by default 100

        Returns
        -------
        Tuple[float, int]
            minimum loss value and the epoch number
        """
        # move the training and validation data to the device
        x_train = x_train.to(self.device)
        y_train = y_train.to(self.device)
        if x_val is not None:
            x_val = x_val.to(self.device)
            y_val = y_val.to(self.device)
        # record the minimum loss of the validation data
        min_loss = np.inf
        # loader for mini-batch
        if batch_size is None:
            self.batch_size = x_train.shape[0]
            self.num_scale = 1.0
        else:
            self.batch_size = batch_size
            self.num_scale = x_train.shape[0] / self.batch_size

        loader = DataLoader(
            dataset=list(zip(x_train, y_train)),
            batch_size=self.batch_size,
            shuffle=True
        )

        # begin the training process
        for epoch in range(num_epochs):

            # set the network to training mode
            self.net.train()
            # running loss for the training data
            running_loss_train = 0
            for X_batch, y_batch in loader:
                # set gradient of params to zero
                self.optimizer.zero_grad()
                # get prediction from network
                pred = self.net.forward(X_batch)
                # calculate the loss value for the batch
                loss = self.loss_function(pred, y_batch)
                # back-propagation
                loss.backward()
                # update the weights
                self.optimizer.step()
                # accumulate the loss value
                running_loss_train += loss.item() * X_batch.size(0)

            # average the loss value
            loss_train = running_loss_train / x_train.size(0)
            if x_val is not None:
                self.net.eval()
                y_val_pred = self.net.forward(x_val)
                loss_val = self.loss_function(y_val, y_val_pred)
                if loss_val.item() < min_loss:
                    # save the model with the best validation loss
                    min_loss = loss_val.item()
                    best_epoch = epoch
                    self.best_net = copy.deepcopy(self.net)

            else:
                loss_val = torch.tensor(0.0)
                self.best_net = copy.deepcopy(self.net)
                min_loss = loss_val.item()
                best_epoch = epoch

            if verbose and epoch % print_iter == 0:
                self._print(epoch, num_epochs,
                            loss_train, loss_val.item())

        return min_loss, best_epoch

    def predict(self, x: Tensor) -> Tensor:
        """predict the output of the network

        Parameters
        ----------
        x : Tensor
            input data

        Returns
        -------
        Tensor
            predicted output data
        """
        self.best_net.eval()
        y = self.best_net.forward(x.to(self.device))

        return y.detach()

    def recurrent_forward(self,
                          x: torch.Tensor,
                          hx: torch.Tensor = None) -> List:
        """forward of low-fidelity RNN, such that hidden states are returned.
        Then, the hidden states are used as the inputs for the transfer layer
        and also used for residual BNN training

        Parameters
        ----------
        x : torch.Tensor
            input data (scaled low-fidelity data or high-fidelity data in
            low-fidelity scale)
        hx : torch.Tensor, optional
            hidden state, by default None

        Returns
        -------
        List
            hidden states of the RNN at each time step
        """

        self.best_net.eval()
        with torch.no_grad():
            outs, _ = self.best_net.gru(x.to(self.device), hx)

        return outs

    def _print(
        self,
        epoch: int,
        num_epoch: int,
        loss_train: float,
        loss_val: float
    ) -> None:
        """print the loss values during training at certain epochs

        Parameters
        ----------
        epoch : int
            the current epoch
        num_epoch : int
            total number of epochs
        loss_train : float
            training loss value at the current epoch
        loss_val : float
            validation loss value at the current epoch
        """

        print(
            "Epoch/Total: %d/%d, Train Loss: %.3e, Val Loss: %.3e"
            % (
                epoch,
                num_epoch,
                loss_train,
                loss_val,
            )
        )
_print(epoch: int, num_epoch: int, loss_train: float, loss_val: float) -> None ¤

print the loss values during training at certain epochs

Parameters:

Name Type Description Default
epoch int

the current epoch

 required
num_epoch int

total number of epochs

 required
loss_train float

training loss value at the current epoch

 required
loss_val float

validation loss value at the current epoch

 required
Source code in src/MFVeBRNN/method/rnn_trainer.py 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
def _print(
    self,
    epoch: int,
    num_epoch: int,
    loss_train: float,
    loss_val: float
) -> None:
    """print the loss values during training at certain epochs

    Parameters
    ----------
    epoch : int
        the current epoch
    num_epoch : int
        total number of epochs
    loss_train : float
        training loss value at the current epoch
    loss_val : float
        validation loss value at the current epoch
    """

    print(
        "Epoch/Total: %d/%d, Train Loss: %.3e, Val Loss: %.3e"
        % (
            epoch,
            num_epoch,
            loss_train,
            loss_val,
        )
    )
configure_loss_function(loss_name: str = 'MSE') -> None ¤

configure the loss function for the training process

Parameters:

Name Type Description Default
loss_name str

name of the loss function, by default "MSE"

 'MSE'

Raises:

Type Description
ValueError

Undefined loss function
Source code in src/MFVeBRNN/method/rnn_trainer.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def configure_loss_function(self, loss_name: str = "MSE") -> None:
    """configure the loss function for the training process

    Parameters
    ----------
    loss_name : str, optional
        name of the loss function, by default "MSE"

    Raises
    ------
    ValueError
        Undefined loss function
    """
    if loss_name == "MSE":
        self.loss_function = torch.nn.MSELoss()
    elif loss_name == "MAE":
        self.loss_function = torch.nn.L1Loss()
    else:
        raise ValueError("Undefined loss function")
configure_optimizer_info(optimizer_name: str = 'Adam', lr: float = 0.001, weight_decay: float = 1e-06) -> None ¤

define optimizer of the low-fidelity deterministic RNN

Parameters:

Name Type Description Default
optimizer_name str

name of the optimizer, by default "Adam"

 'Adam'
lr float

learning rate for the optimizer, by default 1e-3

 0.001
weight_decay float

weight decay for the optimizer, by default 1e-6

 1e-06
Source code in src/MFVeBRNN/method/rnn_trainer.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def configure_optimizer_info(
    self,
    optimizer_name: str = "Adam",
    lr: float = 1e-3,
    weight_decay: float = 1e-6
) -> None:
    """define optimizer of the low-fidelity deterministic RNN

    Parameters
    ----------
    optimizer_name : str, optional
        name of the optimizer, by default "Adam"
    lr : float, optional
        learning rate for the optimizer, by default 1e-3
    weight_decay : float, optional
        weight decay for the optimizer, by default 1e-6
    """
    if optimizer_name == "Adam":
        self.optimizer = torch.optim.Adam(
            self.net.parameters(),
            lr=lr,
            weight_decay=weight_decay
        )
    elif optimizer_name == "SGD":
        self.lf_optimizer = torch.optim.SGD(
            self.net.parameters(),
            lr=lr,
            weight_decay=weight_decay
        )
    else:
        raise ValueError("Undefined optimizer")
predict(x: Tensor) -> Tensor ¤

predict the output of the network

Parameters:

Name Type Description Default
x Tensor

input data

 required

Returns:

Type Description
Tensor

predicted output data
Source code in src/MFVeBRNN/method/rnn_trainer.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def predict(self, x: Tensor) -> Tensor:
    """predict the output of the network

    Parameters
    ----------
    x : Tensor
        input data

    Returns
    -------
    Tensor
        predicted output data
    """
    self.best_net.eval()
    y = self.best_net.forward(x.to(self.device))

    return y.detach()
recurrent_forward(x: Tensor, hx: Tensor = None) -> typing.List ¤

forward of low-fidelity RNN, such that hidden states are returned. Then, the hidden states are used as the inputs for the transfer layer and also used for residual BNN training

Parameters:

Name Type Description Default
x Tensor

input data (scaled low-fidelity data or high-fidelity data in low-fidelity scale)

 required
hx Tensor

hidden state, by default None

 None

Returns:

Type Description
List

hidden states of the RNN at each time step
Source code in src/MFVeBRNN/method/rnn_trainer.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
def recurrent_forward(self,
                      x: torch.Tensor,
                      hx: torch.Tensor = None) -> List:
    """forward of low-fidelity RNN, such that hidden states are returned.
    Then, the hidden states are used as the inputs for the transfer layer
    and also used for residual BNN training

    Parameters
    ----------
    x : torch.Tensor
        input data (scaled low-fidelity data or high-fidelity data in
        low-fidelity scale)
    hx : torch.Tensor, optional
        hidden state, by default None

    Returns
    -------
    List
        hidden states of the RNN at each time step
    """

    self.best_net.eval()
    with torch.no_grad():
        outs, _ = self.best_net.gru(x.to(self.device), hx)

    return outs
train(x_train: Tensor, y_train: Tensor, num_epochs: int, batch_size: int = None, x_val: Tensor = None, y_val: Tensor = None, verbose: bool = True, print_iter: int = 100) -> typing.Tuple[float, int] ¤

train the network

Parameters:

Name Type Description Default
x_train Tensor

input training data

 required
y_train Tensor

output training data

 required
num_epochs int

number of epochs for training

 required
batch_size int

batch size for mini-batch training, by default None

 None
x_val Tensor

input validation data, by default None

 None
y_val Tensor

output validation data, by default None

 None
verbose bool

print the training information, by default True

 True
print_iter int

print the training information every certain epochs, by default 100

 100

Returns:

Type Description
Tuple[float, int]

minimum loss value and the epoch number
Source code in src/MFVeBRNN/method/rnn_trainer.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def train(
    self,
    x_train: Tensor,
    y_train: Tensor,
    num_epochs: int,
    batch_size: int = None,
    x_val: Tensor = None,
    y_val: Tensor = None,
    verbose: bool = True,
    print_iter: int = 100,
) -> Tuple[float, int]:
    """train the network

    Parameters
    ----------
    x_train : Tensor
        input training data
    y_train : Tensor
        output training data
    num_epochs : int
        number of epochs for training
    batch_size : int, optional
        batch size for mini-batch training, by default None
    x_val : Tensor, optional
        input validation data, by default None
    y_val : Tensor, optional
        output validation data, by default None
    verbose : bool, optional
        print the training information, by default True
    print_iter : int, optional
        print the training information every certain epochs, by default 100

    Returns
    -------
    Tuple[float, int]
        minimum loss value and the epoch number
    """
    # move the training and validation data to the device
    x_train = x_train.to(self.device)
    y_train = y_train.to(self.device)
    if x_val is not None:
        x_val = x_val.to(self.device)
        y_val = y_val.to(self.device)
    # record the minimum loss of the validation data
    min_loss = np.inf
    # loader for mini-batch
    if batch_size is None:
        self.batch_size = x_train.shape[0]
        self.num_scale = 1.0
    else:
        self.batch_size = batch_size
        self.num_scale = x_train.shape[0] / self.batch_size

    loader = DataLoader(
        dataset=list(zip(x_train, y_train)),
        batch_size=self.batch_size,
        shuffle=True
    )

    # begin the training process
    for epoch in range(num_epochs):

        # set the network to training mode
        self.net.train()
        # running loss for the training data
        running_loss_train = 0
        for X_batch, y_batch in loader:
            # set gradient of params to zero
            self.optimizer.zero_grad()
            # get prediction from network
            pred = self.net.forward(X_batch)
            # calculate the loss value for the batch
            loss = self.loss_function(pred, y_batch)
            # back-propagation
            loss.backward()
            # update the weights
            self.optimizer.step()
            # accumulate the loss value
            running_loss_train += loss.item() * X_batch.size(0)

        # average the loss value
        loss_train = running_loss_train / x_train.size(0)
        if x_val is not None:
            self.net.eval()
            y_val_pred = self.net.forward(x_val)
            loss_val = self.loss_function(y_val, y_val_pred)
            if loss_val.item() < min_loss:
                # save the model with the best validation loss
                min_loss = loss_val.item()
                best_epoch = epoch
                self.best_net = copy.deepcopy(self.net)

        else:
            loss_val = torch.tensor(0.0)
            self.best_net = copy.deepcopy(self.net)
            min_loss = loss_val.item()
            best_epoch = epoch

        if verbose and epoch % print_iter == 0:
            self._print(epoch, num_epochs,
                        loss_train, loss_val.item())

    return min_loss, best_epoch

MFVeBRNN.method.VeBRNNTrainer ¤

VeBRNN trainer, with supporting different RNN architectures.

Source code in src/MFVeBRNN/method/vebnn_trainer.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
class VeBRNNTrainer(SGMCMCTrainer):
    """VeBRNN trainer, with supporting different RNN architectures."""

    def __init__(
        self,
        mean_net: MeanNet,
        var_net: GammaVarNet,
        device: torch.device = torch.device("cpu"),
        job_id: int = 1,
    ) -> None:
        """Variance estimation Bayesian neural network training using
        Stochastic

        Parameters
        ----------
        mean_net : MeanNet
            mean neural network for the VeBNN
        var_net : GammaVarNet
            variance estimation neural network for VeBNN
        device : torch.device, optional
            device for training, by default torch.device("cpu")
        job_id : int, optional
            job ID for the training, by default 1
        """
        super().__init__(
            mean_net=mean_net,
            var_net=var_net,
            device=device,
            job_id=job_id,
        )

    def recurrent_forward(self,
                          x: torch.Tensor,
                          return_var: bool = False) -> torch.Tensor:
        """forward of low-fidelity RNN, such that hidden states are returned

        Parameters
        ----------
        x : torch.Tensor
            input data for the forward pass
        return_var : bool, optional
            whether to return the variance estimation, by default False

        Returns
        -------
        torch.Tensor
            output data for the forward pass
        """

        hidden_states = []
        for state_dict in self.mean_nets:
            # Create a fresh model instance
            temp_model = copy.deepcopy(self.un_trained_mean_net)
            temp_model.load_state_dict(state_dict)
            with torch.no_grad():
                # get the hidden state from the first
                hidden_state, _ = list(temp_model.net.children())[0](x)
            hidden_states.append(hidden_state)
        hidden_states = torch.stack(hidden_states, dim=1)
        if return_var:
            hidden_states_mean = torch.mean(hidden_states, dim=1)
            hidden_states_var = torch.var(hidden_states, dim=1)
            return hidden_states_mean.detach(), hidden_states_var.detach()
        else:
            return torch.mean(hidden_states, dim=1)
recurrent_forward(x: Tensor, return_var: bool = False) -> Tensor ¤

forward of low-fidelity RNN, such that hidden states are returned

Parameters:

Name Type Description Default
x Tensor

input data for the forward pass

 required
return_var bool

whether to return the variance estimation, by default False

 False

Returns:

Type Description
Tensor

output data for the forward pass
Source code in src/MFVeBRNN/method/vebnn_trainer.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def recurrent_forward(self,
                      x: torch.Tensor,
                      return_var: bool = False) -> torch.Tensor:
    """forward of low-fidelity RNN, such that hidden states are returned

    Parameters
    ----------
    x : torch.Tensor
        input data for the forward pass
    return_var : bool, optional
        whether to return the variance estimation, by default False

    Returns
    -------
    torch.Tensor
        output data for the forward pass
    """

    hidden_states = []
    for state_dict in self.mean_nets:
        # Create a fresh model instance
        temp_model = copy.deepcopy(self.un_trained_mean_net)
        temp_model.load_state_dict(state_dict)
        with torch.no_grad():
            # get the hidden state from the first
            hidden_state, _ = list(temp_model.net.children())[0](x)
        hidden_states.append(hidden_state)
    hidden_states = torch.stack(hidden_states, dim=1)
    if return_var:
        hidden_states_mean = torch.mean(hidden_states, dim=1)
        hidden_states_var = torch.var(hidden_states, dim=1)
        return hidden_states_mean.detach(), hidden_states_var.detach()
    else:
        return torch.mean(hidden_states, dim=1)

MFVeBRNN.method.MFNestRNNTrainer ¤

Source code in src/MFVeBRNN/method/mf_nest_rnn_trainer.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
class MFNestRNNTrainer:

    def __init__(self,
                 net: torch.nn.Module,
                 pre_trained_lf_model: RNNTrainer | VeBRNNTrainer,
                 device: torch.device = torch.device("cpu"),
                 seed: int = 0,
                 nest_option: str = "hidden",
                 ) -> None:
        """initialize the trainer for the high-fidelity model

        Parameters
        ----------
        net : torch.nn.Module
            The high-fidelity neural network
        pre_trained_lf_model : RNNTrainer | VeBRNNTrainer
            The pre-trained low-fidelity model
        device : torch.device, optional
            device for training, by default torch.device("cpu")
        seed : int, optional
            random seed for reproducibility, by default 0
        nest_option : str, optional
            nested option, by default "hidden" or "output"
        """
        self.device = device
        self.net = net.to(self.device)
        # load the pre-trained low-fidelity model
        self.lf_model: RNNTrainer | VeBRNNTrainer = pre_trained_lf_model
        # move the self.lf_model to the device
        self.lf_model.device = self.device
        if isinstance(self.lf_model, RNNTrainer):
            self.lf_model.best_net = self.lf_model.best_net.to(self.device)
        elif isinstance(self.lf_model, VeBRNNTrainer):
            self.lf_model.mean_net = self.lf_model.mean_net.to(self.device)
            self.lf_model.var_net = self.lf_model.var_net.to(self.device)
        else:
            raise ValueError("Undefined low-fidelity model type")
        # set the seed and nest option
        self.seed = seed
        self.nest_option = nest_option

    def configure_optimizer_info(self,
                                 optimizer_name: str = "Adam",
                                 lr: float = 0.001,
                                 weight_decay: float = 0.0):
        """configure optimizers

        Parameters
        ----------
        optimizer_name : str, optional
            name of the optimizer, by default "Adam"
        lr : float, optional
            learning rate for the optimizer, by default 0.001
        weight_decay : float, optional
            weight decay for the optimizer, by default 0.0

        Raises
        ------
        ValueError
            Undefined optimizer
        """

        if optimizer_name == "Adam":
            self.optimizer = torch.optim.Adam(
                self.net.parameters(), lr=lr, weight_decay=weight_decay)
        elif optimizer_name == "SGD":
            self.optimizer = torch.optim.SGD(
                self.net.parameters(), lr=lr, weight_decay=weight_decay)
        else:
            raise ValueError("Optimizer not supported")

    def configure_loss_function(self, loss_name: str = "MSE") -> None:
        """configure the loss function for the training process

        Parameters
        ----------
        loss_name : str, optional
            name of the loss function, by default "MSE"

        Raises
        ------
        ValueError
            Undefined loss function
        """
        if loss_name == "MSE":
            self.loss_function = torch.nn.MSELoss()
        elif loss_name == "MAE":
            self.loss_function = torch.nn.L1Loss()
        else:
            raise ValueError("Undefined loss function")

    def train(self,
              hx_train: Tensor,
              hy_train: Tensor,
              num_epochs: int,
              batch_size: int = None,
              hx_val: Tensor = None,
              hy_val: Tensor = None,
              verbose: bool = True,
              print_iter: int = 100) -> Tuple[float]:
        """train the multi-fidelity rnn model

        Parameters
        ----------
        hx_train : Tensor
            high-fidelity training input data
        hy_train : Tensor
            high-fidelity training output data
        num_epochs : int
            number of epochs to train
        batch_size : int, optional
            size of the mini-batch, by default None
        hx_val : Tensor, optional
            high-fidelity validation input data, by default None
        hy_val : Tensor, optional
            high-fidelity validation output data, by default None
        verbose : bool, optional
            whether to print training information, by default True
        print_iter : int, optional
            frequency of printing training information, by default 100


        """
        # set the data to the device
        hx_train = hx_train.to(self.device)
        hy_train = hy_train.to(self.device)
        if hx_val is not None:
            hx_val = hx_val.to(self.device)
            hy_val = hy_val.to(self.device)

        # re-arrange the training and validation data
        hx_train = self._re_arrange_input(hx_train)
        if hx_val is not None:
            hx_val = self._re_arrange_input(hx_val)

        # record the minimum loss of the validation data
        min_loss = np.inf
        # loader for mini-batch
        if batch_size is None:
            self.batch_size = hx_train.shape[0]
            self.num_scale = 1.0
        else:
            self.batch_size = batch_size
            self.num_scale = hx_train.shape[0] / self.batch_size

        loader = DataLoader(
            dataset=list(zip(hx_train, hy_train)),
            batch_size=self.batch_size,
            shuffle=True
        )

        # begin the training process
        for epoch in range(num_epochs):

            # set the network to training mode
            self.net.train()
            # running loss for the training data
            running_loss_train = 0
            for X_batch, y_batch in loader:
                # set gradient of params to zero
                self.optimizer.zero_grad()
                # get prediction from network
                pred = self.net.forward(X_batch)
                # calculate the loss value for the batch
                loss = self.loss_function(pred, y_batch)
                # back propagation
                loss.backward()
                # update the weights
                self.optimizer.step()
                # accumulate the loss value
                running_loss_train += loss.item() * X_batch.size(0)

            # average the loss value
            loss_train = running_loss_train / hx_train.size(0)
            if hx_val is not None:
                self.net.eval()
                y_val_pred = self.net.forward(hx_val)
                loss_val = self.loss_function(hy_val, y_val_pred)
                if loss_val.item() < min_loss:
                    # save the model with the best validation loss
                    min_loss = loss_val.item()
                    best_epoch = epoch
                    self.best_net = copy.deepcopy(self.net)
            else:
                loss_val = torch.tensor(0.0)
                self.best_net = copy.deepcopy(self.net)
                min_loss = loss_val.item()
                best_epoch = epoch

            if verbose and epoch % print_iter == 0:
                self._print(epoch, num_epochs,
                            loss_train, loss_val.item())

        return min_loss, best_epoch

    def hf_predict(self, x: Tensor) -> Tensor:
        """predict the output of the network

        Parameters
        ----------
        x : Tensor
            input data

        Returns
        -------
        Tensor
            predicted output data
        """
        # get the re-arranged input data
        x = self._re_arrange_input(x.to(self.device))
        self.best_net.eval()
        y = self.best_net.forward(x)

        return y.detach()

    def lf_predict(self, x: Tensor, return_var: bool = False) -> Tensor:
        """predict the output of the network

        Parameters
        ----------
        x : Tensor
            input data

        Returns
        -------
        Tensor
            predicted output data
        """

        if isinstance(self.lf_model, RNNTrainer):
            y = self.lf_model.predict(x.to(self.device))

        elif isinstance(self.lf_model, VeBRNNTrainer):
            y, var_epistemic = self.lf_model.bayes_predict(
                x.to(self.device))
            var_aleatoric = self.lf_model.aleatoric_variance_predict(
                x.to(self.device))
            if return_var:
                return y, var_aleatoric, var_epistemic
        else:
            raise ValueError("Undefined low-fidelity model")

        return y


    def _re_arrange_input(self,
                          x: Tensor) -> Tensor:
        """re-arrange the input data for the training process

        Parameters
        ----------
        x : Tensor
            input data for the training process or prediction

        Returns
        -------
        Tensor
            the re-arranged input data

        Raises
        ------
        ValueError
            Undefined nest option
        """

        if self.nest_option == "output":

            # predict the output of the low-fidelity model
            re_hx_input = self.lf_model.predict(x)
            # concatenate the data
            x = torch.cat((x, re_hx_input), dim=2)

            return x

        elif self.nest_option == "hidden":

            # predict the output of the low-fidelity model
            re_hx_input = self.lf_model.recurrent_forward(x)
            re_hx_input = re_hx_input.detach()
            # concatenate the data
            x = torch.cat((x, re_hx_input), dim=-1)

            return x

        else:
            raise ValueError("Undefined nest option")

    def _print(
        self,
        epoch: int,
        num_epoch: int,
        loss_train: float,
        loss_val: float
    ) -> None:
        """print the loss values during training at certain epochs

        Parameters
        ----------
        epoch : int
            the current epoch
        num_epoch : int
            total number of epochs
        loss_train : float
            training loss value at the current epoch
        loss_val : float
            validation loss value at the current epoch
        """

        print(
            "Epoch/Total: %d/%d, Train Loss: %.3e, Val Loss: %.3e"
            % (
                epoch,
                num_epoch,
                loss_train,
                loss_val,
            )
        )
_print(epoch: int, num_epoch: int, loss_train: float, loss_val: float) -> None ¤

print the loss values during training at certain epochs

Parameters:

Name Type Description Default
epoch int

the current epoch

 required
num_epoch int

total number of epochs

 required
loss_train float

training loss value at the current epoch

 required
loss_val float

validation loss value at the current epoch

 required
Source code in src/MFVeBRNN/method/mf_nest_rnn_trainer.py 
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
def _print(
    self,
    epoch: int,
    num_epoch: int,
    loss_train: float,
    loss_val: float
) -> None:
    """print the loss values during training at certain epochs

    Parameters
    ----------
    epoch : int
        the current epoch
    num_epoch : int
        total number of epochs
    loss_train : float
        training loss value at the current epoch
    loss_val : float
        validation loss value at the current epoch
    """

    print(
        "Epoch/Total: %d/%d, Train Loss: %.3e, Val Loss: %.3e"
        % (
            epoch,
            num_epoch,
            loss_train,
            loss_val,
        )
    )
_re_arrange_input(x: Tensor) -> Tensor ¤

re-arrange the input data for the training process

Parameters:

Name Type Description Default
x Tensor

input data for the training process or prediction

 required

Returns:

Type Description
Tensor

the re-arranged input data

Raises:

Type Description
ValueError

Undefined nest option
Source code in src/MFVeBRNN/method/mf_nest_rnn_trainer.py 
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
def _re_arrange_input(self,
                      x: Tensor) -> Tensor:
    """re-arrange the input data for the training process

    Parameters
    ----------
    x : Tensor
        input data for the training process or prediction

    Returns
    -------
    Tensor
        the re-arranged input data

    Raises
    ------
    ValueError
        Undefined nest option
    """

    if self.nest_option == "output":

        # predict the output of the low-fidelity model
        re_hx_input = self.lf_model.predict(x)
        # concatenate the data
        x = torch.cat((x, re_hx_input), dim=2)

        return x

    elif self.nest_option == "hidden":

        # predict the output of the low-fidelity model
        re_hx_input = self.lf_model.recurrent_forward(x)
        re_hx_input = re_hx_input.detach()
        # concatenate the data
        x = torch.cat((x, re_hx_input), dim=-1)

        return x

    else:
        raise ValueError("Undefined nest option")
configure_loss_function(loss_name: str = 'MSE') -> None ¤

configure the loss function for the training process

Parameters:

Name Type Description Default
loss_name str

name of the loss function, by default "MSE"

 'MSE'

Raises:

Type Description
ValueError

Undefined loss function
Source code in src/MFVeBRNN/method/mf_nest_rnn_trainer.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def configure_loss_function(self, loss_name: str = "MSE") -> None:
    """configure the loss function for the training process

    Parameters
    ----------
    loss_name : str, optional
        name of the loss function, by default "MSE"

    Raises
    ------
    ValueError
        Undefined loss function
    """
    if loss_name == "MSE":
        self.loss_function = torch.nn.MSELoss()
    elif loss_name == "MAE":
        self.loss_function = torch.nn.L1Loss()
    else:
        raise ValueError("Undefined loss function")
configure_optimizer_info(optimizer_name: str = 'Adam', lr: float = 0.001, weight_decay: float = 0.0) ¤

configure optimizers

Parameters:

Name Type Description Default
optimizer_name str

name of the optimizer, by default "Adam"

 'Adam'
lr float

learning rate for the optimizer, by default 0.001

 0.001
weight_decay float

weight decay for the optimizer, by default 0.0

 0.0

Raises:

Type Description
ValueError

Undefined optimizer
Source code in src/MFVeBRNN/method/mf_nest_rnn_trainer.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def configure_optimizer_info(self,
                             optimizer_name: str = "Adam",
                             lr: float = 0.001,
                             weight_decay: float = 0.0):
    """configure optimizers

    Parameters
    ----------
    optimizer_name : str, optional
        name of the optimizer, by default "Adam"
    lr : float, optional
        learning rate for the optimizer, by default 0.001
    weight_decay : float, optional
        weight decay for the optimizer, by default 0.0

    Raises
    ------
    ValueError
        Undefined optimizer
    """

    if optimizer_name == "Adam":
        self.optimizer = torch.optim.Adam(
            self.net.parameters(), lr=lr, weight_decay=weight_decay)
    elif optimizer_name == "SGD":
        self.optimizer = torch.optim.SGD(
            self.net.parameters(), lr=lr, weight_decay=weight_decay)
    else:
        raise ValueError("Optimizer not supported")
hf_predict(x: Tensor) -> Tensor ¤

predict the output of the network

Parameters:

Name Type Description Default
x Tensor

input data

 required

Returns:

Type Description
Tensor

predicted output data
Source code in src/MFVeBRNN/method/mf_nest_rnn_trainer.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
def hf_predict(self, x: Tensor) -> Tensor:
    """predict the output of the network

    Parameters
    ----------
    x : Tensor
        input data

    Returns
    -------
    Tensor
        predicted output data
    """
    # get the re-arranged input data
    x = self._re_arrange_input(x.to(self.device))
    self.best_net.eval()
    y = self.best_net.forward(x)

    return y.detach()
lf_predict(x: Tensor, return_var: bool = False) -> Tensor ¤

predict the output of the network

Parameters:

Name Type Description Default
x Tensor

input data

 required

Returns:

Type Description
Tensor

predicted output data
Source code in src/MFVeBRNN/method/mf_nest_rnn_trainer.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def lf_predict(self, x: Tensor, return_var: bool = False) -> Tensor:
    """predict the output of the network

    Parameters
    ----------
    x : Tensor
        input data

    Returns
    -------
    Tensor
        predicted output data
    """

    if isinstance(self.lf_model, RNNTrainer):
        y = self.lf_model.predict(x.to(self.device))

    elif isinstance(self.lf_model, VeBRNNTrainer):
        y, var_epistemic = self.lf_model.bayes_predict(
            x.to(self.device))
        var_aleatoric = self.lf_model.aleatoric_variance_predict(
            x.to(self.device))
        if return_var:
            return y, var_aleatoric, var_epistemic
    else:
        raise ValueError("Undefined low-fidelity model")

    return y
train(hx_train: Tensor, hy_train: Tensor, num_epochs: int, batch_size: int = None, hx_val: Tensor = None, hy_val: Tensor = None, verbose: bool = True, print_iter: int = 100) -> typing.Tuple[float] ¤

train the multi-fidelity rnn model

Parameters:

Name Type Description Default
hx_train Tensor

high-fidelity training input data

 required
hy_train Tensor

high-fidelity training output data

 required
num_epochs int

number of epochs to train

 required
batch_size int

size of the mini-batch, by default None

 None
hx_val Tensor

high-fidelity validation input data, by default None

 None
hy_val Tensor

high-fidelity validation output data, by default None

 None
verbose bool

whether to print training information, by default True

 True
print_iter int

frequency of printing training information, by default 100

 100
Source code in src/MFVeBRNN/method/mf_nest_rnn_trainer.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def train(self,
          hx_train: Tensor,
          hy_train: Tensor,
          num_epochs: int,
          batch_size: int = None,
          hx_val: Tensor = None,
          hy_val: Tensor = None,
          verbose: bool = True,
          print_iter: int = 100) -> Tuple[float]:
    """train the multi-fidelity rnn model

    Parameters
    ----------
    hx_train : Tensor
        high-fidelity training input data
    hy_train : Tensor
        high-fidelity training output data
    num_epochs : int
        number of epochs to train
    batch_size : int, optional
        size of the mini-batch, by default None
    hx_val : Tensor, optional
        high-fidelity validation input data, by default None
    hy_val : Tensor, optional
        high-fidelity validation output data, by default None
    verbose : bool, optional
        whether to print training information, by default True
    print_iter : int, optional
        frequency of printing training information, by default 100


    """
    # set the data to the device
    hx_train = hx_train.to(self.device)
    hy_train = hy_train.to(self.device)
    if hx_val is not None:
        hx_val = hx_val.to(self.device)
        hy_val = hy_val.to(self.device)

    # re-arrange the training and validation data
    hx_train = self._re_arrange_input(hx_train)
    if hx_val is not None:
        hx_val = self._re_arrange_input(hx_val)

    # record the minimum loss of the validation data
    min_loss = np.inf
    # loader for mini-batch
    if batch_size is None:
        self.batch_size = hx_train.shape[0]
        self.num_scale = 1.0
    else:
        self.batch_size = batch_size
        self.num_scale = hx_train.shape[0] / self.batch_size

    loader = DataLoader(
        dataset=list(zip(hx_train, hy_train)),
        batch_size=self.batch_size,
        shuffle=True
    )

    # begin the training process
    for epoch in range(num_epochs):

        # set the network to training mode
        self.net.train()
        # running loss for the training data
        running_loss_train = 0
        for X_batch, y_batch in loader:
            # set gradient of params to zero
            self.optimizer.zero_grad()
            # get prediction from network
            pred = self.net.forward(X_batch)
            # calculate the loss value for the batch
            loss = self.loss_function(pred, y_batch)
            # back propagation
            loss.backward()
            # update the weights
            self.optimizer.step()
            # accumulate the loss value
            running_loss_train += loss.item() * X_batch.size(0)

        # average the loss value
        loss_train = running_loss_train / hx_train.size(0)
        if hx_val is not None:
            self.net.eval()
            y_val_pred = self.net.forward(hx_val)
            loss_val = self.loss_function(hy_val, y_val_pred)
            if loss_val.item() < min_loss:
                # save the model with the best validation loss
                min_loss = loss_val.item()
                best_epoch = epoch
                self.best_net = copy.deepcopy(self.net)
        else:
            loss_val = torch.tensor(0.0)
            self.best_net = copy.deepcopy(self.net)
            min_loss = loss_val.item()
            best_epoch = epoch

        if verbose and epoch % print_iter == 0:
            self._print(epoch, num_epochs,
                        loss_train, loss_val.item())

    return min_loss, best_epoch

MFVeBRNN.method.MFResidualRNNTrainer ¤

multi-fidelity residual rnn trainer.

Source code in src/MFVeBRNN/method/mf_residual_rnn_trainer.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
class MFResidualRNNTrainer:
    """multi-fidelity residual rnn trainer.
    """

    def __init__(
        self,
        net: torch.nn.Module,
        pre_trained_lf_model: RNNTrainer | VeBRNNTrainer,
        device: torch.device = torch.device("cpu"),
        seed: int = 0,
        nest_option: str = "hidden",
    ) -> None:
        """initialize the trainer for the high-fidelity model

        Parameters
        ----------
        net : torch.nn.Module
            the high-fidelity neural network
        pre_trained_lf_model : RNNTrainer | VeBRNNTrainer
            the pre-trained low-fidelity model
        device : torch.device, optional
            the device to train the model on, by default torch.device("cpu")
        seed : int, optional
            the random seed for reproducibility, by default 0
        nest_option : str, optional
            nested option, by default "hidden" or "output"
        """
        self.device = device
        # load the net to the device
        self.net = net.to(self.device)
        # load the pre-trained low-fidelity model to the device
        self.lf_model: RNNTrainer | VeBRNNTrainer = pre_trained_lf_model

        # load to the device
        self.lf_model.device = self.device
        if isinstance(self.lf_model, RNNTrainer):
            self.lf_model.best_net = self.lf_model.best_net.to(self.device)
        elif isinstance(self.lf_model, VeBRNNTrainer):
            self.lf_model.mean_net = self.lf_model.mean_net.to(self.device)
            self.lf_model.var_net = self.lf_model.var_net.to(self.device)
        else:
            raise ValueError("Undefined low-fidelity model type")

        # set the seed and nest option
        self.seed = seed
        self.nest_option = nest_option

    def configure_optimizer_info(self,
                                 optimizer_name: str = "Adam",
                                 lr: float = 0.001,
                                 weight_decay: float = 0.0):
        """configure optimizer.

        Parameters
        ----------
        optimizer_name : str, optional
            optimizer name, by default "Adam"
        lr : float, optional
            learning rate, by default 0.001
        weight_decay : float, optional
            weight decay, by default 0.0

        Raises
        ------
        ValueError
            Undefined optimizer
        """

        if optimizer_name == "Adam":
            self.optimizer = torch.optim.Adam(
                self.net.parameters(), lr=lr, weight_decay=weight_decay)
        elif optimizer_name == "SGD":
            self.optimizer = torch.optim.SGD(
                self.net.parameters(), lr=lr, weight_decay=weight_decay)
        else:
            raise ValueError("Optimizer not supported")

    def configure_loss_function(self, loss_name: str = "MSE") -> None:
        """configure the loss function for the training process

        Parameters
        ----------
        loss_name : str, optional
            name of the loss function, by default "MSE"

        Raises
        ------
        ValueError
            Undefined loss function
        """
        if loss_name == "MSE":
            self.loss_function = torch.nn.MSELoss()
        elif loss_name == "MAE":
            self.loss_function = torch.nn.L1Loss()
        else:
            raise ValueError("Undefined loss function")

    def train(self,
              hx_train: Tensor,
              hy_train: Tensor,
              num_epochs: int,
              batch_size: int = None,
              hx_val: Tensor = None,
              hy_val: Tensor = None,
              verbose: bool = True,
              print_iter: int = 100,) -> Tuple[float, int]:
        """train the high-fidelity model

        Parameters
        ----------
        hx_train : Tensor
            high-fidelity train inputs
        hy_train : Tensor
            high-fidelity train outputs
        num_epochs : int
            num of epochs
        batch_size : int, optional
            batch size, by default None
        hx_val : Tensor, optional
            high-fidelity validation inputs, by default None
        hy_val : Tensor, optional
            high-fidelity validation inputs, by default None
        verbose : bool, optional
            verbose or not, by default True
        print_iter : int, optional
            print iteration, by default 100

        Returns
        -------
        Tuple[float, int]
            best validation loss and the corresponding epoch
        """

        # set the data to the device
        hx_train = hx_train.to(self.device)
        hy_train = hy_train.to(self.device)
        if hx_val is not None:
            hx_val = hx_val.to(self.device)
            hy_val = hy_val.to(self.device)

        # re-arrange the training and validation data for output
        hy_train = self._calculate_residual(hx_train, hy_train)
        if hx_val is not None:
            hy_val = self._calculate_residual(hx_val, hy_val)

        # re-arrange the training and validation data
        hx_train = self._re_arrange_input(hx_train)
        if hx_val is not None:
            hx_val = self._re_arrange_input(hx_val)
        min_loss = np.inf
        # loader for mini-batch
        if batch_size is None:
            self.batch_size = hx_train.shape[0]
            self.num_scale = 1.0
        else:
            self.batch_size = batch_size
            self.num_scale = hx_train.shape[0] / self.batch_size

        loader = DataLoader(
            dataset=list(zip(hx_train, hy_train)),
            batch_size=self.batch_size,
            shuffle=True
        )

        # begin the training process
        for epoch in range(num_epochs):

            # set the network to training mode
            self.net.train()
            # running loss for the training data
            running_loss_train = 0
            for X_batch, y_batch in loader:
                # set gradient of params to zero
                self.optimizer.zero_grad()
                # get prediction from network
                pred = self.net.forward(X_batch)
                # calculate the loss value for the batch
                loss = self.loss_function(pred, y_batch)
                # back propagation
                loss.backward()
                # update the weights
                self.optimizer.step()
                # accumulate the loss value
                running_loss_train += loss.item() * X_batch.size(0)

            # average the loss value
            loss_train = running_loss_train / hx_train.size(0)
            if hx_val is not None:
                self.net.eval()
                y_val_pred = self.net.forward(hx_val)
                loss_val = self.loss_function(hy_val, y_val_pred)
                if loss_val.item() < min_loss:
                    # save the model with the best validation loss
                    min_loss = loss_val.item()
                    best_epoch = epoch
                    self.best_net = copy.deepcopy(self.net)

            else:
                loss_val = torch.tensor(0.0)
                self.best_net = copy.deepcopy(self.net)
                min_loss = loss_val.item()
                best_epoch = epoch

            if verbose and epoch % print_iter == 0:
                self._print(epoch, num_epochs,
                            loss_train, loss_val.item())

        return min_loss, best_epoch

    def hf_predict(self, x: Tensor) -> Tensor:
        """predict the output of the network

        Parameters
        ----------
        x : Tensor
            input data

        Returns
        -------
        Tensor
            predicted output data
        """
        x = x.to(self.device)
        y_lf = self.lf_predict(x)
        # get the re-arranged input data
        x = self._re_arrange_input(x)
        self.best_net.eval()
        diff = self.best_net.forward(x)

        y = y_lf + diff

        return y.detach()

    def lf_predict(self, x: Tensor, return_var: bool = False) -> Tensor:
        """predict the output of the network

        Parameters
        ----------
        x : Tensor
            input data

        Returns
        -------
        Tensor
            predicted output data
        """

        if isinstance(self.lf_model, RNNTrainer):
            y = self.lf_model.predict(x.to(self.device))

        elif isinstance(self.lf_model, VeBRNNTrainer):
            y, var_epistemic = self.lf_model.bayes_predict(
                x.to(self.device))
            var_aleatoric = self.lf_model.aleatoric_variance_predict(
                x.to(self.device))
            if return_var:
                return y, var_aleatoric, var_epistemic
        else:
            raise ValueError("Undefined low-fidelity model")

        return y

    def _re_arrange_input(self,
                          x: Tensor) -> Tensor:
        """re-arrange the input data for the training process

        Parameters
        ----------
        x : Tensor
            input data for the training process or prediction

        Returns
        -------
        Tensor
            the re-arranged input data

        Raises
        ------
        ValueError
            Undefined nest option
        """

        if self.nest_option == "original":

            return x

        elif self.nest_option == "hidden":

            # predict the output of the low-fidelity model
            re_hx_input = self.lf_model.recurrent_forward(x)
            # re_hx_input = torch.stack(re_hx_input, dim=1)
            re_hx_input = re_hx_input.detach()
            # concatenate the data
            x = torch.cat((x, re_hx_input), dim=-1)

            return x

        else:
            raise ValueError("Undefined nest option")

    def _calculate_residual(self,
                            x: Tensor,
                            y: Tensor) -> Tensor:
        """calculate residual between the predicted output and the true output

        Parameters
        ----------
        x : Tensor
            input data
        y : Tensor
            true output data

        Returns
        -------
        Tensor
            residual between the predicted output and the true output
        """

        y_pred = self.lf_predict(x)
        residual = y - y_pred

        return residual.detach()

    def _print(
        self,
        epoch: int,
        num_epoch: int,
        loss_train: float,
        loss_val: float
    ) -> None:
        """print the loss values during training at certain epochs

        Parameters
        ----------
        epoch : int
            the current epoch
        num_epoch : int
            total number of epochs
        loss_train : float
            training loss value at the current epoch
        loss_val : float
            validation loss value at the current epoch
        """

        print(
            "Epoch/Total: %d/%d, Train Loss: %.3e, Val Loss: %.3e"
            % (
                epoch,
                num_epoch,
                loss_train,
                loss_val,
            )
        )
_calculate_residual(x: Tensor, y: Tensor) -> Tensor ¤

calculate residual between the predicted output and the true output

Parameters:

Name Type Description Default
x Tensor

input data

 required
y Tensor

true output data

 required

Returns:

Type Description
Tensor

residual between the predicted output and the true output
Source code in src/MFVeBRNN/method/mf_residual_rnn_trainer.py 
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
def _calculate_residual(self,
                        x: Tensor,
                        y: Tensor) -> Tensor:
    """calculate residual between the predicted output and the true output

    Parameters
    ----------
    x : Tensor
        input data
    y : Tensor
        true output data

    Returns
    -------
    Tensor
        residual between the predicted output and the true output
    """

    y_pred = self.lf_predict(x)
    residual = y - y_pred

    return residual.detach()
_print(epoch: int, num_epoch: int, loss_train: float, loss_val: float) -> None ¤

print the loss values during training at certain epochs

Parameters:

Name Type Description Default
epoch int

the current epoch

 required
num_epoch int

total number of epochs

 required
loss_train float

training loss value at the current epoch

 required
loss_val float

validation loss value at the current epoch

 required
Source code in src/MFVeBRNN/method/mf_residual_rnn_trainer.py 
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
def _print(
    self,
    epoch: int,
    num_epoch: int,
    loss_train: float,
    loss_val: float
) -> None:
    """print the loss values during training at certain epochs

    Parameters
    ----------
    epoch : int
        the current epoch
    num_epoch : int
        total number of epochs
    loss_train : float
        training loss value at the current epoch
    loss_val : float
        validation loss value at the current epoch
    """

    print(
        "Epoch/Total: %d/%d, Train Loss: %.3e, Val Loss: %.3e"
        % (
            epoch,
            num_epoch,
            loss_train,
            loss_val,
        )
    )
_re_arrange_input(x: Tensor) -> Tensor ¤

re-arrange the input data for the training process

Parameters:

Name Type Description Default
x Tensor

input data for the training process or prediction

 required

Returns:

Type Description
Tensor

the re-arranged input data

Raises:

Type Description
ValueError

Undefined nest option
Source code in src/MFVeBRNN/method/mf_residual_rnn_trainer.py 
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
def _re_arrange_input(self,
                      x: Tensor) -> Tensor:
    """re-arrange the input data for the training process

    Parameters
    ----------
    x : Tensor
        input data for the training process or prediction

    Returns
    -------
    Tensor
        the re-arranged input data

    Raises
    ------
    ValueError
        Undefined nest option
    """

    if self.nest_option == "original":

        return x

    elif self.nest_option == "hidden":

        # predict the output of the low-fidelity model
        re_hx_input = self.lf_model.recurrent_forward(x)
        # re_hx_input = torch.stack(re_hx_input, dim=1)
        re_hx_input = re_hx_input.detach()
        # concatenate the data
        x = torch.cat((x, re_hx_input), dim=-1)

        return x

    else:
        raise ValueError("Undefined nest option")
configure_loss_function(loss_name: str = 'MSE') -> None ¤

configure the loss function for the training process

Parameters:

Name Type Description Default
loss_name str

name of the loss function, by default "MSE"

 'MSE'

Raises:

Type Description
ValueError

Undefined loss function
Source code in src/MFVeBRNN/method/mf_residual_rnn_trainer.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def configure_loss_function(self, loss_name: str = "MSE") -> None:
    """configure the loss function for the training process

    Parameters
    ----------
    loss_name : str, optional
        name of the loss function, by default "MSE"

    Raises
    ------
    ValueError
        Undefined loss function
    """
    if loss_name == "MSE":
        self.loss_function = torch.nn.MSELoss()
    elif loss_name == "MAE":
        self.loss_function = torch.nn.L1Loss()
    else:
        raise ValueError("Undefined loss function")
configure_optimizer_info(optimizer_name: str = 'Adam', lr: float = 0.001, weight_decay: float = 0.0) ¤

configure optimizer.

Parameters:

Name Type Description Default
optimizer_name str

optimizer name, by default "Adam"

 'Adam'
lr float

learning rate, by default 0.001

 0.001
weight_decay float

weight decay, by default 0.0

 0.0

Raises:

Type Description
ValueError

Undefined optimizer
Source code in src/MFVeBRNN/method/mf_residual_rnn_trainer.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def configure_optimizer_info(self,
                             optimizer_name: str = "Adam",
                             lr: float = 0.001,
                             weight_decay: float = 0.0):
    """configure optimizer.

    Parameters
    ----------
    optimizer_name : str, optional
        optimizer name, by default "Adam"
    lr : float, optional
        learning rate, by default 0.001
    weight_decay : float, optional
        weight decay, by default 0.0

    Raises
    ------
    ValueError
        Undefined optimizer
    """

    if optimizer_name == "Adam":
        self.optimizer = torch.optim.Adam(
            self.net.parameters(), lr=lr, weight_decay=weight_decay)
    elif optimizer_name == "SGD":
        self.optimizer = torch.optim.SGD(
            self.net.parameters(), lr=lr, weight_decay=weight_decay)
    else:
        raise ValueError("Optimizer not supported")
hf_predict(x: Tensor) -> Tensor ¤

predict the output of the network

Parameters:

Name Type Description Default
x Tensor

input data

 required

Returns:

Type Description
Tensor

predicted output data
Source code in src/MFVeBRNN/method/mf_residual_rnn_trainer.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
def hf_predict(self, x: Tensor) -> Tensor:
    """predict the output of the network

    Parameters
    ----------
    x : Tensor
        input data

    Returns
    -------
    Tensor
        predicted output data
    """
    x = x.to(self.device)
    y_lf = self.lf_predict(x)
    # get the re-arranged input data
    x = self._re_arrange_input(x)
    self.best_net.eval()
    diff = self.best_net.forward(x)

    y = y_lf + diff

    return y.detach()
lf_predict(x: Tensor, return_var: bool = False) -> Tensor ¤

predict the output of the network

Parameters:

Name Type Description Default
x Tensor

input data

 required

Returns:

Type Description
Tensor

predicted output data
Source code in src/MFVeBRNN/method/mf_residual_rnn_trainer.py 
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
def lf_predict(self, x: Tensor, return_var: bool = False) -> Tensor:
    """predict the output of the network

    Parameters
    ----------
    x : Tensor
        input data

    Returns
    -------
    Tensor
        predicted output data
    """

    if isinstance(self.lf_model, RNNTrainer):
        y = self.lf_model.predict(x.to(self.device))

    elif isinstance(self.lf_model, VeBRNNTrainer):
        y, var_epistemic = self.lf_model.bayes_predict(
            x.to(self.device))
        var_aleatoric = self.lf_model.aleatoric_variance_predict(
            x.to(self.device))
        if return_var:
            return y, var_aleatoric, var_epistemic
    else:
        raise ValueError("Undefined low-fidelity model")

    return y
train(hx_train: Tensor, hy_train: Tensor, num_epochs: int, batch_size: int = None, hx_val: Tensor = None, hy_val: Tensor = None, verbose: bool = True, print_iter: int = 100) -> typing.Tuple[float, int] ¤

train the high-fidelity model

Parameters:

Name Type Description Default
hx_train Tensor

high-fidelity train inputs

 required
hy_train Tensor

high-fidelity train outputs

 required
num_epochs int

num of epochs

 required
batch_size int

batch size, by default None

 None
hx_val Tensor

high-fidelity validation inputs, by default None

 None
hy_val Tensor

high-fidelity validation inputs, by default None

 None
verbose bool

verbose or not, by default True

 True
print_iter int

print iteration, by default 100

 100

Returns:

Type Description
Tuple[float, int]

best validation loss and the corresponding epoch
Source code in src/MFVeBRNN/method/mf_residual_rnn_trainer.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
def train(self,
          hx_train: Tensor,
          hy_train: Tensor,
          num_epochs: int,
          batch_size: int = None,
          hx_val: Tensor = None,
          hy_val: Tensor = None,
          verbose: bool = True,
          print_iter: int = 100,) -> Tuple[float, int]:
    """train the high-fidelity model

    Parameters
    ----------
    hx_train : Tensor
        high-fidelity train inputs
    hy_train : Tensor
        high-fidelity train outputs
    num_epochs : int
        num of epochs
    batch_size : int, optional
        batch size, by default None
    hx_val : Tensor, optional
        high-fidelity validation inputs, by default None
    hy_val : Tensor, optional
        high-fidelity validation inputs, by default None
    verbose : bool, optional
        verbose or not, by default True
    print_iter : int, optional
        print iteration, by default 100

    Returns
    -------
    Tuple[float, int]
        best validation loss and the corresponding epoch
    """

    # set the data to the device
    hx_train = hx_train.to(self.device)
    hy_train = hy_train.to(self.device)
    if hx_val is not None:
        hx_val = hx_val.to(self.device)
        hy_val = hy_val.to(self.device)

    # re-arrange the training and validation data for output
    hy_train = self._calculate_residual(hx_train, hy_train)
    if hx_val is not None:
        hy_val = self._calculate_residual(hx_val, hy_val)

    # re-arrange the training and validation data
    hx_train = self._re_arrange_input(hx_train)
    if hx_val is not None:
        hx_val = self._re_arrange_input(hx_val)
    min_loss = np.inf
    # loader for mini-batch
    if batch_size is None:
        self.batch_size = hx_train.shape[0]
        self.num_scale = 1.0
    else:
        self.batch_size = batch_size
        self.num_scale = hx_train.shape[0] / self.batch_size

    loader = DataLoader(
        dataset=list(zip(hx_train, hy_train)),
        batch_size=self.batch_size,
        shuffle=True
    )

    # begin the training process
    for epoch in range(num_epochs):

        # set the network to training mode
        self.net.train()
        # running loss for the training data
        running_loss_train = 0
        for X_batch, y_batch in loader:
            # set gradient of params to zero
            self.optimizer.zero_grad()
            # get prediction from network
            pred = self.net.forward(X_batch)
            # calculate the loss value for the batch
            loss = self.loss_function(pred, y_batch)
            # back propagation
            loss.backward()
            # update the weights
            self.optimizer.step()
            # accumulate the loss value
            running_loss_train += loss.item() * X_batch.size(0)

        # average the loss value
        loss_train = running_loss_train / hx_train.size(0)
        if hx_val is not None:
            self.net.eval()
            y_val_pred = self.net.forward(hx_val)
            loss_val = self.loss_function(hy_val, y_val_pred)
            if loss_val.item() < min_loss:
                # save the model with the best validation loss
                min_loss = loss_val.item()
                best_epoch = epoch
                self.best_net = copy.deepcopy(self.net)

        else:
            loss_val = torch.tensor(0.0)
            self.best_net = copy.deepcopy(self.net)
            min_loss = loss_val.item()
            best_epoch = epoch

        if verbose and epoch % print_iter == 0:
            self._print(epoch, num_epochs,
                        loss_train, loss_val.item())

    return min_loss, best_epoch

MFVeBRNN.method.MFNestVeBRNNTrainer ¤

Source code in src/MFVeBRNN/method/mf_nest_vebrnn_trainer.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
class MFNestVeBRNNTrainer:

    def __init__(self,
                 mean_net: MeanNet,
                 var_net: GammaVarNet,
                 pre_trained_lf_model: RNNTrainer | VeBRNNTrainer,
                 device: torch.device = torch.device("cpu"),
                 job_id: int = 0,
                 nest_option: str = "hidden",
                 ) -> None:
        """initialize the trainer for the high-fidelity model

        mean_net : MeanNet
            mean neural network for the VeBNN
        var_net : GammaVarNet
            variance estimation neural network for VeBNN
        pre_trained_lf_model : RNNTrainer | VeBRNNTrainer
            the pre-trained low-fidelity model, which can be either RNNTrainer
            or VeBRNNTrainer
        device : torch.device, optional
            device for training, by default torch.device("cpu")
        job_id : int, optional
            job ID for the training, by default 0
        nest_option : str, optional
            nested option, by default "hidden" or "output"
        """
        # define the device
        self.device = device
        # load the pre-trained low-fidelity model
        self.lf_model: RNNTrainer | VeBRNNTrainer = pre_trained_lf_model
        # load the pre-trained low-fidelity model to the device
        self.lf_model.device = self.device
        if isinstance(self.lf_model, RNNTrainer):
            self.lf_model.best_net = self.lf_model.best_net.to(self.device)
        elif isinstance(self.lf_model, VeBRNNTrainer):
            self.lf_model.mean_net = self.lf_model.mean_net.to(self.device)
            self.lf_model.var_net = self.lf_model.var_net.to(self.device)

        # get the mean and variance network architecture
        self.un_trained_mean_net = mean_net.to(self.device)
        self.un_trained_var_net = var_net.to(self.device)
        # seed of the model
        self.job_id = job_id
        # the nested option
        self.nest_option = nest_option

        # init the high-fidelity trainer
        self.hf_vebrnn_trainer = self._init_hf_trainer()


    def cooperative_train(self,
                          x_train: Tensor,
                          y_train: Tensor,
                          iteration: int,
                          init_config={
                              "loss_name": "MSE",
                              "optimizer_name": "Adam",
                              "lr": 1e-3,
                              "weight_decay": 1e-6,
                              "num_epochs": 1000,
                              "batch_size": 200,
                              "verbose": False,
                              "print_iter": 50,
                              "split_ratio": 0.8,
                          },
                          var_config={
                              "optimizer_name": "Adam",
                              "lr": 1e-3,
                              "num_epochs": 1000,
                              "batch_size": 200,
                              "verbose": True,
                              "print_iter": 50,
                              "early_stopping": False,
                              "early_stopping_iter": 100,
                              "early_stopping_tol": 1e-4,
                          },
                          sampler_config={
                              "sampler": "pSGLD",
                              "lr": 1e-3,
                              "gamma": 0.9999,
                              "num_epochs": 2000,
                              "mix_epochs": 10,
                              "burn_in_epochs": 500,
                              "batch_size": 200,
                              "verbose": False,
                              "print_iter": 100,
                          },
                          delete_model_raw_data=True,
                          ) -> None:
        """train the high-fidelity model with the cooperative training process
        Parameters
        ----------
        x_train : Tensor
            input training data
        y_train : Tensor
            output training data
        iteration : int
            the iteration of the cooperative training process
        init_config : dict, optional
            configuration for the initialization of the high-fidelity model, by
            default is a dictionary with the following keys and values:
            {
                "loss_name": "MSE",
                "optimizer_name": "Adam",
                "lr": 1e-3,
                "weight_decay": 1e-6,
                "num_epochs": 1000,
                "batch_size": 200,
                "verbose": False,
                "print_iter": 50,
                "split_ratio": 0.8,
            }
        var_config : dict, optional
            configuration for the variance estimation of the high-fidelity
            model, by default is a dictionary with the following keys and
            values:
            {
                "optimizer_name": "Adam",
                "lr": 1e-3,
                "num_epochs": 1000,
                "batch_size": 200,
                "verbose": True,
                "print_iter": 50,
                "early_stopping": False,
                "early_stopping_iter": 100,
                "early_stopping_tol": 1e-4,
            }
        sampler_config : dict, optional
            configuration for the sampler of the high-fidelity model,
            by default is a dictionary with the following keys and values:
            {
                "sampler": "pSGLD",
                "lr": 1e-3,
                "gamma": 0.9999,
                "num_epochs": 2000,
                "mix_epochs": 10,
                "burn_in_epochs": 500,
                "batch_size": 200,
                "verbose": False,
                "print_iter": 100,
            }
        delete_model_raw_data : bool, optional
            whether to delete the raw data of the model after training,
            by default True
        """
        x_train = x_train.to(self.device)
        y_train = y_train.to(self.device)

        # re-arrange the input data
        x_train = self._re_arrange_input(x_train)

        # train the residual model with the VeBNN trainer
        self.hf_vebrnn_trainer.cooperative_train(
            x_train=x_train,
            y_train=y_train,
            iteration=iteration,
            init_config=init_config,
            var_config=var_config,
            sampler_config=sampler_config,
            delete_model_raw_data=delete_model_raw_data,
        )

    def hf_bayes_predict(
        self,
        x: torch.Tensor,
        save_ppd: bool = False,
    ) -> Tuple[Tensor, Tensor]:
        """Predict the mean and variance of the output at the scaled data.

        Parameters
        ----------
        x : torch.Tensor
            Test data points.
        save_ppd : bool, optional
            Whether to save ppd or not (default is False).

        Returns
        -------
        Tuple[Tensor, Tensor]
            Predicted mean and variance at the scaled space.
        """
        x = x.to(self.device)

        x = self._re_arrange_input(x)

        # get the prediction from the residual model
        y_pred_mean, y_pred_var = (
            self.hf_vebrnn_trainer.bayes_predict(x, save_ppd=save_ppd)
        )

        if save_ppd:
            self.responses = self.hf_vebrnn_trainer.responses

        return y_pred_mean.detach(), y_pred_var.detach()

    def hf_aleatoric_variance_predict(self, x: Tensor) -> Tensor:
        """Predict the aleatoric variance of the output at the scaled data.

        Parameters
        ----------
        x : Tensor
            Test data points.

        Returns
        -------
        Tensor
            Predicted aleatoric variance at the scaled space.
        """
        x = x.to(self.device)
        # get the re-arranged input data
        x = self._re_arrange_input(x)

        # get the aleatoric variance prediction from the residual model
        var_aleatoric = self.hf_vebrnn_trainer.aleatoric_variance_predict(x)

        return var_aleatoric.detach()

    def lf_predict(self, x: Tensor, return_var: bool = False) -> Tensor:
        """predict the output of the network

        Parameters
        ----------
        x : Tensor
            input data

        Returns
        -------
        Tensor
            predicted output data
        """

        if isinstance(self.lf_model, RNNTrainer):
            y = self.lf_model.predict(x.to(self.device))
        elif isinstance(self.lf_model, VeBRNNTrainer):
            y, var_epistemic = self.lf_model.bayes_predict(
                x.to(self.device))
            var_aleatoric = self.lf_model.aleatoric_variance_predict(
                x.to(self.device))
            if return_var:
                return y, var_aleatoric, var_epistemic

        return y

    def _re_arrange_input(self,
                          x: Tensor) -> Tensor:
        """re-arrange the input data for the training process

        Parameters
        ----------
        x : Tensor
            input data for the training process or prediction

        Returns
        -------
        Tensor
            the re-arranged input data

        Raises
        ------
        ValueError
            Undefined nest option
        """

        if self.nest_option == "output":

            # predict the output of the low-fidelity model
            re_hx_input = self.lf_model.predict(x)
            # concatenate the data
            x = torch.cat((x, re_hx_input), dim=2)

            return x

        elif self.nest_option == "hidden":

            # predict the output of the low-fidelity model
            re_hx_input = self.lf_model.recurrent_forward(x)
            re_hx_input = re_hx_input.detach()
            # concatenate the data
            x = torch.cat((x, re_hx_input), dim=-1)

            return x

        else:
            raise ValueError("Undefined nest option")


    def _init_hf_trainer(self) -> VeBRNNTrainer:
        """init the high-fidelity trainer.

        Returns
        -------
        VeBRNNTrainer
            high fidelity trainer.
        """
        vebrnn_trainer = VeBRNNTrainer(
            mean_net=self.un_trained_mean_net,
            var_net=self.un_trained_var_net,
            device=self.device,
            job_id=self.job_id,
        )
        return vebrnn_trainer
_init_hf_trainer() -> VeBRNNTrainer ¤

init the high-fidelity trainer.

Returns:

Type Description
VeBRNNTrainer

high fidelity trainer.
Source code in src/MFVeBRNN/method/mf_nest_vebrnn_trainer.py 
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
def _init_hf_trainer(self) -> VeBRNNTrainer:
    """init the high-fidelity trainer.

    Returns
    -------
    VeBRNNTrainer
        high fidelity trainer.
    """
    vebrnn_trainer = VeBRNNTrainer(
        mean_net=self.un_trained_mean_net,
        var_net=self.un_trained_var_net,
        device=self.device,
        job_id=self.job_id,
    )
    return vebrnn_trainer
_re_arrange_input(x: Tensor) -> Tensor ¤

re-arrange the input data for the training process

Parameters:

Name Type Description Default
x Tensor

input data for the training process or prediction

 required

Returns:

Type Description
Tensor

the re-arranged input data

Raises:

Type Description
ValueError

Undefined nest option
Source code in src/MFVeBRNN/method/mf_nest_vebrnn_trainer.py 
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
def _re_arrange_input(self,
                      x: Tensor) -> Tensor:
    """re-arrange the input data for the training process

    Parameters
    ----------
    x : Tensor
        input data for the training process or prediction

    Returns
    -------
    Tensor
        the re-arranged input data

    Raises
    ------
    ValueError
        Undefined nest option
    """

    if self.nest_option == "output":

        # predict the output of the low-fidelity model
        re_hx_input = self.lf_model.predict(x)
        # concatenate the data
        x = torch.cat((x, re_hx_input), dim=2)

        return x

    elif self.nest_option == "hidden":

        # predict the output of the low-fidelity model
        re_hx_input = self.lf_model.recurrent_forward(x)
        re_hx_input = re_hx_input.detach()
        # concatenate the data
        x = torch.cat((x, re_hx_input), dim=-1)

        return x

    else:
        raise ValueError("Undefined nest option")
cooperative_train(x_train: Tensor, y_train: Tensor, iteration: int, init_config={'loss_name': 'MSE', 'optimizer_name': 'Adam', 'lr': 0.001, 'weight_decay': 1e-06, 'num_epochs': 1000, 'batch_size': 200, 'verbose': False, 'print_iter': 50, 'split_ratio': 0.8}, var_config={'optimizer_name': 'Adam', 'lr': 0.001, 'num_epochs': 1000, 'batch_size': 200, 'verbose': True, 'print_iter': 50, 'early_stopping': False, 'early_stopping_iter': 100, 'early_stopping_tol': 0.0001}, sampler_config={'sampler': 'pSGLD', 'lr': 0.001, 'gamma': 0.9999, 'num_epochs': 2000, 'mix_epochs': 10, 'burn_in_epochs': 500, 'batch_size': 200, 'verbose': False, 'print_iter': 100}, delete_model_raw_data=True) -> None ¤

train the high-fidelity model with the cooperative training process

Parameters:

Name Type Description Default
x_train Tensor

input training data

 required
y_train Tensor

output training data

 required
iteration int

the iteration of the cooperative training process

 required
init_config dict

configuration for the initialization of the high-fidelity model, by default is a dictionary with the following keys and values: { "loss_name": "MSE", "optimizer_name": "Adam", "lr": 1e-3, "weight_decay": 1e-6, "num_epochs": 1000, "batch_size": 200, "verbose": False, "print_iter": 50, "split_ratio": 0.8, }

 {'loss_name': 'MSE', 'optimizer_name': 'Adam', 'lr': 0.001, 'weight_decay': 1e-06, 'num_epochs': 1000, 'batch_size': 200, 'verbose': False, 'print_iter': 50, 'split_ratio': 0.8}
var_config dict

configuration for the variance estimation of the high-fidelity model, by default is a dictionary with the following keys and values: { "optimizer_name": "Adam", "lr": 1e-3, "num_epochs": 1000, "batch_size": 200, "verbose": True, "print_iter": 50, "early_stopping": False, "early_stopping_iter": 100, "early_stopping_tol": 1e-4, }

 {'optimizer_name': 'Adam', 'lr': 0.001, 'num_epochs': 1000, 'batch_size': 200, 'verbose': True, 'print_iter': 50, 'early_stopping': False, 'early_stopping_iter': 100, 'early_stopping_tol': 0.0001}
sampler_config dict

configuration for the sampler of the high-fidelity model, by default is a dictionary with the following keys and values: { "sampler": "pSGLD", "lr": 1e-3, "gamma": 0.9999, "num_epochs": 2000, "mix_epochs": 10, "burn_in_epochs": 500, "batch_size": 200, "verbose": False, "print_iter": 100, }

 {'sampler': 'pSGLD', 'lr': 0.001, 'gamma': 0.9999, 'num_epochs': 2000, 'mix_epochs': 10, 'burn_in_epochs': 500, 'batch_size': 200, 'verbose': False, 'print_iter': 100}
delete_model_raw_data bool

whether to delete the raw data of the model after training, by default True

 True
Source code in src/MFVeBRNN/method/mf_nest_vebrnn_trainer.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def cooperative_train(self,
                      x_train: Tensor,
                      y_train: Tensor,
                      iteration: int,
                      init_config={
                          "loss_name": "MSE",
                          "optimizer_name": "Adam",
                          "lr": 1e-3,
                          "weight_decay": 1e-6,
                          "num_epochs": 1000,
                          "batch_size": 200,
                          "verbose": False,
                          "print_iter": 50,
                          "split_ratio": 0.8,
                      },
                      var_config={
                          "optimizer_name": "Adam",
                          "lr": 1e-3,
                          "num_epochs": 1000,
                          "batch_size": 200,
                          "verbose": True,
                          "print_iter": 50,
                          "early_stopping": False,
                          "early_stopping_iter": 100,
                          "early_stopping_tol": 1e-4,
                      },
                      sampler_config={
                          "sampler": "pSGLD",
                          "lr": 1e-3,
                          "gamma": 0.9999,
                          "num_epochs": 2000,
                          "mix_epochs": 10,
                          "burn_in_epochs": 500,
                          "batch_size": 200,
                          "verbose": False,
                          "print_iter": 100,
                      },
                      delete_model_raw_data=True,
                      ) -> None:
    """train the high-fidelity model with the cooperative training process
    Parameters
    ----------
    x_train : Tensor
        input training data
    y_train : Tensor
        output training data
    iteration : int
        the iteration of the cooperative training process
    init_config : dict, optional
        configuration for the initialization of the high-fidelity model, by
        default is a dictionary with the following keys and values:
        {
            "loss_name": "MSE",
            "optimizer_name": "Adam",
            "lr": 1e-3,
            "weight_decay": 1e-6,
            "num_epochs": 1000,
            "batch_size": 200,
            "verbose": False,
            "print_iter": 50,
            "split_ratio": 0.8,
        }
    var_config : dict, optional
        configuration for the variance estimation of the high-fidelity
        model, by default is a dictionary with the following keys and
        values:
        {
            "optimizer_name": "Adam",
            "lr": 1e-3,
            "num_epochs": 1000,
            "batch_size": 200,
            "verbose": True,
            "print_iter": 50,
            "early_stopping": False,
            "early_stopping_iter": 100,
            "early_stopping_tol": 1e-4,
        }
    sampler_config : dict, optional
        configuration for the sampler of the high-fidelity model,
        by default is a dictionary with the following keys and values:
        {
            "sampler": "pSGLD",
            "lr": 1e-3,
            "gamma": 0.9999,
            "num_epochs": 2000,
            "mix_epochs": 10,
            "burn_in_epochs": 500,
            "batch_size": 200,
            "verbose": False,
            "print_iter": 100,
        }
    delete_model_raw_data : bool, optional
        whether to delete the raw data of the model after training,
        by default True
    """
    x_train = x_train.to(self.device)
    y_train = y_train.to(self.device)

    # re-arrange the input data
    x_train = self._re_arrange_input(x_train)

    # train the residual model with the VeBNN trainer
    self.hf_vebrnn_trainer.cooperative_train(
        x_train=x_train,
        y_train=y_train,
        iteration=iteration,
        init_config=init_config,
        var_config=var_config,
        sampler_config=sampler_config,
        delete_model_raw_data=delete_model_raw_data,
    )
hf_aleatoric_variance_predict(x: Tensor) -> Tensor ¤

Predict the aleatoric variance of the output at the scaled data.

Parameters:

Name Type Description Default
x Tensor

Test data points.

 required

Returns:

Type Description
Tensor

Predicted aleatoric variance at the scaled space.
Source code in src/MFVeBRNN/method/mf_nest_vebrnn_trainer.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def hf_aleatoric_variance_predict(self, x: Tensor) -> Tensor:
    """Predict the aleatoric variance of the output at the scaled data.

    Parameters
    ----------
    x : Tensor
        Test data points.

    Returns
    -------
    Tensor
        Predicted aleatoric variance at the scaled space.
    """
    x = x.to(self.device)
    # get the re-arranged input data
    x = self._re_arrange_input(x)

    # get the aleatoric variance prediction from the residual model
    var_aleatoric = self.hf_vebrnn_trainer.aleatoric_variance_predict(x)

    return var_aleatoric.detach()
hf_bayes_predict(x: Tensor, save_ppd: bool = False) -> typing.Tuple[torch.Tensor, torch.Tensor] ¤

Predict the mean and variance of the output at the scaled data.

Parameters:

Name Type Description Default
x Tensor

Test data points.

 required
save_ppd bool

Whether to save ppd or not (default is False).

 False

Returns:

Type Description
Tuple[Tensor, Tensor]

Predicted mean and variance at the scaled space.
Source code in src/MFVeBRNN/method/mf_nest_vebrnn_trainer.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def hf_bayes_predict(
    self,
    x: torch.Tensor,
    save_ppd: bool = False,
) -> Tuple[Tensor, Tensor]:
    """Predict the mean and variance of the output at the scaled data.

    Parameters
    ----------
    x : torch.Tensor
        Test data points.
    save_ppd : bool, optional
        Whether to save ppd or not (default is False).

    Returns
    -------
    Tuple[Tensor, Tensor]
        Predicted mean and variance at the scaled space.
    """
    x = x.to(self.device)

    x = self._re_arrange_input(x)

    # get the prediction from the residual model
    y_pred_mean, y_pred_var = (
        self.hf_vebrnn_trainer.bayes_predict(x, save_ppd=save_ppd)
    )

    if save_ppd:
        self.responses = self.hf_vebrnn_trainer.responses

    return y_pred_mean.detach(), y_pred_var.detach()
lf_predict(x: Tensor, return_var: bool = False) -> Tensor ¤

predict the output of the network

Parameters:

Name Type Description Default
x Tensor

input data

 required

Returns:

Type Description
Tensor

predicted output data
Source code in src/MFVeBRNN/method/mf_nest_vebrnn_trainer.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def lf_predict(self, x: Tensor, return_var: bool = False) -> Tensor:
    """predict the output of the network

    Parameters
    ----------
    x : Tensor
        input data

    Returns
    -------
    Tensor
        predicted output data
    """

    if isinstance(self.lf_model, RNNTrainer):
        y = self.lf_model.predict(x.to(self.device))
    elif isinstance(self.lf_model, VeBRNNTrainer):
        y, var_epistemic = self.lf_model.bayes_predict(
            x.to(self.device))
        var_aleatoric = self.lf_model.aleatoric_variance_predict(
            x.to(self.device))
        if return_var:
            return y, var_aleatoric, var_epistemic

    return y

MFVeBRNN.method.MFResidualVeBRNNTrainer ¤

Source code in src/MFVeBRNN/method/mf_residual_vebrnn_trainer.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
class MFResidualVeBRNNTrainer:

    def __init__(self,
                 mean_net: MeanNet,
                 var_net: GammaVarNet,
                 pre_trained_lf_model: RNNTrainer | VeBRNNTrainer,
                 device: torch.device = torch.device("cpu"),
                 job_id: int = 0,
                 nest_option: str = "hidden",
                 ) -> None:
        """initialize the trainer for the high-fidelity model

        Parameters
        ----------
        mean_net : MeanNet
            the mean network for the high-fidelity model
        var_net : GammaVarNet
            the variance network for the high-fidelity model
        pre_trained_lf_model : RNNTrainer | VeBRNNTrainer
            the pre-trained low-fidelity model
        device : torch.device, optional
            the device to train the model on, by default torch.device("cpu")
        job_id : int, optional
            the job ID for reproducibility, by default 0
        nest_option : str, optional
            the nested option, by default "hidden"
        """
        # define the device
        self.device = device
        # load the pre-trained low-fidelity model
        self.lf_model: RNNTrainer | VeBRNNTrainer = pre_trained_lf_model
        # load the pre-trained low-fidelity model to the device
        self.lf_model.device = self.device
        if isinstance(self.lf_model, RNNTrainer):
            self.lf_model.best_net = self.lf_model.best_net.to(self.device)
        elif isinstance(self.lf_model, VeBRNNTrainer):
            self.lf_model.mean_net = self.lf_model.mean_net.to(self.device)
            self.lf_model.var_net = self.lf_model.var_net.to(self.device)

        # get the mean and variance network architecture
        self.un_trained_mean_net = mean_net.to(self.device)
        self.un_trained_var_net = var_net.to(self.device)
        # seed of the model
        self.job_id = job_id
        # the nested option
        self.nest_option = nest_option

        # init the high-fidelity trainer
        self.hf_vebrnn_trainer = self._init_hf_trainer()


    def cooperative_train(self,
                          x_train: Tensor,
                          y_train: Tensor,
                          iteration: int,
                          init_config={
                              "loss_name": "MSE",
                              "optimizer_name": "Adam",
                              "lr": 1e-3,
                              "weight_decay": 1e-6,
                              "num_epochs": 1000,
                              "batch_size": 200,
                              "verbose": False,
                              "print_iter": 50,
                              "split_ratio": 0.8,
                          },
                          var_config={
                              "optimizer_name": "Adam",
                              "lr": 1e-3,
                              "num_epochs": 1000,
                              "batch_size": 200,
                              "verbose": True,
                              "print_iter": 50,
                              "early_stopping": False,
                              "early_stopping_iter": 100,
                              "early_stopping_tol": 1e-4,
                          },
                          sampler_config={
                              "sampler": "pSGLD",
                              "lr": 1e-3,
                              "gamma": 0.9999,
                              "num_epochs": 2000,   # SGMCMC epochs
                              "mix_epochs": 10,     # thinning interval
                              "burn_in_epochs": 500,
                              "batch_size": 200,
                              "verbose": False,
                              "print_iter": 100,
                          },
                          delete_model_raw_data=True,
                          ) -> None:
        """Train the high-fidelity model cooperatively with low-fidelity model.

        Parameters
        ----------
        x_train : Tensor
            Training input data
        y_train : Tensor
            Training output data
        iteration : int
            Number of iterations
        init_config : dict, optional
            Initial training configuration
        var_config : dict, optional
            Variance network training configuration
        sampler_config : dict, optional
            Sampler configuration for SGMCMC
        delete_model_raw_data : bool, optional
            Whether to delete raw data after training
        """
        x_train = x_train.to(self.device)
        y_train = y_train.to(self.device)
        # re-arrange the input data
        y_train = self._calculate_residual(x_train, y_train)
        x_train = self._re_arrange_input(x_train)
        # train the residual model with the VeBNN trainer
        self.hf_vebrnn_trainer.cooperative_train(
            x_train=x_train,
            y_train=y_train,
            iteration=iteration,
            init_config=init_config,
            var_config=var_config,
            sampler_config=sampler_config,
            delete_model_raw_data=delete_model_raw_data,
        )

    def hf_bayes_predict(
        self,
        x: torch.Tensor,
        save_ppd: bool = False,
    ) -> Tuple[Tensor, Tensor]:
        """Predict the mean and variance of the output at the scaled data.

        Parameters
        ----------
        x : torch.Tensor
            Test data points.
        save_ppd : bool, optional
            Whether to save ppd or not (default is False).

        Returns
        -------
        Tuple[Tensor, Tensor]
            Predicted mean and variance at the scaled space.
        """
        x = x.to(self.device)
        y_lf = self.lf_predict(x)
        # get the re-arranged input data
        x = self._re_arrange_input(x)

        # get the prediction from the residual model
        y_pred_mean, y_pred_var = (
            self.hf_vebrnn_trainer.bayes_predict(x, save_ppd=save_ppd)
        )

        if save_ppd:
            self.responses = self.hf_vebrnn_trainer.responses + y_lf

        # Add the low-fidelity model prediction
        y_pred_mean += y_lf

        return y_pred_mean.detach(), y_pred_var.detach()

    def hf_aleatoric_variance_predict(self, x: Tensor) -> Tensor:
        """Predict the aleatoric variance of the output at the scaled data.

        Parameters
        ----------
        x : Tensor
            Test data points.

        Returns
        -------
        Tensor
            Predicted aleatoric variance at the scaled space.
        """
        x = x.to(self.device)
        # get the re-arranged input data
        x = self._re_arrange_input(x)

        # get the aleatoric variance prediction from the residual model
        var_aleatoric = self.hf_vebrnn_trainer.aleatoric_variance_predict(x)

        return var_aleatoric.detach()

    def lf_predict(self, x: Tensor, return_var: bool = False) -> Tensor:
        """predict the output of the network

        Parameters
        ----------
        x : Tensor
            input data

        Returns
        -------
        Tensor
            predicted output data
        """

        if isinstance(self.lf_model, RNNTrainer):
            y = self.lf_model.predict(x.to(self.device))
        elif isinstance(self.lf_model, VeBRNNTrainer):
            y, var_epistemic = self.lf_model.bayes_predict(
                x.to(self.device))
            var_aleatoric = self.lf_model.aleatoric_variance_predict(
                x.to(self.device))
            if return_var:
                return y, var_aleatoric, var_epistemic

        return y

    def _re_arrange_input(self,
                          x: Tensor) -> Tensor:
        """re-arrange the input data for the training process

        Parameters
        ----------
        x : Tensor
            input data for the training process or prediction

        Returns
        -------
        Tensor
            the re-arranged input data

        Raises
        ------
        ValueError
            Undefined nest option
        """

        if self.nest_option == "original":

            return x

        elif self.nest_option == "hidden":

            # predict the output of the low-fidelity model
            re_hx_input = self.lf_model.recurrent_forward(x)
            re_hx_input = re_hx_input.detach()
            # concatenate the data
            x = torch.cat((x, re_hx_input), dim=-1)

            return x

        else:
            raise ValueError("Undefined nest option")

    def _calculate_residual(self,
                            x: Tensor,
                            y: Tensor) -> Tensor:
        """calculate residual between the predicted output and the true output

        Parameters
        ----------
        x : Tensor
            input data
        y : Tensor
            true output data

        Returns
        -------
        Tensor
            residual between the predicted output and the true output
        """

        y_pred = self.lf_predict(x)
        residual = y - y_pred

        return residual.detach()


    def _init_hf_trainer(self) -> VeBRNNTrainer:
        """init the high-fidelity trainer.

        Returns
        -------
        VeBRNNTrainer
            high fidelity trainer.
        """
        vebrnn_trainer = VeBRNNTrainer(
            mean_net=self.un_trained_mean_net,
            var_net=self.un_trained_var_net,
            device=self.device,
            job_id=self.job_id,
        )
        return vebrnn_trainer
_calculate_residual(x: Tensor, y: Tensor) -> Tensor ¤

calculate residual between the predicted output and the true output

Parameters:

Name Type Description Default
x Tensor

input data

 required
y Tensor

true output data

 required

Returns:

Type Description
Tensor

residual between the predicted output and the true output
Source code in src/MFVeBRNN/method/mf_residual_vebrnn_trainer.py 
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
def _calculate_residual(self,
                        x: Tensor,
                        y: Tensor) -> Tensor:
    """calculate residual between the predicted output and the true output

    Parameters
    ----------
    x : Tensor
        input data
    y : Tensor
        true output data

    Returns
    -------
    Tensor
        residual between the predicted output and the true output
    """

    y_pred = self.lf_predict(x)
    residual = y - y_pred

    return residual.detach()
_init_hf_trainer() -> VeBRNNTrainer ¤

init the high-fidelity trainer.

Returns:

Type Description
VeBRNNTrainer

high fidelity trainer.
Source code in src/MFVeBRNN/method/mf_residual_vebrnn_trainer.py 
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
def _init_hf_trainer(self) -> VeBRNNTrainer:
    """init the high-fidelity trainer.

    Returns
    -------
    VeBRNNTrainer
        high fidelity trainer.
    """
    vebrnn_trainer = VeBRNNTrainer(
        mean_net=self.un_trained_mean_net,
        var_net=self.un_trained_var_net,
        device=self.device,
        job_id=self.job_id,
    )
    return vebrnn_trainer
_re_arrange_input(x: Tensor) -> Tensor ¤

re-arrange the input data for the training process

Parameters:

Name Type Description Default
x Tensor

input data for the training process or prediction

 required

Returns:

Type Description
Tensor

the re-arranged input data

Raises:

Type Description
ValueError

Undefined nest option
Source code in src/MFVeBRNN/method/mf_residual_vebrnn_trainer.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
def _re_arrange_input(self,
                      x: Tensor) -> Tensor:
    """re-arrange the input data for the training process

    Parameters
    ----------
    x : Tensor
        input data for the training process or prediction

    Returns
    -------
    Tensor
        the re-arranged input data

    Raises
    ------
    ValueError
        Undefined nest option
    """

    if self.nest_option == "original":

        return x

    elif self.nest_option == "hidden":

        # predict the output of the low-fidelity model
        re_hx_input = self.lf_model.recurrent_forward(x)
        re_hx_input = re_hx_input.detach()
        # concatenate the data
        x = torch.cat((x, re_hx_input), dim=-1)

        return x

    else:
        raise ValueError("Undefined nest option")
cooperative_train(x_train: Tensor, y_train: Tensor, iteration: int, init_config={'loss_name': 'MSE', 'optimizer_name': 'Adam', 'lr': 0.001, 'weight_decay': 1e-06, 'num_epochs': 1000, 'batch_size': 200, 'verbose': False, 'print_iter': 50, 'split_ratio': 0.8}, var_config={'optimizer_name': 'Adam', 'lr': 0.001, 'num_epochs': 1000, 'batch_size': 200, 'verbose': True, 'print_iter': 50, 'early_stopping': False, 'early_stopping_iter': 100, 'early_stopping_tol': 0.0001}, sampler_config={'sampler': 'pSGLD', 'lr': 0.001, 'gamma': 0.9999, 'num_epochs': 2000, 'mix_epochs': 10, 'burn_in_epochs': 500, 'batch_size': 200, 'verbose': False, 'print_iter': 100}, delete_model_raw_data=True) -> None ¤

Train the high-fidelity model cooperatively with low-fidelity model.

Parameters:

Name Type Description Default
x_train Tensor

Training input data

 required
y_train Tensor

Training output data

 required
iteration int

Number of iterations

 required
init_config dict

Initial training configuration

 {'loss_name': 'MSE', 'optimizer_name': 'Adam', 'lr': 0.001, 'weight_decay': 1e-06, 'num_epochs': 1000, 'batch_size': 200, 'verbose': False, 'print_iter': 50, 'split_ratio': 0.8}
var_config dict

Variance network training configuration

 {'optimizer_name': 'Adam', 'lr': 0.001, 'num_epochs': 1000, 'batch_size': 200, 'verbose': True, 'print_iter': 50, 'early_stopping': False, 'early_stopping_iter': 100, 'early_stopping_tol': 0.0001}
sampler_config dict

Sampler configuration for SGMCMC

 {'sampler': 'pSGLD', 'lr': 0.001, 'gamma': 0.9999, 'num_epochs': 2000, 'mix_epochs': 10, 'burn_in_epochs': 500, 'batch_size': 200, 'verbose': False, 'print_iter': 100}
delete_model_raw_data bool

Whether to delete raw data after training

 True
Source code in src/MFVeBRNN/method/mf_residual_vebrnn_trainer.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def cooperative_train(self,
                      x_train: Tensor,
                      y_train: Tensor,
                      iteration: int,
                      init_config={
                          "loss_name": "MSE",
                          "optimizer_name": "Adam",
                          "lr": 1e-3,
                          "weight_decay": 1e-6,
                          "num_epochs": 1000,
                          "batch_size": 200,
                          "verbose": False,
                          "print_iter": 50,
                          "split_ratio": 0.8,
                      },
                      var_config={
                          "optimizer_name": "Adam",
                          "lr": 1e-3,
                          "num_epochs": 1000,
                          "batch_size": 200,
                          "verbose": True,
                          "print_iter": 50,
                          "early_stopping": False,
                          "early_stopping_iter": 100,
                          "early_stopping_tol": 1e-4,
                      },
                      sampler_config={
                          "sampler": "pSGLD",
                          "lr": 1e-3,
                          "gamma": 0.9999,
                          "num_epochs": 2000,   # SGMCMC epochs
                          "mix_epochs": 10,     # thinning interval
                          "burn_in_epochs": 500,
                          "batch_size": 200,
                          "verbose": False,
                          "print_iter": 100,
                      },
                      delete_model_raw_data=True,
                      ) -> None:
    """Train the high-fidelity model cooperatively with low-fidelity model.

    Parameters
    ----------
    x_train : Tensor
        Training input data
    y_train : Tensor
        Training output data
    iteration : int
        Number of iterations
    init_config : dict, optional
        Initial training configuration
    var_config : dict, optional
        Variance network training configuration
    sampler_config : dict, optional
        Sampler configuration for SGMCMC
    delete_model_raw_data : bool, optional
        Whether to delete raw data after training
    """
    x_train = x_train.to(self.device)
    y_train = y_train.to(self.device)
    # re-arrange the input data
    y_train = self._calculate_residual(x_train, y_train)
    x_train = self._re_arrange_input(x_train)
    # train the residual model with the VeBNN trainer
    self.hf_vebrnn_trainer.cooperative_train(
        x_train=x_train,
        y_train=y_train,
        iteration=iteration,
        init_config=init_config,
        var_config=var_config,
        sampler_config=sampler_config,
        delete_model_raw_data=delete_model_raw_data,
    )
hf_aleatoric_variance_predict(x: Tensor) -> Tensor ¤

Predict the aleatoric variance of the output at the scaled data.

Parameters:

Name Type Description Default
x Tensor

Test data points.

 required

Returns:

Type Description
Tensor

Predicted aleatoric variance at the scaled space.
Source code in src/MFVeBRNN/method/mf_residual_vebrnn_trainer.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def hf_aleatoric_variance_predict(self, x: Tensor) -> Tensor:
    """Predict the aleatoric variance of the output at the scaled data.

    Parameters
    ----------
    x : Tensor
        Test data points.

    Returns
    -------
    Tensor
        Predicted aleatoric variance at the scaled space.
    """
    x = x.to(self.device)
    # get the re-arranged input data
    x = self._re_arrange_input(x)

    # get the aleatoric variance prediction from the residual model
    var_aleatoric = self.hf_vebrnn_trainer.aleatoric_variance_predict(x)

    return var_aleatoric.detach()
hf_bayes_predict(x: Tensor, save_ppd: bool = False) -> typing.Tuple[torch.Tensor, torch.Tensor] ¤

Predict the mean and variance of the output at the scaled data.

Parameters:

Name Type Description Default
x Tensor

Test data points.

 required
save_ppd bool

Whether to save ppd or not (default is False).

 False

Returns:

Type Description
Tuple[Tensor, Tensor]

Predicted mean and variance at the scaled space.
Source code in src/MFVeBRNN/method/mf_residual_vebrnn_trainer.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def hf_bayes_predict(
    self,
    x: torch.Tensor,
    save_ppd: bool = False,
) -> Tuple[Tensor, Tensor]:
    """Predict the mean and variance of the output at the scaled data.

    Parameters
    ----------
    x : torch.Tensor
        Test data points.
    save_ppd : bool, optional
        Whether to save ppd or not (default is False).

    Returns
    -------
    Tuple[Tensor, Tensor]
        Predicted mean and variance at the scaled space.
    """
    x = x.to(self.device)
    y_lf = self.lf_predict(x)
    # get the re-arranged input data
    x = self._re_arrange_input(x)

    # get the prediction from the residual model
    y_pred_mean, y_pred_var = (
        self.hf_vebrnn_trainer.bayes_predict(x, save_ppd=save_ppd)
    )

    if save_ppd:
        self.responses = self.hf_vebrnn_trainer.responses + y_lf

    # Add the low-fidelity model prediction
    y_pred_mean += y_lf

    return y_pred_mean.detach(), y_pred_var.detach()
lf_predict(x: Tensor, return_var: bool = False) -> Tensor ¤

predict the output of the network

Parameters:

Name Type Description Default
x Tensor

input data

 required

Returns:

Type Description
Tensor

predicted output data
Source code in src/MFVeBRNN/method/mf_residual_vebrnn_trainer.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def lf_predict(self, x: Tensor, return_var: bool = False) -> Tensor:
    """predict the output of the network

    Parameters
    ----------
    x : Tensor
        input data

    Returns
    -------
    Tensor
        predicted output data
    """

    if isinstance(self.lf_model, RNNTrainer):
        y = self.lf_model.predict(x.to(self.device))
    elif isinstance(self.lf_model, VeBRNNTrainer):
        y, var_epistemic = self.lf_model.bayes_predict(
            x.to(self.device))
        var_aleatoric = self.lf_model.aleatoric_variance_predict(
            x.to(self.device))
        if return_var:
            return y, var_aleatoric, var_epistemic

    return y

Datasets¤

MFVeBRNN.dataset.SingleFidelityDataset ¤

Load the dataset for the plasticity problem

Source code in src/MFVeBRNN/dataset/load_dataset.py 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
class SingleFidelityDataset:
    """Load the dataset for the plasticity problem"""

    def __init__(self,
                 train_data_path: str = None,
                 id_ground_truth: bool = True,
                 id_ground_truth_data_path: str = None,
                 id_test_data_path: str = None,
                 ood_ground_truth: bool = False,
                 ood_ground_truth_data_path: str = None,
                 ood_test_data_path: str = None) -> None:
        """ load single-fidelity dataset

        Parameters
        ----------
        train_data_path : str, optional
            path of training dataset, by default None
        id_ground_truth : bool, optional
            whether to load in-distribution ground truth data, by default True
        id_ground_truth_data_path : str, optional
            path of in-distribution ground truth dataset, which is to be used
            for testing the aleatoric uncertainty. Usually, it is generated by
            SVE simulation, by default None
        id_test_data_path : str, optional
            path of in-distribution ground truth dataset, which is to be used
            for testing the mean. Usually, it is generated by RVE simulation,
            by default None
        ood_ground_truth : bool, optional
            whether to load out-of-distribution ground truth data,
            by default False
        ood_ground_truth_data_path : str, optional
            path of out-of-distribution ground truth dataset,
            which is to be used for testing the aleatoric uncertainty
            with larger strain amplitudes.
            Usually, it is generated by SVE simulation, by default None
        ood_test_data_path : str, optional
            path of out-of-distribution test dataset, which is to be used
            for testing the mean with larger strain amplitudes. Usually, it is
            generated by RVE simulation, by default None
        """

        # get the path of the repository
        self.fpath = Path(__file__).parent.parent.as_posix()
        # load the data for training
        self.dataset: pd.DataFrame = pd.read_pickle(
            self.fpath + "/dataset/data/" + train_data_path)
        # number of samples in dataset
        self.num_samples = len(self.dataset)
        # Convert to torch tensors
        self.X, self.Y = self.convert_data_to_torch(dataset=self.dataset)
        # scale the dataset
        self.scale_dataset()

        if id_ground_truth:
            # load the in-distribution test dataset
            if id_test_data_path is not None:
                self.id_test_dataset: pd.DataFrame = pd.read_pickle(
                    self.fpath + "/dataset/data/" + id_test_data_path)
                # number of samples in test dataset (ground truth)
                self.n_id_test = len(self.id_test_dataset)
                # load the lf ground truth dataset
                self.get_id_test_data()
            else:
                print("No in-distribution test dataset is provided.")
            if id_ground_truth_data_path is not None:
                self.id_ground_truth: pd.DataFrame = pd.read_pickle(
                    self.fpath + "/dataset/data/" + id_ground_truth_data_path)
                self.get_id_ground_truth()
            else:
                print("No in-distribution ground truth dataset is provided.")
        else:
            self.n_id_test = 0
            print("No in-distribution ground truth data is provided.")

        # load the out-of-distribution ground truth dataset
        if ood_ground_truth:
            # load the out-of-distribution hf test dataset
            if ood_test_data_path is not None:
                self.ood_hf_test_dataset: pd.DataFrame = pd.read_pickle(
                    self.fpath + "/dataset/data/" + ood_test_data_path)
                # number of samples in test dataset (ground truth)
                self.n_ood_hf_test = len(self.ood_hf_test_dataset)
                # load the lf ground truth dataset
                self.get_ood_test_data()
            else:
                print("No out-of-distribution hf test dataset is provided.")
            if ood_ground_truth_data_path is not None:
                self.ood_ground_truth: pd.DataFrame = pd.read_pickle(
                    self.fpath + "/dataset/data/" + ood_ground_truth_data_path)
                self.get_ood_ground_truth()
            else:
                print(
                    "No out-of-distribution ground truth dataset is "
                    "provided."
                )

        else:
            print("No out-of-distribution lf ground truth data is provided.")
            self.n_ood_hf_test = 0

        # print a message for the user
        print("=============================================================")
        print("The dataset is loaded successfully.")
        print(
            f"Number of training samples: {self.num_samples}")
        print(
            f"Number of in-distribution test samples: {self.n_id_test}")
        print(f"Number of out-of-distribution test samples: "
              f"{self.n_ood_hf_test}")
        print("=============================================================")

    def get_train_val_split(self,
                            num_train: int,
                            num_val: int,
                            seed: int = 1) -> None:
        """Get the processed data for training, validation

        Parameters
        ----------
        num_train : int
            Number of training data
        num_val : int
            Number of validation data
        seed: int
            seed of selecting the training paths
        """

        total_samples = len(self.dataset)
        requested_total = num_train + num_val

        if requested_total > total_samples:
            raise ValueError(
                "Requested split exceeds the total number of samples.")

        # Shuffle the indices with certain seed
        indices = torch.randperm(
            total_samples, generator=torch.Generator().manual_seed(seed))
        # Compute split boundaries
        train_end = num_train
        val_end = train_end + num_val

        # Assign data splits using non-overlapping indices
        train_indices = indices[:train_end]
        val_indices = indices[train_end:val_end]

        self.x_train = self.strain_normalized[train_indices]
        self.x_val = self.strain_normalized[val_indices]

        self.y_train = self.stress_normalized[train_indices]
        self.y_val = self.stress_normalized[val_indices]

    def get_id_test_data(self) -> None:
        """Get the processed data for testing
        """
        # convert to torch
        self.ID_X, self.ID_Y = self.convert_data_to_torch(
            self.id_test_dataset)

        # scale the high-fidelity test data
        self.x_id_gt_scaled = (self.ID_X-self.X_mean)/self.X_std
        self.y_id_gt_scaled = (self.ID_Y - self.Y_mean)/self.Y_std

    def get_id_ground_truth(self) -> None:
        """get the processed data for testing
        """

        # first convert to torch
        x_test = []
        y_test_mean = []
        y_test_var = []
        for ii in range(self.n_id_test):
            x_test.append((
                torch.FloatTensor(
                    self.id_ground_truth['strain_mean'].iloc[ii]).flatten(
                    start_dim=1)[:, [0, 1, 3]]
            ).unsqueeze(0))
            y_test_mean.append((
                torch.FloatTensor(
                    self.id_ground_truth['stress_mean'].iloc[ii]).flatten(
                    start_dim=1)[:, [0, 1, 3]]
            ).unsqueeze(0))
            y_test_var.append((
                torch.FloatTensor(
                    self.id_ground_truth['stress_var'].iloc[ii]).flatten(
                    start_dim=1)[:, [0, 1, 3]]
            ).unsqueeze(0))

        # original scale
        self.x_id_gt = torch.cat(x_test, dim=0)
        self.y_id_gt_mean = torch.cat(y_test_mean, dim=0)
        self.y_id_gt_var = torch.cat(y_test_var, dim=0)
        # scale the data
        self.x_id_gt_scaled = (self.x_id_gt-self.X_mean)/self.X_std
        self.y_id_gt_mean_scaled = (
            self.y_id_gt_mean - self.Y_mean)/self.Y_std
        self.y_id_gt_var_scaled = self.y_id_gt_var / self.Y_std**2


    def get_ood_test_data(self) -> None:
        """Get the processed data for testing
        """
        # convert to torch
        self.OOD_X, self.OOD_Y = self.convert_data_to_torch(
            self.ood_hf_test_dataset)

        # scale the high-fidelity test data
        self.x_ood_gt = (self.OOD_X-self.X_mean)/self.X_std
        self.y_ood_gt = (self.OOD_Y - self.Y_mean)/self.Y_std

    def get_ood_ground_truth(self) -> None:
        """get the processed data for testing
        """

        # first convert to torch
        x_test = []
        y_test_mean = []
        y_test_var = []
        for ii in range(self.n_ood_hf_test):
            x_test.append((
                torch.FloatTensor(
                    self.ood_ground_truth['strain_mean'].iloc[ii]).flatten(
                    start_dim=1)[:, [0, 1, 3]]
            ).unsqueeze(0))
            y_test_mean.append((
                torch.FloatTensor(
                    self.ood_ground_truth['stress_mean'].iloc[ii]).flatten(
                    start_dim=1)[:, [0, 1, 3]]
            ).unsqueeze(0))
            y_test_var.append((
                torch.FloatTensor(
                    self.ood_ground_truth['stress_var'].iloc[ii]).flatten(
                    start_dim=1)[:, [0, 1, 3]]
            ).unsqueeze(0))

        # original scale
        self.x_ood_gt = torch.cat(x_test, dim=0)
        self.y_ood_gt_mean = torch.cat(y_test_mean, dim=0)
        self.y_ood_gt_var = torch.cat(y_test_var, dim=0)
        # scale it
        self.x_ood_gt_scaled = (self.x_ood_gt-self.X_mean)/self.X_std
        self.y_ood_gt_mean_scaled = (
            self.y_ood_gt_mean - self.Y_mean)/self.Y_std
        self.y_ood_gt_var_scaled = self.y_ood_gt_var / self.Y_std**2

    def plot_training_data(self,
                           index: int,
                           save_figure: bool = False) -> None:
        """plot the training data

        index: int
            index of the path
        save_figure: bool
            save the figure to disk or not
        """

        # get the data
        fig, ax = plt.subplots(2, 3, figsize=(12, 5))
        pparam = dict(ylabel=r"$E_{11}$")
        ax[0, 0].plot(self.dataset["strain"][index][:, 0, 0],
                      color="#0077BB",
                      linewidth=2)
        ax[0, 0].set(**pparam)
        # set the limits of the y axis
        pparam = dict(ylabel=r"$E_{12}$")
        ax[0, 1].plot(self.dataset["strain"][index][:, 0, 1],
                      color="#0077BB",
                      linewidth=2)
        ax[0, 1].set(**pparam)
        pparam = dict(ylabel=r"$E_{22}$")
        ax[0, 2].plot(self.dataset["strain"][index][:, 1, 1],
                      color="#0077BB",
                      linewidth=2)
        ax[0, 2].set(**pparam)
        pparam = dict(xlabel="Pseudo time", ylabel=r"$\sigma_{11}$ (MPa)")
        ax[1, 0].plot(self.dataset["stress"][index][:, 0, 0],
                      color="#0077BB",
                      linewidth=2)

        ax[1, 0].set(**pparam)
        pparam = dict(xlabel="Pseudo time", ylabel=r"$\sigma_{12}$ (MPa)")
        ax[1, 1].plot(self.dataset["stress"][index][:, 0, 1],
                      color="#0077BB",
                      linewidth=2)

        ax[1, 1].set(**pparam)
        pparam = dict(xlabel="Pseudo time", ylabel=r"$\sigma_{22}$ (MPa)")
        ax[1, 2].plot(self.dataset["stress"][index][:, 1, 1],
                      color="#0077BB",
                      linewidth=2)
        ax[1, 2].set(**pparam)
        # set the fontsize of the axes
        for i in range(2):
            for j in range(3):
                ax[i, j].tick_params(axis='both', which='major', labelsize=12)
                ax[i, j].tick_params(axis='both', which='minor', labelsize=12)
        # set the linewidth of the axes
        for i in range(2):
            for j in range(3):
                for axis in ['top', 'bottom', 'left', 'right']:
                    ax[i, j].spines[axis].set_linewidth(1.5)
        # set the fontsize of the labels
        for i in range(2):
            for j in range(3):
                ax[i, j].set_xlabel(ax[i, j].get_xlabel(), fontsize=14)
                ax[i, j].set_ylabel(ax[i, j].get_ylabel(), fontsize=14)
                ax[i, j].yaxis.set_major_formatter(
                    formatter)
        # adjust the space between the subplots
        plt.subplots_adjust(wspace=0.32, hspace=0.25)
        # save the figure
        if save_figure:
            plt.savefig(f"train_data_{index}.png",
                        dpi=300, bbox_inches="tight")
            plt.savefig(f"train_data_{index}.svg",
                        dpi=300, bbox_inches="tight")
        else:
            plt.show()

    def plot_testing_data(self,
                          index: int,
                          test_data: str = "id",
                          save_figure: bool = False) -> None:
        """plot the testing data

        Parameters
        ----------
        index : int
            index of the data
        test_data : str, optional
            which test data to plot, "id" for in-distribution data and "ood"
        save_figure : bool, optional
            save the figure, by default False

        """

        # get the data
        if test_data == "id":
            # check the id_ground_truth attribute is available
            if (hasattr(self, "id_ground_truth")
                    and self.id_ground_truth is not None):
                rve_data_strain_mean = (
                    self.id_ground_truth["strain_mean"].iloc[index]
                )
                rve_data_stress_mean = (
                    self.id_ground_truth["stress_mean"].iloc[index]
                )
                rve_data_stress_std = (
                    self.id_ground_truth["stress_var"].iloc[index] ** 0.5
                )
            else:
                rve_data_strain_mean = None
                rve_data_stress_mean = None
                rve_data_stress_std = None

            # get the high-fidelity data
            test_strain = self.id_test_dataset["strain"].iloc[index]
            test_stress = self.id_test_dataset["stress"].iloc[index]
            # length of the data
            length_of_strain = test_strain.shape[0]

        elif test_data == "ood":
            if (hasattr(self, "ood_ground_truth")
                    and self.ood_ground_truth is not None):
                rve_data_strain_mean = (
                    self.ood_ground_truth["strain_mean"].iloc[index]
                )
                rve_data_stress_mean = (
                    self.ood_ground_truth["stress_mean"].iloc[index]
                )
                rve_data_stress_std = (
                    self.ood_ground_truth["stress_var"].iloc[index] ** 0.5
                )
            else:
                rve_data_strain_mean = None
                rve_data_stress_mean = None
                rve_data_stress_std = None

            # get the high-fidelity data
            test_strain = self.ood_hf_test_dataset["strain"].iloc[index]
            test_stress = self.ood_hf_test_dataset["stress"].iloc[index]

            # length of the data
            length_of_strain = test_strain.shape[0]

        fig, ax = plt.subplots(2, 3, figsize=(12, 5))
        # set the title of the whole figure
        pparam = dict(ylabel=r"$E_{11}$")
        ax[0, 0].plot(test_strain[:, 0, 0],
                      color="#0077BB",
                      linewidth=2,
                      )
        ax[0, 0].set(**pparam)
        # set the limits of the y axis
        pparam = dict(ylabel=r"$E_{12}$")
        ax[0, 1].plot(test_strain[:, 0, 1],
                      color="#0077BB",
                      linewidth=2,
                      )

        ax[0, 1].set(**pparam)
        pparam = dict(ylabel=r"$E_{22}$")
        ax[0, 2].plot(test_strain[:, 1, 1],
                      color="#0077BB",
                      linewidth=2,
                      )

        ax[0, 2].set(**pparam)
        pparam = dict(xlabel="Pseudo time", ylabel=r"$\sigma_{11}$ (MPa)")
        ax[1, 0].plot(test_stress[:, 0, 0], color="#CC3311", linewidth=2)
        if rve_data_strain_mean is not None:
            ax[1, 0].plot(rve_data_stress_mean[:, 0, 0],
                          color="#0077BB",
                          linewidth=2,
                          linestyle="--")
            ax[1, 0].fill_between(
                np.arange(length_of_strain),
                np.subtract(
                    rve_data_stress_mean[:, 0, 0],
                    1.96 * rve_data_stress_std[:, 0, 0],
                ),
                np.add(
                    rve_data_stress_mean[:, 0, 0],
                    1.96 * rve_data_stress_std[:, 0, 0],
                ),
                color="#0077BB",
                edgecolor="none",
                alpha=0.5,
            )

        ax[1, 0].set(**pparam)

        pparam = dict(xlabel="Pseudo time", ylabel=r"$\sigma_{12}$ (MPa)")
        ax[1, 1].plot(test_stress[:, 0, 1],
                      color="#CC3311", linewidth=2,
                      )
        if rve_data_strain_mean is not None:
            ax[1, 1].plot(rve_data_stress_mean[:, 0, 1],
                          color="#0077BB",
                          linewidth=2,
                          linestyle="--")

            ax[1, 1].fill_between(
                np.arange(length_of_strain),
                np.subtract(
                    rve_data_stress_mean[:, 0, 1],
                    1.96 * rve_data_stress_std[:, 0, 1],
                ),
                np.add(
                    rve_data_stress_mean[:, 0, 1],
                    1.96 * rve_data_stress_std[:, 0, 1],
                ),
                color="#0077BB",
                edgecolor="none",
                alpha=0.5,
            )
        ax[1, 1].set(**pparam)

        pparam = dict(xlabel="Pseudo time", ylabel=r"$\sigma_{22}$ (MPa)")
        ax[1, 2].plot(test_stress[:, 1, 1], color="#CC3311", linewidth=2)
        if rve_data_strain_mean is not None:
            ax[1, 2].plot(rve_data_stress_mean[:, 1, 1],
                          color="#0077BB",
                          linewidth=2,
                          linestyle="--")
            ax[1, 2].fill_between(
                np.arange(length_of_strain),
                np.subtract(
                    rve_data_stress_mean[:, 1, 1],
                    1.96 * rve_data_stress_std[:, 1, 1],
                ),
                np.add(
                    rve_data_stress_mean[:, 1, 1],
                    1.96 * rve_data_stress_std[:, 1, 1],
                ),
                color="#0077BB",
                edgecolor="none",
                alpha=0.5,
            )

        ax[1, 2].set(**pparam)
        ax[1, 2].legend([
            "Ground Truth (RVE)",
            "Ground Truth Mean (SVE)",
            "Ground Truth 95% CI"
        ], loc="lower right", fontsize=10, edgecolor="k", frameon=True)

        # set the fontsize of the axes
        for i in range(2):
            for j in range(3):
                ax[i, j].tick_params(axis='both', which='major', labelsize=12)
                ax[i, j].tick_params(axis='both', which='minor', labelsize=12)
        # set the linewidth of the axes
        for i in range(2):
            for j in range(3):
                for axis in ['top', 'bottom', 'left', 'right']:
                    ax[i, j].spines[axis].set_linewidth(1)
        # set the fontsize of the labels
        for i in range(2):
            for j in range(3):
                ax[i, j].set_xlabel(ax[i, j].get_xlabel(), fontsize=14)
                ax[i, j].set_ylabel(ax[i, j].get_ylabel(), fontsize=14)
                ax[i, j].yaxis.set_major_formatter(
                    formatter)
        # adjust the space between the subplots
        plt.subplots_adjust(wspace=0.32, hspace=0.25)
        # save the figure
        if save_figure:
            plt.savefig(f"test_data_{index}.svg",
                        dpi=300, bbox_inches="tight")
            plt.savefig(f"test_data_{index}.pdf",
                        dpi=300, bbox_inches="tight")
        else:
            plt.show()

    def convert_data_to_torch(
        self,
        dataset: pd.DataFrame,
    ) -> Tuple[torch.Tensor, torch.Tensor]:
        """convert the data to torch tensors
        Parameters
        ----------
        dataset : pd.DataFrame
            dataset to be converted

        Returns
        -------
        Tuple[torch.Tensor, torch.Tensor]
            X and Y tensors
        """
        # empty lists for data
        X, Y = [], []
        # gen number of samples
        num_samples = len(dataset)
        # get the data
        for ii in range(num_samples):
            X.append(
                (torch.FloatTensor(dataset['strain'].iloc[ii]).flatten(
                    start_dim=1)[:, [0, 1, 3]]).unsqueeze(0))
            Y.append(
                (torch.FloatTensor(dataset['stress'].iloc[ii]).flatten(
                    start_dim=1)[:, [0, 1, 3]]).unsqueeze(0))
        # concatenate all date together
        X, Y = torch.cat(X, dim=0), torch.cat(Y, dim=0)

        return X, Y

    def scale_dataset(self) -> None:
        """scale the dataset
        """
        self.strain_normalized, self.X_mean, self.X_std = \
            self._normalize_data(
                data=self.X)
        self.stress_normalized, self.Y_mean, self.Y_std = \
            self._normalize_data(
                data=self.Y)

    def scale_back_inputs(self, input_data: Tensor) -> Tensor:
        """scale back the input data"""
        return input_data * self.X_std + self.X_mean

    def scale_back_outputs(self, output_data: Tensor) -> Tensor:
        """scale back the output data"""
        return output_data * self.Y_std + self.Y_mean

    def scale_back_variance(self, output_data: Tensor) -> Tensor:
        """scale back the variance data"""
        return output_data * self.Y_std**2

    @staticmethod
    def _normalize_data(data: Tensor) -> Tuple[Tensor, Tensor, Tensor]:
        """normalize the dataset

        Parameters
        ----------
        data : torch.Tensor
            data to be normalized

        Returns
        -------
        Tuple[Tensor, Tensor, Tensor]
            normalized data, mean, and standard deviation
        """
        dim = (0, 1)
        data_mean = data.mean(dim=dim, keepdim=True)
        data_std = data.std(dim=dim, unbiased=False, keepdim=True)

        data_normalized = (data - data_mean) / data_std

        return data_normalized, data_mean, data_std
_normalize_data(data: Tensor) -> typing.Tuple[torch.Tensor, torch.Tensor, torch.Tensor] staticmethod ¤

normalize the dataset

Parameters:

Name Type Description Default
data Tensor

data to be normalized

 required

Returns:

Type Description
Tuple[Tensor, Tensor, Tensor]

normalized data, mean, and standard deviation
Source code in src/MFVeBRNN/dataset/load_dataset.py 
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
@staticmethod
def _normalize_data(data: Tensor) -> Tuple[Tensor, Tensor, Tensor]:
    """normalize the dataset

    Parameters
    ----------
    data : torch.Tensor
        data to be normalized

    Returns
    -------
    Tuple[Tensor, Tensor, Tensor]
        normalized data, mean, and standard deviation
    """
    dim = (0, 1)
    data_mean = data.mean(dim=dim, keepdim=True)
    data_std = data.std(dim=dim, unbiased=False, keepdim=True)

    data_normalized = (data - data_mean) / data_std

    return data_normalized, data_mean, data_std
convert_data_to_torch(dataset: DataFrame) -> typing.Tuple[torch.Tensor, torch.Tensor] ¤

convert the data to torch tensors

Parameters:

Name Type Description Default
dataset DataFrame

dataset to be converted

 required

Returns:

Type Description
Tuple[Tensor, Tensor]

X and Y tensors
Source code in src/MFVeBRNN/dataset/load_dataset.py 
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
def convert_data_to_torch(
    self,
    dataset: pd.DataFrame,
) -> Tuple[torch.Tensor, torch.Tensor]:
    """convert the data to torch tensors
    Parameters
    ----------
    dataset : pd.DataFrame
        dataset to be converted

    Returns
    -------
    Tuple[torch.Tensor, torch.Tensor]
        X and Y tensors
    """
    # empty lists for data
    X, Y = [], []
    # gen number of samples
    num_samples = len(dataset)
    # get the data
    for ii in range(num_samples):
        X.append(
            (torch.FloatTensor(dataset['strain'].iloc[ii]).flatten(
                start_dim=1)[:, [0, 1, 3]]).unsqueeze(0))
        Y.append(
            (torch.FloatTensor(dataset['stress'].iloc[ii]).flatten(
                start_dim=1)[:, [0, 1, 3]]).unsqueeze(0))
    # concatenate all date together
    X, Y = torch.cat(X, dim=0), torch.cat(Y, dim=0)

    return X, Y
get_id_ground_truth() -> None ¤

get the processed data for testing

Source code in src/MFVeBRNN/dataset/load_dataset.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
def get_id_ground_truth(self) -> None:
    """get the processed data for testing
    """

    # first convert to torch
    x_test = []
    y_test_mean = []
    y_test_var = []
    for ii in range(self.n_id_test):
        x_test.append((
            torch.FloatTensor(
                self.id_ground_truth['strain_mean'].iloc[ii]).flatten(
                start_dim=1)[:, [0, 1, 3]]
        ).unsqueeze(0))
        y_test_mean.append((
            torch.FloatTensor(
                self.id_ground_truth['stress_mean'].iloc[ii]).flatten(
                start_dim=1)[:, [0, 1, 3]]
        ).unsqueeze(0))
        y_test_var.append((
            torch.FloatTensor(
                self.id_ground_truth['stress_var'].iloc[ii]).flatten(
                start_dim=1)[:, [0, 1, 3]]
        ).unsqueeze(0))

    # original scale
    self.x_id_gt = torch.cat(x_test, dim=0)
    self.y_id_gt_mean = torch.cat(y_test_mean, dim=0)
    self.y_id_gt_var = torch.cat(y_test_var, dim=0)
    # scale the data
    self.x_id_gt_scaled = (self.x_id_gt-self.X_mean)/self.X_std
    self.y_id_gt_mean_scaled = (
        self.y_id_gt_mean - self.Y_mean)/self.Y_std
    self.y_id_gt_var_scaled = self.y_id_gt_var / self.Y_std**2
get_id_test_data() -> None ¤

Get the processed data for testing

Source code in src/MFVeBRNN/dataset/load_dataset.py 
198
199
200
201
202
203
204
205
206
207
def get_id_test_data(self) -> None:
    """Get the processed data for testing
    """
    # convert to torch
    self.ID_X, self.ID_Y = self.convert_data_to_torch(
        self.id_test_dataset)

    # scale the high-fidelity test data
    self.x_id_gt_scaled = (self.ID_X-self.X_mean)/self.X_std
    self.y_id_gt_scaled = (self.ID_Y - self.Y_mean)/self.Y_std
get_ood_ground_truth() -> None ¤

get the processed data for testing

Source code in src/MFVeBRNN/dataset/load_dataset.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def get_ood_ground_truth(self) -> None:
    """get the processed data for testing
    """

    # first convert to torch
    x_test = []
    y_test_mean = []
    y_test_var = []
    for ii in range(self.n_ood_hf_test):
        x_test.append((
            torch.FloatTensor(
                self.ood_ground_truth['strain_mean'].iloc[ii]).flatten(
                start_dim=1)[:, [0, 1, 3]]
        ).unsqueeze(0))
        y_test_mean.append((
            torch.FloatTensor(
                self.ood_ground_truth['stress_mean'].iloc[ii]).flatten(
                start_dim=1)[:, [0, 1, 3]]
        ).unsqueeze(0))
        y_test_var.append((
            torch.FloatTensor(
                self.ood_ground_truth['stress_var'].iloc[ii]).flatten(
                start_dim=1)[:, [0, 1, 3]]
        ).unsqueeze(0))

    # original scale
    self.x_ood_gt = torch.cat(x_test, dim=0)
    self.y_ood_gt_mean = torch.cat(y_test_mean, dim=0)
    self.y_ood_gt_var = torch.cat(y_test_var, dim=0)
    # scale it
    self.x_ood_gt_scaled = (self.x_ood_gt-self.X_mean)/self.X_std
    self.y_ood_gt_mean_scaled = (
        self.y_ood_gt_mean - self.Y_mean)/self.Y_std
    self.y_ood_gt_var_scaled = self.y_ood_gt_var / self.Y_std**2
get_ood_test_data() -> None ¤

Get the processed data for testing

Source code in src/MFVeBRNN/dataset/load_dataset.py 
245
246
247
248
249
250
251
252
253
254
def get_ood_test_data(self) -> None:
    """Get the processed data for testing
    """
    # convert to torch
    self.OOD_X, self.OOD_Y = self.convert_data_to_torch(
        self.ood_hf_test_dataset)

    # scale the high-fidelity test data
    self.x_ood_gt = (self.OOD_X-self.X_mean)/self.X_std
    self.y_ood_gt = (self.OOD_Y - self.Y_mean)/self.Y_std
get_train_val_split(num_train: int, num_val: int, seed: int = 1) -> None ¤

Get the processed data for training, validation

Parameters:

Name Type Description Default
num_train int

Number of training data

 required
num_val int

Number of validation data

 required
seed int

seed of selecting the training paths

 1
Source code in src/MFVeBRNN/dataset/load_dataset.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def get_train_val_split(self,
                        num_train: int,
                        num_val: int,
                        seed: int = 1) -> None:
    """Get the processed data for training, validation

    Parameters
    ----------
    num_train : int
        Number of training data
    num_val : int
        Number of validation data
    seed: int
        seed of selecting the training paths
    """

    total_samples = len(self.dataset)
    requested_total = num_train + num_val

    if requested_total > total_samples:
        raise ValueError(
            "Requested split exceeds the total number of samples.")

    # Shuffle the indices with certain seed
    indices = torch.randperm(
        total_samples, generator=torch.Generator().manual_seed(seed))
    # Compute split boundaries
    train_end = num_train
    val_end = train_end + num_val

    # Assign data splits using non-overlapping indices
    train_indices = indices[:train_end]
    val_indices = indices[train_end:val_end]

    self.x_train = self.strain_normalized[train_indices]
    self.x_val = self.strain_normalized[val_indices]

    self.y_train = self.stress_normalized[train_indices]
    self.y_val = self.stress_normalized[val_indices]
plot_testing_data(index: int, test_data: str = 'id', save_figure: bool = False) -> None ¤

plot the testing data

Parameters:

Name Type Description Default
index int

index of the data

 required
test_data str

which test data to plot, "id" for in-distribution data and "ood"

 'id'
save_figure bool

save the figure, by default False

 False
Source code in src/MFVeBRNN/dataset/load_dataset.py 
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
def plot_testing_data(self,
                      index: int,
                      test_data: str = "id",
                      save_figure: bool = False) -> None:
    """plot the testing data

    Parameters
    ----------
    index : int
        index of the data
    test_data : str, optional
        which test data to plot, "id" for in-distribution data and "ood"
    save_figure : bool, optional
        save the figure, by default False

    """

    # get the data
    if test_data == "id":
        # check the id_ground_truth attribute is available
        if (hasattr(self, "id_ground_truth")
                and self.id_ground_truth is not None):
            rve_data_strain_mean = (
                self.id_ground_truth["strain_mean"].iloc[index]
            )
            rve_data_stress_mean = (
                self.id_ground_truth["stress_mean"].iloc[index]
            )
            rve_data_stress_std = (
                self.id_ground_truth["stress_var"].iloc[index] ** 0.5
            )
        else:
            rve_data_strain_mean = None
            rve_data_stress_mean = None
            rve_data_stress_std = None

        # get the high-fidelity data
        test_strain = self.id_test_dataset["strain"].iloc[index]
        test_stress = self.id_test_dataset["stress"].iloc[index]
        # length of the data
        length_of_strain = test_strain.shape[0]

    elif test_data == "ood":
        if (hasattr(self, "ood_ground_truth")
                and self.ood_ground_truth is not None):
            rve_data_strain_mean = (
                self.ood_ground_truth["strain_mean"].iloc[index]
            )
            rve_data_stress_mean = (
                self.ood_ground_truth["stress_mean"].iloc[index]
            )
            rve_data_stress_std = (
                self.ood_ground_truth["stress_var"].iloc[index] ** 0.5
            )
        else:
            rve_data_strain_mean = None
            rve_data_stress_mean = None
            rve_data_stress_std = None

        # get the high-fidelity data
        test_strain = self.ood_hf_test_dataset["strain"].iloc[index]
        test_stress = self.ood_hf_test_dataset["stress"].iloc[index]

        # length of the data
        length_of_strain = test_strain.shape[0]

    fig, ax = plt.subplots(2, 3, figsize=(12, 5))
    # set the title of the whole figure
    pparam = dict(ylabel=r"$E_{11}$")
    ax[0, 0].plot(test_strain[:, 0, 0],
                  color="#0077BB",
                  linewidth=2,
                  )
    ax[0, 0].set(**pparam)
    # set the limits of the y axis
    pparam = dict(ylabel=r"$E_{12}$")
    ax[0, 1].plot(test_strain[:, 0, 1],
                  color="#0077BB",
                  linewidth=2,
                  )

    ax[0, 1].set(**pparam)
    pparam = dict(ylabel=r"$E_{22}$")
    ax[0, 2].plot(test_strain[:, 1, 1],
                  color="#0077BB",
                  linewidth=2,
                  )

    ax[0, 2].set(**pparam)
    pparam = dict(xlabel="Pseudo time", ylabel=r"$\sigma_{11}$ (MPa)")
    ax[1, 0].plot(test_stress[:, 0, 0], color="#CC3311", linewidth=2)
    if rve_data_strain_mean is not None:
        ax[1, 0].plot(rve_data_stress_mean[:, 0, 0],
                      color="#0077BB",
                      linewidth=2,
                      linestyle="--")
        ax[1, 0].fill_between(
            np.arange(length_of_strain),
            np.subtract(
                rve_data_stress_mean[:, 0, 0],
                1.96 * rve_data_stress_std[:, 0, 0],
            ),
            np.add(
                rve_data_stress_mean[:, 0, 0],
                1.96 * rve_data_stress_std[:, 0, 0],
            ),
            color="#0077BB",
            edgecolor="none",
            alpha=0.5,
        )

    ax[1, 0].set(**pparam)

    pparam = dict(xlabel="Pseudo time", ylabel=r"$\sigma_{12}$ (MPa)")
    ax[1, 1].plot(test_stress[:, 0, 1],
                  color="#CC3311", linewidth=2,
                  )
    if rve_data_strain_mean is not None:
        ax[1, 1].plot(rve_data_stress_mean[:, 0, 1],
                      color="#0077BB",
                      linewidth=2,
                      linestyle="--")

        ax[1, 1].fill_between(
            np.arange(length_of_strain),
            np.subtract(
                rve_data_stress_mean[:, 0, 1],
                1.96 * rve_data_stress_std[:, 0, 1],
            ),
            np.add(
                rve_data_stress_mean[:, 0, 1],
                1.96 * rve_data_stress_std[:, 0, 1],
            ),
            color="#0077BB",
            edgecolor="none",
            alpha=0.5,
        )
    ax[1, 1].set(**pparam)

    pparam = dict(xlabel="Pseudo time", ylabel=r"$\sigma_{22}$ (MPa)")
    ax[1, 2].plot(test_stress[:, 1, 1], color="#CC3311", linewidth=2)
    if rve_data_strain_mean is not None:
        ax[1, 2].plot(rve_data_stress_mean[:, 1, 1],
                      color="#0077BB",
                      linewidth=2,
                      linestyle="--")
        ax[1, 2].fill_between(
            np.arange(length_of_strain),
            np.subtract(
                rve_data_stress_mean[:, 1, 1],
                1.96 * rve_data_stress_std[:, 1, 1],
            ),
            np.add(
                rve_data_stress_mean[:, 1, 1],
                1.96 * rve_data_stress_std[:, 1, 1],
            ),
            color="#0077BB",
            edgecolor="none",
            alpha=0.5,
        )

    ax[1, 2].set(**pparam)
    ax[1, 2].legend([
        "Ground Truth (RVE)",
        "Ground Truth Mean (SVE)",
        "Ground Truth 95% CI"
    ], loc="lower right", fontsize=10, edgecolor="k", frameon=True)

    # set the fontsize of the axes
    for i in range(2):
        for j in range(3):
            ax[i, j].tick_params(axis='both', which='major', labelsize=12)
            ax[i, j].tick_params(axis='both', which='minor', labelsize=12)
    # set the linewidth of the axes
    for i in range(2):
        for j in range(3):
            for axis in ['top', 'bottom', 'left', 'right']:
                ax[i, j].spines[axis].set_linewidth(1)
    # set the fontsize of the labels
    for i in range(2):
        for j in range(3):
            ax[i, j].set_xlabel(ax[i, j].get_xlabel(), fontsize=14)
            ax[i, j].set_ylabel(ax[i, j].get_ylabel(), fontsize=14)
            ax[i, j].yaxis.set_major_formatter(
                formatter)
    # adjust the space between the subplots
    plt.subplots_adjust(wspace=0.32, hspace=0.25)
    # save the figure
    if save_figure:
        plt.savefig(f"test_data_{index}.svg",
                    dpi=300, bbox_inches="tight")
        plt.savefig(f"test_data_{index}.pdf",
                    dpi=300, bbox_inches="tight")
    else:
        plt.show()
plot_training_data(index: int, save_figure: bool = False) -> None ¤

plot the training data

index: int index of the path save_figure: bool save the figure to disk or not

Source code in src/MFVeBRNN/dataset/load_dataset.py 
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
def plot_training_data(self,
                       index: int,
                       save_figure: bool = False) -> None:
    """plot the training data

    index: int
        index of the path
    save_figure: bool
        save the figure to disk or not
    """

    # get the data
    fig, ax = plt.subplots(2, 3, figsize=(12, 5))
    pparam = dict(ylabel=r"$E_{11}$")
    ax[0, 0].plot(self.dataset["strain"][index][:, 0, 0],
                  color="#0077BB",
                  linewidth=2)
    ax[0, 0].set(**pparam)
    # set the limits of the y axis
    pparam = dict(ylabel=r"$E_{12}$")
    ax[0, 1].plot(self.dataset["strain"][index][:, 0, 1],
                  color="#0077BB",
                  linewidth=2)
    ax[0, 1].set(**pparam)
    pparam = dict(ylabel=r"$E_{22}$")
    ax[0, 2].plot(self.dataset["strain"][index][:, 1, 1],
                  color="#0077BB",
                  linewidth=2)
    ax[0, 2].set(**pparam)
    pparam = dict(xlabel="Pseudo time", ylabel=r"$\sigma_{11}$ (MPa)")
    ax[1, 0].plot(self.dataset["stress"][index][:, 0, 0],
                  color="#0077BB",
                  linewidth=2)

    ax[1, 0].set(**pparam)
    pparam = dict(xlabel="Pseudo time", ylabel=r"$\sigma_{12}$ (MPa)")
    ax[1, 1].plot(self.dataset["stress"][index][:, 0, 1],
                  color="#0077BB",
                  linewidth=2)

    ax[1, 1].set(**pparam)
    pparam = dict(xlabel="Pseudo time", ylabel=r"$\sigma_{22}$ (MPa)")
    ax[1, 2].plot(self.dataset["stress"][index][:, 1, 1],
                  color="#0077BB",
                  linewidth=2)
    ax[1, 2].set(**pparam)
    # set the fontsize of the axes
    for i in range(2):
        for j in range(3):
            ax[i, j].tick_params(axis='both', which='major', labelsize=12)
            ax[i, j].tick_params(axis='both', which='minor', labelsize=12)
    # set the linewidth of the axes
    for i in range(2):
        for j in range(3):
            for axis in ['top', 'bottom', 'left', 'right']:
                ax[i, j].spines[axis].set_linewidth(1.5)
    # set the fontsize of the labels
    for i in range(2):
        for j in range(3):
            ax[i, j].set_xlabel(ax[i, j].get_xlabel(), fontsize=14)
            ax[i, j].set_ylabel(ax[i, j].get_ylabel(), fontsize=14)
            ax[i, j].yaxis.set_major_formatter(
                formatter)
    # adjust the space between the subplots
    plt.subplots_adjust(wspace=0.32, hspace=0.25)
    # save the figure
    if save_figure:
        plt.savefig(f"train_data_{index}.png",
                    dpi=300, bbox_inches="tight")
        plt.savefig(f"train_data_{index}.svg",
                    dpi=300, bbox_inches="tight")
    else:
        plt.show()
scale_back_inputs(input_data: Tensor) -> Tensor ¤

scale back the input data

Source code in src/MFVeBRNN/dataset/load_dataset.py 
603
604
605
def scale_back_inputs(self, input_data: Tensor) -> Tensor:
    """scale back the input data"""
    return input_data * self.X_std + self.X_mean
scale_back_outputs(output_data: Tensor) -> Tensor ¤

scale back the output data

Source code in src/MFVeBRNN/dataset/load_dataset.py 
607
608
609
def scale_back_outputs(self, output_data: Tensor) -> Tensor:
    """scale back the output data"""
    return output_data * self.Y_std + self.Y_mean
scale_back_variance(output_data: Tensor) -> Tensor ¤

scale back the variance data

Source code in src/MFVeBRNN/dataset/load_dataset.py 
611
612
613
def scale_back_variance(self, output_data: Tensor) -> Tensor:
    """scale back the variance data"""
    return output_data * self.Y_std**2
scale_dataset() -> None ¤

scale the dataset

Source code in src/MFVeBRNN/dataset/load_dataset.py 
593
594
595
596
597
598
599
600
601
def scale_dataset(self) -> None:
    """scale the dataset
    """
    self.strain_normalized, self.X_mean, self.X_std = \
        self._normalize_data(
            data=self.X)
    self.stress_normalized, self.Y_mean, self.Y_std = \
        self._normalize_data(
            data=self.Y)

MFVeBRNN.dataset.MultiFidelityDataset ¤

load multi-fidelity dataset

Source code in src/MFVeBRNN/dataset/load_dataset.py 
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
class MultiFidelityDataset:
    """load multi-fidelity dataset
    """

    def __init__(self,
                 lf_train_data_path: str = None,
                 hf_train_data_path: str = None,
                 id_ground_truth: bool = True,
                 id_lf_ground_truth_data_path: str = None,
                 id_hf_test_data_path: str = None,
                 ood_ground_truth: bool = False,
                 ood_lf_ground_truth_data_path: str = None,
                 ood_hf_test_data_path: str = None) -> None:
        """Multi-fidelity dataset for plasticity law datasets

        Parameters
        ----------
        lf_train_data_path : str, optional
            path for low-fidelity training data, by default None
        hf_train_data_path : str, optional
            path for high-fidelity training data, by default None
        id_ground_truth : bool, optional
            whether to include in-distribution ground truth data,
            by default True
        id_lf_ground_truth_data_path : str, optional
            path for in-distribution low-fidelity ground truth data,
            by default None
        id_hf_test_data_path : str, optional
            path for in-distribution high-fidelity test data, by default None
        ood_ground_truth : bool, optional
            whether to include out-of-distribution ground truth data,
            by default False
        ood_lf_ground_truth_data_path : str, optional
            path for out-of-distribution low-fidelity ground truth data,
            by default None
        ood_hf_test_data_path : str, optional
            path for out-of-distribution high-fidelity test data,
            by default None
        """


        # get the path of the repository
        self.fpath = Path(__file__).parent.parent.as_posix()

        # load the low-fidelity data for training
        self.lf_dataset: pd.DataFrame = pd.read_pickle(
            self.fpath + "/dataset/data/" + lf_train_data_path)
        # number of samples in dataset
        self.lf_num_samples = len(self.lf_dataset)
        # Convert to torch tensors
        self.LX, self.LY = self.convert_data_to_torch(dataset=self.lf_dataset)
        self.scale_dataset()
        # load the high-fidelity data for training
        self.hf_dataset: pd.DataFrame = pd.read_pickle(
            self.fpath + "/dataset/data/" + hf_train_data_path)
        # number of samples in dataset
        self.hf_num_samples = len(self.hf_dataset)
        # Convert to torch tensors
        self.HX, self.HY = self.convert_data_to_torch(
            dataset=self.hf_dataset)
        # scale the high-fidelity data
        self.hx_scaled = (self.HX - self.LX_mean)/self.LX_std
        self.hy_scaled = (self.HY - self.LY_mean)/self.LY_std

        if id_ground_truth:
            # load the in-distribution hf test dataset
            if id_hf_test_data_path is not None:
                self.id_hf_test_dataset: pd.DataFrame = pd.read_pickle(
                    self.fpath + "/dataset/data/" + id_hf_test_data_path)
                self.n_id_hf_test = len(self.id_hf_test_dataset)
                self.get_id_hf_test_data()
            else:
                print("No in-distribution hf test dataset is provided.")
            if id_lf_ground_truth_data_path is not None:
                self.id_ground_truth: pd.DataFrame = pd.read_pickle(
                    self.fpath
                    + "/dataset/data/"
                    + id_lf_ground_truth_data_path
                )
                self.get_id_lf_ground_truth()
            else:
                print(
                    "No in-distribution lf ground truth "
                    "dataset is provided."
                )

        else:
            self.n_id_hf_test = 0
            print("No lf ground truth data is provided.")

        # load the out-of-distribution lf ground truth dataset
        if ood_ground_truth:
            # load the out-of-distribution hf test dataset
            if ood_hf_test_data_path is not None:
                self.ood_hf_test_dataset: pd.DataFrame = pd.read_pickle(
                    self.fpath + "/dataset/data/" + ood_hf_test_data_path)
                # number of samples in test dataset (ground truth)
                self.n_ood_hf_test = len(self.ood_hf_test_dataset)
                # load the lf ground truth dataset
                self.get_ood_hf_test_data()
            else:
                print("No out-of-distribution hf test dataset is provided.")
            if ood_lf_ground_truth_data_path is not None:
                self.ood_ground_truth: pd.DataFrame = pd.read_pickle(
                    self.fpath
                    + "/dataset/data/"
                    + ood_lf_ground_truth_data_path
                )
                self.get_ood_lf_ground_truth()
            else:
                print(
                    "No out-of-distribution lf ground truth dataset is "
                    "provided."
                )

        else:
            print("No out-of-distribution lf ground truth data is provided.")
            self.n_ood_hf_test = 0

        # print a message for the user
        print("=============================================================")
        print("The dataset is loaded successfully.")
        print(
            f"Number of low-fidelity training samples: {self.lf_num_samples}")
        print(
            f"Number of high-fidelity training samples: {self.hf_num_samples}")
        print(
            f"Number of in-distribution test samples: {self.n_id_hf_test}")
        print(f"Number of out-of-distribution test samples: "
              f"{self.n_ood_hf_test}")
        print("=============================================================")

    def get_lf_train_val_split(self,
                               num_lf_train: int,
                               num_lf_val: int,
                               seed: int = 1) -> None:
        """Get the processed data for training, validation, and testing

        Parameters
        ----------
        num_lf_train : int
            Number of low-fidelity training data
        num_lf_val : int
            Number of low-fidelity validation data
        seed: int
            seed of selecting the training paths
        """

        total_lf_samples = len(self.lf_dataset)
        requested_total = num_lf_train + num_lf_val

        if requested_total > total_lf_samples:
            raise ValueError(
                "Requested split exceeds the total number of samples.")

        # Shuffle the indices with certain seed
        indices = torch.randperm(
            total_lf_samples, generator=torch.Generator().manual_seed(seed))
        # Compute split boundaries
        train_end = num_lf_train
        val_end = train_end + num_lf_val

        # Assign data splits using non-overlapping indices
        train_indices = indices[:train_end]
        val_indices = indices[train_end:val_end]

        self.lx_train = self.strain_normalized[train_indices]
        self.lx_val = self.strain_normalized[val_indices]

        self.ly_train = self.stress_normalized[train_indices]
        self.ly_val = self.stress_normalized[val_indices]

    def get_hf_train_val_split(self,
                               num_hf_train: int,
                               num_hf_val: int,
                               seed: int = 1) -> None:
        """Get the processed data for training, validation, and testing
        Parameters
        ----------
        num_hf_train : int
            Number of training data
        num_hf_val : int
            Number of validation data
        seed: int
            seed of selecting the training paths
        """
        total_hf_samples = len(self.hf_dataset)
        requested_total = num_hf_train + num_hf_val
        if requested_total > total_hf_samples:
            raise ValueError(
                "Requested split exceeds the total number of samples.")
        # Shuffle the indices with certain seed
        indices = torch.randperm(
            total_hf_samples, generator=torch.Generator().manual_seed(seed))

        # Compute split boundaries
        train_end = num_hf_train
        val_end = train_end + num_hf_val
        # Assign data splits using non-overlapping indices
        train_indices = indices[:train_end]
        val_indices = indices[train_end:val_end]
        # get the data
        self.hx_train = self.hx_scaled[train_indices]
        self.hx_val = self.hx_scaled[val_indices]
        self.hy_train = self.hy_scaled[train_indices]
        self.hy_val = self.hy_scaled[val_indices]

    def get_id_hf_test_data(self) -> None:
        """Get the processed data for testing
        """
        # convert to torch
        self.ID_HX, self.ID_HY = self.convert_data_to_torch(
            self.id_hf_test_dataset)

        # scale the high-fidelity test data
        self.hx_id_gt_scaled = (self.ID_HX - self.LX_mean) / self.LX_std
        self.hy_id_gt_scaled = (self.ID_HY - self.LY_mean) / self.LY_std

    def get_id_lf_ground_truth(self) -> None:
        """get the processed data for testing
        """

        # first convert to torch
        x_test = []
        y_test_mean = []
        y_test_var = []
        for ii in range(self.n_id_hf_test):
            x_test.append((
                torch.FloatTensor(
                    self.id_ground_truth['strain_mean'].iloc[ii]).flatten(
                    start_dim=1)[:, [0, 1, 3]]
            ).unsqueeze(0))
            y_test_mean.append((
                torch.FloatTensor(
                    self.id_ground_truth['stress_mean'].iloc[ii]).flatten(
                    start_dim=1)[:, [0, 1, 3]]
            ).unsqueeze(0))
            y_test_var.append((
                torch.FloatTensor(
                    self.id_ground_truth['stress_var'].iloc[ii]).flatten(
                    start_dim=1)[:, [0, 1, 3]]
            ).unsqueeze(0))

        # original scale
        self.lx_id_gt = torch.cat(x_test, dim=0)
        self.ly_id_gt_mean = torch.cat(y_test_mean, dim=0)
        self.ly_id_gt_var = torch.cat(y_test_var, dim=0)
        # scale it
        self.lx_id_gt_scaled = (self.lx_id_gt-self.LX_mean)/self.LX_std
        self.ly_id_gt_mean_scaled = (
            self.ly_id_gt_mean - self.LY_mean)/self.LY_std
        self.ly_id_gt_var_scaled = self.ly_id_gt_var / self.LY_std**2

    def get_ood_hf_test_data(self) -> None:
        """Get the processed data for testing
        """
        # convert to torch
        self.OOD_HX, self.OOD_HY = self.convert_data_to_torch(
            self.ood_hf_test_dataset)
        # scale the high-fidelity test data
        self.hx_ood_gt_scaled = (self.OOD_HX - self.LX_mean) / self.LX_std
        self.hy_ood_gt_scaled = (self.OOD_HY - self.LY_mean) / self.LY_std

    def get_ood_lf_ground_truth(self) -> None:
        """get the processed data for testing
        """

        # first convert to torch
        x_test = []
        y_test_mean = []
        y_test_var = []
        for ii in range(self.n_ood_hf_test):
            x_test.append((
                torch.FloatTensor(
                    self.ood_ground_truth['strain_mean'].iloc[ii]).flatten(
                    start_dim=1)[:, [0, 1, 3]]
            ).unsqueeze(0))
            y_test_mean.append((
                torch.FloatTensor(
                    self.ood_ground_truth['stress_mean'].iloc[ii]).flatten(
                    start_dim=1)[:, [0, 1, 3]]
            ).unsqueeze(0))
            y_test_var.append((
                torch.FloatTensor(
                    self.ood_ground_truth['stress_var'].iloc[ii]).flatten(
                    start_dim=1)[:, [0, 1, 3]]
            ).unsqueeze(0))

        # original scale
        self.lx_ood_gt = torch.cat(x_test, dim=0)
        self.ly_ood_gt_mean = torch.cat(y_test_mean, dim=0)
        self.ly_ood_gt_var = torch.cat(y_test_var, dim=0)
        # scale it
        self.lx_ood_gt_scaled = (self.lx_ood_gt-self.LX_mean)/self.LX_std
        self.ly_ood_gt_mean_scaled = (
            self.ly_ood_gt_mean - self.LY_mean)/self.LY_std
        self.ly_ood_gt_var_scaled = self.ly_ood_gt_var / self.LY_std**2

    def plot_training_data(self,
                           index: int,
                           fidelity: str = "lf",
                           save_figure: bool = False,
                           ) -> None:
        """plot the training data

        index: int
            index of the path
        fidelity: str
            which fidelity to plot, "lf" for low-fidelity data and "hf" for
            high-fidelity data
        save_figure: bool
            save the figure to disk or not
        """

        if fidelity == "hf":
            dataset_to_use = self.hf_dataset
        else:
            dataset_to_use = self.lf_dataset

        # get the data
        fig, ax = plt.subplots(2, 3, figsize=(12, 5))
        pparam = dict(ylabel=r"$E_{11}$")
        ax[0, 0].plot(dataset_to_use["strain"].iloc[index][:, 0, 0],
                      color="#0077BB",
                      linewidth=2)
        ax[0, 0].set(**pparam)
        # set the limits of the y axis
        pparam = dict(ylabel=r"$E_{12}$")
        ax[0, 1].plot(dataset_to_use["strain"].iloc[index][:, 0, 1],
                      color="#0077BB",
                      linewidth=2)
        ax[0, 1].set(**pparam)
        pparam = dict(ylabel=r"$E_{22}$")
        ax[0, 2].plot(dataset_to_use["strain"].iloc[index][:, 1, 1],
                      color="#0077BB",
                      linewidth=2)
        ax[0, 2].set(**pparam)
        pparam = dict(xlabel="Time step", ylabel=r"$\sigma_{11}$ (MPa)")
        ax[1, 0].plot(dataset_to_use["stress"].iloc[index][:, 0, 0],
                      color="#0077BB",
                      linewidth=2)

        ax[1, 0].set(**pparam)
        pparam = dict(xlabel="Time step", ylabel=r"$\sigma_{12}$ (MPa)")
        ax[1, 1].plot(dataset_to_use["stress"].iloc[index][:, 0, 1],
                      color="#0077BB",
                      linewidth=2)

        ax[1, 1].set(**pparam)
        pparam = dict(xlabel="Time step", ylabel=r"$\sigma_{22}$ (MPa)")
        ax[1, 2].plot(dataset_to_use["stress"].iloc[index][:, 1, 1],
                      color="#0077BB",
                      linewidth=2)
        ax[1, 2].set(**pparam)
        # set the fontsize of the axes
        for i in range(2):
            for j in range(3):
                ax[i, j].tick_params(axis='both', which='major', labelsize=12)
                ax[i, j].tick_params(axis='both', which='minor', labelsize=12)
        # set the linewidth of the axes
        for i in range(2):
            for j in range(3):
                for axis in ['top', 'bottom', 'left', 'right']:
                    ax[i, j].spines[axis].set_linewidth(1.5)
        # set the fontsize of the labels
        for i in range(2):
            for j in range(3):
                ax[i, j].set_xlabel(ax[i, j].get_xlabel(), fontsize=14)
                ax[i, j].set_ylabel(ax[i, j].get_ylabel(), fontsize=14)
                ax[i, j].yaxis.set_major_formatter(
                    formatter)
        # adjust the space between the subplots
        plt.subplots_adjust(wspace=0.32, hspace=0.25)
        # save the figure
        if save_figure:
            plt.savefig(f"train_data_{index}.png",
                        dpi=300, bbox_inches="tight")
            plt.savefig(f"train_data_{index}.svg",
                        dpi=300, bbox_inches="tight")
        else:
            plt.show()

    def plot_testing_data(self,
                          index: int,
                          test_data: str = "id",
                          save_figure: bool = False) -> None:
        """plot the data

        Parameters
        ----------
        index : int
            index of the data
        test_data : str, optional
            type of test data to plot, by default "id"
        save_figure : bool, optional
            save the figure, by default False
        """

        # get the data
        if test_data == "id":
            # check the id_ground_truth attribute is available
            if hasattr(self, "id_ground_truth"):
                rve_data_strain_mean = (
                    self.id_ground_truth["strain_mean"].iloc[index]
                )
                rve_data_stress_mean = (
                    self.id_ground_truth["stress_mean"].iloc[index]
                )
                rve_data_stress_std = (
                    self.id_ground_truth["stress_var"].iloc[index] ** 0.5
                )
            else:
                rve_data_strain_mean = None
                rve_data_stress_mean = None
                rve_data_stress_std = None
            # get the high-fidelity data
            hf_test_train = self.id_hf_test_dataset["strain"].iloc[index]
            hf_test_stress = self.id_hf_test_dataset["stress"].iloc[index]

            # length of the data
            length_of_strain = hf_test_train.shape[0]
        elif test_data == "ood":
            if hasattr(self, "ood_ground_truth"):
                rve_data_strain_mean = (
                    self.ood_ground_truth["strain_mean"].iloc[index]
                )
                rve_data_stress_mean = (
                    self.ood_ground_truth["stress_mean"].iloc[index]
                )
                rve_data_stress_std = (
                    self.ood_ground_truth["stress_var"].iloc[index] ** 0.5
                )
            else:
                rve_data_strain_mean = None
                rve_data_stress_mean = None
                rve_data_stress_std = None

            # get the high-fidelity data
            hf_test_train = self.ood_hf_test_dataset["strain"].iloc[index]
            hf_test_stress = self.ood_hf_test_dataset["stress"].iloc[index]

            # length of the data
            length_of_strain = hf_test_train.shape[0]

        fig, ax = plt.subplots(2, 3, figsize=(12, 5))
        # set the title of the whole figure
        pparam = dict(ylabel=r"$E_{11}$")
        ax[0, 0].plot(hf_test_train[:, 0, 0],
                      color="#0077BB",
                      linewidth=2,
                      label="rve")
        ax[0, 0].set(**pparam)
        # set the limits of the y axis
        pparam = dict(ylabel=r"$E_{12}$")
        ax[0, 1].plot(hf_test_train[:, 0, 1],
                      color="#0077BB",
                      linewidth=2,
                      label="rve")

        ax[0, 1].set(**pparam)
        pparam = dict(ylabel=r"$E_{22}$")
        ax[0, 2].plot(hf_test_train[:, 1, 1],
                      color="#0077BB",
                      linewidth=2,
                      label="rve")

        ax[0, 2].set(**pparam)
        pparam = dict(xlabel="Time step", ylabel=r"$\sigma_{11}$ (MPa)")
        ax[1, 0].plot(hf_test_stress[:, 0, 0], color="#CC3311",
                      linewidth=2, label="hf")
        if rve_data_strain_mean is not None:
            ax[1, 0].plot(rve_data_stress_mean[:, 0, 0],
                          color="#0077BB",
                          linewidth=2,
                          linestyle="--")
            ax[1, 0].fill_between(
                np.arange(length_of_strain),
                np.subtract(
                    rve_data_stress_mean[:, 0, 0],
                    1.96 * rve_data_stress_std[:, 0, 0],
                ),
                np.add(
                    rve_data_stress_mean[:, 0, 0],
                    1.96 * rve_data_stress_std[:, 0, 0],
                ),
                color="#0077BB",
                edgecolor="none",
                alpha=0.5,
            )

        ax[1, 0].set(**pparam)

        pparam = dict(xlabel="Time step", ylabel=r"$\sigma_{12}$ (MPa)")
        ax[1, 1].plot(hf_test_stress[:, 0, 1],
                      color="#CC3311", linewidth=2,
                      label="hf")
        if rve_data_strain_mean is not None:
            ax[1, 1].plot(rve_data_stress_mean[:, 0, 1],
                          color="#0077BB",
                          linewidth=2,
                          linestyle="--",
                          label="rve")

            ax[1, 1].fill_between(
                np.arange(length_of_strain),
                np.subtract(
                    rve_data_stress_mean[:, 0, 1],
                    1.96 * rve_data_stress_std[:, 0, 1],
                ),
                np.add(
                    rve_data_stress_mean[:, 0, 1],
                    1.96 * rve_data_stress_std[:, 0, 1],
                ),
                color="#0077BB",
                edgecolor="none",
                alpha=0.5,
            )
        ax[1, 1].set(**pparam)

        pparam = dict(xlabel="Time step", ylabel=r"$\sigma_{22}$ (MPa)")
        ax[1, 2].plot(hf_test_stress[:, 1, 1], color="#CC3311",
                      linewidth=2, label="hf")
        if rve_data_strain_mean is not None:
            ax[1, 2].plot(rve_data_stress_mean[:, 1, 1],
                          color="#0077BB",
                          linewidth=2,
                          linestyle="--",
                          label="RVE")
            ax[1, 2].fill_between(
                np.arange(length_of_strain),
                np.subtract(
                    rve_data_stress_mean[:, 1, 1],
                    1.96 * rve_data_stress_std[:, 1, 1],
                ),
                np.add(
                    rve_data_stress_mean[:, 1, 1],
                    1.96 * rve_data_stress_std[:, 1, 1],
                ),
                color="#0077BB",
                edgecolor="none",
                alpha=0.5,
            )

        ax[1, 2].set(**pparam)
        ax[1, 2].legend([
            "Ground Truth (RVE)",
            "Ground Truth Mean (SVE)",
            "Ground Truth 95% CI"
        ], loc="lower right", fontsize=10, edgecolor="k", frameon=True)

        # set the fontsize of the axes
        for i in range(2):
            for j in range(3):
                ax[i, j].tick_params(axis='both', which='major', labelsize=12)
                ax[i, j].tick_params(axis='both', which='minor', labelsize=12)
        # set the linewidth of the axes
        for i in range(2):
            for j in range(3):
                for axis in ['top', 'bottom', 'left', 'right']:
                    ax[i, j].spines[axis].set_linewidth(1)
        # set the fontsize of the labels
        for i in range(2):
            for j in range(3):
                ax[i, j].set_xlabel(ax[i, j].get_xlabel(), fontsize=14)
                ax[i, j].set_ylabel(ax[i, j].get_ylabel(), fontsize=14)
                ax[i, j].yaxis.set_major_formatter(
                    formatter)
        # adjust the space between the subplots
        plt.subplots_adjust(wspace=0.32, hspace=0.25)
        # save the figure
        if save_figure:
            plt.savefig(f"test_data_{index}.svg",
                        dpi=300, bbox_inches="tight")
            plt.savefig(f"test_data_{index}.pdf",
                        dpi=300, bbox_inches="tight")
        else:
            plt.show()

    def convert_data_to_torch(
        self,
        dataset: pd.DataFrame,
    ) -> Tuple[torch.Tensor, torch.Tensor]:
        """convert data to torch

        Parameters
        ----------
        dataset : pd.DataFrame
            dataset data frame

        Returns
        -------
        Tuple[torch.Tensor, torch.Tensor]
            A tuple of data
        """
        # empty lists for data
        X, Y = [], []
        # gen number of samples
        num_samples = len(dataset)
        # get the data
        for ii in range(num_samples):
            X.append(
                (torch.FloatTensor(dataset['strain'].iloc[ii]).flatten(
                    start_dim=1)[:, [0, 1, 3]]).unsqueeze(0))
            Y.append(
                (torch.FloatTensor(dataset['stress'].iloc[ii]).flatten(
                    start_dim=1)[:, [0, 1, 3]]).unsqueeze(0))
        # concatenate all date together
        X, Y = torch.cat(X, dim=0), torch.cat(Y, dim=0)

        return X, Y

    def scale_dataset(self) -> None:
        """scale the dataset
        """
        self.strain_normalized, self.LX_mean, self.LX_std = \
            self._normalize_data(
                data=self.LX)
        self.stress_normalized, self.LY_mean, self.LY_std = \
            self._normalize_data(
                data=self.LY)

    def scale_back_inputs(self, input_data: Tensor) -> Tensor:
        """scale back the input data"""
        return input_data * self.LX_std + self.LX_mean

    def scale_back_outputs(self, output_data: Tensor) -> Tensor:
        """scale back the output data"""
        return output_data * self.LY_std + self.LY_mean

    def scale_back_variance(self, output_data: Tensor) -> Tensor:
        """scale back the variance data"""
        return output_data * self.LY_std**2

    @staticmethod
    def _normalize_data(data: Tensor) -> Tuple[Tensor, Tensor, Tensor]:
        """normalize the dataset

        Parameters
        ----------
        data : Tensor
            data to be normalized

        Returns
        -------
        Tuple[Tensor, Tensor, Tensor]
            normalized data, mean, and standard deviation
        """
        dim = (0, 1)
        data_mean = data.mean(dim=dim, keepdim=True)
        data_std = data.std(dim=dim, unbiased=False, keepdim=True)

        data_normalized = (data - data_mean) / data_std

        return data_normalized, data_mean, data_std
_normalize_data(data: Tensor) -> typing.Tuple[torch.Tensor, torch.Tensor, torch.Tensor] staticmethod ¤

normalize the dataset

Parameters:

Name Type Description Default
data Tensor

data to be normalized

 required

Returns:

Type Description
Tuple[Tensor, Tensor, Tensor]

normalized data, mean, and standard deviation
Source code in src/MFVeBRNN/dataset/load_dataset.py 
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
@staticmethod
def _normalize_data(data: Tensor) -> Tuple[Tensor, Tensor, Tensor]:
    """normalize the dataset

    Parameters
    ----------
    data : Tensor
        data to be normalized

    Returns
    -------
    Tuple[Tensor, Tensor, Tensor]
        normalized data, mean, and standard deviation
    """
    dim = (0, 1)
    data_mean = data.mean(dim=dim, keepdim=True)
    data_std = data.std(dim=dim, unbiased=False, keepdim=True)

    data_normalized = (data - data_mean) / data_std

    return data_normalized, data_mean, data_std
convert_data_to_torch(dataset: DataFrame) -> typing.Tuple[torch.Tensor, torch.Tensor] ¤

convert data to torch

Parameters:

Name Type Description Default
dataset DataFrame

dataset data frame

 required

Returns:

Type Description
Tuple[Tensor, Tensor]

A tuple of data
Source code in src/MFVeBRNN/dataset/load_dataset.py 
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
def convert_data_to_torch(
    self,
    dataset: pd.DataFrame,
) -> Tuple[torch.Tensor, torch.Tensor]:
    """convert data to torch

    Parameters
    ----------
    dataset : pd.DataFrame
        dataset data frame

    Returns
    -------
    Tuple[torch.Tensor, torch.Tensor]
        A tuple of data
    """
    # empty lists for data
    X, Y = [], []
    # gen number of samples
    num_samples = len(dataset)
    # get the data
    for ii in range(num_samples):
        X.append(
            (torch.FloatTensor(dataset['strain'].iloc[ii]).flatten(
                start_dim=1)[:, [0, 1, 3]]).unsqueeze(0))
        Y.append(
            (torch.FloatTensor(dataset['stress'].iloc[ii]).flatten(
                start_dim=1)[:, [0, 1, 3]]).unsqueeze(0))
    # concatenate all date together
    X, Y = torch.cat(X, dim=0), torch.cat(Y, dim=0)

    return X, Y
get_hf_train_val_split(num_hf_train: int, num_hf_val: int, seed: int = 1) -> None ¤

Get the processed data for training, validation, and testing

Parameters:

Name Type Description Default
num_hf_train int

Number of training data

 required
num_hf_val int

Number of validation data

 required
seed int

seed of selecting the training paths

 1
Source code in src/MFVeBRNN/dataset/load_dataset.py 
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
def get_hf_train_val_split(self,
                           num_hf_train: int,
                           num_hf_val: int,
                           seed: int = 1) -> None:
    """Get the processed data for training, validation, and testing
    Parameters
    ----------
    num_hf_train : int
        Number of training data
    num_hf_val : int
        Number of validation data
    seed: int
        seed of selecting the training paths
    """
    total_hf_samples = len(self.hf_dataset)
    requested_total = num_hf_train + num_hf_val
    if requested_total > total_hf_samples:
        raise ValueError(
            "Requested split exceeds the total number of samples.")
    # Shuffle the indices with certain seed
    indices = torch.randperm(
        total_hf_samples, generator=torch.Generator().manual_seed(seed))

    # Compute split boundaries
    train_end = num_hf_train
    val_end = train_end + num_hf_val
    # Assign data splits using non-overlapping indices
    train_indices = indices[:train_end]
    val_indices = indices[train_end:val_end]
    # get the data
    self.hx_train = self.hx_scaled[train_indices]
    self.hx_val = self.hx_scaled[val_indices]
    self.hy_train = self.hy_scaled[train_indices]
    self.hy_val = self.hy_scaled[val_indices]
get_id_hf_test_data() -> None ¤

Get the processed data for testing

Source code in src/MFVeBRNN/dataset/load_dataset.py 
845
846
847
848
849
850
851
852
853
854
def get_id_hf_test_data(self) -> None:
    """Get the processed data for testing
    """
    # convert to torch
    self.ID_HX, self.ID_HY = self.convert_data_to_torch(
        self.id_hf_test_dataset)

    # scale the high-fidelity test data
    self.hx_id_gt_scaled = (self.ID_HX - self.LX_mean) / self.LX_std
    self.hy_id_gt_scaled = (self.ID_HY - self.LY_mean) / self.LY_std
get_id_lf_ground_truth() -> None ¤

get the processed data for testing

Source code in src/MFVeBRNN/dataset/load_dataset.py 
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
def get_id_lf_ground_truth(self) -> None:
    """get the processed data for testing
    """

    # first convert to torch
    x_test = []
    y_test_mean = []
    y_test_var = []
    for ii in range(self.n_id_hf_test):
        x_test.append((
            torch.FloatTensor(
                self.id_ground_truth['strain_mean'].iloc[ii]).flatten(
                start_dim=1)[:, [0, 1, 3]]
        ).unsqueeze(0))
        y_test_mean.append((
            torch.FloatTensor(
                self.id_ground_truth['stress_mean'].iloc[ii]).flatten(
                start_dim=1)[:, [0, 1, 3]]
        ).unsqueeze(0))
        y_test_var.append((
            torch.FloatTensor(
                self.id_ground_truth['stress_var'].iloc[ii]).flatten(
                start_dim=1)[:, [0, 1, 3]]
        ).unsqueeze(0))

    # original scale
    self.lx_id_gt = torch.cat(x_test, dim=0)
    self.ly_id_gt_mean = torch.cat(y_test_mean, dim=0)
    self.ly_id_gt_var = torch.cat(y_test_var, dim=0)
    # scale it
    self.lx_id_gt_scaled = (self.lx_id_gt-self.LX_mean)/self.LX_std
    self.ly_id_gt_mean_scaled = (
        self.ly_id_gt_mean - self.LY_mean)/self.LY_std
    self.ly_id_gt_var_scaled = self.ly_id_gt_var / self.LY_std**2
get_lf_train_val_split(num_lf_train: int, num_lf_val: int, seed: int = 1) -> None ¤

Get the processed data for training, validation, and testing

Parameters:

Name Type Description Default
num_lf_train int

Number of low-fidelity training data

 required
num_lf_val int

Number of low-fidelity validation data

 required
seed int

seed of selecting the training paths

 1
Source code in src/MFVeBRNN/dataset/load_dataset.py 
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
def get_lf_train_val_split(self,
                           num_lf_train: int,
                           num_lf_val: int,
                           seed: int = 1) -> None:
    """Get the processed data for training, validation, and testing

    Parameters
    ----------
    num_lf_train : int
        Number of low-fidelity training data
    num_lf_val : int
        Number of low-fidelity validation data
    seed: int
        seed of selecting the training paths
    """

    total_lf_samples = len(self.lf_dataset)
    requested_total = num_lf_train + num_lf_val

    if requested_total > total_lf_samples:
        raise ValueError(
            "Requested split exceeds the total number of samples.")

    # Shuffle the indices with certain seed
    indices = torch.randperm(
        total_lf_samples, generator=torch.Generator().manual_seed(seed))
    # Compute split boundaries
    train_end = num_lf_train
    val_end = train_end + num_lf_val

    # Assign data splits using non-overlapping indices
    train_indices = indices[:train_end]
    val_indices = indices[train_end:val_end]

    self.lx_train = self.strain_normalized[train_indices]
    self.lx_val = self.strain_normalized[val_indices]

    self.ly_train = self.stress_normalized[train_indices]
    self.ly_val = self.stress_normalized[val_indices]
get_ood_hf_test_data() -> None ¤

Get the processed data for testing

Source code in src/MFVeBRNN/dataset/load_dataset.py 
891
892
893
894
895
896
897
898
899
def get_ood_hf_test_data(self) -> None:
    """Get the processed data for testing
    """
    # convert to torch
    self.OOD_HX, self.OOD_HY = self.convert_data_to_torch(
        self.ood_hf_test_dataset)
    # scale the high-fidelity test data
    self.hx_ood_gt_scaled = (self.OOD_HX - self.LX_mean) / self.LX_std
    self.hy_ood_gt_scaled = (self.OOD_HY - self.LY_mean) / self.LY_std
get_ood_lf_ground_truth() -> None ¤

get the processed data for testing

Source code in src/MFVeBRNN/dataset/load_dataset.py 
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
def get_ood_lf_ground_truth(self) -> None:
    """get the processed data for testing
    """

    # first convert to torch
    x_test = []
    y_test_mean = []
    y_test_var = []
    for ii in range(self.n_ood_hf_test):
        x_test.append((
            torch.FloatTensor(
                self.ood_ground_truth['strain_mean'].iloc[ii]).flatten(
                start_dim=1)[:, [0, 1, 3]]
        ).unsqueeze(0))
        y_test_mean.append((
            torch.FloatTensor(
                self.ood_ground_truth['stress_mean'].iloc[ii]).flatten(
                start_dim=1)[:, [0, 1, 3]]
        ).unsqueeze(0))
        y_test_var.append((
            torch.FloatTensor(
                self.ood_ground_truth['stress_var'].iloc[ii]).flatten(
                start_dim=1)[:, [0, 1, 3]]
        ).unsqueeze(0))

    # original scale
    self.lx_ood_gt = torch.cat(x_test, dim=0)
    self.ly_ood_gt_mean = torch.cat(y_test_mean, dim=0)
    self.ly_ood_gt_var = torch.cat(y_test_var, dim=0)
    # scale it
    self.lx_ood_gt_scaled = (self.lx_ood_gt-self.LX_mean)/self.LX_std
    self.ly_ood_gt_mean_scaled = (
        self.ly_ood_gt_mean - self.LY_mean)/self.LY_std
    self.ly_ood_gt_var_scaled = self.ly_ood_gt_var / self.LY_std**2
plot_testing_data(index: int, test_data: str = 'id', save_figure: bool = False) -> None ¤

plot the data

Parameters:

Name Type Description Default
index int

index of the data

 required
test_data str

type of test data to plot, by default "id"

 'id'
save_figure bool

save the figure, by default False

 False
Source code in src/MFVeBRNN/dataset/load_dataset.py 
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
def plot_testing_data(self,
                      index: int,
                      test_data: str = "id",
                      save_figure: bool = False) -> None:
    """plot the data

    Parameters
    ----------
    index : int
        index of the data
    test_data : str, optional
        type of test data to plot, by default "id"
    save_figure : bool, optional
        save the figure, by default False
    """

    # get the data
    if test_data == "id":
        # check the id_ground_truth attribute is available
        if hasattr(self, "id_ground_truth"):
            rve_data_strain_mean = (
                self.id_ground_truth["strain_mean"].iloc[index]
            )
            rve_data_stress_mean = (
                self.id_ground_truth["stress_mean"].iloc[index]
            )
            rve_data_stress_std = (
                self.id_ground_truth["stress_var"].iloc[index] ** 0.5
            )
        else:
            rve_data_strain_mean = None
            rve_data_stress_mean = None
            rve_data_stress_std = None
        # get the high-fidelity data
        hf_test_train = self.id_hf_test_dataset["strain"].iloc[index]
        hf_test_stress = self.id_hf_test_dataset["stress"].iloc[index]

        # length of the data
        length_of_strain = hf_test_train.shape[0]
    elif test_data == "ood":
        if hasattr(self, "ood_ground_truth"):
            rve_data_strain_mean = (
                self.ood_ground_truth["strain_mean"].iloc[index]
            )
            rve_data_stress_mean = (
                self.ood_ground_truth["stress_mean"].iloc[index]
            )
            rve_data_stress_std = (
                self.ood_ground_truth["stress_var"].iloc[index] ** 0.5
            )
        else:
            rve_data_strain_mean = None
            rve_data_stress_mean = None
            rve_data_stress_std = None

        # get the high-fidelity data
        hf_test_train = self.ood_hf_test_dataset["strain"].iloc[index]
        hf_test_stress = self.ood_hf_test_dataset["stress"].iloc[index]

        # length of the data
        length_of_strain = hf_test_train.shape[0]

    fig, ax = plt.subplots(2, 3, figsize=(12, 5))
    # set the title of the whole figure
    pparam = dict(ylabel=r"$E_{11}$")
    ax[0, 0].plot(hf_test_train[:, 0, 0],
                  color="#0077BB",
                  linewidth=2,
                  label="rve")
    ax[0, 0].set(**pparam)
    # set the limits of the y axis
    pparam = dict(ylabel=r"$E_{12}$")
    ax[0, 1].plot(hf_test_train[:, 0, 1],
                  color="#0077BB",
                  linewidth=2,
                  label="rve")

    ax[0, 1].set(**pparam)
    pparam = dict(ylabel=r"$E_{22}$")
    ax[0, 2].plot(hf_test_train[:, 1, 1],
                  color="#0077BB",
                  linewidth=2,
                  label="rve")

    ax[0, 2].set(**pparam)
    pparam = dict(xlabel="Time step", ylabel=r"$\sigma_{11}$ (MPa)")
    ax[1, 0].plot(hf_test_stress[:, 0, 0], color="#CC3311",
                  linewidth=2, label="hf")
    if rve_data_strain_mean is not None:
        ax[1, 0].plot(rve_data_stress_mean[:, 0, 0],
                      color="#0077BB",
                      linewidth=2,
                      linestyle="--")
        ax[1, 0].fill_between(
            np.arange(length_of_strain),
            np.subtract(
                rve_data_stress_mean[:, 0, 0],
                1.96 * rve_data_stress_std[:, 0, 0],
            ),
            np.add(
                rve_data_stress_mean[:, 0, 0],
                1.96 * rve_data_stress_std[:, 0, 0],
            ),
            color="#0077BB",
            edgecolor="none",
            alpha=0.5,
        )

    ax[1, 0].set(**pparam)

    pparam = dict(xlabel="Time step", ylabel=r"$\sigma_{12}$ (MPa)")
    ax[1, 1].plot(hf_test_stress[:, 0, 1],
                  color="#CC3311", linewidth=2,
                  label="hf")
    if rve_data_strain_mean is not None:
        ax[1, 1].plot(rve_data_stress_mean[:, 0, 1],
                      color="#0077BB",
                      linewidth=2,
                      linestyle="--",
                      label="rve")

        ax[1, 1].fill_between(
            np.arange(length_of_strain),
            np.subtract(
                rve_data_stress_mean[:, 0, 1],
                1.96 * rve_data_stress_std[:, 0, 1],
            ),
            np.add(
                rve_data_stress_mean[:, 0, 1],
                1.96 * rve_data_stress_std[:, 0, 1],
            ),
            color="#0077BB",
            edgecolor="none",
            alpha=0.5,
        )
    ax[1, 1].set(**pparam)

    pparam = dict(xlabel="Time step", ylabel=r"$\sigma_{22}$ (MPa)")
    ax[1, 2].plot(hf_test_stress[:, 1, 1], color="#CC3311",
                  linewidth=2, label="hf")
    if rve_data_strain_mean is not None:
        ax[1, 2].plot(rve_data_stress_mean[:, 1, 1],
                      color="#0077BB",
                      linewidth=2,
                      linestyle="--",
                      label="RVE")
        ax[1, 2].fill_between(
            np.arange(length_of_strain),
            np.subtract(
                rve_data_stress_mean[:, 1, 1],
                1.96 * rve_data_stress_std[:, 1, 1],
            ),
            np.add(
                rve_data_stress_mean[:, 1, 1],
                1.96 * rve_data_stress_std[:, 1, 1],
            ),
            color="#0077BB",
            edgecolor="none",
            alpha=0.5,
        )

    ax[1, 2].set(**pparam)
    ax[1, 2].legend([
        "Ground Truth (RVE)",
        "Ground Truth Mean (SVE)",
        "Ground Truth 95% CI"
    ], loc="lower right", fontsize=10, edgecolor="k", frameon=True)

    # set the fontsize of the axes
    for i in range(2):
        for j in range(3):
            ax[i, j].tick_params(axis='both', which='major', labelsize=12)
            ax[i, j].tick_params(axis='both', which='minor', labelsize=12)
    # set the linewidth of the axes
    for i in range(2):
        for j in range(3):
            for axis in ['top', 'bottom', 'left', 'right']:
                ax[i, j].spines[axis].set_linewidth(1)
    # set the fontsize of the labels
    for i in range(2):
        for j in range(3):
            ax[i, j].set_xlabel(ax[i, j].get_xlabel(), fontsize=14)
            ax[i, j].set_ylabel(ax[i, j].get_ylabel(), fontsize=14)
            ax[i, j].yaxis.set_major_formatter(
                formatter)
    # adjust the space between the subplots
    plt.subplots_adjust(wspace=0.32, hspace=0.25)
    # save the figure
    if save_figure:
        plt.savefig(f"test_data_{index}.svg",
                    dpi=300, bbox_inches="tight")
        plt.savefig(f"test_data_{index}.pdf",
                    dpi=300, bbox_inches="tight")
    else:
        plt.show()
plot_training_data(index: int, fidelity: str = 'lf', save_figure: bool = False) -> None ¤

plot the training data

index: int index of the path fidelity: str which fidelity to plot, "lf" for low-fidelity data and "hf" for high-fidelity data save_figure: bool save the figure to disk or not

Source code in src/MFVeBRNN/dataset/load_dataset.py 
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
def plot_training_data(self,
                       index: int,
                       fidelity: str = "lf",
                       save_figure: bool = False,
                       ) -> None:
    """plot the training data

    index: int
        index of the path
    fidelity: str
        which fidelity to plot, "lf" for low-fidelity data and "hf" for
        high-fidelity data
    save_figure: bool
        save the figure to disk or not
    """

    if fidelity == "hf":
        dataset_to_use = self.hf_dataset
    else:
        dataset_to_use = self.lf_dataset

    # get the data
    fig, ax = plt.subplots(2, 3, figsize=(12, 5))
    pparam = dict(ylabel=r"$E_{11}$")
    ax[0, 0].plot(dataset_to_use["strain"].iloc[index][:, 0, 0],
                  color="#0077BB",
                  linewidth=2)
    ax[0, 0].set(**pparam)
    # set the limits of the y axis
    pparam = dict(ylabel=r"$E_{12}$")
    ax[0, 1].plot(dataset_to_use["strain"].iloc[index][:, 0, 1],
                  color="#0077BB",
                  linewidth=2)
    ax[0, 1].set(**pparam)
    pparam = dict(ylabel=r"$E_{22}$")
    ax[0, 2].plot(dataset_to_use["strain"].iloc[index][:, 1, 1],
                  color="#0077BB",
                  linewidth=2)
    ax[0, 2].set(**pparam)
    pparam = dict(xlabel="Time step", ylabel=r"$\sigma_{11}$ (MPa)")
    ax[1, 0].plot(dataset_to_use["stress"].iloc[index][:, 0, 0],
                  color="#0077BB",
                  linewidth=2)

    ax[1, 0].set(**pparam)
    pparam = dict(xlabel="Time step", ylabel=r"$\sigma_{12}$ (MPa)")
    ax[1, 1].plot(dataset_to_use["stress"].iloc[index][:, 0, 1],
                  color="#0077BB",
                  linewidth=2)

    ax[1, 1].set(**pparam)
    pparam = dict(xlabel="Time step", ylabel=r"$\sigma_{22}$ (MPa)")
    ax[1, 2].plot(dataset_to_use["stress"].iloc[index][:, 1, 1],
                  color="#0077BB",
                  linewidth=2)
    ax[1, 2].set(**pparam)
    # set the fontsize of the axes
    for i in range(2):
        for j in range(3):
            ax[i, j].tick_params(axis='both', which='major', labelsize=12)
            ax[i, j].tick_params(axis='both', which='minor', labelsize=12)
    # set the linewidth of the axes
    for i in range(2):
        for j in range(3):
            for axis in ['top', 'bottom', 'left', 'right']:
                ax[i, j].spines[axis].set_linewidth(1.5)
    # set the fontsize of the labels
    for i in range(2):
        for j in range(3):
            ax[i, j].set_xlabel(ax[i, j].get_xlabel(), fontsize=14)
            ax[i, j].set_ylabel(ax[i, j].get_ylabel(), fontsize=14)
            ax[i, j].yaxis.set_major_formatter(
                formatter)
    # adjust the space between the subplots
    plt.subplots_adjust(wspace=0.32, hspace=0.25)
    # save the figure
    if save_figure:
        plt.savefig(f"train_data_{index}.png",
                    dpi=300, bbox_inches="tight")
        plt.savefig(f"train_data_{index}.svg",
                    dpi=300, bbox_inches="tight")
    else:
        plt.show()
scale_back_inputs(input_data: Tensor) -> Tensor ¤

scale back the input data

Source code in src/MFVeBRNN/dataset/load_dataset.py 
1259
1260
1261
def scale_back_inputs(self, input_data: Tensor) -> Tensor:
    """scale back the input data"""
    return input_data * self.LX_std + self.LX_mean
scale_back_outputs(output_data: Tensor) -> Tensor ¤

scale back the output data

Source code in src/MFVeBRNN/dataset/load_dataset.py 
1263
1264
1265
def scale_back_outputs(self, output_data: Tensor) -> Tensor:
    """scale back the output data"""
    return output_data * self.LY_std + self.LY_mean
scale_back_variance(output_data: Tensor) -> Tensor ¤

scale back the variance data

Source code in src/MFVeBRNN/dataset/load_dataset.py 
1267
1268
1269
def scale_back_variance(self, output_data: Tensor) -> Tensor:
    """scale back the variance data"""
    return output_data * self.LY_std**2
scale_dataset() -> None ¤

scale the dataset

Source code in src/MFVeBRNN/dataset/load_dataset.py 
1249
1250
1251
1252
1253
1254
1255
1256
1257
def scale_dataset(self) -> None:
    """scale the dataset
    """
    self.strain_normalized, self.LX_mean, self.LX_std = \
        self._normalize_data(
            data=self.LX)
    self.stress_normalized, self.LY_mean, self.LY_std = \
        self._normalize_data(
            data=self.LY)