Skip to content

Core Models

physXAI.models.models

Attributes

MODEL_CLASS_REGISTRY: dict[str, Type[AbstractModel]] = dict() module-attribute

Classes

AbstractModel

Bases: ABC

Source code in physXAI/models/models.py 
 18
 19
 20
 21
 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
 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
class AbstractModel(ABC):

    @abstractmethod
    def generate_model(self, **kwargs):
        """
        Abstract method to be implemented by subclasses.
        Should generate and return an instance of the specific model.
        `kwargs` can be used to pass necessary parameters, e.g., `td` (TrainingData or TrainingDataMultiStep).
        """
        pass

    @abstractmethod
    def compile_model(self, model):
        """
        Abstract method for model compilation.
        Relevant for models like Keras neural networks. For scikit-learn models,
        this might be a pass-through or not applicable.
        """
        pass

    @abstractmethod
    def fit_model(self, model, td: TrainingDataGeneric):
        """
        Abstract method to fit the model to the training data.

        Args:
            model: The model instance to be trained.
            td (TrainingDataGeneric): The TrainingData object.
        """
        pass

    @staticmethod
    @abstractmethod
    def evaluate(model, td: TrainingDataGeneric):
        pass

    @abstractmethod
    def plot(self, td: TrainingDataGeneric):
        """
         Abstract method for generating and displaying plots related to model performance.

         Args:
             td (TrainingDataGeneric): The TrainingData object containing true values and predictions.
         """
        pass

    @abstractmethod
    def save_model(self, model, save_path: str):
        """
        Abstract method for saving the trained model.

        Args:
            model: The trained model instance to save.
            save_path (str): The path where the model should be saved.
        """
        pass

    @abstractmethod
    def load_model(self, load_path: str):
        """
        Abstract method for loading a pre-trained model.

        Args:
            load_path (str): The path from which to load the model.

        Returns:
            The loaded model instance.
        """
        pass

    def pipeline(self, td: TrainingDataGeneric, save_path: str = None, plot: bool = True, save_model: bool = True):
        """
          Defines a standard pipeline for single-step models:
          1. Generate model
          2. Compile model (if applicable)
          3. Fit model
          4. Evaluate model
          5. Plot results
          6. Save model (if save_path is provided)

          Args:
              td (TrainingDataGeneric): The training data.
              save_path (str, optional): Path to save the trained model. Defaults to None (Saving path from Logger).
              plot (bool, optional): Whether to plot the results. Defaults to True.
              save_model (bool, optional): Whether to save the trained model. Defaults to True.

          Returns:
              The trained model instance.
          """

        model = self.generate_model(td=td)
        self.compile_model(model)
        self.fit_model(model, td)

        self.evaluate(model, td)
        if plot:
            self.plot(td)
        if save_model:
            self.save_model(model, save_path=save_path)

        return model

    def online_pipeline(self, td: TrainingDataGeneric, load_path: str, save_path: str = None,
                        plot: bool = True, save_model: bool = True):
        """
        Implements an "online" training pipeline: loads a pre-existing model,
        further trains it on new data, evaluates, plots, and saves it back.

        Args:
            td (TrainingDataGeneric): New training data.
            load_path (str): Path to the pre-existing model.
            save_path (str, optional): Path to save the trained model. Defaults to None (Saving path from Logger).
            plot (bool, optional): Whether to plot the results. Defaults to True.
            save_model (bool, optional): Whether to save the trained model. Defaults to True.

        Returns:
            The updated and saved model.
        """

        model = self.load_model(load_path)
        self.compile_model(model)

        model = self.online_pipeline_internal(td, model)

        if plot:
            self.plot(td)
        if save_model:
            self.save_model(model, save_path=save_path)

        return model

    def online_pipeline_internal(self, td: TrainingDataGeneric, model):
        """
        Implements an "online" training pipeline: trains a pre-existing model on new data

        Args:
            td (TrainingDataGeneric): New training data.
            model: The model to train.

        Returns:
            The updated and saved model.
        """
        self.fit_model(model, td)
        self.evaluate(model, td)

        return model

    def get_config(self) -> dict:
        return {
            '__class_name__': self.__class__.__name__,
        }

    @classmethod
    @abstractmethod
    def from_config(cls, config: dict) -> 'AbstractModel':
        pass

    @staticmethod
    def model_from_config(item_conf: dict) -> 'AbstractModel':
        """
        Factory function to create a model object from its configuration dictionary.

        Args:
            item_conf (dict): The configuration dictionary for a model.
                              Must contain 'class_name' and other necessary parameters.

        Returns:
            AbstractModel: An instance of the appropriate model subclass.

        Raises:
            KeyError: If 'class_name' is not in `item_conf` or if the class_name is not in `MODEL_CLASS_REGISTRY`.
        """
        class_name = item_conf['__class_name__']
        model_class = MODEL_CLASS_REGISTRY[class_name]
        m = model_class.from_config(item_conf)
        return m
