01_Experiment_Fine_Tuning_CNN.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Fine-Tuning of CNN models\n",
    "\n",
    "In order to transfer learning from a pre-trained models available in [Keras](https://keras.io/api/applications/) to solve specific problems, this project shows how to apply a famous machine learning technique called **fine-tuning**.\n",
    "\n",
    "**Problem: [Domestic Garbage Detection](https://www.kaggle.com/datasets/farzadnekouei/trash-type-image-dataset/)**\n",
    "\n",
    "## Steps\n",
    "* Import the dataset;\n",
    "* Exploration and preprocessing\n",
    "* Compare 3 (or more) pre-trained models\n",
    "* Evaluating the best model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Importig libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import tensorflow as tf\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "import keras_tuner as kt\n",
    "import prettytable\n",
    "import seaborn as sns\n",
    "\n",
    "from sklearn.metrics import confusion_matrix, accuracy_score, f1_score, precision_score, recall_score, roc_auc_score, log_loss\n",
    "from datetime import datetime"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Defining general constants"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "BATCH_SIZE=32\n",
    "IMAGE_SIZE=(384, 512)\n",
    "EPOCHS=200\n",
    "SEED=10 \n",
    "\n",
    "CURRENT_PATH = os.getcwd()\n",
    "DATASET_PATH = os.path.join(CURRENT_PATH, 'datasets', 'trash_type_dataset')\n",
    "# DATASET_PATH = os.path.join(CURRENT_PATH, 'datasets', 'dev')\n",
    "TRAIN_PATH = os.path.join(DATASET_PATH, 'train')\n",
    "VALIDATION_PATH = os.path.join(DATASET_PATH, 'validation')\n",
    "TEST_PATH = os.path.join(DATASET_PATH, 'test')\n",
    "\n",
    "now_is = datetime.now().strftime('%d_%m_%YT%H_%M_%S')\n",
    "EXPERIMENT_NAME = f\"experiment_{now_is}\"\n",
    "BEST_MODEL_PATH = os.path.join(CURRENT_PATH, 'models', EXPERIMENT_NAME)\n",
    "\n",
    "print(f\"Executing the experiment '{EXPERIMENT_NAME}'\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Importing dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def import_dataset(dataset_path: str, batch_size: int = None):\n",
    "    dataset = tf.keras.utils.image_dataset_from_directory(dataset_path,\n",
    "                                                        labels='inferred',\n",
    "                                                        label_mode='int',\n",
    "                                                        batch_size=batch_size,\n",
    "                                                        image_size=IMAGE_SIZE)\n",
    "    return dataset\n",
    "\n",
    "def import_datasets():\n",
    "    print(f\"[INFO] Importing training dataset from {TRAIN_PATH}\")\n",
    "    train_dataset = import_dataset(TRAIN_PATH, BATCH_SIZE)\n",
    "\n",
    "    print(f\"[INFO] Importing validation dataset from {VALIDATION_PATH}\")\n",
    "    validation_dataset = import_dataset(VALIDATION_PATH, BATCH_SIZE)\n",
    "\n",
    "    print(f\"[INFO] Importing test dataset from {TEST_PATH}\")\n",
    "    test_dataset = import_dataset(TEST_PATH, batch_size=None)\n",
    "\n",
    "    return train_dataset, validation_dataset, test_dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_dataset, validation_dataset, test_dataset = import_datasets()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exploring the dataset"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "What are the classes?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "classes = train_dataset.class_names\n",
    "print(f\"The classes are {classes}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Preview of the samples"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_sample_of_dataset(dataset):\n",
    "    image, label = next(iter(dataset))\n",
    "    return image[0].numpy().astype(np.uint8), classes[label[0].numpy()]\n",
    "\n",
    "def plot_n_samples(dataset, n: int, ncols = 4):\n",
    "    samples = []\n",
    "    nrows = (n // ncols) + 1\n",
    "    figure, axis = plt.subplots(ncols=ncols, nrows=nrows, figsize=(12,8))\n",
    "\n",
    "    figure.suptitle(\"Samples - Garbage Image Classification\")\n",
    "\n",
    "    for _ in range(len(samples)):\n",
    "        samples.append(get_sample_of_dataset(dataset))\n",
    "\n",
    "    for ax in axis.reshape(-1):\n",
    "        image, label = get_sample_of_dataset(dataset)\n",
    "        ax.set_title(f\"[class={label}]\")\n",
    "        ax.imshow(image)\n",
    "        ax.axis(\"off\")\n",
    "\n",
    "    plt.subplots_adjust(wspace=0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plot_n_samples(train_dataset, 10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Preparing experiments\n",
    "For this project, some [available models](https://keras.io/api/applications/#available-models) were chosen to be compared using [Keras](https://keras.io/) framework. Below, it is described each used model:\n",
    "\n",
    "|Model|Size (MB)|Top-1 Accuracy (ImageNet)|\n",
    "|---|---|---|\n",
    "|EfficientNetB7|256|84.3%|\n",
    "|EfficientNetV2S|88|83.9%|\n",
    "|ConvNeXtXLarge|1310|86.7%|\n",
    "\n",
    "Before start the experiments, let's define some configurations and functions in order to re-use funcionalities."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class FineTuningModel:\n",
    "    def __init__(self, \n",
    "                 title: str, \n",
    "                 base_model, \n",
    "                 input_shape = IMAGE_SIZE, \n",
    "                 batch_size = BATCH_SIZE):\n",
    "        self.title = title\n",
    "        self.base_model = base_model\n",
    "        self.input_shape = input_shape\n",
    "        self.batch_size = batch_size\n",
    "        self.best_model = None\n",
    "        self.history = None\n",
    "        self._freeze_base_model()\n",
    "\n",
    "    def _freeze_base_model(self):\n",
    "        self.base_model.trainable = False\n",
    "\n",
    "    def _build(self, hp: kt.HyperParameters):\n",
    "        dense_units = hp.Choice('dense_units', values=[256, 128, 64])\n",
    "        \n",
    "        # Conv Layers using a pre-trained model\n",
    "        inputs = tf.keras.Input(shape=(*self.input_shape, 3))\n",
    "        x = self.base_model(inputs, training=False)\n",
    "        \n",
    "        # Dense layers\n",
    "        x = tf.keras.layers.Dropout(0.2)(x)\n",
    "        x = tf.keras.layers.GlobalAveragePooling2D()(x)\n",
    "        x = tf.keras.layers.Dense(dense_units, activation='relu')(x)\n",
    "        outputs = tf.keras.layers.Dense(len(classes), activation='softmax')(x)\n",
    "        \n",
    "        # Compiling the model\n",
    "        model = tf.keras.Model(inputs, outputs)\n",
    "        model = self._compile(model, hp)\n",
    "        return model\n",
    "\n",
    "    def _compile(self, model: tf.keras.Model, hp: kt.HyperParameters):\n",
    "        learning_rate = hp.Choice('learning_rate', values=[0.01, 0.001])\n",
    "        model.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=learning_rate),\n",
    "                      loss='sparse_categorical_crossentropy',\n",
    "                      metrics=[tf.keras.metrics.SparseCategoricalAccuracy(name='accuracy')])\n",
    "        return model\n",
    "\n",
    "    def _get_best_model(self, train, validation, best_hps: list, callbacks: list):\n",
    "        best_hp = best_hps[0]\n",
    "        model = self._build(best_hp)\n",
    "        self.history = model.fit(train, \n",
    "                  validation_data=validation, \n",
    "                  batch_size=BATCH_SIZE,\n",
    "                  epochs=EPOCHS, \n",
    "                  callbacks=callbacks,\n",
    "                  verbose=0)\n",
    "        return model\n",
    "\n",
    "    def train(self, train: tf.data.Dataset, validation: tf.data.Dataset):\n",
    "        print(f\"[INFO] Running experiment [{self.title}]\")\n",
    "        callbacks = [tf.keras.callbacks.EarlyStopping(monitor='val_loss', patience=10, restore_best_weights=True)]\n",
    "        tuner = kt.Hyperband(hypermodel=self._build,\n",
    "                            objective=kt.Objective('val_accuracy', direction='max'),\n",
    "                            overwrite=True,\n",
    "                            directory='checkpoints',\n",
    "                            project_name=self.title,\n",
    "                            max_epochs=EPOCHS)\n",
    "        \n",
    "        tuner.search(train, \n",
    "                     batch_size=BATCH_SIZE, \n",
    "                     validation_data=validation,\n",
    "                     callbacks=callbacks)\n",
    "        \n",
    "        self.best_model = self._get_best_model(train,\n",
    "                                                validation,\n",
    "                                                tuner.get_best_hyperparameters(5),\n",
    "                                                callbacks)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Defining the experiments"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "input_shape = (*IMAGE_SIZE, 3)\n",
    "experiments = [\n",
    "    FineTuningModel('EfficientNetB7', \n",
    "                    tf.keras.applications.EfficientNetB7(include_top=False, input_shape=input_shape)),\n",
    "    # FineTuningModel('EfficientNetV2S', \n",
    "    #                 tf.keras.applications.EfficientNetV2S(include_top=False, input_shape=input_shape))\n",
    "]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Optimizing the datasets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_dataset = train_dataset.prefetch(tf.data.AUTOTUNE).cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Training"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for experiment in experiments:\n",
    "    experiment.train(train=train_dataset, validation=validation_dataset)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Evaluating the best experiment"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Comparing the training history"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "number_of_experiments = len(experiments)\n",
    "\n",
    "figure, axis = plt.subplots(nrows=number_of_experiments, ncols=2, squeeze=False)\n",
    "figure.set_figwidth(12)\n",
    "figure.set_figheight(number_of_experiments*4)\n",
    "\n",
    "for index in range(number_of_experiments):\n",
    "    experiment = experiments[index]\n",
    "    history = experiment.history.history\n",
    "    title = experiment.title\n",
    "    \n",
    "    # Plotting loss\n",
    "    axis[index][0].set_title(f'{title} [loss]')\n",
    "    axis[index][0].plot(history['loss'], label=\"train\")\n",
    "    axis[index][0].plot(history['val_loss'], label=\"validation\")\n",
    "    axis[index][0].legend()\n",
    "    \n",
    "    # Plotting accuracy\n",
    "    axis[index][1].set_title(f'{title} [accuracy]')\n",
    "    axis[index][1].plot(history['accuracy'], label=\"train\")\n",
    "    axis[index][1].plot(history['val_accuracy'], label=\"validation\")\n",
    "    axis[index][1].legend()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Evaluating the best experiment based on the testing partition. First of all, let's create the truth array with the labels of the testing dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "number_os_tests = len(test_dataset)\n",
    "y_true = np.ndarray(shape=(number_os_tests,), dtype=np.int8)\n",
    "index = 0\n",
    "for _, y in test_dataset:\n",
    "    y = y.numpy()\n",
    "    y_true[index] = y\n",
    "    index += 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Predicting..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_dataset_as_batch = test_dataset.batch(1)\n",
    "probabilities = np.ndarray(shape=(number_of_experiments, number_os_tests, len(classes)), dtype=np.float32)\n",
    "predictions = np.ndarray(shape=(number_of_experiments, number_os_tests, ), dtype=np.int8)\n",
    "\n",
    "index = 0\n",
    "for experiment in experiments:\n",
    "    best_model = experiment.best_model\n",
    "    \n",
    "    # Making predictions\n",
    "    experiment_probabilities = best_model.predict(test_dataset_as_batch, verbose=0)\n",
    "    experiment_predictions = [np.argmax(probability) for probability in experiment_probabilities]\n",
    "    \n",
    "    # Storing the predictions \n",
    "    probabilities[index] = experiment_probabilities\n",
    "    predictions[index] = experiment_predictions\n",
    "    \n",
    "    index += 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Plotting the metrics of each experiment's result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def format_float(value: float, percent=True):\n",
    "    if percent:\n",
    "        return f'{value*100:.2f}%'\n",
    "    return f'{value:.2f}'\n",
    "\n",
    "index = 0\n",
    "table = prettytable.PrettyTable()\n",
    "table.title = 'Fine-Tuning pre-trained models -- Results'\n",
    "table.field_names = ['Experiment', 'Accuracy', 'F1-Score', 'Precision', 'Recall', 'ROC AUC', 'Log loss']\n",
    "\n",
    "for experiment in experiments:\n",
    "    title = experiment.title\n",
    "    y_pred = predictions[index]\n",
    "    y_probabilities = probabilities[index]\n",
    "\n",
    "    # Calculating the metrics\n",
    "    accuracy = accuracy_score(y_true=y_true, y_pred=y_pred)\n",
    "    f1 = f1_score(y_true=y_true, y_pred=y_pred, average='macro', zero_division=0)\n",
    "    precision = precision_score(y_true=y_true, y_pred=y_pred, average='macro', zero_division=0)\n",
    "    recall = recall_score(y_true=y_true, y_pred=y_pred, average='macro')\n",
    "    logl = log_loss(y_true=y_true, y_pred=y_probabilities)\n",
    "    rocauc = roc_auc_score(y_true=y_true, y_score=y_probabilities, multi_class='ovr')\n",
    "    \n",
    "    # Adding the metrics to table\n",
    "    table.add_row([title, \n",
    "                   format_float(accuracy), \n",
    "                   format_float(f1), \n",
    "                   format_float(precision), \n",
    "                   format_float(recall),\n",
    "                   format_float(logl, percent=False),\n",
    "                   format_float(rocauc, percent=False)])\n",
    "\n",
    "    index += 1\n",
    "\n",
    "print(table)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Plotting the confusion matrix of each experiment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pallet = sns.color_palette(\"ch:s=.25,rot=-.25\", as_cmap=True)\n",
    "\n",
    "nrows = (number_of_experiments // 2)\n",
    "if number_of_experiments % 2 != 0:\n",
    "    nrows += 1 \n",
    "\n",
    "figure, axis = plt.subplots(nrows=nrows, ncols=2, squeeze=False)\n",
    "figure.suptitle('Confusion matrix')\n",
    "figure.set_figwidth(12)\n",
    "figure.set_figheight(nrows * 4)\n",
    "figure.tight_layout(w_pad = 1, h_pad=4)\n",
    "\n",
    "index_col = 0\n",
    "for index in range(number_of_experiments):\n",
    "    index_row = index // 2\n",
    "    experiment = experiments[index]\n",
    "    y_pred = predictions[index]\n",
    "    title = experiment.title\n",
    "\n",
    "    # Calculating the confusion matrix\n",
    "    cm = confusion_matrix(y_true=y_true, y_pred=y_pred)\n",
    "    \n",
    "    # Plotting confusion matrix\n",
    "    axis[index_row, index_col].set_title(title)\n",
    "    sns.heatmap(ax=axis[index_row, index_col], \n",
    "                data=cm, \n",
    "                xticklabels=classes, \n",
    "                yticklabels=classes, \n",
    "                fmt='.0f',\n",
    "                cmap=pallet,\n",
    "                annot=True)\n",
    "    \n",
    "    index_col = 1 if index_col == 0 else 0\n",
    "\n",
    "if number_of_experiments % 2 != 0:\n",
    "    axis[-1][1].axis('off')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Saving each trained model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for experiment in experiments:\n",
    "    title = experiment.title\n",
    "    print(f\"[INFO] Saving {title}\")\n",
    "    best_model = experiment.best_model\n",
    "    model_name = f\"Model_{title}.keras\"\n",
    "    model_path = os.path.join(BEST_MODEL_PATH, model_name)\n",
    "    best_model.save(model_path, overwrite=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Conclusion"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "ml",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.18"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}