Transfer Learning with Deep CNNs

# Importing Libraries
import os
import numpy as np
import matplotlib.pyplot as plt
import tensorflow as tf
import seaborn as sns
import json
from sklearn.metrics import classification_report
from tensorflow.keras.applications.mobilenet_v2 import preprocess_input
from tensorflow.keras.layers import GlobalAveragePooling2D, Dense, Dropout
from tensorflow.keras.models import Model
from tensorflow.keras.callbacks import EarlyStopping, ModelCheckpoint
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.applications import MobileNetV2

print("All the required libraries are imported.\n")

# Defining Paths & Parameters
base_path = '/kaggle/input/oxford-102-flower-dataset/102 flower/flowers'
json_path = '/kaggle/input/oxford-102-flower-dataset/102 flower/cat_to_name.json'
img_size = 224
batch_size = 32
epochs = 150
patience = 15
learning_rate = 1e-4
output_dir = './mobilenetv2_outputs'
os.makedirs(output_dir, exist_ok=True)

print(f"Parameters for training are set as:")
print(f"  Image Size: {img_size}")
print(f"  Batch Size: {batch_size}")
print(f"  Epochs: {epochs}")
print(f"  Patience: {patience}")
print(f"  Learning Rate: {learning_rate}\n")

# Visualizing Dataset
with open(json_path, 'r') as f:
    real_names = json.load(f)

numeric_classes = sorted(os.listdir(os.path.join(base_path, 'train')))
num_classes = len(numeric_classes)
class_names = [real_names[c] for c in numeric_classes]

print("\n================ CLASS MAPPING TABLE ================")
print(f"Number of classes: {num_classes}\n")
print(f"{'Folder':<10} | {'Flower Name'}")
print("-" * 50)
for f, cname in zip(numeric_classes, class_names):
    print(f"{f:<10} | {cname}")
print("======================================================\n")

train_raw = tf.keras.utils.image_dataset_from_directory(
    os.path.join(base_path, "train"),
    image_size=(img_size, img_size),
    batch_size=9,          # small batch for visualization
    label_mode='int',
    shuffle=True,
    verbose=0)

# Showing samples from dataset
def show_samples(ds_raw, numeric_class_list, name_map, n=9):
    plt.figure(figsize=(8, 8))

    for batch_x, batch_y in ds_raw.take(1):
        imgs = batch_x.numpy().astype("uint8")
        labels = batch_y.numpy()
        break

    rows = int(np.ceil(np.sqrt(n)))
    for i in range(n):
        ax = plt.subplot(rows, rows, i+1)
        plt.imshow(imgs[i])
        folder_id = numeric_class_list[int(labels[i])]
        cls_name = name_map[folder_id]
        ax.set_title(cls_name, fontsize=9)
        ax.axis("off")

    plt.tight_layout()
    plt.show()

print("\nShowing random sample images from the dataset...\n")
show_samples(train_raw, numeric_classes, real_names, n=9)

# Model Training

# Loading dataset
train_ds = tf.keras.utils.image_dataset_from_directory(
    os.path.join(base_path, "train"),
    image_size=(img_size, img_size),
    batch_size=batch_size,
    label_mode='int',
    shuffle=True)

valid_ds = tf.keras.utils.image_dataset_from_directory(
    os.path.join(base_path, "valid"),
    image_size=(img_size, img_size),
    batch_size=batch_size,
    label_mode='int',
    shuffle=True)

test_ds = tf.keras.utils.image_dataset_from_directory(
    os.path.join(base_path, "test"),
    image_size=(img_size, img_size),
    batch_size=batch_size,
    label_mode='int',
    shuffle=False)

# Preprocess function
def preprocess(image, label):
    image = preprocess_input(tf.cast(image, tf.float32))
    label = tf.one_hot(label, num_classes)
    return image, label

train_ds = train_ds.map(preprocess).prefetch(tf.data.AUTOTUNE)
valid_ds = valid_ds.map(preprocess).prefetch(tf.data.AUTOTUNE)
test_ds = test_ds.map(preprocess).prefetch(tf.data.AUTOTUNE)

# Function to set trainable layers
def set_trainable_layers(base_model, unfreeze_percent):
    total_layers = len(base_model.layers)
    num_unfreeze = int(total_layers * unfreeze_percent)
    
    for i, layer in enumerate(base_model.layers):
        if i < total_layers - num_unfreeze:
            layer.trainable = False
        else:
            layer.trainable = True
    
    print(f"Total layers in base model: {total_layers}")
    print(f"Unfreeze percent: {unfreeze_percent*100:.1f}%")
    print(f"No. of layers unfreeze: {num_unfreeze}")
    print(f"Layers freeze: {total_layers - num_unfreeze}")
    if unfreeze_percent != 0:
        print("\nList of unfreeze layers:")
        for i, layer in enumerate(base_model.layers[-num_unfreeze:]):
            print(f"  {i + (total_layers - num_unfreeze)}: {layer.name}")

# Build model
base_model = MobileNetV2(include_top=False, weights='imagenet', input_shape=(img_size, img_size, 3))

# Unfreeze last 0% of layers
set_trainable_layers(base_model, 0.00)

