First commit

app.py

@@ -0,0 +1,83 @@
+import torch, torchvision
+from torchvision import transforms
+import numpy as np
+import gradio as gr
+from PIL import Image
+from pytorch_grad_cam import GradCAM
+from pytorch_grad_cam.utils.image import show_cam_on_image
+from resnet import ResNet18
+import gradio as gr
+
+model = ResNet18()
+model.load_state_dict(torch.load("model.ckpt", map_location=torch.device('cpu')), strict=False)
+
+inv_normalize = transforms.Normalize(
+    mean=[-0.50/0.23, -0.50/0.23, -0.50/0.23],
+    std=[1/0.23, 1/0.23, 1/0.23]
+)
+classes = ('plane', 'car', 'bird', 'cat', 'deer',
+           'dog', 'frog', 'horse', 'ship', 'truck')
+
+def resize_image_pil(image, new_width, new_height):
+
+    # Convert to PIL image
+    img = Image.fromarray(np.array(image))
+    
+    # Get original size
+    width, height = img.size
+
+    # Calculate scale
+    width_scale = new_width / width
+    height_scale = new_height / height 
+    scale = min(width_scale, height_scale)
+
+    # Resize
+    resized = img.resize((int(width*scale), int(height*scale)), Image.NEAREST)
+    
+    # Crop to exact size
+    resized = resized.crop((0, 0, new_width, new_height))
+
+    return resized
+
+def inference(input_img, transparency = 0.5, target_layer_number = -1):
+    input_img = resize_image_pil(input_img, 32, 32)
+    
+    input_img = np.array(input_img)
+    org_img = input_img
+    input_img = input_img.reshape((32, 32, 3))
+    transform = transforms.ToTensor()
+    input_img = transform(input_img)
+    input_img = input_img
+    input_img = input_img.unsqueeze(0)
+    outputs = model(input_img)
+    softmax = torch.nn.Softmax(dim=0)
+    o = softmax(outputs.flatten())
+    confidences = {classes[i]: float(o[i]) for i in range(10)}
+    _, prediction = torch.max(outputs, 1)
+    target_layers = [model.layer2[target_layer_number]]
+    cam = GradCAM(model=model, target_layers=target_layers)
+    grayscale_cam = cam(input_tensor=input_img, targets=None)
+    grayscale_cam = grayscale_cam[0, :]
+    visualization = show_cam_on_image(org_img/255, grayscale_cam, use_rgb=True, image_weight=transparency)
+    return classes[prediction[0].item()], visualization, confidences
+
+title = "CIFAR10 trained on ResNet18 Model with GradCAM"
+description = "A simple Gradio interface to infer on ResNet model, and get GradCAM results"
+examples = [["cat.jpg", 0.5, -1], ["dog.jpg", 0.5, -1]]
+demo = gr.Interface(
+    inference, 
+    inputs = [
+        gr.Image(width=256, height=256, label="Input Image"), gr.Slider
+        (0, 1, value = 0.5, label="Overall Opacity of Image"), 
+        gr.Slider(-2, -1, value = -2, step=1, label="Which Layer?")
+        ], 
+    outputs = [
+        "text", 
+        gr.Image(width=256, height=256, label="Output"),
+        gr.Label(num_top_classes=3)
+        ],
+    title = title,
+    description = description,
+    examples = examples,
+)
+demo.launch()

model.ckpt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:940f21f828787740b7b275a45b29051806977b52570b4e2afbb50a3f1dd04cab
+size 89492032

requirements.txt