Functions
generate_model(**kwargs) abstractmethod

Abstract method to be implemented by subclasses. Should generate and return an instance of the specific model. kwargs can be used to pass necessary parameters, e.g., td (TrainingData or TrainingDataMultiStep).

Source code in physXAI/models/models.py 
20
21
22
23
24
25
26
27
@abstractmethod
def generate_model(self, **kwargs):
    """
    Abstract method to be implemented by subclasses.
    Should generate and return an instance of the specific model.
    `kwargs` can be used to pass necessary parameters, e.g., `td` (TrainingData or TrainingDataMultiStep).
    """
    pass
compile_model(model) abstractmethod

Abstract method for model compilation. Relevant for models like Keras neural networks. For scikit-learn models, this might be a pass-through or not applicable.

Source code in physXAI/models/models.py 
29
30
31
32
33
34
35
36
@abstractmethod
def compile_model(self, model):
    """
    Abstract method for model compilation.
    Relevant for models like Keras neural networks. For scikit-learn models,
    this might be a pass-through or not applicable.
    """
    pass
fit_model(model, td: TrainingDataGeneric) abstractmethod

Abstract method to fit the model to the training data.

Parameters:

Name Type Description Default
model

The model instance to be trained.

 required
td TrainingDataGeneric

The TrainingData object.

 required
Source code in physXAI/models/models.py 
38
39
40
41
42
43
44
45
46
47
@abstractmethod
def fit_model(self, model, td: TrainingDataGeneric):
    """
    Abstract method to fit the model to the training data.

    Args:
        model: The model instance to be trained.
        td (TrainingDataGeneric): The TrainingData object.
    """
    pass
evaluate(model, td: TrainingDataGeneric) abstractmethod staticmethod
Source code in physXAI/models/models.py 
49
50
51
52
@staticmethod
@abstractmethod
def evaluate(model, td: TrainingDataGeneric):
    pass
plot(td: TrainingDataGeneric) abstractmethod

Abstract method for generating and displaying plots related to model performance.

Parameters:

Name Type Description Default
td TrainingDataGeneric

The TrainingData object containing true values and predictions.

 required
Source code in physXAI/models/models.py 
54
55
56
57
58
59
60
61
62
@abstractmethod
def plot(self, td: TrainingDataGeneric):
    """
     Abstract method for generating and displaying plots related to model performance.

     Args:
         td (TrainingDataGeneric): The TrainingData object containing true values and predictions.
     """
    pass
save_model(model, save_path: str) abstractmethod

Abstract method for saving the trained model.

Parameters:

Name Type Description Default
model

The trained model instance to save.

 required
save_path str

The path where the model should be saved.

 required
Source code in physXAI/models/models.py 
64
65
66
67
68
69
70
71
72
73
@abstractmethod
def save_model(self, model, save_path: str):
    """
    Abstract method for saving the trained model.

    Args:
        model: The trained model instance to save.
        save_path (str): The path where the model should be saved.
    """
    pass
load_model(load_path: str) abstractmethod

Abstract method for loading a pre-trained model.

Parameters:

Name Type Description Default
load_path str

The path from which to load the model.

 required

Returns:

Type Description

The loaded model instance.
Source code in physXAI/models/models.py 
75
76
77
78
79
80
81
82
83
84
85
86
@abstractmethod
def load_model(self, load_path: str):
    """
    Abstract method for loading a pre-trained model.

    Args:
        load_path (str): The path from which to load the model.

    Returns:
        The loaded model instance.
    """
    pass
pipeline(td: TrainingDataGeneric, save_path: str = None, plot: bool = True, save_model: bool = True)

Defines a standard pipeline for single-step models: 1. Generate model 2. Compile model (if applicable) 3. Fit model 4. Evaluate model 5. Plot results 6. Save model (if save_path is provided)

