DeiT-III
In [ ]:
from google.colab import drive
drive.mount('/content/drive')
Mounted at /content/drive
In [ ]:
!pip install timm grad-cam --quiet
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 0.0/44.5 kB ? eta -:--:-- ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 44.5/44.5 kB 2.6 MB/s eta 0:00:00 Installing build dependencies ... done Getting requirements to build wheel ... done Preparing metadata (pyproject.toml) ... done Building wheel for grad-cam (pyproject.toml) ... done
In [ ]:
import os
import zipfile
from google.colab import drive
# ============================================================
# CONFIGURATION
# ============================================================
ZIP_PATH = "/content/drive/MyDrive/Colab Notebooks/final-thesis/dataset_split_224.zip"
EXTRACT_ROOT = "/content"
# ============================================================
# CHECK ZIP FILE
# ============================================================
if not os.path.exists(ZIP_PATH):
raise FileNotFoundError(
f"❌ ZIP file not found:\n{ZIP_PATH}"
)
print("ZIP file found:")
print(ZIP_PATH)
# ============================================================
# EXTRACT
# ============================================================
print("\nExtracting dataset...")
with zipfile.ZipFile(
ZIP_PATH,
"r"
) as zip_ref:
zip_ref.extractall(
EXTRACT_ROOT
)
# ============================================================
# VERIFY
# ============================================================
DATASET_DIR = "/content/dataset_split_224"
print("\n" + "=" * 70)
print("DATASET EXTRACTION COMPLETED")
print("=" * 70)
if os.path.exists(DATASET_DIR):
print(f"✓ Dataset ready:")
print(DATASET_DIR)
for split in ["train", "val", "test"]:
split_path = os.path.join(
DATASET_DIR,
split
)
if os.path.exists(split_path):
class_count = len(
[
d for d in os.listdir(split_path)
if os.path.isdir(
os.path.join(split_path, d)
)
]
)
image_count = sum(
len(files)
for _, _, files in os.walk(split_path)
)
print(
f"✓ {split:5s} : "
f"{image_count} images | "
f"{class_count} classes"
)
else:
print(
f"⚠️ {split} folder not found"
)
else:
print("❌ Extraction failed.")
print("=" * 70)
ZIP file found: /content/drive/MyDrive/Colab Notebooks/final-thesis/dataset_split_224.zip Extracting dataset... ====================================================================== DATASET EXTRACTION COMPLETED ====================================================================== ✓ Dataset ready: /content/dataset_split_224 ✓ train : 7308 images | 28 classes ✓ val : 382 images | 28 classes ✓ test : 421 images | 28 classes ======================================================================
In [ ]:
# ============================================================
# CELL 1 — DeiT-III Training Configuration & DataLoader
# T4 Optimized | Reproducible | Resume-Friendly
#
# NOTE:
# Letterboxing has ALREADY been applied to the complete dataset.
# Therefore, NO additional resizing, cropping or letterboxing.
# Normalization follows the timm model configuration.
#
# This cell also defines ALL shared paths (checkpoints, history,
# final model) so Cell 2 and Cell 3 can run independently.
# ============================================================
# ============================================================
# 1. IMPORTS
# ============================================================
import os
import time
import copy
import json
import random
import cv2
import numpy as np
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, Dataset
from torchvision import datasets
import timm
from timm.data import resolve_model_data_config
# Avoid CPU thread oversubscription inside DataLoader workers
cv2.setNumThreads(0)
# ============================================================
# 2. REPRODUCIBILITY
# ============================================================
SEED = 42
random.seed(SEED)
np.random.seed(SEED)
torch.manual_seed(SEED)
if torch.cuda.is_available():
torch.cuda.manual_seed(SEED)
torch.cuda.manual_seed_all(SEED)
torch.backends.cudnn.deterministic = True
torch.backends.cudnn.benchmark = False
def seed_worker(worker_id):
"""Make every DataLoader worker reproducible."""
worker_seed = torch.initial_seed() % 2**32
np.random.seed(worker_seed)
random.seed(worker_seed)
loader_generator = torch.Generator()
loader_generator.manual_seed(SEED)
# ============================================================
# 3. DEVICE CONFIGURATION
# ============================================================
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print("\n" + "=" * 70)
print("DEVICE CONFIGURATION")
print("=" * 70)
print(f"Device : {device}")
if torch.cuda.is_available():
gpu_name = torch.cuda.get_device_name(0)
gpu_memory = torch.cuda.get_device_properties(0).total_memory / 1024**3
print(f"GPU : {gpu_name}")
print(f"CUDA Memory : {gpu_memory:.2f} GB")
print("=" * 70)
# ============================================================
# 4. DATASET PATHS
# ============================================================
train_dir = "/content/dataset_split_224/train"
val_dir = "/content/dataset_split_224/val"
test_dir = "/content/dataset_split_224/test"
# ============================================================
# 5. HYPERPARAMETERS
# ============================================================
IMG_SIZE = 224
BATCH_SIZE = 16
# Training stages
EPOCHS_STAGE1 = 5
EPOCHS_STAGE2 = 25
PATIENCE = 4
# DataLoader
NUM_WORKERS = 2
# Learning rates
LR_STAGE1 = 5e-4
LR_BACKBONE = 5e-6
LR_HEAD = 5e-5
# Regularization
WEIGHT_DECAY = 1e-4
# Optional light augmentation (train split only). Default OFF so the
# experiment stays identical to before; set True to enable random
# horizontal flip (only if flipping keeps labels valid for your data).
USE_AUGMENTATION = False
# Mixed precision (T4 supports fp16, NOT bf16)
USE_AMP = torch.cuda.is_available()
# Model configuration
DISPLAY_NAME = "DeiT-III"
TIMM_NAME = "deit3_base_patch16_224"
# ============================================================
# 6. OUTPUT / CHECKPOINT DIRECTORIES & SHARED PATHS
# ============================================================
OUTPUT_DIR = (
"/content/drive/MyDrive/Colab Notebooks/"
"final-thesis/Model_Comparison_Results_DeiT-III"
)
os.makedirs(OUTPUT_DIR, exist_ok=True)
CHECKPOINT_DIR = os.path.join(OUTPUT_DIR, "checkpoints")
os.makedirs(CHECKPOINT_DIR, exist_ok=True)
# Shared by Cell 2 (training) and Cell 3 (evaluation)
LATEST_CHECKPOINT = os.path.join(CHECKPOINT_DIR, "latest_checkpoint.pt")
BEST_CHECKPOINT = os.path.join(CHECKPOINT_DIR, "best_checkpoint.pt")
FINAL_MODEL_PATH = os.path.join(OUTPUT_DIR, f"{DISPLAY_NAME}_best_macro_f1.pt")
HISTORY_PATH = os.path.join(OUTPUT_DIR, f"{DISPLAY_NAME}_training_history.json")
print("\n" + "=" * 70)
print("OUTPUT CONFIGURATION")
print("=" * 70)
print(f"Output directory : {OUTPUT_DIR}")
print(f"Checkpoint directory : {CHECKPOINT_DIR}")
print("=" * 70)
# ============================================================
# 7. CLASS NAMES (taken from ImageFolder so they always match
# the labels the Dataset will produce)
# ============================================================
_train_probe = datasets.ImageFolder(train_dir)
class_names = list(_train_probe.classes)
num_classes = len(class_names)
if num_classes == 0:
raise ValueError(f"No class folders found in:\n{train_dir}")
train_targets = np.array(_train_probe.targets)
print("\n" + "=" * 70)
print("CLASS INFORMATION")
print("=" * 70)
print(f"Number of classes : {num_classes}")
# ============================================================
# 8. CALCULATE CLASS WEIGHTS ("balanced": N / (n_classes * count))
# Computed from the SAME file list ImageFolder uses.
# ============================================================
class_counts = np.bincount(train_targets, minlength=num_classes)
total_train = int(class_counts.sum())
class_weights = np.ones(num_classes, dtype=np.float32)
for i in range(num_classes):
if class_counts[i] > 0:
class_weights[i] = total_train / (num_classes * class_counts[i])
else:
print(f"⚠️ Class '{class_names[i]}' has 0 training images.")
class_weights_tensor = torch.tensor(
class_weights, dtype=torch.float32, device=device
)
print("\nClass weights:")
print(f"{'ID':<4}| {'Class':<30} | {'Count':>6} | Weight")
for i, cls in enumerate(class_names):
print(f"{i:02d} | {cls:<30} | {class_counts[i]:>6} | {class_weights[i]:.4f}")
print("=" * 70)
# ============================================================
# 9. TIMM MODEL DATA CONFIGURATION
# ============================================================
print("\n" + "=" * 70)
print("RESOLVING TIMM MODEL CONFIGURATION")
print("=" * 70)
config_model = timm.create_model(
TIMM_NAME,
pretrained=False,
num_classes=num_classes
)
data_config = resolve_model_data_config(config_model)
MODEL_INPUT_SIZE = tuple(data_config["input_size"])
MODEL_MEAN = np.array(data_config["mean"], dtype=np.float32).reshape(1, 1, 3)
MODEL_STD = np.array(data_config["std"], dtype=np.float32).reshape(1, 1, 3)
MODEL_INTERPOLATION = data_config.get("interpolation", "bicubic")
MODEL_CROP_PCT = data_config.get("crop_pct", None)
print(f"Model name : {TIMM_NAME}")
print(f"Input size : {MODEL_INPUT_SIZE}")
print(f"Normalization mean : {MODEL_MEAN.flatten().tolist()}")
print(f"Normalization std : {MODEL_STD.flatten().tolist()}")
print(f"Interpolation : {MODEL_INTERPOLATION}")
print(f"Crop percentage : {MODEL_CROP_PCT}")
if MODEL_INPUT_SIZE != (3, IMG_SIZE, IMG_SIZE):
raise ValueError(
f"The model expects input size {MODEL_INPUT_SIZE}, "
f"but this dataset is configured for (3, {IMG_SIZE}, {IMG_SIZE})."
)
del config_model
if torch.cuda.is_available():
torch.cuda.empty_cache()
print("\n✓ Model-specific normalization configuration resolved.")
print("✓ Additional resize/crop/letterbox disabled.")
print("=" * 70)
# ============================================================
# 10. DATASET DEFINITION
# ============================================================
class PreprocessedImageDataset(Dataset):
def __init__(self, root_dir, img_size=224, augment=False):
self.dataset = datasets.ImageFolder(root_dir)
self.targets = self.dataset.targets
self.img_size = img_size
self.augment = augment
self.mean = MODEL_MEAN
self.std = MODEL_STD
def __len__(self):
return len(self.dataset)
def __getitem__(self, idx):
img_path, label = self.dataset.samples[idx]
image = cv2.imread(img_path)
if image is None:
raise RuntimeError(f"Could not read image:\n{img_path}")
image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
h, w = image.shape[:2]
if h != self.img_size or w != self.img_size:
raise ValueError(
f"Unexpected image size {w}x{h} for:\n{img_path}\n"
f"Expected preprocessed size: {self.img_size}x{self.img_size}"
)
# Optional augmentation (train only)
if self.augment and random.random() < 0.5:
image = np.ascontiguousarray(image[:, ::-1])
image = image.astype(np.float32) / 255.0
# TIMM model-specific normalization
image = (image - self.mean) / self.std
image = np.ascontiguousarray(np.transpose(image, (2, 0, 1)))
image_tensor = torch.from_numpy(image).float()
label_tensor = torch.tensor(label, dtype=torch.long)
return image_tensor, label_tensor
# ============================================================
# 11. BUILD DATALOADERS
# ============================================================
def build_dataloaders():
train_ds = PreprocessedImageDataset(train_dir, IMG_SIZE, augment=USE_AUGMENTATION)
val_ds = PreprocessedImageDataset(val_dir, IMG_SIZE, augment=False)
test_ds = PreprocessedImageDataset(test_dir, IMG_SIZE, augment=False)
expected_class_to_idx = {name: idx for idx, name in enumerate(class_names)}
for split_name, ds in [("Train", train_ds), ("Validation", val_ds), ("Test", test_ds)]:
if ds.dataset.class_to_idx != expected_class_to_idx:
raise ValueError(f"{split_name} class mapping differs from training mapping.")
pin_memory = (device.type == "cuda")
persistent = (NUM_WORKERS > 0)
common = dict(
batch_size=BATCH_SIZE,
num_workers=NUM_WORKERS,
pin_memory=pin_memory,
persistent_workers=persistent,
worker_init_fn=seed_worker,
)
train_loader = DataLoader(train_ds, shuffle=True, generator=loader_generator, **common)
val_loader = DataLoader(val_ds, shuffle=False, **common)
test_loader = DataLoader(test_ds, shuffle=False, **common)
return train_loader, val_loader, test_loader
train_loader, val_loader, test_loader = build_dataloaders()
# ============================================================
# 12. DATASET SUMMARY
# ============================================================
print("\n" + "=" * 70)
print("DATASET SUMMARY")
print("=" * 70)
print(f"Train images : {len(train_loader.dataset)}")
print(f"Validation : {len(val_loader.dataset)}")
print(f"Test images : {len(test_loader.dataset)}")
print(f"\nTrain batches : {len(train_loader)}")
print(f"Val batches : {len(val_loader)}")
print(f"Test batches : {len(test_loader)}")
print(f"\nBatch size : {BATCH_SIZE}")
print(f"Image size : {IMG_SIZE} x {IMG_SIZE}")
print(f"AMP enabled : {USE_AMP}")
print(f"Workers : {NUM_WORKERS}")
print(f"Augmentation : {USE_AUGMENTATION}")
print("\nPreprocessing : Already letterboxed")
print("Normalization : TIMM model configuration")
print("=" * 70)
# ============================================================
# 13. VERIFY ONE BATCH
# ============================================================
print("\n" + "=" * 70)
print("DATA PIPELINE VERIFICATION")
print("=" * 70)
sample_images, sample_labels = next(iter(train_loader))
print(f"Batch image shape : {tuple(sample_images.shape)}")
print(f"Batch label shape : {tuple(sample_labels.shape)}")
print(f"Image dtype : {sample_images.dtype}")
print(f"Label dtype : {sample_labels.dtype}")
print(f"Pixel range : [{sample_images.min():.3f}, {sample_images.max():.3f}]")
if tuple(sample_images.shape[1:]) == (3, IMG_SIZE, IMG_SIZE):
print("\n✓ Data pipeline verification passed.")
else:
print("\n⚠️ Sample shape differs from expected:")
print(f"Expected: (B, 3, {IMG_SIZE}, {IMG_SIZE})")
print("=" * 70)
====================================================================== DEVICE CONFIGURATION ====================================================================== Device : cuda GPU : Tesla T4 CUDA Memory : 14.56 GB ====================================================================== ====================================================================== OUTPUT CONFIGURATION ====================================================================== Output directory : /content/drive/MyDrive/Colab Notebooks/final-thesis/Model_Comparison_Results_DeiT-III Checkpoint directory : /content/drive/MyDrive/Colab Notebooks/final-thesis/Model_Comparison_Results_DeiT-III/checkpoints ====================================================================== ====================================================================== CLASS INFORMATION ====================================================================== Number of classes : 28 Class weights: ID | Class | Count | Weight 00 | Bottle_Gourd_Flower_Blossom_Blight | 172 | 1.5174 01 | Bottle_Gourd_Flower_Healthy | 196 | 1.3316 02 | Bottle_Gourd_Fruit_Borer | 240 | 1.0875 03 | Bottle_Gourd_Fruit_Healthy | 244 | 1.0697 04 | Bottle_Gourd_Leaf_Healthy | 276 | 0.9457 05 | Bottle_Gourd_Leaf_Wilt | 236 | 1.1059 06 | Eggplant_Fruit_Borer | 172 | 1.5174 07 | Eggplant_Fruit_Healthy | 288 | 0.9062 08 | Eggplant_Leaf_Healthy | 620 | 0.4210 09 | Eggplant_Leaf_Mosaic_Virus | 288 | 0.9062 10 | Eggplant_Leaf_Wilt | 200 | 1.3050 11 | Okra_Leaf_Healthy | 212 | 1.2311 12 | Okra_Leaf_Yellow_Vein_Mosaic_Virus | 244 | 1.0697 13 | Papaya_Fruit_Anthracnose | 248 | 1.0524 14 | Papaya_Fruit_Healthy | 240 | 1.0875 15 | Papaya_Leaf_Curl_Virus | 248 | 1.0524 16 | Papaya_Leaf_Healthy | 336 | 0.7768 17 | Papaya_Leaf_Mosaic_Virus | 180 | 1.4500 18 | Papaya_Leaf_Spot | 296 | 0.8818 19 | Pointed_Gourd_Fruit_Healthy | 216 | 1.2083 20 | Pointed_Gourd_Fruit_Rot | 144 | 1.8125 21 | Pointed_Gourd_Leaf_Healthy | 316 | 0.8259 22 | Pointed_Gourd_Leaf_Potassium_Deficiency | 232 | 1.1250 23 | Teasel_Gourd_Fruit_Healthy | 372 | 0.7016 24 | Teasel_Gourd_Fruit_Rot | 336 | 0.7768 25 | Teasel_Gourd_Leaf_Blight | 168 | 1.5536 26 | Teasel_Gourd_Leaf_Healthy | 416 | 0.6274 27 | Teasel_Gourd_Leaf_Spot | 172 | 1.5174 ====================================================================== ====================================================================== RESOLVING TIMM MODEL CONFIGURATION ====================================================================== Model name : deit3_base_patch16_224 Input size : (3, 224, 224) Normalization mean : [0.48500001430511475, 0.4560000002384186, 0.4059999883174896] Normalization std : [0.2290000021457672, 0.2240000069141388, 0.22499999403953552] Interpolation : bicubic Crop percentage : 0.9 ✓ Model-specific normalization configuration resolved. ✓ Additional resize/crop/letterbox disabled. ====================================================================== ====================================================================== DATASET SUMMARY ====================================================================== Train images : 7308 Validation : 382 Test images : 421 Train batches : 457 Val batches : 24 Test batches : 27 Batch size : 16 Image size : 224 x 224 AMP enabled : True Workers : 2 Augmentation : False Preprocessing : Already letterboxed Normalization : TIMM model configuration ====================================================================== ====================================================================== DATA PIPELINE VERIFICATION ====================================================================== Batch image shape : (16, 3, 224, 224) Batch label shape : (16,) Image dtype : torch.float32 Label dtype : torch.int64 Pixel range : [-2.118, 2.640] ✓ Data pipeline verification passed. ======================================================================
In [ ]:
# ============================================================
# CELL 2 — DeiT-III Two-Stage Fine-Tuning
# Keras-Style Live Progress + AMP + Checkpoint + Resume
# + Early Stopping + Best Model
#
# DEPENDENCY NOTE:
# Needs CELL 1 (config, dataloaders, shared paths).
# ============================================================
import os
import sys
import time
import json
from contextlib import nullcontext
import torch
import torch.nn as nn
import torch.nn.functional as F
import timm
from sklearn.metrics import f1_score
print("\n" + "=" * 80)
print(f"🚀 Running Two-Stage Fine-Tuning — {DISPLAY_NAME}")
print("=" * 80)
# ============================================================
# 1. MODEL INITIALIZATION
# ============================================================
model = timm.create_model(
TIMM_NAME,
pretrained=True,
num_classes=num_classes
).to(device)
print(f"\n✓ Model loaded: {TIMM_NAME}")
print(f"✓ Input size: {IMG_SIZE} × {IMG_SIZE}")
# ============================================================
# 2. LOSS FUNCTION
# - `criterion` (class-weighted) is used for TRAINING only.
# - Reported train/val loss is the plain (unweighted) CE so the
# two curves are comparable and exact per-sample averages.
# ============================================================
criterion = nn.CrossEntropyLoss(weight=class_weights_tensor)
# ============================================================
# 3. AMP GRAD SCALER & AUTOCAST CONTEXT
# ============================================================
try:
scaler = torch.amp.GradScaler("cuda", enabled=USE_AMP)
except Exception:
scaler = torch.cuda.amp.GradScaler(enabled=USE_AMP)
def autocast_context():
if USE_AMP:
return torch.autocast(device_type="cuda", dtype=torch.float16)
return nullcontext()
# ============================================================
# 4. LIVE PROGRESS DISPLAY (Keras-Style)
# ============================================================
def print_progress(current, total, start_time, train_loss, train_acc, bar_length=24):
progress = current / total if total > 0 else 0.0
filled = int(bar_length * progress)
bar = "━" * filled + "░" * (bar_length - filled)
elapsed = time.time() - start_time
remaining = (elapsed / current) * (total - current) if current > 0 else 0.0
sys.stdout.write(
f"\r{current}/{total} {bar} {progress * 100:5.1f}% "
f"- {elapsed:.0f}s - ETA {remaining:.0f}s "
f"- accuracy: {train_acc:.4f} - loss: {train_loss:.4f}"
)
sys.stdout.flush()
# ============================================================
# 5. VALIDATION FUNCTION
# ============================================================
def validate_model(model, val_loader):
model.eval()
loss_sum = 0.0
correct = 0
total = 0
y_true, y_pred = [], []
with torch.no_grad():
for images, labels in val_loader:
images = images.to(device, non_blocking=True)
labels = labels.to(device, non_blocking=True)
with autocast_context():
outputs = model(images)
outputs = outputs.float()
loss_sum += F.cross_entropy(outputs, labels, reduction="sum").item()
preds = outputs.argmax(dim=1)
correct += (preds == labels).sum().item()
total += labels.size(0)
y_true.extend(labels.cpu().numpy())
y_pred.extend(preds.cpu().numpy())
val_loss = loss_sum / max(total, 1)
val_acc = correct / max(total, 1)
macro_f1 = f1_score(y_true, y_pred, average="macro", zero_division=0)
return val_loss, val_acc, macro_f1
# ============================================================
# 6. MODEL FREEZING / UNFREEZING UTILITIES
# ============================================================
def get_head_param_ids(model):
return {id(p) for p in model.get_classifier().parameters()}
def get_head_parameter_names(model):
ids = get_head_param_ids(model)
return [n for n, p in model.named_parameters() if id(p) in ids]
def freeze_backbone(model):
for p in model.parameters():
p.requires_grad = False
head_params = list(model.get_classifier().parameters())
if len(head_params) == 0:
raise ValueError("No classifier parameters found.")
for p in head_params:
p.requires_grad = True
return head_params
def unfreeze_all(model):
for p in model.parameters():
p.requires_grad = True
# ============================================================
# 7. CHECKPOINTING (atomic write; best-checkpoint is model-only
# to keep Drive writes small and fast)
# ============================================================
ATOMIC_SAVE = False # keep False on Google Drive (prevents duplicate files)
def save_checkpoint(
path, model, optimizer, scheduler, scaler,
epoch, stage, best_macro_f1, epochs_no_improve, history,
include_training_state=True
):
checkpoint = {
"model_state_dict": model.state_dict(),
"epoch": epoch,
"stage": stage,
"best_macro_f1": best_macro_f1,
"epochs_no_improve": epochs_no_improve,
"history": history,
"class_names": class_names,
"num_classes": num_classes,
"img_size": IMG_SIZE,
"batch_size": BATCH_SIZE,
"seed": SEED,
"model_name": TIMM_NAME,
"head_parameter_names": get_head_parameter_names(model),
}
if include_training_state:
checkpoint["optimizer_state_dict"] = optimizer.state_dict()
checkpoint["scheduler_state_dict"] = scheduler.state_dict()
checkpoint["scaler_state_dict"] = scaler.state_dict()
if ATOMIC_SAVE:
# Temp file + rename. NOT recommended on Google Drive: Drive can keep
# the old file and create a 2nd file with the same name.
tmp_path = path + ".tmp"
torch.save(checkpoint, tmp_path)
os.replace(tmp_path, path)
else:
# Direct overwrite -> Drive keeps exactly ONE file per name
torch.save(checkpoint, path)
# ============================================================
# 8. CHECKPOINT COMPATIBILITY & RESUME STATE LOADING
# ============================================================
start_stage = 1
start_epoch = 0
best_macro_f1 = 0.0
epochs_no_improve = 0
history = {"stage1": [], "stage2": []}
optimizer = None
scheduler = None
resume_ckpt = None
if os.path.exists(LATEST_CHECKPOINT):
# Load on CPU (saves ~1 GB of T4 GPU memory)
resume_ckpt = torch.load(LATEST_CHECKPOINT, map_location="cpu", weights_only=False)
compatible = (
resume_ckpt.get("model_name") == TIMM_NAME
and resume_ckpt.get("class_names") == class_names
and resume_ckpt.get("img_size") == IMG_SIZE
and resume_ckpt.get("head_parameter_names") == get_head_parameter_names(model)
)
if not compatible:
raise RuntimeError(
"Incompatible checkpoint detected. Back up or remove "
f"latest_checkpoint.pt and best_checkpoint.pt from {CHECKPOINT_DIR}."
)
print("\n" + "=" * 80)
print("🔄 CHECKPOINT FOUND — RESUMING TRAINING")
print("=" * 80)
model.load_state_dict(resume_ckpt["model_state_dict"])
best_macro_f1 = resume_ckpt.get("best_macro_f1", 0.0)
epochs_no_improve = resume_ckpt.get("epochs_no_improve", 0)
history = resume_ckpt.get("history", {"stage1": [], "stage2": []})
saved_stage = resume_ckpt["stage"]
saved_epoch = resume_ckpt["epoch"]
if saved_stage not in (1, 2):
raise ValueError(f"Unknown checkpoint stage: {saved_stage}")
start_stage = saved_stage
start_epoch = saved_epoch + 1
print(f"Saved stage : {saved_stage}")
print(f"Saved epoch : {saved_epoch + 1}")
print(f"Best Macro F1 : {best_macro_f1:.4f}")
print(f"No improvement : {epochs_no_improve}")
print(f"\n→ Resuming Stage {start_stage} from Epoch {start_epoch + 1}")
else:
print("\n✓ No checkpoint found. Starting training from scratch.")
# ============================================================
# 9. SHARED EPOCH LOGIC
# ============================================================
def train_one_epoch(optimizer, epoch, total_epochs):
model.train()
epoch_start = time.time()
loss_sum, correct, seen, skipped = 0.0, 0, 0, 0
n_batches = len(train_loader)
print(f"\nEpoch {epoch + 1}/{total_epochs}")
for batch_idx, (images, labels) in enumerate(train_loader, start=1):
images = images.to(device, non_blocking=True)
labels = labels.to(device, non_blocking=True)
optimizer.zero_grad(set_to_none=True)
with autocast_context():
outputs = model(images)
outputs = outputs.float()
loss = criterion(outputs, labels)
if not torch.isfinite(loss):
skipped += 1
continue
scaler.scale(loss).backward()
scaler.unscale_(optimizer)
torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
scaler.step(optimizer)
scaler.update()
with torch.no_grad():
loss_sum += F.cross_entropy(outputs.detach(), labels, reduction="sum").item()
correct += (outputs.argmax(dim=1) == labels).sum().item()
seen += labels.size(0)
print_progress(
batch_idx, n_batches, epoch_start,
loss_sum / seen, correct / seen
)
print()
if skipped:
print(f"⚠️ Skipped {skipped} batch(es) with non-finite loss.")
if seen == 0:
raise RuntimeError("All batches were skipped (non-finite loss).")
return loss_sum / seen, correct / seen, epoch_start
def run_stage(stage, total_epochs, first_epoch, use_early_stopping):
"""Runs epochs of one stage. Returns True if early stopping fired."""
global best_macro_f1, epochs_no_improve
key = f"stage{stage}"
n_batches = len(train_loader)
for epoch in range(first_epoch, total_epochs):
train_loss, train_acc, epoch_start = train_one_epoch(optimizer, epoch, total_epochs)
# LR actually used during this epoch (read BEFORE scheduler.step)
lrs = [g["lr"] for g in optimizer.param_groups]
scheduler.step()
val_loss, val_acc, macro_f1 = validate_model(model, val_loader)
epoch_time = time.time() - epoch_start
step_ms = epoch_time / n_batches * 1000
print(
f"{n_batches}/{n_batches} ━━━━━━━━━━━━━━━━━━━━━━━━ "
f"{epoch_time:.0f}s {step_ms:.0f}ms/step "
f"- accuracy: {train_acc:.4f} - loss: {train_loss:.4f} "
f"- val_accuracy: {val_acc:.4f} - val_loss: {val_loss:.4f} "
f"- val_macro_f1: {macro_f1:.4f}"
)
record = {
"epoch": epoch + 1,
"train_loss": train_loss,
"train_acc": train_acc,
"val_loss": val_loss,
"val_acc": val_acc,
"macro_f1": macro_f1,
"time_sec": epoch_time,
}
if stage == 1:
record["lr"] = lrs[0]
else:
record["backbone_lr"] = lrs[0]
record["head_lr"] = lrs[1]
history[key].append(record)
if macro_f1 > best_macro_f1:
best_macro_f1 = macro_f1
epochs_no_improve = 0
print(f"🏆 New Best Macro F1: {best_macro_f1:.4f}")
save_checkpoint(
BEST_CHECKPOINT, model, optimizer, scheduler, scaler,
epoch, stage, best_macro_f1, epochs_no_improve, history,
include_training_state=False
)
elif use_early_stopping:
epochs_no_improve += 1
print(f"⏳ No improvement: {epochs_no_improve}/{PATIENCE}")
save_checkpoint(
LATEST_CHECKPOINT, model, optimizer, scheduler, scaler,
epoch, stage, best_macro_f1, epochs_no_improve, history
)
with open(HISTORY_PATH, "w") as f:
json.dump(history, f, indent=4)
if use_early_stopping and epochs_no_improve >= PATIENCE:
print("\n" + "=" * 80)
print("🛑 EARLY STOPPING TRIGGERED")
print(f"No Macro F1 improvement for {PATIENCE} epochs.")
print("=" * 80)
return True
return False
# ============================================================
# 10. STAGE 1 — CLASSIFIER HEAD ONLY
# ============================================================
if start_stage == 1:
print("\n" + "=" * 80)
print("🟦 STAGE 1 — CLASSIFIER HEAD ONLY")
print("=" * 80)
head_params = freeze_backbone(model)
trainable_params = [p for p in model.parameters() if p.requires_grad]
print(f"Trainable parameters : {sum(p.numel() for p in trainable_params):,}")
print(f"Classifier parameters: {sum(p.numel() for p in head_params):,}")
optimizer = torch.optim.AdamW(trainable_params, lr=LR_STAGE1, weight_decay=WEIGHT_DECAY)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=EPOCHS_STAGE1)
if resume_ckpt is not None and resume_ckpt["stage"] == 1:
optimizer.load_state_dict(resume_ckpt["optimizer_state_dict"])
scheduler.load_state_dict(resume_ckpt["scheduler_state_dict"])
scaler.load_state_dict(resume_ckpt["scaler_state_dict"])
print("✓ Stage-1 optimizer/scheduler/scaler restored.")
run_stage(1, EPOCHS_STAGE1, start_epoch, use_early_stopping=False)
if os.path.exists(BEST_CHECKPOINT):
best_ckpt = torch.load(BEST_CHECKPOINT, map_location="cpu", weights_only=False)
model.load_state_dict(best_ckpt["model_state_dict"])
print(f"\n✓ Restored best Stage-1 model (Macro F1: {best_ckpt['best_macro_f1']:.4f})")
del best_ckpt
start_stage = 2
start_epoch = 0
epochs_no_improve = 0
# ============================================================
# 11. STAGE 2 — FULL FINE-TUNING
# ============================================================
if start_stage == 2:
print("\n" + "=" * 80)
print("🟩 STAGE 2 — FULL FINE-TUNING")
print("=" * 80)
unfreeze_all(model)
head_ids = get_head_param_ids(model)
backbone_params = [p for p in model.parameters() if id(p) not in head_ids]
head_params = [p for p in model.parameters() if id(p) in head_ids]
print(f"Backbone parameters : {sum(p.numel() for p in backbone_params):,}")
print(f"Head parameters : {sum(p.numel() for p in head_params):,}")
optimizer = torch.optim.AdamW(
[
{"params": backbone_params, "lr": LR_BACKBONE},
{"params": head_params, "lr": LR_HEAD},
],
weight_decay=WEIGHT_DECAY
)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=EPOCHS_STAGE2)
if resume_ckpt is not None and resume_ckpt["stage"] == 2:
optimizer.load_state_dict(resume_ckpt["optimizer_state_dict"])
scheduler.load_state_dict(resume_ckpt["scheduler_state_dict"])
scaler.load_state_dict(resume_ckpt["scaler_state_dict"])
print("✓ Stage-2 optimizer/scheduler/scaler restored.")
if epochs_no_improve >= PATIENCE:
print("\n🛑 Early stopping was already triggered in a previous run. Skipping Stage 2.")
else:
run_stage(2, EPOCHS_STAGE2, start_epoch, use_early_stopping=True)
del resume_ckpt
if torch.cuda.is_available():
torch.cuda.empty_cache()
# ============================================================
# 12. FINAL MODEL & ARTIFACT SAVING
# ============================================================
print("\n" + "=" * 80)
print("🏆 LOADING BEST MODEL & SAVING ARTIFACTS")
print("=" * 80)
if not os.path.exists(BEST_CHECKPOINT):
raise FileNotFoundError(f"Best checkpoint not found:\n{BEST_CHECKPOINT}")
best_ckpt = torch.load(BEST_CHECKPOINT, map_location="cpu", weights_only=False)
model.load_state_dict(best_ckpt["model_state_dict"])
best_macro_f1 = best_ckpt["best_macro_f1"]
del best_ckpt
print(f"✓ Best validation Macro F1: {best_macro_f1:.4f}")
torch.save(model.state_dict(), FINAL_MODEL_PATH)
print(f"✓ Final best model saved:\n{FINAL_MODEL_PATH}")
with open(HISTORY_PATH, "w") as f:
json.dump(history, f, indent=4)
print(f"✓ Training history saved:\n{HISTORY_PATH}")
# ============================================================
# 13. TRAINING SUMMARY
# ============================================================
print("\n" + "=" * 80)
print("✨ TRAINING COMPLETE")
print("=" * 80)
print(f"Stage 1 epochs completed : {len(history['stage1'])}")
print(f"Stage 2 epochs completed : {len(history['stage2'])}")
print(f"Best Validation Macro F1 : {best_macro_f1:.4f}")
print(f"AMP enabled : {USE_AMP}")
print(f"\nBest checkpoint : {BEST_CHECKPOINT}")
print(f"Latest checkpoint : {LATEST_CHECKPOINT}")
print(f"Training history : {HISTORY_PATH}")
print("=" * 80)
================================================================================ 🚀 Running Two-Stage Fine-Tuning — DeiT-III ================================================================================
Warning: You are sending unauthenticated requests to the HF Hub. Please set a HF_TOKEN to enable higher rate limits and faster downloads. WARNING:huggingface_hub.utils._http:Warning: You are sending unauthenticated requests to the HF Hub. Please set a HF_TOKEN to enable higher rate limits and faster downloads.
✓ Model loaded: deit3_base_patch16_224 ✓ Input size: 224 × 224 ✓ No checkpoint found. Starting training from scratch. ================================================================================ 🟦 STAGE 1 — CLASSIFIER HEAD ONLY ================================================================================ Trainable parameters : 21,532 Classifier parameters: 21,532 Epoch 1/5 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% - 31s - ETA 0s - accuracy: 0.8389 - loss: 0.7823 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 32s 71ms/step - accuracy: 0.8389 - loss: 0.7823 - val_accuracy: 0.8979 - val_loss: 0.3401 - val_macro_f1: 0.8877 🏆 New Best Macro F1: 0.8877 Epoch 2/5 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% - 30s - ETA 0s - accuracy: 0.9640 - loss: 0.1754 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 31s 68ms/step - accuracy: 0.9640 - loss: 0.1754 - val_accuracy: 0.9188 - val_loss: 0.2242 - val_macro_f1: 0.9057 🏆 New Best Macro F1: 0.9057 Epoch 3/5 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% - 23s - ETA 0s - accuracy: 0.9789 - loss: 0.1063 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 24s 53ms/step - accuracy: 0.9789 - loss: 0.1063 - val_accuracy: 0.9607 - val_loss: 0.1835 - val_macro_f1: 0.9541 🏆 New Best Macro F1: 0.9541 Epoch 4/5 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% - 28s - ETA 0s - accuracy: 0.9851 - loss: 0.0803 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 30s 65ms/step - accuracy: 0.9851 - loss: 0.0803 - val_accuracy: 0.9529 - val_loss: 0.1713 - val_macro_f1: 0.9456 Epoch 5/5 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% - 24s - ETA 0s - accuracy: 0.9882 - loss: 0.0707 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 25s 55ms/step - accuracy: 0.9882 - loss: 0.0707 - val_accuracy: 0.9555 - val_loss: 0.1722 - val_macro_f1: 0.9457 ✓ Restored best Stage-1 model (Macro F1: 0.9541) ================================================================================ 🟩 STAGE 2 — FULL FINE-TUNING ================================================================================ Backbone parameters : 85,816,320 Head parameters : 21,532 Epoch 1/25 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% - 79s - ETA 0s - accuracy: 0.9915 - loss: 0.0314 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 80s 175ms/step - accuracy: 0.9915 - loss: 0.0314 - val_accuracy: 0.9686 - val_loss: 0.1161 - val_macro_f1: 0.9620 🏆 New Best Macro F1: 0.9620 Epoch 2/25 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% - 78s - ETA 0s - accuracy: 0.9996 - loss: 0.0014 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 79s 172ms/step - accuracy: 0.9996 - loss: 0.0014 - val_accuracy: 0.9738 - val_loss: 0.1160 - val_macro_f1: 0.9682 🏆 New Best Macro F1: 0.9682 Epoch 3/25 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% - 78s - ETA 0s - accuracy: 1.0000 - loss: 0.0002 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 79s 173ms/step - accuracy: 1.0000 - loss: 0.0002 - val_accuracy: 0.9738 - val_loss: 0.1192 - val_macro_f1: 0.9671 ⏳ No improvement: 1/4 Epoch 4/25 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% - 77s - ETA 0s - accuracy: 1.0000 - loss: 0.0000 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 78s 171ms/step - accuracy: 1.0000 - loss: 0.0000 - val_accuracy: 0.9712 - val_loss: 0.1200 - val_macro_f1: 0.9652 ⏳ No improvement: 2/4 Epoch 5/25 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% - 77s - ETA 0s - accuracy: 1.0000 - loss: 0.0000 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 78s 171ms/step - accuracy: 1.0000 - loss: 0.0000 - val_accuracy: 0.9712 - val_loss: 0.1213 - val_macro_f1: 0.9652 ⏳ No improvement: 3/4 Epoch 6/25 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% - 77s - ETA 0s - accuracy: 1.0000 - loss: 0.0000 457/457 ━━━━━━━━━━━━━━━━━━━━━━━━ 78s 171ms/step - accuracy: 1.0000 - loss: 0.0000 - val_accuracy: 0.9712 - val_loss: 0.1215 - val_macro_f1: 0.9652 ⏳ No improvement: 4/4 ================================================================================ 🛑 EARLY STOPPING TRIGGERED No Macro F1 improvement for 4 epochs. ================================================================================ ================================================================================ 🏆 LOADING BEST MODEL & SAVING ARTIFACTS ================================================================================ ✓ Best validation Macro F1: 0.9682 ✓ Final best model saved: /content/drive/MyDrive/Colab Notebooks/final-thesis/Model_Comparison_Results_DeiT-III/DeiT-III_best_macro_f1.pt ✓ Training history saved: /content/drive/MyDrive/Colab Notebooks/final-thesis/Model_Comparison_Results_DeiT-III/DeiT-III_training_history.json ================================================================================ ✨ TRAINING COMPLETE ================================================================================ Stage 1 epochs completed : 5 Stage 2 epochs completed : 6 Best Validation Macro F1 : 0.9682 AMP enabled : True Best checkpoint : /content/drive/MyDrive/Colab Notebooks/final-thesis/Model_Comparison_Results_DeiT-III/checkpoints/best_checkpoint.pt Latest checkpoint : /content/drive/MyDrive/Colab Notebooks/final-thesis/Model_Comparison_Results_DeiT-III/checkpoints/latest_checkpoint.pt Training history : /content/drive/MyDrive/Colab Notebooks/final-thesis/Model_Comparison_Results_DeiT-III/DeiT-III_training_history.json ================================================================================
In [ ]:
import os
import json
try:
with open(HISTORY_PATH, 'r') as f:
data = json.load(f)
grand_total_time = 0
grand_total_epochs = 0
print("--- Stage-by-Stage Training Time Report ---")
# Loop through each stage (e.g., stage1, stage2)
for stage_name, epochs_list in data.items():
if not epochs_list:
continue
stage_time = sum(epoch.get("time_sec", 0) for epoch in epochs_list)
stage_epochs = len(epochs_list)
stage_avg = stage_time / stage_epochs if stage_epochs > 0 else 0
grand_total_time += stage_time
grand_total_epochs += stage_epochs
# Print individual stage report
print(f"\n[{stage_name.upper()}]")
print(f" - Epochs: {stage_epochs}")
print(f" - Total Time: {stage_time:.2f} seconds ({stage_time/60:.2f} minutes)")
print(f" - Avg Time per Epoch: {stage_avg:.2f} seconds")
# Calculate overall grand total in hours, minutes, seconds
hours = int(grand_total_time // 3600)
minutes = int((grand_total_time % 3600) // 60)
seconds = grand_total_time % 3600 % 60
print("\n" + "="*40)
print("--- OVERALL TRAINING SUMMARY ---")
print(f"Total Epochs Across All Stages: {grand_total_epochs}")
if hours > 0:
print(f"Grand Total Time: {hours} hours, {minutes} minutes, {seconds:.2f} seconds")
else:
print(f"Grand Total Time: {minutes} minutes, {seconds:.2f} seconds ({grand_total_time:.2f} total seconds)")
print("="*40)
except FileNotFoundError:
print(f"Error: File not found at {HISTORY_PATH}. Please check your path variables.")
--- Stage-by-Stage Training Time Report --- [STAGE1] - Epochs: 5 - Total Time: 142.68 seconds (2.38 minutes) - Avg Time per Epoch: 28.54 seconds [STAGE2] - Epochs: 6 - Total Time: 471.62 seconds (7.86 minutes) - Avg Time per Epoch: 78.60 seconds ======================================== --- OVERALL TRAINING SUMMARY --- Total Epochs Across All Stages: 11 Grand Total Time: 10 minutes, 14.30 seconds (614.30 total seconds) ========================================
In [ ]:
# ============================================================
# CELL 3 — FINAL TEST EVALUATION (Independent / Resume-Friendly)
# DeiT-III
#
# DEPENDENCY NOTE:
# Depends ONLY on CELL 1 (config, shared paths, dataloaders).
# Does NOT require CELL 2 in the same session — loads weights
# directly from best_checkpoint.pt (or the final .pt file).
# ============================================================
import os
import sys
import json
import math
import random
import subprocess
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from matplotlib.ticker import MaxNLocator
import seaborn as sns
import torch
import timm
from sklearn.metrics import (
classification_report,
confusion_matrix,
roc_curve,
precision_recall_curve,
auc,
roc_auc_score,
f1_score,
accuracy_score,
balanced_accuracy_score,
matthews_corrcoef,
average_precision_score
)
try:
from pytorch_grad_cam import EigenCAM
except ImportError:
subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", "grad-cam"])
from pytorch_grad_cam import EigenCAM
from pytorch_grad_cam.utils.image import show_cam_on_image
# ============================================================
# 1. EVALUATION DIRECTORY SETUP
# ============================================================
EVAL_DIR = os.path.join(OUTPUT_DIR, "Evaluation")
os.makedirs(EVAL_DIR, exist_ok=True)
n_classes = len(class_names)
print("=" * 80)
print(f"FINAL TEST EVALUATION (Independent) — {DISPLAY_NAME}")
print("=" * 80)
# ============================================================
# 2. LOAD TRAINING HISTORY (if available)
# ============================================================
print("\n" + "=" * 80)
print("LOADING TRAINING HISTORY")
print("=" * 80)
history = {"stage1": [], "stage2": []}
if os.path.exists(HISTORY_PATH):
with open(HISTORY_PATH, "r") as f:
history = json.load(f)
print(f"✓ Training history loaded from:\n{HISTORY_PATH}")
else:
print("⚠️ Training history file not found. Curves will be skipped.")
stage1_history = history.get("stage1", [])
stage2_history = history.get("stage2", [])
# ============================================================
# 3. TRAINING CURVES
# ============================================================
rows = []
for r in stage1_history:
rows.append({"stage": 1, "epoch": int(r["epoch"]), **{k: r.get(k) for k in ("train_loss", "val_loss", "train_acc", "val_acc")}})
stage1_completed = len(stage1_history)
for r in stage2_history:
rows.append({"stage": 2, "epoch": stage1_completed + int(r["epoch"]), **{k: r.get(k) for k in ("train_loss", "val_loss", "train_acc", "val_acc")}})
history_df = pd.DataFrame(rows)
if len(history_df) > 0:
print("\n" + "=" * 80)
print("📉 TRAINING vs VALIDATION LOSS CURVE")
print("=" * 80)
fig, ax = plt.subplots(figsize=(12, 7))
ax.plot(
history_df["epoch"],
history_df["train_loss"],
marker="o",
linewidth=2,
markersize=4,
label="Training Loss"
)
ax.plot(
history_df["epoch"],
history_df["val_loss"],
marker="s",
linewidth=2,
markersize=4,
label="Validation Loss"
)
# Stage transition
if len(stage1_history) > 0 and len(stage2_history) > 0:
transition_epoch = len(stage1_history) + 0.5
ax.axvline(
transition_epoch,
color="gray",
linestyle="--",
linewidth=1.5,
alpha=0.8,
label="Stage 1 → Stage 2"
)
ax.set_title(
f"Training vs Validation Loss — {DISPLAY_NAME}",
fontsize=16,
fontweight="bold"
)
ax.set_xlabel("Epoch", fontsize=12)
ax.set_ylabel("Loss", fontsize=12)
ax.set_xticks(history_df["epoch"].astype(int))
ax.grid(linestyle="--", alpha=0.4)
ax.legend(fontsize=10)
fig.tight_layout()
loss_curve_path = os.path.join(
EVAL_DIR,
f"{DISPLAY_NAME}_Training_vs_Validation_Loss.png"
)
fig.savefig(loss_curve_path, dpi=300, bbox_inches="tight")
plt.show()
plt.close(fig)
print(f"✓ Loss curve saved:\n{loss_curve_path}")
else:
print("⚠️ No training history available. Loss curve skipped.")
if len(history_df) > 0:
print("\n" + "=" * 80)
print("📈 TRAINING vs VALIDATION ACCURACY CURVE")
print("=" * 80)
if (
"val_acc" in history_df.columns
and not history_df["val_acc"].isna().all()
):
fig, ax = plt.subplots(figsize=(12, 7))
ax.plot(
history_df["epoch"],
history_df["train_acc"],
marker="o",
linewidth=2,
markersize=4,
label="Training Accuracy"
)
ax.plot(
history_df["epoch"],
history_df["val_acc"],
marker="s",
linewidth=2,
markersize=4,
label="Validation Accuracy"
)
# Stage transition
if len(stage1_history) > 0 and len(stage2_history) > 0:
transition_epoch = len(stage1_history) + 0.5
ax.axvline(
transition_epoch,
color="gray",
linestyle="--",
linewidth=1.5,
alpha=0.8,
label="Stage 1 → Stage 2"
)
ax.set_title(
f"Training vs Validation Accuracy — {DISPLAY_NAME}",
fontsize=16,
fontweight="bold"
)
ax.set_xlabel("Epoch", fontsize=12)
ax.set_ylabel("Accuracy", fontsize=12)
ax.set_ylim(0, 1.05)
ax.set_xticks(history_df["epoch"].astype(int))
ax.grid(linestyle="--", alpha=0.4)
ax.legend(fontsize=10)
fig.tight_layout()
accuracy_curve_path = os.path.join(
EVAL_DIR,
f"{DISPLAY_NAME}_Training_vs_Validation_Accuracy.png"
)
fig.savefig(accuracy_curve_path, dpi=300, bbox_inches="tight")
plt.show()
plt.close(fig)
print(f"✓ Accuracy curve saved:\n{accuracy_curve_path}")
else:
print("⚠️ Validation accuracy is missing from training history.")
print(" Accuracy curve was NOT generated because plotting Macro F1 as validation accuracy would be incorrect.")
# ============================================================
# 4. MODEL INITIALIZATION & CHECKPOINT LOADING
# ============================================================
print("\n" + "=" * 80)
print(f"INITIALIZING MODEL ({TIMM_NAME}) & LOADING CHECKPOINT")
print("=" * 80)
model = timm.create_model(TIMM_NAME, pretrained=False, num_classes=num_classes)
best_val_f1 = float("nan")
if os.path.exists(BEST_CHECKPOINT):
ckpt = torch.load(BEST_CHECKPOINT, map_location="cpu", weights_only=False)
model.load_state_dict(ckpt["model_state_dict"])
best_val_f1 = ckpt.get("best_macro_f1", float("nan"))
del ckpt
print("✓ Loaded model weights from best_checkpoint.pt")
print(f"Saved Validation Macro F1: {best_val_f1:.4f}")
elif os.path.exists(FINAL_MODEL_PATH):
model.load_state_dict(torch.load(FINAL_MODEL_PATH, map_location="cpu", weights_only=True))
print("✓ Loaded model weights from final model path.")
else:
raise FileNotFoundError(
f"No saved model found at:\n1) {BEST_CHECKPOINT}\n2) {FINAL_MODEL_PATH}\n"
"Please ensure training was run at least once."
)
model = model.to(device)
model.eval()
# ============================================================
# 5. TEST INFERENCE
# ============================================================
print("\n" + "=" * 80)
print("RUNNING TEST INFERENCE")
print("=" * 80)
y_true, y_pred, y_probs = [], [], []
test_ds = test_loader.dataset
image_paths = [s[0] for s in test_ds.dataset.samples] # same order as loader (shuffle=False)
with torch.no_grad():
for images, labels in test_loader:
images = images.to(device, non_blocking=True)
if device.type == "cuda":
with torch.autocast(device_type="cuda", dtype=torch.float16):
outputs = model(images)
else:
outputs = model(images)
probs = torch.softmax(outputs.float(), dim=1)
y_true.extend(labels.numpy())
y_pred.extend(probs.argmax(dim=1).cpu().numpy())
y_probs.extend(probs.cpu().numpy())
y_true = np.asarray(y_true)
y_pred = np.asarray(y_pred)
y_probs = np.asarray(y_probs)
print(f"✓ Test samples evaluated: {len(y_true)}")
# ============================================================
# 6. METRICS
# ============================================================
accuracy = accuracy_score(y_true, y_pred)
balanced_acc = balanced_accuracy_score(y_true, y_pred)
macro_f1 = f1_score(y_true, y_pred, average="macro", zero_division=0)
weighted_f1 = f1_score(y_true, y_pred, average="weighted", zero_division=0)
micro_f1 = f1_score(y_true, y_pred, average="micro", zero_division=0)
mcc = matthews_corrcoef(y_true, y_pred)
# One-hot labels (works for any n_classes, incl. 2)
y_true_bin = np.eye(n_classes, dtype=int)[y_true]
support = y_true_bin.sum(axis=0)
# Per-class ROC-AUC / PR-AUC (NaN if a class has no positives or no negatives in test)
per_class_roc_auc, per_class_pr_auc = [], []
for i in range(n_classes):
if 0 < support[i] < len(y_true):
per_class_roc_auc.append(roc_auc_score(y_true_bin[:, i], y_probs[:, i]))
per_class_pr_auc.append(average_precision_score(y_true_bin[:, i], y_probs[:, i]))
else:
per_class_roc_auc.append(np.nan)
per_class_pr_auc.append(np.nan)
per_class_roc_auc = np.array(per_class_roc_auc)
per_class_pr_auc = np.array(per_class_pr_auc)
valid = ~np.isnan(per_class_roc_auc)
# Computed manually so a class missing from the test set cannot crash sklearn
macro_roc_auc = float(np.nanmean(per_class_roc_auc))
weighted_roc_auc = float(np.average(per_class_roc_auc[valid], weights=support[valid]))
micro_roc_auc = float(roc_auc_score(y_true_bin.ravel(), y_probs.ravel()))
macro_pr_auc = float(np.nanmean(per_class_pr_auc))
print("\n" + "=" * 80)
print("TEST RESULTS SUMMARY")
print("=" * 80)
print(f"Accuracy : {accuracy:.4f}")
print(f"Balanced Accuracy : {balanced_acc:.4f}")
print(f"Macro F1 : {macro_f1:.4f}")
print(f"Weighted F1 : {weighted_f1:.4f}")
print(f"Micro F1 : {micro_f1:.4f}")
print(f"Macro ROC-AUC : {macro_roc_auc:.4f}")
print(f"Weighted ROC-AUC : {weighted_roc_auc:.4f}")
print(f"Micro ROC-AUC : {micro_roc_auc:.4f}")
print(f"Macro PR-AUC : {macro_pr_auc:.4f}")
print(f"MCC : {mcc:.4f}")
# ---- Save overall metrics
overall = {
"model": DISPLAY_NAME,
"val_macro_f1_best": best_val_f1,
"test_accuracy": accuracy,
"test_balanced_accuracy": balanced_acc,
"test_macro_f1": macro_f1,
"test_weighted_f1": weighted_f1,
"test_micro_f1": micro_f1,
"test_macro_roc_auc": macro_roc_auc,
"test_weighted_roc_auc": weighted_roc_auc,
"test_micro_roc_auc": micro_roc_auc,
"test_macro_pr_auc": macro_pr_auc,
"test_mcc": mcc,
"test_samples": int(len(y_true)),
}
pd.DataFrame([overall]).to_csv(os.path.join(EVAL_DIR, f"{DISPLAY_NAME}_overall_metrics.csv"), index=False)
# ---- Classification report & per-class metrics
report_dict = classification_report(
y_true, y_pred, labels=np.arange(n_classes),
target_names=class_names, zero_division=0, output_dict=True
)
pd.DataFrame(report_dict).transpose().to_csv(
os.path.join(EVAL_DIR, f"{DISPLAY_NAME}_classification_report.csv")
)
per_class_df = pd.DataFrame([{
"class": cls,
"precision": report_dict[cls]["precision"],
"recall": report_dict[cls]["recall"],
"f1_score": report_dict[cls]["f1-score"],
"support": report_dict[cls]["support"],
"roc_auc": per_class_roc_auc[i],
"pr_auc": per_class_pr_auc[i],
} for i, cls in enumerate(class_names)])
per_class_df.to_csv(os.path.join(EVAL_DIR, f"{DISPLAY_NAME}_per_class_metrics.csv"), index=False)
# ---- Per-image predictions (useful for error analysis)
pred_df = pd.DataFrame({
"image_path": image_paths,
"true_label": [class_names[i] for i in y_true],
"pred_label": [class_names[i] for i in y_pred],
"confidence": y_probs.max(axis=1),
"correct": (y_true == y_pred),
})
pred_df.to_csv(os.path.join(EVAL_DIR, f"{DISPLAY_NAME}_test_predictions.csv"), index=False)
print(f"✓ Misclassified test images: {int((~pred_df['correct']).sum())}")
# ============================================================
# 7. CONFUSION MATRICES (raw + row-normalized)
# ============================================================
cm = confusion_matrix(y_true, y_pred, labels=np.arange(n_classes))
cm_norm = cm / np.clip(cm.sum(axis=1, keepdims=True), 1, None)
for data, fmt, name, title in [
(cm, "d", "confusion_matrix", "Confusion Matrix"),
(cm_norm, ".2f", "confusion_matrix_normalized", "Normalized Confusion Matrix"),
]:
plt.figure(figsize=(18, 15))
sns.heatmap(data, annot=(n_classes <= 30), fmt=fmt, cmap="Blues",
xticklabels=class_names, yticklabels=class_names)
plt.title(f"{title} — {DISPLAY_NAME}", fontsize=16, fontweight="bold")
plt.xlabel("Predicted")
plt.ylabel("True")
plt.xticks(rotation=90)
plt.yticks(rotation=0)
plt.savefig(os.path.join(EVAL_DIR, f"{DISPLAY_NAME}_{name}.png"), dpi=300, bbox_inches="tight")
plt.show()
plt.close()
# ============================================================
# 8. ROC & PRECISION-RECALL CURVES
# ============================================================
plt.figure(figsize=(14, 11))
for i in range(n_classes):
if not valid[i]:
continue
fpr, tpr, _ = roc_curve(y_true_bin[:, i], y_probs[:, i])
plt.plot(fpr, tpr, linewidth=1.7, label=f"{class_names[i]} (AUC={auc(fpr, tpr):.3f})")
plt.plot([0, 1], [0, 1], "k--", linewidth=1.5)
plt.xlabel("False Positive Rate")
plt.ylabel("True Positive Rate")
plt.title(f"Multiclass ROC Curves — {DISPLAY_NAME}", fontsize=15, fontweight="bold")
plt.legend(loc="center left", bbox_to_anchor=(1.02, 0.5), fontsize=8)
plt.tight_layout()
plt.savefig(os.path.join(EVAL_DIR, f"{DISPLAY_NAME}_ROC_AUC_curves.png"), dpi=300, bbox_inches="tight")
plt.show()
plt.close()
plt.figure(figsize=(14, 11))
for i in range(n_classes):
if support[i] == 0:
continue
prec, rec, _ = precision_recall_curve(y_true_bin[:, i], y_probs[:, i])
plt.plot(rec, prec, linewidth=1.7, label=f"{class_names[i]} (AP={per_class_pr_auc[i]:.3f})")
plt.xlabel("Recall")
plt.ylabel("Precision")
plt.title(f"Multiclass Precision-Recall Curves — {DISPLAY_NAME}", fontsize=15, fontweight="bold")
plt.legend(loc="center left", bbox_to_anchor=(1.02, 0.5), fontsize=8)
plt.tight_layout()
plt.savefig(os.path.join(EVAL_DIR, f"{DISPLAY_NAME}_PR_curves.png"), dpi=300, bbox_inches="tight")
plt.show()
plt.close()
# ============================================================
# 9. EIGEN-CAM : 3 INDEPENDENT FULL-CLASS GRID IMAGES (7x4) - DEiT-III
# ============================================================
import os
import random
import numpy as np
import matplotlib.pyplot as plt
import torch
from pytorch_grad_cam import EigenCAM
from pytorch_grad_cam.utils.image import show_cam_on_image
# Robust reshape transform tailored specifically for DEiT-III / Vision Transformers (ViT)
def reshape_transform_deit(tensor):
# tensor shape: [batch_size, seq_len, embedding_dim]
# DEiT/ViT has a class token, so seq_len = 1 + grid_size * grid_size
if len(tensor.shape) == 3:
b, seq_len, c = tensor.shape
# Exclude the class token and keep only the patch tokens
# If seq_len = 1 + 14*14 (e.g., 197), drop the class token (index 0)
grid_size = int((seq_len - 1) ** 0.5)
if grid_size * grid_size == seq_len - 1:
# Drop the class token and reshape the patch tokens
tensor = tensor[:, 1:, :]
return tensor.reshape(b, grid_size, grid_size, c).permute(0, 3, 1, 2)
else:
# Fallback: if there is no class token or a different size
grid_size = int(seq_len ** 0.5)
return tensor.reshape(b, grid_size, grid_size, c).permute(0, 3, 1, 2)
elif len(tensor.shape) == 4:
return tensor.permute(0, 3, 1, 2)
return tensor
# Target layer selection for DEiT-III (Standard ViT blocks)
# Selecting the LayerNorm or Attention block of the last encoder layer is ideal
if hasattr(model, "blocks"):
target_layers = [model.blocks[-1].norm1]
elif hasattr(model, "transformer") and hasattr(model.transformer, "resblocks"):
target_layers = [model.transformer.resblocks[-1].ln_1]
else:
# Fallback to the last encoder block norm
target_layers = [list(model.children())[-2]]
print("EigenCAM target layer:")
print(target_layers[0])
# Initialize EigenCAM for DEiT-III
cam = EigenCAM(model=model, target_layers=target_layers, reshape_transform=reshape_transform_deit)
# Group test dataset indices by class
class_to_indices = {i: [] for i in range(num_classes)}
for idx, dataset_item in enumerate(test_loader.dataset):
if isinstance(dataset_item, tuple):
_, label = dataset_item
else:
label = dataset_item["label"]
class_to_indices[int(label)].append(idx)
mean = MODEL_MEAN.squeeze()
std = MODEL_STD.squeeze()
NUM_RUNS = 3
ROWS, COLS = 7, 4 # Grid size for multi-class visual evaluation
# Generate 3 independent full-class grid images
for run_id in range(1, NUM_RUNS + 1):
print(f"\nGenerating EigenCAM grid image {run_id}/{NUM_RUNS}...")
fig, axes = plt.subplots(ROWS, COLS, figsize=(20, 32))
axes = axes.flatten()
for class_idx in range(num_classes):
available_indices = class_to_indices[class_idx]
if len(available_indices) == 0:
axes[class_idx].axis("off")
continue
chosen_idx = random.choice(available_indices)
dataset_item = test_loader.dataset[chosen_idx]
if isinstance(dataset_item, tuple):
img_tensor, true_label = dataset_item
else:
img_tensor = dataset_item["pixel_values"][0]
true_label = dataset_item["label"]
input_tensor = img_tensor.unsqueeze(0).to(device)
# Model Prediction
with torch.no_grad():
if device.type == "cuda":
with torch.autocast(device_type="cuda", dtype=torch.float16):
output = model(input_tensor)
else:
output = model(input_tensor)
probs = torch.softmax(output.float(), dim=1)
pred_label = int(probs.argmax(dim=1).item())
confidence = float(probs[0, pred_label].item())
# Compute EigenCAM heatmap with targets=None
grayscale_cam = cam(input_tensor=input_tensor, targets=None)[0]
# Denormalize Image
img_np = img_tensor.cpu().numpy().transpose(1, 2, 0)
img_np = img_np * std + mean
img_np = np.clip(img_np, 0, 1)
# Overlay Heatmap onto Image
cam_image = show_cam_on_image(img_np, grayscale_cam, use_rgb=True)
# Plotting
ax = axes[class_idx]
ax.imshow(cam_image)
is_correct = (pred_label == int(true_label))
status_symbol = "✓" if is_correct else "✗"
title_color = "green" if is_correct else "red"
ax.set_title(
f"[{status_symbol}] {class_names[class_idx]}\n"
f"Pred: {class_names[pred_label]} ({confidence:.1%})",
fontsize=9,
fontweight="bold",
color=title_color
)
ax.axis("off")
for idx in range(num_classes, len(axes)):
axes[idx].axis("off")
plt.tight_layout()
save_path = os.path.join(EVAL_DIR, f"{DISPLAY_NAME}_EigenCAM_Grid_{run_id}.png")
plt.savefig(save_path, dpi=300, bbox_inches="tight")
plt.show()
plt.close()
print(f"✓ Grid {run_id} successfully saved: {save_path}")
# Release hooks
cam.activations_and_grads.release()
print("\nAll EigenCAM grid images generated and saved successfully!")
if torch.cuda.is_available():
torch.cuda.empty_cache()
print("\n" + "=" * 80)
print("✨ INDEPENDENT EVALUATION COMPLETED SUCCESSFULLY!")
print(f"📁 Artifacts saved in: {EVAL_DIR}")
print("=" * 80)
================================================================================ FINAL TEST EVALUATION (Independent) — DeiT-III ================================================================================ ================================================================================ LOADING TRAINING HISTORY ================================================================================ ✓ Training history loaded from: /content/drive/MyDrive/Colab Notebooks/final-thesis/Model_Comparison_Results_DeiT-III/DeiT-III_training_history.json ================================================================================ 📉 TRAINING vs VALIDATION LOSS CURVE ================================================================================
✓ Loss curve saved: /content/drive/MyDrive/Colab Notebooks/final-thesis/Model_Comparison_Results_DeiT-III/Evaluation/DeiT-III_Training_vs_Validation_Loss.png ================================================================================ 📈 TRAINING vs VALIDATION ACCURACY CURVE ================================================================================
✓ Accuracy curve saved: /content/drive/MyDrive/Colab Notebooks/final-thesis/Model_Comparison_Results_DeiT-III/Evaluation/DeiT-III_Training_vs_Validation_Accuracy.png ================================================================================ INITIALIZING MODEL (deit3_base_patch16_224) & LOADING CHECKPOINT ================================================================================ ✓ Loaded model weights from best_checkpoint.pt Saved Validation Macro F1: 0.9682 ================================================================================ RUNNING TEST INFERENCE ================================================================================ ✓ Test samples evaluated: 421 ================================================================================ TEST RESULTS SUMMARY ================================================================================ Accuracy : 0.9786 Balanced Accuracy : 0.9795 Macro F1 : 0.9778 Weighted F1 : 0.9784 Micro F1 : 0.9786 Macro ROC-AUC : 0.9998 Weighted ROC-AUC : 0.9998 Micro ROC-AUC : 0.9999 Macro PR-AUC : 0.9963 MCC : 0.9778 ✓ Misclassified test images: 9
EigenCAM target layer: LayerNorm((768,), eps=1e-06, elementwise_affine=True) Generating EigenCAM grid image 1/3...
✓ Grid 1 successfully saved: /content/drive/MyDrive/Colab Notebooks/final-thesis/Model_Comparison_Results_DeiT-III/Evaluation/DeiT-III_EigenCAM_Grid_1.png Generating EigenCAM grid image 2/3...