@@ -0,0 +1,116 @@
+aiofiles==23.2.1
+aiohttp==3.9.5
+aiosignal==1.3.1
+altair==5.3.0
+annotated-types==0.6.0
+anyio==4.3.0
+asttokens==2.4.1
+attrs==23.2.0
+certifi==2024.2.2
+charset-normalizer==3.3.2
+click==8.1.7
+colorama==0.4.6
+comm==0.2.2
+contourpy==1.2.1
+cycler==0.12.1
+debugpy==1.8.1
+decorator==5.1.1
+executing==2.0.1
+fastapi==0.110.2
+ffmpy==0.3.2
+filelock==3.13.1
+fonttools==4.51.0
+frozenlist==1.4.1
+fsspec==2024.2.0
+grad-cam==1.5.0
+gradio==4.28.3
+gradio_client==0.16.0
+h11==0.14.0
+httpcore==1.0.5
+httpx==0.27.0
+huggingface-hub==0.22.2
+idna==3.7
+importlib_resources==6.4.0
+intel-openmp==2021.4.0
+ipykernel==6.29.4
+ipython==8.24.0
+jedi==0.19.1
+Jinja2==3.1.3
+joblib==1.4.0
+jsonschema==4.21.1
+jsonschema-specifications==2023.12.1
+jupyter_client==8.6.1
+jupyter_core==5.7.2
+kiwisolver==1.4.5
+lightning==2.2.3
+lightning-utilities==0.11.2
+markdown-it-py==3.0.0
+MarkupSafe==2.1.5
+matplotlib==3.8.4
+matplotlib-inline==0.1.7
+mdurl==0.1.2
+mkl==2021.4.0
+mpmath==1.3.0
+multidict==6.0.5
+nest-asyncio==1.6.0
+networkx==3.2.1
+numpy==1.26.3
+opencv-python==4.9.0.80
+orjson==3.10.1
+packaging==24.0
+pandas==2.2.2
+parso==0.8.4
+pillow==10.2.0
+platformdirs==4.2.1
+prompt-toolkit==3.0.43
+psutil==5.9.8
+pure-eval==0.2.2
+pydantic==2.7.1
+pydantic_core==2.18.2
+pydub==0.25.1
+Pygments==2.17.2
+pyparsing==3.1.2
+python-dateutil==2.9.0.post0
+python-multipart==0.0.9
+pytorch-lightning==2.2.3
+pytz==2024.1
+pywin32==306
+PyYAML==6.0.1
+pyzmq==26.0.2
+referencing==0.35.0
+requests==2.31.0
+rich==13.7.1
+rpds-py==0.18.0
+ruff==0.4.2
+scikit-learn==1.4.2
+scipy==1.13.0
+semantic-version==2.10.0
+setuptools==69.5.1
+shellingham==1.5.4
+six==1.16.0
+sniffio==1.3.1
+stack-data==0.6.3
+starlette==0.37.2
+sympy==1.12
+tbb==2021.11.0
+threadpoolctl==3.4.0
+tomlkit==0.12.0
+toolz==0.12.1
+torch==2.3.0+cu121
+torch-lr-finder==0.2.1
+torchaudio==2.3.0+cu121
+torchmetrics==1.3.2
+torchsummary==1.5.1
+torchvision==0.18.0+cu121
+tornado==6.4
+tqdm==4.66.2
+traitlets==5.14.3
+ttach==0.0.3
+typer==0.12.3
+typing_extensions==4.9.0
+tzdata==2024.1
+urllib3==2.2.1
+uvicorn==0.29.0
+wcwidth==0.2.13
+websockets==11.0.3
+yarl==1.9.4

resnet.py