Parameters:

Name Type Description Default
td TrainingDataGeneric

The training data.

 required
save_path str

Path to save the trained model. Defaults to None (Saving path from Logger).

 None
plot bool

Whether to plot the results. Defaults to True.

 True
save_model bool

Whether to save the trained model. Defaults to True.

 True

Returns:

Type Description

The trained model instance.
Source code in physXAI/models/models.py 
 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
def pipeline(self, td: TrainingDataGeneric, save_path: str = None, plot: bool = True, save_model: bool = True):
    """
      Defines a standard pipeline for single-step models:
      1. Generate model
      2. Compile model (if applicable)
      3. Fit model
      4. Evaluate model
      5. Plot results
      6. Save model (if save_path is provided)

      Args:
          td (TrainingDataGeneric): The training data.
          save_path (str, optional): Path to save the trained model. Defaults to None (Saving path from Logger).
          plot (bool, optional): Whether to plot the results. Defaults to True.
          save_model (bool, optional): Whether to save the trained model. Defaults to True.

      Returns:
          The trained model instance.
      """

    model = self.generate_model(td=td)
    self.compile_model(model)
    self.fit_model(model, td)

    self.evaluate(model, td)
    if plot:
        self.plot(td)
    if save_model:
        self.save_model(model, save_path=save_path)

    return model
online_pipeline(td: TrainingDataGeneric, load_path: str, save_path: str = None, plot: bool = True, save_model: bool = True)

Implements an "online" training pipeline: loads a pre-existing model, further trains it on new data, evaluates, plots, and saves it back.

Parameters:

Name Type Description Default
td TrainingDataGeneric

New training data.

 required
load_path str

Path to the pre-existing model.

 required
save_path str

Path to save the trained model. Defaults to None (Saving path from Logger).

 None
plot bool

Whether to plot the results. Defaults to True.

 True
save_model bool

Whether to save the trained model. Defaults to True.

 True

Returns:

Type Description

The updated and saved model.
Source code in physXAI/models/models.py 
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
def online_pipeline(self, td: TrainingDataGeneric, load_path: str, save_path: str = None,
                    plot: bool = True, save_model: bool = True):
    """
    Implements an "online" training pipeline: loads a pre-existing model,
    further trains it on new data, evaluates, plots, and saves it back.

    Args:
        td (TrainingDataGeneric): New training data.
        load_path (str): Path to the pre-existing model.
        save_path (str, optional): Path to save the trained model. Defaults to None (Saving path from Logger).
        plot (bool, optional): Whether to plot the results. Defaults to True.
        save_model (bool, optional): Whether to save the trained model. Defaults to True.

    Returns:
        The updated and saved model.
    """

    model = self.load_model(load_path)
    self.compile_model(model)

    model = self.online_pipeline_internal(td, model)

    if plot:
        self.plot(td)
    if save_model:
        self.save_model(model, save_path=save_path)

    return model
online_pipeline_internal(td: TrainingDataGeneric, model)

Implements an "online" training pipeline: trains a pre-existing model on new data

Parameters:

Name Type Description Default
td TrainingDataGeneric

New training data.

 required
model

The model to train.

 required

Returns:

Type Description

The updated and saved model.
Source code in physXAI/models/models.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def online_pipeline_internal(self, td: TrainingDataGeneric, model):
    """
    Implements an "online" training pipeline: trains a pre-existing model on new data

    Args:
        td (TrainingDataGeneric): New training data.
        model: The model to train.

    Returns:
        The updated and saved model.
    """
    self.fit_model(model, td)
    self.evaluate(model, td)

    return model
get_config() -> dict
Source code in physXAI/models/models.py 
165
166
167
168
def get_config(self) -> dict:
    return {
        '__class_name__': self.__class__.__name__,
    }
from_config(config: dict) -> AbstractModel abstractmethod classmethod
Source code in physXAI/models/models.py 
170
171
172
173
@classmethod
@abstractmethod
def from_config(cls, config: dict) -> 'AbstractModel':
    pass
model_from_config(item_conf: dict) -> AbstractModel staticmethod

Factory function to create a model object from its configuration dictionary.

Parameters:

Name Type Description Default
item_conf dict

The configuration dictionary for a model. Must contain 'class_name' and other necessary parameters.

 required

Returns:

