26 integrate modular anns

executables/bestest_hydronic_heat_pump/P_hp.py

@@ -36,7 +36,7 @@
 # It is recommended to rename features, so that they can be easily added to the input list
 
 # Create Training data
-prep = PreprocessingSingleStep(inputs, output)
+prep = PreprocessingSingleStep(inputs=inputs, output=output)
 # Process Training data
 td = prep.pipeline(file_path)
 

executables/bestest_hydronic_heat_pump/P_hp_FeatureSelection.py

@@ -33,7 +33,7 @@
 
 # Generic Preprocessing Pipeline
 # Model is output model, so single step evaluation is choosen
-prep = PreprocessingSingleStep(inputs, output)
+prep = PreprocessingSingleStep(inputs=inputs, output=output)
 
 # Generic Model
 m = LinearRegressionModel()

executables/bestest_hydronic_heat_pump/P_hp_modular.py

@@ -0,0 +1,41 @@
+from physXAI.models.modular.modular_ann import ModularANN, ModularModel
+from physXAI.preprocessing.preprocessing import PreprocessingSingleStep
+from physXAI.preprocessing.constructed import Feature
+from physXAI.models.ann.ann_design import ClassicalANNModel
+from physXAI.utils.logging import Logger
+
+
+"""
+Creates modular models to predict the power of the heat pump using the Boptest data.
+"""
+
+Logger.setup_logger(folder_name='P_hp_modular', override=True)
+
+file_path = r"data/bestest_hydronic_heat_pump/pid_data.csv"
+
+inputs = ['oveHeaPumY_u', 'Func(logistic)', 'weaSta_reaWeaTDryBul_y', 'reaTZon_y']
+output = 'reaPHeaPum_y'
+
+oveHeaPumY_u = Feature('oveHeaPumY_u')
+func_logistic = Feature('Func(logistic)')
+TDryBul = Feature('weaSta_reaWeaTDryBul_y')
+TZon = Feature('reaTZon_y')
+
+prep = PreprocessingSingleStep(inputs=inputs, output=output)
+td = prep.pipeline(file_path)
+
+"""Example usages of modular models"""
+y = ModularModel(
+    model=ClassicalANNModel(),
+    inputs=[oveHeaPumY_u.input() / func_logistic.input(), func_logistic.input() ** 2, TDryBul.input(), TZon.input()]
+)
+m = ModularANN(architecture=y, rescale_output=True)
+
+# Training pipeline
+model = m.pipeline(td)
+
+
+# Log setup of preprocessing and model as json
+Logger.log_setup(prep, m)
+# Log training data as pickle
+Logger.save_training_data(td)

executables/bestest_hydronic_heat_pump/P_hp_pinn.py

@@ -39,7 +39,7 @@
 pinn.rename('pinn')
 
 # Create Training data
-prep = PreprocessingSingleStep(inputs, output)
+prep = PreprocessingSingleStep(inputs=inputs, output=output)
 # Process Training data
 td = prep.pipeline(file_path)
 

executables/bestest_hydronic_heat_pump/TAir.py

@@ -31,7 +31,7 @@
 x3.lag(2)  # oveHeaPumY_u_lag1, oveHeaPumY_u_lag2
 
 # Create Training data
-prep = PreprocessingSingleStep(inputs, output)
+prep = PreprocessingSingleStep(inputs=inputs, output=output)
 # Process Training data
 td = prep.pipeline(file_path)
 

executables/bestest_hydronic_heat_pump/TAir_FeatureSelection.py

@@ -34,7 +34,7 @@
 # Generic Preprocessing Pipeline
 # Model is state model, so multi-step evaluation is choosen
 # See example TAir_evaluateMultiStep.py for more information