@@ -0,0 +1,122 @@
+"""
+ResNet in PyTorch.
+For Pre-activation ResNet, see 'preact_resnet.py'.
+
+Reference:
+[1] Kaiming He, Xiangyu Zhang, Shaoqing Ren, Jian Sun
+    Deep Residual Learning for Image Recognition. arXiv:1512.03385
+"""
+import os
+import torch
+import utils
+import torch.nn as nn
+import torch.nn.functional as F
+
+from torchmetrics import Accuracy
+from torchvision.datasets import CIFAR10
+from pytorch_lightning import LightningModule
+from torch.utils.data import DataLoader, random_split
+
+
+class BasicBlock(nn.Module):
+    expansion = 1
+
+    def __init__(self, in_planes, planes, stride=1):
+        super(BasicBlock, self).__init__()
+        self.conv1 = nn.Conv2d(in_planes, planes, kernel_size=3, stride=stride, padding=1, bias=False)
+        self.bn1 = nn.BatchNorm2d(planes)
+        self.conv2 = nn.Conv2d(planes, planes, kernel_size=3, stride=1, padding=1, bias=False)
+        self.bn2 = nn.BatchNorm2d(planes)
+
+        self.shortcut = nn.Sequential()
+        if stride != 1 or in_planes != self.expansion*planes:
+            self.shortcut = nn.Sequential(
+                nn.Conv2d(in_planes, self.expansion*planes, kernel_size=1, stride=stride, bias=False),
+                nn.BatchNorm2d(self.expansion*planes)
+            )
+
+    def forward(self, x):
+        out = F.relu(self.bn1(self.conv1(x)))
+        out = self.bn2(self.conv2(out))
+        out += self.shortcut(x)
+        out = F.relu(out)
+        return out
+
+
+class ResNet(LightningModule):
+    def __init__(self, block, num_blocks, num_classes=10, loss='cross_entropy', learning_rate=2e-4, momentum=0.9, optimizer="SGD",
+                 epochs=20):
+        super(ResNet, self).__init__()
+        self.in_planes = 64
+
+        self.conv1 = nn.Conv2d(3, 64, kernel_size=3, stride=1, padding=1, bias=False)
+        self.bn1 = nn.BatchNorm2d(64)
+        self.layer1 = self._make_layer(block, 64, num_blocks[0], stride=1)
+        self.layer2 = self._make_layer(block, 128, num_blocks[1], stride=2)
+        self.layer3 = self._make_layer(block, 256, num_blocks[2], stride=2)
+        self.layer4 = self._make_layer(block, 512, num_blocks[3], stride=2)
+        self.linear = nn.Linear(512*block.expansion, num_classes)
+        self.accuracy = Accuracy(task="multiclass", num_classes=num_classes)
+        self.learning_rate = learning_rate
+        self.optimizer = optimizer
+        self.momentum = momentum
+        self.loss = utils.get_criterion(loss)
+        self.epochs = epochs
+
+    def _make_layer(self, block, planes, num_blocks, stride):
+        strides = [stride] + [1]*(num_blocks-1)
+        layers = []
+        for stride in strides:
+            layers.append(block(self.in_planes, planes, stride))
+            self.in_planes = planes * block.expansion
+        return nn.Sequential(*layers)
+
+    def forward(self, x):
+        out = F.relu(self.bn1(self.conv1(x)))
+        out = self.layer1(out)
+        out = self.layer2(out)
+        out = self.layer3(out)
+        out = self.layer4(out)
+        out = F.avg_pool2d(out, 4)
+        out = out.view(out.size(0), -1)
+        out = self.linear(out)
+        return out
+    
+    def training_step(self, batch, batch_idx):
+        x, y = batch
+        loss = self.loss(self(x), y)
+        return loss
+    
+    def validation_step(self, batch, batch_idx):
+        x, y = batch
+        logits = self(x)
+        loss = self.loss(logits, y)
+        preds = torch.argmax(logits, dim=1)
+        self.accuracy(preds, y)
+
+        # Calling self.log will surface up scalars for you in TensorBoard
+        self.log("val_loss", loss, prog_bar=True)
+        self.log("val_acc", self.accuracy, prog_bar=True)
+        return loss
+
+    def test_step(self, batch, batch_idx):
+        # Here we just reuse the validation_step for testing
+        return self.validation_step(batch, batch_idx)
+    
+    def configure_optimizers(self):
+        optimizer = utils.get_optimizer(self, lr=self.learning_rate, momentum=self.momentum, optimizer_type="SGD")
+        max_lr = utils.get_learning_rate(self, optimizer, self.loss, self.trainer.datamodule.train_dataloader())
+        scheduler = utils.get_OneCycleLR_scheduler(optimizer, max_lr=max_lr,  epochs=self.epochs,
+                                           steps_per_epoch=len(self.trainer.datamodule.train_dataloader()), max_at_epoch=5,
+                                           anneal_strategy = 'linear', div_factor=10,
+                                           final_div_factor=1)
+        return [optimizer],[{"scheduler": scheduler, "interval": "step", "frequency": 1}]
+
+def ResNet18(loss='cross_entropy', learning_rate=2e-4, momentum=0.9, optimizer="SGD", epochs=20):
+    return ResNet(BasicBlock, [2, 2, 2, 2], loss=loss, learning_rate=learning_rate, momentum=momentum,
+                  optimizer=optimizer, epochs=epochs)
+
+
+def ResNet34(loss='cross_entropy', learning_rate=2e-4, momentum=0.9, optimizer="SGD", epochs=20):
+    return ResNet(BasicBlock, [3, 4, 6, 3], loss=loss, learning_rate=learning_rate, momentum=momentum,
+                  optimizer=optimizer, epochs=epochs)