Name Type Description
AbstractModel AbstractModel

An instance of the appropriate model subclass.

Raises:

Type Description
KeyError

If 'class_name' is not in item_conf or if the class_name is not in MODEL_CLASS_REGISTRY.
Source code in physXAI/models/models.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
@staticmethod
def model_from_config(item_conf: dict) -> 'AbstractModel':
    """
    Factory function to create a model object from its configuration dictionary.

    Args:
        item_conf (dict): The configuration dictionary for a model.
                          Must contain 'class_name' and other necessary parameters.

    Returns:
        AbstractModel: An instance of the appropriate model subclass.

    Raises:
        KeyError: If 'class_name' is not in `item_conf` or if the class_name is not in `MODEL_CLASS_REGISTRY`.
    """
    class_name = item_conf['__class_name__']
    model_class = MODEL_CLASS_REGISTRY[class_name]
    m = model_class.from_config(item_conf)
    return m

SingleStepModel

Bases: AbstractModel, ABC

Abstract Base Class for single-step prediction models. Defines a common interface and a pipeline for training, evaluating, plotting, and saving models that predict a single output based on input features.

Source code in physXAI/models/models.py 
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
class SingleStepModel(AbstractModel, ABC):
    """
    Abstract Base Class for single-step prediction models.
    Defines a common interface and a pipeline for training, evaluating,
    plotting, and saving models that predict a single output based on input features.
    """

    def __init__(self, **kwargs):
        pass

    @staticmethod
    def evaluate(model, td: TrainingDataGeneric):
        """
        Evaluates the trained model on training, validation (if available), and test sets.
        Predictions are stored in the TrainingData object, and metrics are calculated and stored.

        Args:
            model: The trained model instance.
            td (TrainingDataGeneric): The TrainingData object containing datasets and for storing results.
        """

        y_pred_train = model.predict(td.X_train_single)
        y_pred_test = model.predict(td.X_test_single)
        if td.X_val_single is not None:
            y_pred_val = model.predict(td.X_val_single)
        else:
            y_pred_val = None
        td.add_predictions(y_pred_train, y_pred_val, y_pred_test)

        metrics = Metrics(td)
        if isinstance(td, TrainingData):
            td.add_metrics(metrics)
        elif isinstance(td, TrainingDataMultiStep):
            td.add_single_step_metrics(metrics)
            SingleStepModel.evaluate_multi(model, td)
        else:
            raise NotImplementedError

    @staticmethod
    def evaluate_multi(model, td: TrainingDataMultiStep):
        if td.init_columns[0] != td.output[0]:
            delta_prediction = True
        else:
            delta_prediction = False
        y_pred_train, y_train = SingleStepModel._evaluate_multi_inner_loop(model, td.X_train_features, td.y_train,
                                                                           td.columns, td.init_columns[0],
                                                                           delta_prediction)
        y_pred_test, y_test = SingleStepModel._evaluate_multi_inner_loop(model, td.X_test_features, td.y_test,
                                                                         td.columns, td.init_columns[0],
                                                                         delta_prediction)
        if td.X_val_features is not None:
            y_pred_val, y_val = SingleStepModel._evaluate_multi_inner_loop(model, td.X_val_features, td.y_val,
                                                                           td.columns, td.init_columns[0],
                                                                           delta_prediction)
        else:
            y_pred_val, y_val = None, None
        td.add_predictions(y_pred_train, y_pred_val, y_pred_test)
        td.y_train = y_train
        td.y_val = y_val
        td.y_test = y_test

        metrics = MetricsMultiStep(td)
        td.add_metrics(metrics)

    @staticmethod
    def _evaluate_multi_inner_loop(model, X: np.ndarray, y: np.ndarray, X_columns: list[str],
                                   recursive_output_column: str, delta_prediction: bool = True) \
            -> (np.ndarray, np.ndarray):
        true_vals = np.ndarray(shape=(y.shape[0], y.shape[1], 1), dtype=np.float64)
        preds = np.ndarray(shape=(X.shape[0], X.shape[1], 1), dtype=np.float64)

        assert recursive_output_column in X_columns, (f'Error: Cannot find recursive_output_column '
                                                      f'"{recursive_output_column}" in X_columns')
        if isinstance(model, LinearRegression):
            for b in range(X.shape[0]):
                index = X_columns.index(recursive_output_column)
                current_val = X[b, 0, index]
                current_true_val = current_val
                for t in range(X.shape[1]):
                    pred = float(model.predict(X[b, t, :].reshape(1, -1)))
                    if delta_prediction:
                        current_val += pred
                        current_true_val += y[b, t, 0]
                        pred = current_val
                    else:
                        current_true_val = y[b, t, 0]
                    preds[b, t, 0] = pred
                    true_vals[b, t, 0] = current_true_val

                    for l in range(0, 10):
                        if l == 0:
                            col = recursive_output_column
                        else:
                            col = recursive_output_column + f'_lag{l}'
                        if col in X_columns:
                            index = X_columns.index(col)
                            if t + 1 + l < X.shape[1]:
                                X[b, t + 1 + l, index] = pred
        else:
            index = X_columns.index(recursive_output_column)
            current_val = X[:, 0, index].reshape(-1, 1)
            current_true_val = current_val.copy()
            for t in range(X.shape[1]):
                pred = model.predict(X[:, t, :], verbose=0)
                if delta_prediction:
                    current_val += pred
                    current_true_val += y[:, t, 0].reshape(-1, 1)
                    pred = current_val
                else:
                    current_true_val = y[:, t, 0].reshape(-1, 1)
                preds[:, t, 0] = pred.reshape(-1)
                true_vals[:, t, 0] = current_true_val.reshape(-1)

                for l in range(0, 10):
                    if l == 0:
                        col = recursive_output_column
                    else:
                        col = recursive_output_column + f'_lag{l}'
                    if col in X_columns:
                        index = X_columns.index(col)
                        if t + 1 + l < X.shape[1]:
                            X[:, t + 1 + l, index] = pred.reshape(-1)
        return preds, true_vals

    @classmethod
    def from_config(cls, config: dict) -> 'SingleStepModel':
        return cls(**config)