x = GlobalAveragePooling2D()(base_model.output)
x = Dropout(0.3)(x)
x = Dense(512, activation='relu')(x)
x = Dropout(0.3)(x)
output = Dense(num_classes, activation='softmax')(x)
model = Model(base_model.input, output)
print(f"Total layers in final model: {len(model.layers)}")

def compact_model_summary(model, base_model_name="MobileNetV2", custom_layer_start_idx=-5):
    col1_width = 52
    col2_width = 25
    col3_width = 12

    print(f"{'Layer (type)':<{col1_width}} {'Output Shape':<{col2_width}} {'Param #':>{col3_width}}")
    print("=" * (col1_width + col2_width + col3_width))

    base_output = model.layers[custom_layer_start_idx - 1].output.shape
    total_base_params = sum([l.count_params() for l in model.layers[:custom_layer_start_idx]])
    print(f"{base_model_name.title()} (Functional)".ljust(col1_width), f"{str(base_output):<{col2_width}}", f"{total_base_params:>{col3_width},}")

    for layer in model.layers[custom_layer_start_idx:]:
        name = layer.name
        layer_type = layer.__class__.__name__
        try:
            output_shape = str(layer.output.shape)
        except:
            output_shape = "N/A"
        params = f"{layer.count_params():,}"
        print(f"{name + ' (' + layer_type + ')':<{col1_width}} {output_shape:<{col2_width}} {params:>{col3_width}}")

    print("=" * (col1_width + col2_width + col3_width))

    total_params = model.count_params()
    trainable_params = np.sum([tf.keras.backend.count_params(w) for w in model.trainable_weights])
    non_trainable_params = np.sum([tf.keras.backend.count_params(w) for w in model.non_trainable_weights])

    print(f"Total params: {total_params:,}")
    print(f"Trainable params: {trainable_params:,}")
    print(f"Non-trainable params: {non_trainable_params:,}")

print("\nPrinting Model Summary:")
compact_model_summary(model, base_model_name="MobileNetV2", custom_layer_start_idx=-5)

# Compile model
loss_fn = tf.keras.losses.CategoricalCrossentropy(label_smoothing=0.1)

ckpt_path = os.path.join(output_dir, "best_model.h5")
callbacks = [
    EarlyStopping(monitor='val_loss', patience=patience, restore_best_weights=True, verbose=0),
    ModelCheckpoint(ckpt_path, save_best_only=True, monitor='val_loss', verbose=0)
]

model.compile(optimizer=Adam(learning_rate), loss=loss_fn, metrics=['accuracy'])

# Train model
print("\nStarting training...\n")
history = model.fit(
    train_ds,
    validation_data=valid_ds,
    epochs=epochs,
    callbacks=callbacks,
    verbose=0)
print("\nTraining completed.")

# Model Evaluation
print("\nLoading best model for evaluation...")
best_model = tf.keras.models.load_model(ckpt_path)

print("\nEvaluating on Train:")
train_loss, train_acc = best_model.evaluate(train_ds, verbose=0)
print(f"Train Accuracy: {train_acc*100:.2f}%")

print("\nEvaluating on Validation:")
val_loss, val_acc = best_model.evaluate(valid_ds, verbose=0)
print(f"Validation Accuracy: {val_acc*100:.2f}%")

print("\nEvaluating on Test:")
test_loss, test_acc = best_model.evaluate(test_ds, verbose=0)
print(f"Test Accuracy: {test_acc*100:.2f}%")

# Visualizing Results

# Training curves
def plot_training_curves(history):
    acc = history.history["accuracy"]
    val_acc = history.history["val_accuracy"]
    loss = history.history["loss"]
    val_loss = history.history["val_loss"]
    epochs_range = range(1, len(acc) + 1)

    plt.figure(figsize=(14, 5))

    # Accuracy
    plt.subplot(1, 2, 1)
    plt.plot(epochs_range, acc, label="Train Acc")
    plt.plot(epochs_range, val_acc, label="Val Acc")
    plt.xlabel("Epoch")
    plt.ylabel("Accuracy")
    plt.title("Training vs Validation Accuracy")
    plt.legend()

    # Loss
    plt.subplot(1, 2, 2)
    plt.plot(epochs_range, loss, label="Train Loss")
    plt.plot(epochs_range, val_loss, label="Val Loss")
    plt.xlabel("Epoch")
    plt.ylabel("Loss")
    plt.title("Training vs Validation Loss")
    plt.legend()

    plt.tight_layout()
    plt.show()

print("\nPlotting training curves...")
plot_training_curves(history)

# Classification report
y_true = []
y_pred = []

for x_batch, y_batch in test_ds:
    preds = best_model.predict(x_batch, verbose=0)
    y_pred.extend(np.argmax(preds, axis=1))
    y_true.extend(np.argmax(y_batch.numpy(), axis=1))

y_true = np.array(y_true)
y_pred = np.array(y_pred)

print("\nClassification Report:")
print(classification_report(y_true, y_pred, target_names=class_names, zero_division=0))

# Confusion Matrix
from sklearn.metrics import confusion_matrix
cm = confusion_matrix(y_true, y_pred)

plt.figure(figsize=(30, 30))
sns.heatmap(cm, cmap="Blues", annot=True, fmt='d',
            xticklabels=class_names, yticklabels=class_names)
plt.xlabel("Predicted")
plt.ylabel("True")
plt.title("Confusion Matrix")
plt.tight_layout()
plt.show()