-prep = PreprocessingMultiStep(inputs, output, 48, 0, init_features=['reaTZon_y'],
+prep = PreprocessingMultiStep(inputs=inputs, output=output, label_width=48, warmup_width=0, init_features=['reaTZon_y'],
                               overlapping_sequences=False, batch_size=1)
 
 # Generic Model

executables/bestest_hydronic_heat_pump/TAir_evaluateMultiStep.py

@@ -42,7 +42,7 @@
 overlapping_sequence should be False to avoid duplicate labels for single step prediction
 batch_size should be 1 as batches are processes differently in single step models
 """
-prep = PreprocessingMultiStep(inputs, output, 48, 0, init_features=['reaTZon_y'],
+prep = PreprocessingMultiStep(inputs=inputs, output=output, label_width=48, warmup_width=0, init_features=['reaTZon_y'],
                               overlapping_sequences=False, batch_size=1)
 # Process Training data
 td = prep.pipeline(file_path)

executables/bestest_hydronic_heat_pump/TAir_rnn.py

@@ -29,7 +29,7 @@
 warmup_width = 48
 
 # Create Training data. For RNNs MultiStep training data is required
-prep = PreprocessingMultiStep(inputs, output, label_width, warmup_width, init_features=inits)
+prep = PreprocessingMultiStep(inputs=inputs, output=output, label_width=label_width, warmup_width=warmup_width, init_features=inits)
 # Process Training data
 td = prep.pipeline(file_path)
 

physXAI/__init__.py

@@ -1 +0,0 @@
-

physXAI/models/ann/__init__.py

@@ -1 +0,0 @@
-

physXAI/models/ann/ann_design.py

@@ -1,3 +1,4 @@
+from logging import warning
 import os
 import time
 from abc import ABC, abstractmethod
@@ -48,6 +49,9 @@ def __init__(self, batch_size: int = 32, epochs: int = 1000, learning_rate: floa
         self.random_seed: int = random_seed
         keras.utils.set_random_seed(random_seed)
 
+        self.model_config = dict()
+
+
     @abstractmethod
     def generate_model(self, **kwargs):
         """
@@ -234,12 +238,12 @@ def __init__(self, n_layers: int = 1, n_neurons: Union[int, list[int]] = 32,
         self.activation_function: Union[str, list[str]] = activation_function
         self.rescale_output: bool = rescale_output
 
-        self.model_config = {
+        self.model_config.update({
             "n_layers": self.n_layers,
             "n_neurons": self.n_neurons,
             "activation_function": self.activation_function,
             "rescale_output": self.rescale_output,
-        }
+        })
 
     def generate_model(self, **kwargs):
         """
@@ -290,12 +294,12 @@ def __init__(self, n_layers: int = 1, n_neurons: Union[int, list[int]] = 32, res
         self.n_neurons: Union[int, list[int]] = n_neurons
         self.rescale_output: bool = rescale_output
 
-        self.model_config = {
+        self.model_config.update({
             "n_layers": self.n_layers,
             "n_neurons": self.n_neurons,
             "rescale_output": self.rescale_output,
             "random_state": random_seed
-        }
+        })
 
     def generate_model(self, **kwargs):
         """
@@ -354,12 +358,12 @@ def __init__(self, n_layers: int = 1, n_neurons: Union[int, list[int]] = 32, res
         self.n_neurons: Union[int, list[int]] = n_neurons
         self.rescale_output: bool = rescale_output
 
-        self.model_config = {
+        self.model_config.update({
             "n_layers": self.n_layers,
             "n_neurons": self.n_neurons,
             "rescale_output": self.rescale_output,
             "random_state": random_seed
-        }
+        })
 
     def generate_model(self, **kwargs):
         """
@@ -419,14 +423,14 @@ def __init__(self, n_layers: int = 1, n_neurons: Union[int, list[int]] = 32,
         self.monotonies: dict[str, int] = monotonies
         self.activation_split: list[float] = activation_split
 
-        self.model_config = {
+        self.model_config.update({
             "n_layers": self.n_layers,
             "n_neurons": self.n_neurons,
             "activation_function": self.activation_function,
             "rescale_output": self.rescale_output,
             "monotonicities": self.monotonies,
             "activation_split": activation_split,
-        }
+        })
 
     def generate_model(self, **kwargs):
         """
@@ -494,14 +498,14 @@ def __init__(self, n_layers: int = 1, n_neurons: Union[int, list[int]] = 32,
 
         self.pinn_weights: list[float] = pinn_weights
 
-        self.model_config = {
+        self.model_config.update({
             "n_layers": self.n_layers,
             "n_neurons": self.n_neurons,
             "activation_function": self.activation_function,
             "rescale_output": self.rescale_output,
             "monotonicities": self.monotonies,
             "activation_split": activation_split,
-        }
+        })
 
         # Create pinn loss based on standard losses
         self.pinn_loss = multi_y_loss(keras.losses.MeanSquaredError(name='MSE'), self.pinn_weights, 'mse')
@@ -626,7 +630,6 @@ def get_config(self) -> dict:
         })
         return config
 
-
 @register_model
 class RNNModel(MultiStepModel):
     """

physXAI/models/ann/configs/ann_model_configs.py

@@ -4,10 +4,12 @@
 
 class ClassicalANNConstruction_config(BaseModel):
 
-    n_layers: int = Field(..., gt=0)
+    n_layers: int = Field(..., ge=0)
     n_neurons: Union[int, list[int]] = 32
     activation_function: Union[str, list[str]] = 'softplus'
     rescale_output: bool = True
+    normalize: bool = True
+    n_features: Optional[int] = None
 
     @field_validator('n_neurons')
     def validate_n_neurons(cls, v, info):
@@ -33,6 +35,7 @@ def validate_activation(cls, v, info):
 
 class RBFConstruction_config(ClassicalANNConstruction_config):
 
+    n_layers: int = Field(..., ge=1, le=1)
     random_state: int = 42
     rescale_mean: Optional[float] = Field(
         None, description="Mean value for z-score normalization of outputs"

physXAI/models/ann/keras_models/__init__.py

@@ -1 +0,0 @@
-

physXAI/models/ann/model_construction/ann_models.py

@@ -1,5 +1,4 @@
 import os
-
 import numpy as np
 from physXAI.preprocessing.training_data import TrainingDataGeneric
 from physXAI.models.ann.configs.ann_model_configs import (ClassicalANNConstruction_config,
@@ -35,28 +34,28 @@ def ClassicalANNConstruction(config: dict, td: TrainingDataGeneric):
         n_neurons = [n_neurons] * n_layers
     else:
         assert len(n_neurons) == n_layers
-    n_featues = td.X_train_single.shape[1]
+    if config['n_features'] is not None:
+        n_features = config['n_features']
+    else:
+        n_features = td.X_train_single.shape[1]
     activation_function = config['activation_function']
     # If activation_function is a single string, replicate it for all layers
     if isinstance(activation_function, str):
         activation_function = [activation_function] * n_layers
     else:
         assert len(activation_function) == n_layers
 
-    # Rescaling for output layer
-    rescale_mean = float(np.mean(td.y_train_single))
-    rescale_sigma = float(np.std(td.y_train_single, ddof=1))
-
     # Build artificial neural network as Sequential
     model = keras.Sequential()
 
     # Add input layer
-    model.add(keras.layers.Input(shape=(n_featues,)))
+    model.add(keras.layers.Input(shape=(n_features,)))
 
     # Add normalization layer
-    normalization = keras.layers.Normalization()
-    normalization.adapt(td.X_train_single)
-    model.add(normalization)
+    if config['normalize']:
+        normalization = keras.layers.Normalization()
+        normalization.adapt(td.X_train_single)
+        model.add(normalization)
 
     for i in range(0, n_layers):
         # For each layer add dense
@@ -65,6 +64,9 @@ def ClassicalANNConstruction(config: dict, td: TrainingDataGeneric):
     model.add(keras.layers.Dense(1, activation='linear'))
     # Add rescaling
     if config['rescale_output']:
+        # Rescaling for output layer
+        rescale_mean = float(np.mean(td.y_train_single))
+        rescale_sigma = float(np.std(td.y_train_single, ddof=1))
         model.add(keras.layers.Rescaling(scale=rescale_sigma, offset=rescale_mean))
 
     model.summary()
@@ -98,7 +100,10 @@ def CMNNModelConstruction(config: dict, td: TrainingDataGeneric):
         n_neurons = [n_neurons] * n_layers
     else:
         assert len(n_neurons) == n_layers
-    n_featues = td.X_train_single.shape[1]
+    if config['n_features'] is not None:
+        n_features = config['n_features']
+    else:
+        n_features = td.X_train_single.shape[1]
     activation_function = config['activation_function']
     # If activation_function is a single string, replicate it for all layers
     if isinstance(activation_function, str):
@@ -109,21 +114,20 @@ def CMNNModelConstruction(config: dict, td: TrainingDataGeneric):
     # Get monotonicity constraints
     mono = config['monotonicities']
     if mono is None:
-        monotonicities = [0] * n_featues
+        monotonicities = [0] * n_features
     else:
         monotonicities = [0 if name not in mono.keys() else mono[name] for name in td.columns]
 
-    # Rescaling for output layer
-    rescale_mean = float(np.mean(td.y_train_single))
-    rescale_sigma = float(np.std(td.y_train_single, ddof=1))
-
     # Add input layer
-    input_layer = keras.layers.Input(shape=(n_featues,))
+    input_layer = keras.layers.Input(shape=(n_features,))
 
     # Add normalization layer
-    normalization = keras.layers.Normalization()
-    normalization.adapt(td.X_train_single)
-    x = normalization(input_layer)
+    if config['normalize']:
+        normalization = keras.layers.Normalization()
+        normalization.adapt(td.X_train_single)
+        x = normalization(input_layer)
+    else:
+        x = input_layer
 
     # Add dense layer
     activation_split = config['activation_split']
@@ -169,6 +173,9 @@ def CMNNModelConstruction(config: dict, td: TrainingDataGeneric):
 
     # Add rescaling
     if config['rescale_output']:
+        # Rescaling for output layer
+        rescale_mean = float(np.mean(td.y_train_single))
+        rescale_sigma = float(np.std(td.y_train_single, ddof=1))
         x = keras.layers.Rescaling(scale=rescale_sigma, offset=rescale_mean)(x)
 
     # # Add min / max constraints