Functions
__init__(**kwargs)
Source code in physXAI/models/models.py 
214
215
def __init__(self, **kwargs):
    pass
evaluate(model, td: TrainingDataGeneric) staticmethod

Evaluates the trained model on training, validation (if available), and test sets. Predictions are stored in the TrainingData object, and metrics are calculated and stored.

Parameters:

Name Type Description Default
model

The trained model instance.

 required
td TrainingDataGeneric

The TrainingData object containing datasets and for storing results.

 required
Source code in physXAI/models/models.py 
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
@staticmethod
def evaluate(model, td: TrainingDataGeneric):
    """
    Evaluates the trained model on training, validation (if available), and test sets.
    Predictions are stored in the TrainingData object, and metrics are calculated and stored.

    Args:
        model: The trained model instance.
        td (TrainingDataGeneric): The TrainingData object containing datasets and for storing results.
    """

    y_pred_train = model.predict(td.X_train_single)
    y_pred_test = model.predict(td.X_test_single)
    if td.X_val_single is not None:
        y_pred_val = model.predict(td.X_val_single)
    else:
        y_pred_val = None
    td.add_predictions(y_pred_train, y_pred_val, y_pred_test)

    metrics = Metrics(td)
    if isinstance(td, TrainingData):
        td.add_metrics(metrics)
    elif isinstance(td, TrainingDataMultiStep):
        td.add_single_step_metrics(metrics)
        SingleStepModel.evaluate_multi(model, td)
    else:
        raise NotImplementedError
evaluate_multi(model, td: TrainingDataMultiStep) staticmethod
Source code in physXAI/models/models.py 
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
@staticmethod
def evaluate_multi(model, td: TrainingDataMultiStep):
    if td.init_columns[0] != td.output[0]:
        delta_prediction = True
    else:
        delta_prediction = False
    y_pred_train, y_train = SingleStepModel._evaluate_multi_inner_loop(model, td.X_train_features, td.y_train,
                                                                       td.columns, td.init_columns[0],
                                                                       delta_prediction)
    y_pred_test, y_test = SingleStepModel._evaluate_multi_inner_loop(model, td.X_test_features, td.y_test,
                                                                     td.columns, td.init_columns[0],
                                                                     delta_prediction)
    if td.X_val_features is not None:
        y_pred_val, y_val = SingleStepModel._evaluate_multi_inner_loop(model, td.X_val_features, td.y_val,
                                                                       td.columns, td.init_columns[0],
                                                                       delta_prediction)
    else:
        y_pred_val, y_val = None, None
    td.add_predictions(y_pred_train, y_pred_val, y_pred_test)
    td.y_train = y_train
    td.y_val = y_val
    td.y_test = y_test

    metrics = MetricsMultiStep(td)
    td.add_metrics(metrics)