utils.py

@@ -0,0 +1,237 @@
+#!/usr/bin/env python3
+"""
+Utility Script containing functions to be used for training
+Author: Shilpaj Bhalerao
+"""
+# Standard Library Imports
+import math
+from typing import NoReturn
+
+# Third-Party Imports
+import numpy as np
+import matplotlib.pyplot as plt
+import torch
+from torchsummary import summary
+from torchvision import transforms
+from pytorch_grad_cam import GradCAM
+from pytorch_grad_cam.utils.image import show_cam_on_image
+
+import torch.optim as optim
+import torch.nn.functional as F
+from torch_lr_finder import LRFinder
+
+
+def get_summary(model, input_size: tuple) -> NoReturn:
+    """
+    Function to get the summary of the model architecture
+    :param model: Object of model architecture class
+    :param input_size: Input data shape (Channels, Height, Width)
+    """
+    use_cuda = torch.cuda.is_available()
+    device = torch.device("cuda" if use_cuda else "cpu")
+    network = model.to(device)
+    summary(network, input_size=input_size)
+
+
+def get_misclassified_data(model, device, test_loader):
+    """
+    Function to run the model on test set and return misclassified images
+    :param model: Network Architecture
+    :param device: CPU/GPU
+    :param test_loader: DataLoader for test set
+    """
+    # Prepare the model for evaluation i.e. drop the dropout layer
+    model.eval()
+    model.to(device)
+
+    # List to store misclassified Images
+    misclassified_data = []
+
+    # Reset the gradients
+    with torch.no_grad():
+        # Extract images, labels in a batch
+        for data, target in test_loader:
+
+            # Migrate the data to the device
+            data, target = data.to(device), target.to(device)
+
+            # Extract single image, label from the batch
+            for image, label in zip(data, target):
+
+                # Add batch dimension to the image
+                image = image.unsqueeze(0)
+
+                # Get the model prediction on the image
+                output = model(image)
+
+                # Convert the output from one-hot encoding to a value
+                pred = output.argmax(dim=1, keepdim=True)
+
+                # If prediction is incorrect, append the data
+                if pred != label:
+                    misclassified_data.append((image, label, pred))
+    return misclassified_data
+
+
+# -------------------- GradCam --------------------
+def display_gradcam_output(data: list,
+                           classes: list[str],
+                           inv_normalize: transforms.Normalize,
+                           model,
+                           target_layers,
+                           targets=None,
+                           number_of_samples: int = 10,
+                           transparency: float = 0.60):
+    """
+    Function to visualize GradCam output on the data
+    :param data: List[Tuple(image, label)]
+    :param classes: Name of classes in the dataset
+    :param inv_normalize: Mean and Standard deviation values of the dataset
+    :param model: Model architecture
+    :param target_layers: Layers on which GradCam should be executed
+    :param targets: Classes to be focused on for GradCam
+    :param number_of_samples: Number of images to print
+    :param transparency: Weight of Normal image when mixed with activations
+    """
+    # Plot configuration
+    fig = plt.figure(figsize=(10, 10))
+    x_count = 5
+    y_count = 1 if number_of_samples <= 5 else math.floor(number_of_samples / x_count)
+
+    # Create an object for GradCam
+    cam = GradCAM(model=model, target_layers=target_layers, use_cuda=True)
+
+    # Iterate over number of specified images
+    for i in range(number_of_samples):
+        plt.subplot(y_count, x_count, i + 1)
+        input_tensor = data[i][0]
+
+        # Get the activations of the layer for the images
+        grayscale_cam = cam(input_tensor=input_tensor, targets=targets)
+        grayscale_cam = grayscale_cam[0, :]
+
+        # Get back the original image
+        img = input_tensor.squeeze(0).to('cpu')
+        img = inv_normalize(img)
+        rgb_img = np.transpose(img, (1, 2, 0))
+        rgb_img = rgb_img.numpy()
+
+        # Mix the activations on the original image
+        visualization = show_cam_on_image(rgb_img, grayscale_cam, use_rgb=True, image_weight=transparency)
+
+        # Display the images on the plot
+        plt.imshow(visualization)
+        plt.title(r"Correct: " + classes[data[i][1].item()] + '\n' + 'Output: ' + classes[data[i][2].item()])
+        plt.xticks([])
+        plt.yticks([])
+
+
+def get_optimizer(model, lr, momentum=0, weight_decay=0, optimizer_type='SGD'):
+    """Method to get object of stochastic gradient descent. Used to update weights.
+
+    Args:
+        model (Object): Neural Network model
+        lr (float): Value of learning rate
+        momentum (float): Value of momentum
+        weight_decay (float): Value of weight decay
+        optimizer_type (str): Type of optimizer SGD or ADAM
+
+    Returns:
+        object: Object of optimizer class to update weights
+    """
+    if optimizer_type == 'SGD':
+        optimizer = optim.SGD(model.parameters(), lr=lr, momentum=momentum)
+    elif optimizer_type == 'ADAM':
+        optimizer = optim.Adam(model.parameters(), lr=lr, weight_decay=weight_decay)
+    return optimizer
+
+def get_StepLR_scheduler(optimizer, step_size, gamma):
+    """Method to get object of scheduler class. Used to update learning rate
+
+    Args:
+        optimizer (Object): Object of optimizer
+        step_size (int): Period of learning rate decay
+        gamma (float): Number to multiply with learning rate
+
+    Returns:
+        object: Object of StepLR class to update learning rate
+    """
+    scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=step_size, gamma=gamma, verbose=True)
+    return scheduler
+
+def get_ReduceLROnPlateau_scheduler(optimizer, factor, patience):
+    """Method to get object of scheduler class. Used to update learning rate
+
+    Args:
+        optimizer (Object): Object of optimizer
+        factor (float): Number to multiply with learning rate
+        patience (int): Number of epoch to wait
+
+    Returns:
+        object: Object of StepLR class to update learning rate
+    """
+    scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, factor=factor, patience=patience, verbose=True)
+    return scheduler
+
+def get_OneCycleLR_scheduler(optimizer, max_lr, epochs, steps_per_epoch, max_at_epoch, anneal_strategy, div_factor, final_div_factor):
+    """Method to get object of scheduler class. Used to update learning rate
+
+    Args:
+        optimizer (Object): Object of optimizer
+        max_lr (float): Maximum learning rate to reach during training
+        epochs (float): Total number of epoch
+        steps_per_epoch (int): Total steps in an epoch
+        max_at_epoch (int): Epoch to reach maximum learning rate
+        anneal_strategy (string): Strategy to interpolate between minimum and maximum lr
+        div_factor (int): Divisive factor to calculate intial learning rate
+        final_div_factor (int): Divisive factor to calculate minimum learning rate
+
+    Returns:
+        object: Object of StepLR class to update learning rate
+    """
+    scheduler = optim.lr_scheduler.OneCycleLR(optimizer, max_lr=max_lr,  epochs=epochs,
+                                              steps_per_epoch=steps_per_epoch, 
+                                              pct_start=max_at_epoch/epochs,
+                                              anneal_strategy=anneal_strategy,
+                                              div_factor=div_factor,
+                                              final_div_factor=final_div_factor)
+    return scheduler
+
+def get_criterion(loss_type='cross_entropy'):
+    """Method to get loss calculation ctiterion
+
+    Args:
+        loss_type (str): Type of loss 'nll_loss' or 'cross_entropy' loss
+
+    Returns:
+        object: Object to calculate loss 
+    """
+    if loss_type == 'nll_loss':
+        criterion = F.nll_loss
+    elif loss_type == 'cross_entropy':
+        criterion = F.cross_entropy
+    return criterion
+
+def get_learning_rate(model, optimizer, criterion, trainloader):
+    """Method to find learning rate using LR finder.
+
+    Args:
+        model (Object): Object of model
+        optimizer (Object): Object of optimizer class
+        criterion (Object): Loss function
+        trainloader (Object): Object of dataloader class
+
+    Returns:
+        float: Learning rate suggested by lr finder
+    """
+    # Create object and perform range test
+    lr_finder = LRFinder(model, optimizer, criterion)
+    lr_finder.range_test(trainloader, end_lr=100, num_iter=100)
+
+    # Plot result and store suggested lr
+    plot, suggested_lr = lr_finder.plot()
+
+    # Reset model and optimizer
+    lr_finder.reset()
+
+    return suggested_lr