develop (#47)

* Add freezing methods to fine tuning

* Add set_encoder_decoder method

* Add transfer learning for the AutoClassifier.ipynb for transfer learning functionality
* Update tools.py

* [FIX] InputLayer error when saving and loading the model

* Update transfer learning for the AutoClassifier.ipynb

* Update transfer learning for the AutoClassifier.ipynb

.gitignore

@@ -129,7 +129,7 @@ dmypy.json
 *.sql
 /*.html
 
-*best_model*
+*_model*
 *my_dir*
 # Ignore <files>.txt for testing
 /examples/*.txt

likelihood/VERSION

@@ -1 +1 @@
-1.3.0
+1.3.1

likelihood/models/deep/autoencoders.py

@@ -19,8 +19,10 @@
 
 import keras_tuner
 import tensorflow as tf
+from keras.src.engine.input_layer import InputLayer
 from pandas.core.frame import DataFrame
 from sklearn.manifold import TSNE
+from tensorflow.keras.regularizers import l2
 
 from likelihood.tools import OneHotEncoder
 
@@ -81,6 +83,8 @@ def __init__(self, input_shape_parm, num_classes, units, activation, **kwargs):
             The number of hidden layers in the classifier. Default is 1.
         dropout : `float`
             The dropout rate to use in the classifier. Default is None.
+        l2_reg : `float`
+            The L2 regularization parameter. Default is 0.0.
         """
         super(AutoClassifier, self).__init__()
         self.input_shape_parm = input_shape_parm
@@ -94,32 +98,68 @@ def __init__(self, input_shape_parm, num_classes, units, activation, **kwargs):
         self.classifier_activation = kwargs.get("classifier_activation", "softmax")
         self.num_layers = kwargs.get("num_layers", 1)
         self.dropout = kwargs.get("dropout", None)
+        self.l2_reg = kwargs.get("l2_reg", 0.0)
 
     def build(self, input_shape):
-        self.encoder = tf.keras.Sequential(
-            [
-                tf.keras.layers.Dense(units=self.units, activation=self.activation),
-                tf.keras.layers.Dense(units=int(self.units / 2), activation=self.activation),
-            ]
+        # Encoder with L2 regularization
+        self.encoder = (
+            tf.keras.Sequential(
+                [
+                    tf.keras.layers.Dense(
+                        units=self.units,
+                        activation=self.activation,
+                        kernel_regularizer=l2(self.l2_reg),
+                    ),
+                    tf.keras.layers.Dense(
+                        units=int(self.units / 2),
+                        activation=self.activation,
+                        kernel_regularizer=l2(self.l2_reg),
+                    ),
+                ]
+            )
+            if not self.encoder
+            else self.encoder
         )
 
-        self.decoder = tf.keras.Sequential(
-            [
-                tf.keras.layers.Dense(units=self.units, activation=self.activation),
-                tf.keras.layers.Dense(units=self.input_shape_parm, activation=self.activation),
-            ]
+        # Decoder with L2 regularization
+        self.decoder = (
+            tf.keras.Sequential(
+                [
+                    tf.keras.layers.Dense(
+                        units=self.units,
+                        activation=self.activation,
+                        kernel_regularizer=l2(self.l2_reg),
+                    ),
+                    tf.keras.layers.Dense(
+                        units=self.input_shape_parm,
+                        activation=self.activation,
+                        kernel_regularizer=l2(self.l2_reg),
+                    ),
+                ]
+            )
+            if not self.decoder
+            else self.decoder
         )
 
+        # Classifier with L2 regularization
         self.classifier = tf.keras.Sequential()
         if self.num_layers > 1:
             for _ in range(self.num_layers - 1):
                 self.classifier.add(
-                    tf.keras.layers.Dense(units=self.units, activation=self.activation)
+                    tf.keras.layers.Dense(
+                        units=self.units,
+                        activation=self.activation,
+                        kernel_regularizer=l2(self.l2_reg),
+                    )
                 )
                 if self.dropout:
                     self.classifier.add(tf.keras.layers.Dropout(self.dropout))
         self.classifier.add(
-            tf.keras.layers.Dense(units=self.num_classes, activation=self.classifier_activation)
+            tf.keras.layers.Dense(
+                units=self.num_classes,
+                activation=self.classifier_activation,
+                kernel_regularizer=l2(self.l2_reg),
+            )
         )
 
     def call(self, x):
@@ -129,6 +169,84 @@ def call(self, x):
         classification = self.classifier(combined)
         return classification
 
+    def freeze_encoder_decoder(self):
+        """
+        Freezes the encoder and decoder layers to prevent them from being updated during training.
+        """
+        for layer in self.encoder.layers:
+            layer.trainable = False
+        for layer in self.decoder.layers:
+            layer.trainable = False
+
+    def unfreeze_encoder_decoder(self):
+        """
+        Unfreezes the encoder and decoder layers allowing them to be updated during training.
+        """
+        for layer in self.encoder.layers:
+            layer.trainable = True
+        for layer in self.decoder.layers:
+            layer.trainable = True
+
+    def set_encoder_decoder(self, source_model):
+        """
+        Sets the encoder and decoder layers from another AutoClassifier instance,
+        ensuring compatibility in dimensions.
+
+        Parameters:
+        -----------
+        source_model : AutoClassifier
+            The source model to copy the encoder and decoder layers from.
+
+        Raises:
+        -------
+        ValueError
+            If the input shape or units of the source model do not match.
+        """
+        if not isinstance(source_model, AutoClassifier):
+            raise ValueError("Source model must be an instance of AutoClassifier.")
+
+        # Check compatibility in input shape and units
+        if self.input_shape_parm != source_model.input_shape_parm:
+            raise ValueError(
+                f"Incompatible input shape. Expected {self.input_shape_parm}, got {source_model.input_shape_parm}."
+            )
+        if self.units != source_model.units:
+            raise ValueError(
+                f"Incompatible number of units. Expected {self.units}, got {source_model.units}."
+            )
+        self.encoder, self.decoder = tf.keras.Sequential(), tf.keras.Sequential()
+        # Copy the encoder layers
+        for i, layer in enumerate(source_model.encoder.layers):
+            if isinstance(layer, tf.keras.layers.Dense):  # Make sure it's a Dense layer
+                dummy_input = tf.convert_to_tensor(tf.random.normal([1, layer.input_shape[1]]))
+                dense_layer = tf.keras.layers.Dense(
+                    units=layer.units,
+                    activation=self.activation,
+                    kernel_regularizer=l2(self.l2_reg),
+                )
+                dense_layer.build(dummy_input.shape)
+                self.encoder.add(dense_layer)
+                # Set the weights correctly
+                self.encoder.layers[i].set_weights(layer.get_weights())
+            elif not isinstance(layer, InputLayer):
+                raise ValueError(f"Layer type {type(layer)} not supported for copying.")
+
+        # Copy the decoder layers
+        for i, layer in enumerate(source_model.decoder.layers):
+            if isinstance(layer, tf.keras.layers.Dense):  # Ensure it's a Dense layer
+                dummy_input = tf.convert_to_tensor(tf.random.normal([1, layer.input_shape[1]]))
+                dense_layer = tf.keras.layers.Dense(
+                    units=layer.units,
+                    activation=self.activation,
+                    kernel_regularizer=l2(self.l2_reg),
+                )
+                dense_layer.build(dummy_input.shape)
+                self.decoder.add(dense_layer)
+                # Set the weights correctly
+                self.decoder.layers[i].set_weights(layer.get_weights())
+            elif not isinstance(layer, InputLayer):
+                raise ValueError(f"Layer type {type(layer)} not supported for copying.")
+
     def get_config(self):
         config = {
             "input_shape_parm": self.input_shape_parm,
@@ -138,6 +256,7 @@ def get_config(self):
             "classifier_activation": self.classifier_activation,
             "num_layers": self.num_layers,
             "dropout": self.dropout,
+            "l2_reg": self.l2_reg,
         }
         base_config = super(AutoClassifier, self).get_config()
         return dict(list(base_config.items()) + list(config.items()))
@@ -152,6 +271,7 @@ def from_config(cls, config):
             classifier_activation=config["classifier_activation"],
             num_layers=config["num_layers"],
             dropout=config["dropout"],
+            l2_reg=config["l2_reg"],
         )
 
 
@@ -189,13 +309,15 @@ def call_existing_code(
         The AutoClassifier instance.
     """
     dropout = kwargs.get("dropout", None)
+    l2_reg = kwargs.get("l2_reg", 0.0)
     model = AutoClassifier(
         input_shape_parm=input_shape_parm,
         num_classes=num_classes,
         units=units,
         activation=activation,
         num_layers=num_layers,
         dropout=dropout,
+        l2_reg=l2_reg,
     )
     model.compile(
         optimizer=optimizer,
@@ -242,32 +364,65 @@ def build_model(
             step=2,
         )
         if "units" not in hyperparameters_keys
-        else hyperparameters["units"]
+        else (
+            hp.Choice("units", hyperparameters["units"])
+            if isinstance(hyperparameters["units"], list)
+            else hyperparameters["units"]
+        )
     )
     activation = (
         hp.Choice("activation", ["sigmoid", "relu", "tanh", "selu", "softplus", "softsign"])
         if "activation" not in hyperparameters_keys
-        else hyperparameters["activation"]
+        else (
+            hp.Choice("activation", hyperparameters["activation"])
+            if isinstance(hyperparameters["activation"], list)
+            else hyperparameters["activation"]
+        )
     )
     optimizer = (
         hp.Choice("optimizer", ["sgd", "adam", "adadelta", "rmsprop", "adamax", "adagrad"])
         if "optimizer" not in hyperparameters_keys
-        else hyperparameters["optimizer"]
+        else (
+            hp.Choice("optimizer", hyperparameters["optimizer"])
+            if isinstance(hyperparameters["optimizer"], list)
+            else hyperparameters["optimizer"]
+        )
     )
     threshold = (
         hp.Float("threshold", min_value=0.1, max_value=0.9, sampling="log")
         if "threshold" not in hyperparameters_keys
-        else hyperparameters["threshold"]
+        else (
+            hp.Choice("threshold", hyperparameters["threshold"])
+            if isinstance(hyperparameters["threshold"], list)
+            else hyperparameters["threshold"]
+        )
     )
     num_layers = (
         hp.Int("num_layers", min_value=1, max_value=10, step=1)
         if "num_layers" not in hyperparameters_keys
-        else hyperparameters["num_layers"]
+        else (
+            hp.Choice("num_layers", hyperparameters["num_layers"])
+            if isinstance(hyperparameters["num_layers"], list)
+            else hyperparameters["num_layers"]
+        )
     )
     dropout = (
         hp.Float("dropout", min_value=0.1, max_value=0.9, sampling="log")
         if "dropout" not in hyperparameters_keys
-        else hyperparameters["dropout"]
+        else (
+            hp.Choice("dropout", hyperparameters["dropout"])
+            if isinstance(hyperparameters["dropout"], list)
+            else hyperparameters["dropout"]
+        )
+    )
+    l2_reg = (
+        hp.Float("l2_reg", min_value=1e-6, max_value=0.1, sampling="log")
+        if "l2_reg" not in hyperparameters_keys
+        else (
+            hp.Choice("l2_reg", hyperparameters["l2_reg"])
+            if isinstance(hyperparameters["l2_reg"], list)
+            else hyperparameters["l2_reg"]
+        )
     )
 
     model = call_existing_code(
@@ -279,6 +434,7 @@ def build_model(
         num_classes=num_classes,
         num_layers=num_layers,
         dropout=dropout,
+        l2_reg=l2_reg,
     )
     return model
 
@@ -477,11 +633,31 @@ def predictor_analyzer(
         )
         self.data["class"] = self.classification
         self.data_input["class"] = self.classification
-        radviz(self.data, "class", color=self.colors)
+
+        self.data_normalized = self.data.copy(deep=True)
+        self.data_normalized.iloc[:, :-1] = (
+            2.0
+            * (self.data_normalized.iloc[:, :-1] - self.data_normalized.iloc[:, :-1].min())
+            / (self.data_normalized.iloc[:, :-1].max() - self.data_normalized.iloc[:, :-1].min())
+            - 1
+        )
+        radviz(self.data_normalized, "class", color=self.colors)
         plt.title("Radviz Visualization of Latent Space")
         plt.show()
-
-        radviz(self.data_input, "class", color=self.colors)
+        self.data_input_normalized = self.data_input.copy(deep=True)
+        self.data_input_normalized.iloc[:, :-1] = (
+            2.0
+            * (
+                self.data_input_normalized.iloc[:, :-1]
+                - self.data_input_normalized.iloc[:, :-1].min()
+            )
+            / (
+                self.data_input_normalized.iloc[:, :-1].max()
+                - self.data_input_normalized.iloc[:, :-1].min()
+            )
+            - 1
+        )
+        radviz(self.data_input_normalized, "class", color=self.colors)
         plt.title("Radviz Visualization of Input Data")
         plt.show()
         return self._statistics(self.data_input)

likelihood/tools/tools.py

@@ -979,6 +979,7 @@ def __init__(self) -> None:
     def f_mean(self, y_true: np.ndarray, y_pred: np.ndarray, labels: List[int]) -> float:
         F_vec = self._f1_score(y_true, y_pred, labels)
         mean_f_measure = np.mean(F_vec)
+        mean_f_measure = np.around(mean_f_measure, decimals=4)
 
         for label, f_measure in zip(labels, F_vec):
             print(f"F-measure of label {label} -> {f_measure}")
@@ -1005,9 +1006,9 @@ def resp(self, y_true: np.ndarray, y_pred: np.ndarray, labels: List[int]) -> flo
 
     def _summary_pred(self, y_true: np.ndarray, y_pred: np.ndarray, labels: List[int]) -> None:
         count_mat = self._confu_mat(y_true, y_pred, labels)
-        print("       ", " | ".join(f"--{label}--" for label in labels))
+        print(" " * 6, " | ".join(f"--{label}--" for label in labels))
         for i, label_i in enumerate(labels):
-            row = [f"  {int(count_mat[i, j])}  " for j in range(len(labels))]
+            row = [f" {int(count_mat[i, j]):5d} " for j in range(len(labels))]
             print(f"--{label_i}--|", " | ".join(row))
 
     def _f1_score(self, y_true: np.ndarray, y_pred: np.ndarray, labels: List[int]) -> np.ndarray:
@@ -1023,6 +1024,7 @@ def _f1_score(self, y_true: np.ndarray, y_pred: np.ndarray, labels: List[int]) -
             count_mat.diagonal(), sum_rows, out=np.zeros_like(sum_rows), where=sum_rows != 0
         )
         f1_vec = 2 * ((precision * recall) / (precision + recall))
+        f1_vec = np.around(f1_vec, decimals=4)
 
         return f1_vec
 

requirements.txt

@@ -1,12 +1,14 @@
 black[jupyter]>=24.3.0
-mypy-extensions==1.0.0
-types-openpyxl==3.1.0.15
-pydocstyle==6.3.0
-flake8==6.0.0
-isort==5.12.0
-mypy==1.4.1
+mypy-extensions>=1.0.0
+types-openpyxl>=3.1.0.15
+pydocstyle>=6.3.0
+flake8>=6.0.0
+isort>=5.12.0
+mypy>=1.4.1
 numpy<2.0.0
+pydot==2.0.0
 matplotlib
+graphviz
 pyyaml
 pandas 
 corner