from_config(config: dict) -> SingleStepModel classmethod
Source code in physXAI/models/models.py 
331
332
333
@classmethod
def from_config(cls, config: dict) -> 'SingleStepModel':
    return cls(**config)

LinearRegressionModel

Bases: SingleStepModel

A concrete implementation of SingleStepModel for scikit-learn's Linear Regression.

Source code in physXAI/models/models.py 
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
@register_model
class LinearRegressionModel(SingleStepModel):
    """
    A concrete implementation of SingleStepModel for scikit-learn's Linear Regression.
    """

    def generate_model(self, **kwargs):
        """
        Generates an instance of scikit-learn's LinearRegression model.
        """
        return LinearRegression()

    def fit_model(self, model, td: TrainingDataGeneric):
        """
        Fits the LinearRegression model using the training data from `td`.
        Also records the training time.

        Args:
            model (LinearRegression): The scikit-learn LinearRegression model instance.
            td (TrainingDataGeneric): The TrainingData object.
        """

        start_time = time.perf_counter()
        model.fit(td.X_train_single, td.y_train_single)
        stop_time = time.perf_counter()
        td.add_training_time(stop_time - start_time)

    def compile_model(self, model):
        """
        No compilation step is needed for scikit-learn models.
        """
        pass

    def plot(self, td: TrainingDataGeneric):
        """
        Generates and displays various plots related to model performance.

        Args:
            td (TrainingDataGeneric): The TrainingData object
        """

        fig1 = plot_prediction_correlation(td)
        fig2 = plot_predictions(td)
        fig3 = plot_metrics_table(td)

        if isinstance(td, TrainingData):
            subplots(
                "Linear Regression",
                {"title": "Prediction Correlation", "type": "scatter", "figure": fig1},
                {"title": "Predictions Sorted", "type": "scatter", "figure": fig2},
                {"title": "Performance Metrics", "type": "table", "figure": fig3}
            )
        elif isinstance(td, TrainingDataMultiStep):
            fig4 = plot_multi_rmse(td)
            subplots(
                "Linear Regression",
                # {"title": "Prediction Correlation", "type": "scatter", "figure": fig1},
                {"title": "Predictions Sorted", "type": "scatter", "figure": fig2},
                {"title": "Prediction Step RMSE", "type": "scatter", "figure": fig4},
                {"title": "Performance Metrics", "type": "table", "figure": fig3}
            )
        else:
            raise NotImplementedError

    def save_model(self, model, save_path: str):
        """
        Saves the trained LinearRegression model using joblib.

        Args:
            model (LinearRegression): The trained scikit-learn model.
            save_path (str): The path to save the model.
        """

        if save_path is None:
            save_path = Logger.get_model_savepath()

        if not save_path.endswith('.joblib'):
            save_path += '.joblib'

        save_path = create_full_path(save_path)
        dump(model, save_path)

    def load_model(self, load_path: str):
        """
        Loads a scikit-learn LinearRegression model from a file using joblib.

        Args:
            load_path (str): The path from which to load the model.

        Returns:
            LinearRegression: The loaded scikit-learn model.
        """

        load_path = get_full_path(load_path)
        model = joblib.load(load_path)
        return model
Functions
generate_model(**kwargs)

Generates an instance of scikit-learn's LinearRegression model.

Source code in physXAI/models/models.py 
342
343
344
345
346
def generate_model(self, **kwargs):
    """
    Generates an instance of scikit-learn's LinearRegression model.
    """
    return LinearRegression()
fit_model(model, td: TrainingDataGeneric)

Fits the LinearRegression model using the training data from td. Also records the training time.

Parameters:

Name Type Description Default
model LinearRegression

The scikit-learn LinearRegression model instance.

 required
td TrainingDataGeneric

The TrainingData object.

 required
Source code in physXAI/models/models.py 
348
349
350
351
352
353
354
355
356
357
358
359
360
361
def fit_model(self, model, td: TrainingDataGeneric):
    """
    Fits the LinearRegression model using the training data from `td`.
    Also records the training time.

    Args:
        model (LinearRegression): The scikit-learn LinearRegression model instance.
        td (TrainingDataGeneric): The TrainingData object.
    """

    start_time = time.perf_counter()
    model.fit(td.X_train_single, td.y_train_single)
    stop_time = time.perf_counter()
    td.add_training_time(stop_time - start_time)
