Initial push

.gitattributes

@@ -33,3 +33,4 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+EfficientNet_B7_20percent.pth_wrong_pred.png filter=lfs diff=lfs merge=lfs -text

efficient_b7_20_percent.py

@@ -0,0 +1,432 @@
+import torch
+import torchvision
+import torchinfo
+
+import typing
+import requests
+import os
+import zipfile
+import mlxtend.plotting
+import torchmetrics
+from pathlib import Path
+from timeit import default_timer as timer
+from tqdm.auto import tqdm
+import matplotlib
+
+matplotlib.use("TkAgg")
+from matplotlib import pyplot as plt
+
+
+device = "cuda" if torch.cuda.is_available() else "cpu"
+TRAIN_MODEL = False
+BATCH_SIZE = 32
+LEARNING_RATE = 0.001
+NUM_EPOCH = 10
+MODEL_PATH = Path("models")
+MODEL_NAME = "EfficientNet_B7_20percent.pth"
+MODEL_SAVE_PATH = MODEL_PATH / MODEL_NAME
+
+# Downloading the data here
+data_path = Path("data/")
+image_path = data_path / "pizza_steak_sushi_20_percent"
+
+# If the image folder doesn't exist, download it and prepare it...
+if image_path.is_dir():
+    print(f"{image_path} directory exists.")
+else:
+    print(f"Did not find {image_path} directory, creating one...")
+    image_path.mkdir(parents=True, exist_ok=True)
+
+    # Download pizza, steak, sushi data
+    with open(data_path / "pizza_steak_sushi_20_percent.zip", "wb") as f:
+        request = requests.get(
+            "https://github.com/mrdbourke/pytorch-deep-learning/raw/main/data/pizza_steak_sushi_20_percent.zip"
+        )
+        print("Downloading pizza, steak, sushi data...")
+        f.write(request.content)
+
+    # Unzip pizza, steak, sushi data
+    with zipfile.ZipFile(
+        data_path / "pizza_steak_sushi_20_percent.zip", "r"
+    ) as zip_ref:
+        print("Unzipping pizza, steak, sushi data...")
+        zip_ref.extractall(image_path)
+
+    # Remove .zip file
+    os.remove(data_path / "pizza_steak_sushi_20_percent.zip")
+
+train_dir = image_path / "train"
+test_dir = image_path / "test"
+
+manual_transform = torchvision.transforms.Compose(
+    [
+        torchvision.transforms.Resize((224, 224)),
+        torchvision.transforms.ToTensor(),
+        torchvision.transforms.Normalize(
+            mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]
+        ),
+    ]
+)
+
+
+def create_dataloaders(
+    train_dir: Path,
+    test_dir: Path,
+    batch_size: int,
+    num_workers: int,
+    transform: torchvision.transforms.Compose,
+) -> tuple[
+    torch.utils.data.DataLoader,
+    torch.utils.data.DataLoader,
+    list[str],
+    torchvision.datasets.ImageFolder,
+    torchvision.datasets.ImageFolder,
+]:
+    train_data = torchvision.datasets.ImageFolder(
+        train_dir,
+        transform=transform,
+    )
+    test_data = torchvision.datasets.ImageFolder(
+        test_dir,
+        transform=transform,
+    )
+
+    class_names = train_data.classes
+
+    train_dataloader = torch.utils.data.DataLoader(
+        train_data,
+        batch_size=batch_size,
+        shuffle=True,
+        num_workers=num_workers,
+        pin_memory=True,
+    )
+
+    test_dataloader = torch.utils.data.DataLoader(
+        test_data,
+        batch_size=batch_size,
+        num_workers=num_workers,
+        shuffle=False,
+        pin_memory=True,
+    )
+
+    return (
+        train_dataloader,
+        test_dataloader,
+        class_names,
+        train_data,
+        test_data,
+    )
+
+
+(
+    train_dataloader_manual_transform,
+    test_dataloader_manual_transform,
+    class_names_manual_transform,
+    train_data,
+    test_data,
+) = create_dataloaders(
+    train_dir=train_dir,
+    test_dir=test_dir,
+    num_workers=os.cpu_count() or 0,
+    batch_size=BATCH_SIZE,
+    transform=manual_transform,
+)
+
+weights = torchvision.models.EfficientNet_B7_Weights.DEFAULT
+
+auto_transform = weights.transforms()
+
+(
+    train_dataloader,
+    test_dataloader,
+    class_names,
+    train_data,
+    test_data,
+) = create_dataloaders(
+    train_dir=train_dir,
+    test_dir=test_dir,
+    batch_size=BATCH_SIZE,
+    num_workers=os.cpu_count() or 0,
+    transform=auto_transform,
+)
+
+model = torchvision.models.efficientnet_b7(weights=weights).to(device)
+
+torchinfo.summary(
+    model=model,
+    input_size=(32, 3, 224, 224),
+    col_names=["input_size", "output_size", "num_params", "trainable"],
+    row_settings=["var_names"],
+)
+
+for feature in model.features:
+    print(feature)
+
+for param in model.features.parameters():
+    param.requires_grad = False
+
+print(f"Classifier part has (before changing):\n{model.classifier}")
+
+torch.manual_seed(37)
+torch.cuda.manual_seed(37)
+output_shape = len(class_names)
+model.classifier = torch.nn.Sequential(
+    torch.nn.Dropout(p=0.2, inplace=True),
+    torch.nn.Linear(in_features=2560, out_features=output_shape, bias=True),
+)
+
+print(f"Classifier part has (after changing):\n{model.classifier}")
+
+torchinfo.summary(
+    model=model,
+    input_size=(32, 3, 224, 224),
+    col_names=["input_size", "output_size", "num_params", "trainable"],
+    row_settings=["var_names"],
+)
+
+loss_fn = torch.nn.CrossEntropyLoss()
+optim = torch.optim.Adam(params=model.parameters(), lr=LEARNING_RATE)
+
+
+class Engine:
+    def __init__(
+        self,
+        train_dataloader: torch.utils.data.DataLoader,
+        test_dataloader: torch.utils.data.DataLoader,
+        model: torch.nn.Module,
+        optim: torch.optim.Optimizer,
+        loss_fn: torch.nn.Module,
+        device: typing.Literal["cuda", "cpu"],
+        num_epoch: int,
+    ):
+        self.train_dataloader = train_dataloader
+        self.test_dataloader = test_dataloader
+        self.optim = optim
+        self.loss_fn = loss_fn
+        self.device = device
+        self.num_epoch = num_epoch
+        self.model = model.to(device)
+
+    def _train_step(self) -> tuple[float, float]:
+        self.model.train()
+        loss_train = 0
+        acc_train = 0
+
+        for batch, (X, y) in enumerate(self.train_dataloader):
+            X, y = X.to(self.device), y.to(self.device)
+
+            train_pred = self.model(X)
+            loss = self.loss_fn(train_pred, y)
+
+            loss_train += loss.item()
+
+            optim.zero_grad()
+            loss.backward()
+            optim.step()
+
+            pred_class = torch.argmax(torch.softmax(train_pred, dim=1), dim=1)
+            acc = (pred_class == y).sum().item() / len(pred_class)
+
+            acc_train += acc
+
+            if batch % 2 == 0:
+                print(f"{batch} batches have been processed...")
+
+        loss_train = loss_train / len(self.train_dataloader)
+        acc_train = acc_train / len(self.train_dataloader)
+
+        return loss_train, acc_train
+
+    def _test_step(self) -> tuple[float, float]:
+        self.model.eval()
+        loss_test = 0
+        acc_test = 0
+
+        with torch.inference_mode():
+            for batch, (X, y) in enumerate(self.test_dataloader):
+                X, y = X.to(self.device), y.to(self.device)
+
+                test_pred = self.model(X)
+                loss = self.loss_fn(test_pred, y)
+
+                loss_test += loss.item()
+
+                pred_class = torch.argmax(torch.softmax(test_pred, dim=1), dim=1)
+                acc = (pred_class == y).sum().item() / len(pred_class)
+                acc_test += acc
+
+                if batch % 2 == 0:
+                    print(f"{batch} batches have been processed...")
+
+        loss_test = loss_test / len(self.test_dataloader)
+        acc_test = acc_test / len(self.test_dataloader)
+
+        return loss_test, acc_test
+
+    def train(self) -> tuple[list[float], list[float], list[float], list[float]]:
+        train_loss_list = []
+        test_loss_list = []
+        train_acc_list = []
+        test_acc_list = []
+        for epoch in tqdm(range(self.num_epoch)):
+            print(f"{'*' * 6} EPOCH NUM: {epoch} {'*' * 6}")
+
+            print("Starting the training...")
+            train_loss, train_acc = self._train_step()
+            print("Starting the testing...")
+            test_loss, test_acc = self._test_step()
+            print(
+                f"Train Loss: {train_loss:.3f} | Train Acc: {train_acc:.3f} "
+                f"Test Loss: {test_loss:.3f} | Test Acc: {test_acc:.3f}"
+            )
+
+            train_loss_list.append(train_loss)
+            train_acc_list.append(train_acc)
+            test_loss_list.append(test_loss)
+            test_acc_list.append(test_acc)
+
+        return train_loss_list, train_acc_list, test_loss_list, test_acc_list
+
+
+torch.manual_seed(37)
+torch.cuda.manual_seed(37)
+engine = Engine(
+    train_dataloader=train_dataloader,
+    test_dataloader=test_dataloader,
+    model=model,
+    optim=optim,
+    loss_fn=loss_fn,
+    num_epoch=NUM_EPOCH,
+    device=device,
+)
+
+
+def plot_curves(
+    train_loss: list[float],
+    train_acc: list[float],
+    test_loss: list[float],
+    test_acc: list[float],
+    num_epoch: int,
+):
+    fig, ax = plt.subplots(nrows=1, ncols=2, figsize=(12, 8))
+
+    # Ploting loss curves
+    ax[0].plot(range(num_epoch), train_loss, color="red", label="Train")
+    ax[0].plot(range(num_epoch), test_loss, color="blue", label="Test")
+    ax[0].set(xlabel="Epochs", ylabel="Loss", title="Train vs Test Loss")
+    ax[0].legend()
+
+    # Plotting acc curves
+    ax[1].plot(range(num_epoch), train_acc, color="red", label="Train")
+    ax[1].plot(range(num_epoch), test_acc, color="blue", label="Test")
+    ax[1].set(xlabel="Epochs", ylabel="Accuracy", title="Train vs Test Accuracy")
+    ax[1].legend()
+
+    fig.suptitle("Loss and Accuracy Curve")
+    plt.savefig(f"{MODEL_NAME}_curves.png")
+    plt.show()
+
+
+if TRAIN_MODEL:
+    start_time = timer()
+    train_loss, train_acc, test_loss, test_acc = engine.train()
+    end_time = timer()
+    print(f"INFO: Training process took {end_time - start_time:.3f} seconds.")
+
+    MODEL_PATH.mkdir(parents=True, exist_ok=True)
+    torch.save(obj=model.state_dict(), f=MODEL_SAVE_PATH)
+
+    plot_curves(train_loss, train_acc, test_loss, test_acc, NUM_EPOCH)
+
+else:
+    model.load_state_dict(
+        torch.load(f=MODEL_SAVE_PATH, weights_only=True, map_location=device)
+    )
+
+
+# Plotting the Confusion Matrix
+def give_predictions(
+    test_dataloader: torch.utils.data.DataLoader,
+    model: torch.nn.Module,
+    device: typing.Literal["cuda", "cpu"],
+) -> tuple[torch.Tensor, torch.Tensor]:
+    print("Starting the testing...")
+    model.to(device)
+
+    predictions = []
+    logits_prob = []
+    model.eval()
+    with torch.inference_mode():
+        for X, y in tqdm(test_dataloader, desc="Doing Validation"):
+            X, y = X.to(device), y.to(device)
+
+            logits = model(X)
+
+            pred = torch.argmax(torch.softmax(logits, dim=1), dim=1)
+            logits_prob.append(torch.softmax(logits, dim=1).cpu())
+
+            predictions.append(pred.cpu())
+
+    return torch.cat(predictions), torch.cat(logits_prob)
+
+
+# First we need the prediction on entire dataset
+test_preds, logits_prob = give_predictions(
+    test_dataloader=test_dataloader, model=model, device=device
+)
+
+confmat = torchmetrics.ConfusionMatrix(num_classes=len(class_names), task="multiclass")
+confmat_tensor = confmat(preds=test_preds, target=torch.tensor(test_data.targets))
+fig, ax = mlxtend.plotting.plot_confusion_matrix(
+    conf_mat=confmat_tensor.numpy(),
+    class_names=class_names,
+    figsize=(10, 7),
+)
+plt.savefig(f"{MODEL_NAME}_confusion_matrix.png")
+plt.show()
+
+# Getting the wrong predictions where the model was most confidient.
+pred_wrong = []
+for i in range(len(test_preds)):
+    if test_preds[i] != test_data.targets[i]:
+        pred_wrong.append([test_data.targets[i], test_preds[i], logits_prob[i], i])
+
+pred_wrong.sort(key=lambda x: x[2][x[1]], reverse=True)
+
+# Creating this so I can get un-normalized data so I can plot the image.
+# otherwise some images will be below zero that is invaild etc.
+test_data_original = torchvision.datasets.ImageFolder(
+    test_dir,
+    transform=None,
+)
+
+if len(pred_wrong) > 2:
+    nrows, ncols = len(pred_wrong) // 2 if len(pred_wrong) // 2 < 5 else 5, 2
+    fig, ax = plt.subplots(nrows=nrows, ncols=ncols, figsize=(12, 8))
+    for rows in range(nrows):
+        for cols in range(ncols):
+            index_1d = rows * ncols + cols
+            image, true_label_index = test_data_original[pred_wrong[index_1d][3]]
+            true_label = class_names[true_label_index]
+            pred_label_index = pred_wrong[index_1d][1]
+            pred_label = class_names[pred_label_index]
+            ax[rows][cols].imshow(image)
+            ax[rows][cols].set_title(
+                f"True: {true_label}:{pred_wrong[index_1d][2][true_label_index]:.2f} | Prediction: {pred_label}:{pred_wrong[index_1d][2][pred_label_index]:.2f}"
+            )
+            ax[rows][cols].axis("off")
+    plt.savefig(f"{MODEL_NAME}_wrong_pred.png")
+    plt.show()
+elif len(pred_wrong) == 1:
+    image, true_label_index = test_data_original[pred_wrong[0][3]]
+    true_label = class_names[true_label_index]
+    pred_label_index = pred_wrong[0][1]
+    pred_label = class_names[pred_label_index]
+    plt.imshow(image)
+
+    plt.title(
+        f"True: {true_label}:{pred_wrong[0][2][true_label_index]:.2f} | Prediction: {pred_label}:{pred_wrong[0][2][pred_label_index]:.2f}"
+    )
+    plt.axis(False)
+    plt.savefig(f"{MODEL_NAME}_wrong_pred.png")
+    plt.show()