compile_model(model)

No compilation step is needed for scikit-learn models.

Source code in physXAI/models/models.py 
363
364
365
366
367
def compile_model(self, model):
    """
    No compilation step is needed for scikit-learn models.
    """
    pass
plot(td: TrainingDataGeneric)

Generates and displays various plots related to model performance.

Parameters:

Name Type Description Default
td TrainingDataGeneric

The TrainingData object

 required
Source code in physXAI/models/models.py 
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
def plot(self, td: TrainingDataGeneric):
    """
    Generates and displays various plots related to model performance.

    Args:
        td (TrainingDataGeneric): The TrainingData object
    """

    fig1 = plot_prediction_correlation(td)
    fig2 = plot_predictions(td)
    fig3 = plot_metrics_table(td)

    if isinstance(td, TrainingData):
        subplots(
            "Linear Regression",
            {"title": "Prediction Correlation", "type": "scatter", "figure": fig1},
            {"title": "Predictions Sorted", "type": "scatter", "figure": fig2},
            {"title": "Performance Metrics", "type": "table", "figure": fig3}
        )
    elif isinstance(td, TrainingDataMultiStep):
        fig4 = plot_multi_rmse(td)
        subplots(
            "Linear Regression",
            # {"title": "Prediction Correlation", "type": "scatter", "figure": fig1},
            {"title": "Predictions Sorted", "type": "scatter", "figure": fig2},
            {"title": "Prediction Step RMSE", "type": "scatter", "figure": fig4},
            {"title": "Performance Metrics", "type": "table", "figure": fig3}
        )
    else:
        raise NotImplementedError
save_model(model, save_path: str)

Saves the trained LinearRegression model using joblib.

Parameters:

Name Type Description Default
model LinearRegression

The trained scikit-learn model.

 required
save_path str

The path to save the model.

 required
Source code in physXAI/models/models.py 
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
def save_model(self, model, save_path: str):
    """
    Saves the trained LinearRegression model using joblib.

    Args:
        model (LinearRegression): The trained scikit-learn model.
        save_path (str): The path to save the model.
    """

    if save_path is None:
        save_path = Logger.get_model_savepath()

    if not save_path.endswith('.joblib'):
        save_path += '.joblib'

    save_path = create_full_path(save_path)
    dump(model, save_path)
load_model(load_path: str)

Loads a scikit-learn LinearRegression model from a file using joblib.

Parameters:

Name Type Description Default
load_path str

The path from which to load the model.

 required

Returns:

Name Type Description
LinearRegression

The loaded scikit-learn model.
Source code in physXAI/models/models.py 
418
419
420
421
422
423
424
425
426
427
428
429
430
431
def load_model(self, load_path: str):
    """
    Loads a scikit-learn LinearRegression model from a file using joblib.

    Args:
        load_path (str): The path from which to load the model.

    Returns:
        LinearRegression: The loaded scikit-learn model.
    """

    load_path = get_full_path(load_path)
    model = joblib.load(load_path)
    return model

MultiStepModel

Bases: AbstractModel, ABC

Abstract Base Class for multi-step prediction models. Defines a common interface and pipeline for models that forecast multiple steps ahead. This class is similar to SingleStepModel but tailored for multi-step data and metrics.

Source code in physXAI/models/models.py 
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
class MultiStepModel(AbstractModel, ABC):
    """
    Abstract Base Class for multi-step prediction models.
    Defines a common interface and pipeline for models that forecast multiple steps ahead.
    This class is similar to SingleStepModel but tailored for multi-step data and metrics.
    """

    def __init__(self, **kwargs):
        pass

    @abstractmethod
    def fit_model(self, model, td: TrainingDataMultiStep):
        """
        Abstract method to fit the model to the training data.

        Args:
            model: The model instance to be trained.
            td (TrainingDataMultiStep): The TrainingData object.
        """
        pass

    @staticmethod
    def evaluate(model, td: TrainingDataMultiStep):
        """
        Evaluates the trained model on training, validation (if available), and test sets.
        Predictions are stored in the TrainingDataMultiStep object, and metrics are calculated and stored.

        Args:
            model: The trained model instance.
            td (TrainingDataMultistep): The TrainingDataMultiStep object containing datasets and for storing results.
        """

        y_pred_train = model.predict(td.X_train)
        if td.X_val is not None:
            y_pred_val = model.predict(td.X_val)
        else:
            y_pred_val = None
        y_pred_test = model.predict(td.X_test)
        td.add_predictions(y_pred_train, y_pred_val, y_pred_test)

        metrics = MetricsMultiStep(td)
        td.add_metrics(metrics)

    @abstractmethod
    def plot(self, td: TrainingDataMultiStep):
        """
         Abstract method for generating and displaying plots related to model performance.

         Args:
             td (TrainingDataMultiStep): The TrainingDataMultiStep object containing true values and predictions.
         """
        pass

    @classmethod
    def from_config(cls, config: dict) -> 'MultiStepModel':
        return cls(**config)
Functions
__init__(**kwargs)
Source code in physXAI/models/models.py 
441
442
def __init__(self, **kwargs):
    pass
fit_model(model, td: TrainingDataMultiStep) abstractmethod

Abstract method to fit the model to the training data.

Parameters:

Name Type Description Default
model

The model instance to be trained.

 required
td TrainingDataMultiStep

The TrainingData object.

 required
Source code in physXAI/models/models.py 
444
445
446
447
448
449
450
451
452
453
@abstractmethod
def fit_model(self, model, td: TrainingDataMultiStep):
    """
    Abstract method to fit the model to the training data.

    Args:
        model: The model instance to be trained.
        td (TrainingDataMultiStep): The TrainingData object.
    """
    pass
evaluate(model, td: TrainingDataMultiStep) staticmethod

Evaluates the trained model on training, validation (if available), and test sets. Predictions are stored in the TrainingDataMultiStep object, and metrics are calculated and stored.

Parameters:

Name Type Description Default
model

The trained model instance.

 required
td TrainingDataMultistep

The TrainingDataMultiStep object containing datasets and for storing results.

 required
Source code in physXAI/models/models.py 
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
@staticmethod
def evaluate(model, td: TrainingDataMultiStep):
    """
    Evaluates the trained model on training, validation (if available), and test sets.
    Predictions are stored in the TrainingDataMultiStep object, and metrics are calculated and stored.

    Args:
        model: The trained model instance.
        td (TrainingDataMultistep): The TrainingDataMultiStep object containing datasets and for storing results.
    """

    y_pred_train = model.predict(td.X_train)
    if td.X_val is not None:
        y_pred_val = model.predict(td.X_val)
    else:
        y_pred_val = None
    y_pred_test = model.predict(td.X_test)
    td.add_predictions(y_pred_train, y_pred_val, y_pred_test)

    metrics = MetricsMultiStep(td)
    td.add_metrics(metrics)
plot(td: TrainingDataMultiStep) abstractmethod

Abstract method for generating and displaying plots related to model performance.

Parameters:

Name Type Description Default
td TrainingDataMultiStep

The TrainingDataMultiStep object containing true values and predictions.

 required
Source code in physXAI/models/models.py 
477
478
479
480
481
482
483
484
485
@abstractmethod
def plot(self, td: TrainingDataMultiStep):
    """
     Abstract method for generating and displaying plots related to model performance.

     Args:
         td (TrainingDataMultiStep): The TrainingDataMultiStep object containing true values and predictions.
     """
    pass
from_config(config: dict) -> MultiStepModel classmethod
Source code in physXAI/models/models.py 
487
488
489
@classmethod
def from_config(cls, config: dict) -> 'MultiStepModel':
    return cls(**config)

Functions

register_model(cls)

A class decorator that registers the decorated class in the MODEL_CLASS_REGISTRY. The class is registered using its name.

Source code in physXAI/models/models.py 
196
197
198
199
200
201
202
203
204
def register_model(cls):  # pragma: no cover
    """
    A class decorator that registers the decorated class in the MODEL_CLASS_REGISTRY.
    The class is registered using its __name__.
    """
    if cls.__name__ in MODEL_CLASS_REGISTRY:
        print(f"Warning: Class '{cls.__name__}' is already registered. Overwriting.")
    MODEL_CLASS_REGISTRY[cls.__name__] = cls
    return cls  # Decorators must return the class (or a replacement)