Skip to content
Snippets Groups Projects
Select Git revision
  • 1267ecb7fdaee73c8f66d15a29c1016b646caafe
  • master default protected
2 results

test_lifoqueue.py

Blame
  • Forked from Vuillemot Romain / INF-TC1
    Source project has a limited visibility.
    ML_TD7.py 17.48 KiB
    """ML-TD7.ipynb
    
    ### **_Introduction au Machine Learning - Enise - Centrale Lyon_**
    
    2024-2025
    
    Emmanuel Dellandréa
    
    # TD7 – Convolutional Neural Networks
    
    The objective of this tutorial is to use the PyTorch library for building, training, and evaluating CNN models.
    
    ## Sequence 1: Training a CNN to classify CIFAR10 images
    
    The goal is to apply a Convolutional Neural Net (CNN) model on the CIFAR10 image dataset and test the accuracy of the model on the basis of image classification.
    
    Be sure to check the PyTorch tutorials and documentation when needed:
    
    https://pytorch.org/tutorials/
    
    https://pytorch.org/docs/stable/index.html
    
    You can test if GPU is available on your machine and thus train on it to speed up the process
    """
    
    import torch
    
    # check if CUDA is available
    train_on_gpu = torch.cuda.is_available()
    
    if not train_on_gpu:
        print("CUDA is not available.  Training on CPU ...")
    else:
        print("CUDA is available!  Training on GPU ...")
    
    """Next we load the CIFAR10 dataset"""
    
    import numpy as np
    from torchvision import datasets, transforms
    from torch.utils.data.sampler import SubsetRandomSampler
    
    # number of subprocesses to use for data loading
    num_workers = 0
    # how many samples per batch to load
    batch_size = 20
    # percentage of training set to use as validation
    valid_size = 0.2
    
    # convert data to a normalized torch.FloatTensor
    transform = transforms.Compose(
        [transforms.ToTensor(), transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))]
    )
    
    # choose the training and test datasets
    train_data = datasets.CIFAR10("data", train=True, download=True, transform=transform)
    test_data = datasets.CIFAR10("data", train=False, download=True, transform=transform)
    
    # obtain training indices that will be used for validation
    num_train = len(train_data)
    indices = list(range(num_train))
    np.random.shuffle(indices)
    split = int(np.floor(valid_size * num_train))
    train_idx, valid_idx = indices[split:], indices[:split]
    
    # define samplers for obtaining training and validation batches
    train_sampler = SubsetRandomSampler(train_idx)
    valid_sampler = SubsetRandomSampler(valid_idx)
    
    # prepare data loaders (combine dataset and sampler)
    train_loader = torch.utils.data.DataLoader(
        train_data, batch_size=batch_size, sampler=train_sampler, num_workers=num_workers
    )
    valid_loader = torch.utils.data.DataLoader(
        train_data, batch_size=batch_size, sampler=valid_sampler, num_workers=num_workers
    )
    test_loader = torch.utils.data.DataLoader(
        test_data, batch_size=batch_size, num_workers=num_workers
    )
    
    # specify the image classes
    classes = [
        "airplane",
        "automobile",
        "bird",
        "cat",
        "deer",
        "dog",
        "frog",
        "horse",
        "ship",
        "truck",
    ]
    
    """CNN definition (this one is an example)"""
    
    import torch.nn as nn
    import torch.nn.functional as F
    
    # define the CNN architecture
    
    
    class Net(nn.Module):
        def __init__(self):
            super(Net, self).__init__()
            self.conv1 = nn.Conv2d(3, 6, 5)
            self.pool = nn.MaxPool2d(2, 2)
            self.conv2 = nn.Conv2d(6, 16, 5)
            self.fc1 = nn.Linear(16 * 5 * 5, 120)
            self.fc2 = nn.Linear(120, 84)
            self.fc3 = nn.Linear(84, 10)
    
        def forward(self, x):
            x = self.pool(F.relu(self.conv1(x)))
            x = self.pool(F.relu(self.conv2(x)))
            x = x.view(-1, 16 * 5 * 5)
            x = F.relu(self.fc1(x))
            x = F.relu(self.fc2(x))
            x = self.fc3(x)
            return x
    
    
    # create a complete CNN
    model = Net()
    print(model)
    # move tensors to GPU if CUDA is available
    if train_on_gpu:
        model.cuda()
    
    """Loss function and training using SGD (Stochastic Gradient Descent) optimizer"""
    
    import torch.optim as optim
    
    criterion = nn.CrossEntropyLoss()  # specify loss function
    optimizer = optim.SGD(model.parameters(), lr=0.01)  # specify optimizer
    
    n_epochs = 30  # number of epochs to train the model
    train_loss_list = []  # list to store loss to visualize
    valid_loss_min = np.Inf  # track change in validation loss
    
    for epoch in range(n_epochs):
        # Keep track of training and validation loss
        train_loss = 0.0
        valid_loss = 0.0
    
        # Train the model
        model.train()
        for data, target in train_loader:
            # Move tensors to GPU if CUDA is available
            if train_on_gpu:
                data, target = data.cuda(), target.cuda()
            # Clear the gradients of all optimized variables
            optimizer.zero_grad()
            # Forward pass: compute predicted outputs by passing inputs to the model
            output = model(data)
            # Calculate the batch loss
            loss = criterion(output, target)
            # Backward pass: compute gradient of the loss with respect to model parameters
            loss.backward()
            # Perform a single optimization step (parameter update)
            optimizer.step()
            # Update training loss
            train_loss += loss.item() * data.size(0)
    
        # Validate the model
        model.eval()
        for data, target in valid_loader:
            # Move tensors to GPU if CUDA is available
            if train_on_gpu:
                data, target = data.cuda(), target.cuda()
            # Forward pass: compute predicted outputs by passing inputs to the model
            output = model(data)
            # Calculate the batch loss
            loss = criterion(output, target)
            # Update average validation loss
            valid_loss += loss.item() * data.size(0)
    
        # Calculate average losses
        train_loss = train_loss / len(train_loader)
        valid_loss = valid_loss / len(valid_loader)
        train_loss_list.append(train_loss)
    
        # Print training/validation statistics
        print(
            "Epoch: {} \tTraining Loss: {:.6f} \tValidation Loss: {:.6f}".format(
                epoch, train_loss, valid_loss
            )
        )
    
        # Save model if validation loss has decreased
        if valid_loss <= valid_loss_min:
            print(
                "Validation loss decreased ({:.6f} --> {:.6f}).  Saving model ...".format(
                    valid_loss_min, valid_loss
                )
            )
            torch.save(model.state_dict(), "model_cifar.pt")
            valid_loss_min = valid_loss
    
    """Does overfit occur? If so, do an early stopping."""
    
    import matplotlib.pyplot as plt
    
    plt.plot(range((len(train_loss_list))), train_loss_list)
    plt.xlabel("Epoch")
    plt.ylabel("Loss")
    plt.title("Performance of Model 1")
    plt.show()
    
    """Now loading the model with the lowest validation loss value
    
    """
    
    # Commented out IPython magic to ensure Python compatibility.
    model.load_state_dict(torch.load("./model_cifar.pt"))
    
    # track test loss
    test_loss = 0.0
    class_correct = list(0.0 for i in range(10))
    class_total = list(0.0 for i in range(10))
    
    model.eval()
    # iterate over test data
    for data, target in test_loader:
        # move tensors to GPU if CUDA is available
        if train_on_gpu:
            data, target = data.cuda(), target.cuda()
        # forward pass: compute predicted outputs by passing inputs to the model
        output = model(data)
        # calculate the batch loss
        loss = criterion(output, target)
        # update test loss
        test_loss += loss.item() * data.size(0)
        # convert output probabilities to predicted class
        _, pred = torch.max(output, 1)
        # compare predictions to true label
        correct_tensor = pred.eq(target.data.view_as(pred))
        correct = (
            np.squeeze(correct_tensor.numpy())
            if not train_on_gpu
            else np.squeeze(correct_tensor.cpu().numpy())
        )
        # calculate test accuracy for each object class
        for i in range(batch_size):
            label = target.data[i]
            class_correct[label] += correct[i].item()
            class_total[label] += 1
    
    # average test loss
    test_loss = test_loss / len(test_loader)
    print("Test Loss: {:.6f}\n".format(test_loss))
    
    for i in range(10):
        if class_total[i] > 0:
            print(
                "Test Accuracy of %5s: %2d%% (%2d/%2d)"
    #             % (
                    classes[i],
                    100 * class_correct[i] / class_total[i],
                    np.sum(class_correct[i]),
                    np.sum(class_total[i]),
                )
            )
        else:
            print("Test Accuracy of %5s: N/A (no training examples)" % (classes[i]))
    
    print(
        "\nTest Accuracy (Overall): %2d%% (%2d/%2d)"
    #     % (
            100.0 * np.sum(class_correct) / np.sum(class_total),
            np.sum(class_correct),
            np.sum(class_total),
        )
    )
    
    """### Experiments:
    
    Build a new network with the following structure.
    
    - It has 3 convolutional layers of kernel size 3 and padding of 1.
    - The first convolutional layer must output 16 channels, the second 32 and the third 64.
    - At each convolutional layer output, we apply a ReLU activation then a MaxPool with kernel size of 2.
    - Then, three fully connected layers, the first two being followed by a ReLU activation.
    - The first fully connected layer will have an output size of 512.
    - The second fully connected layer will have an output size of 64.
    
    Compare the results obtained with this new network to those obtained previously.
    
    ## Sequence 2: Working with pre-trained models.
    
    PyTorch offers several pre-trained models https://pytorch.org/vision/0.8/models.html        
    We will use ResNet50 trained on ImageNet dataset (https://www.image-net.org/index.php). Use the following code with the files `imagenet-simple-labels.json` that contains the imagenet labels and the image dog.png that we will use as test.
    """
    
    import json
    from PIL import Image
    from torchvision import models
    
    # Choose an image to pass through the model
    test_image = "dog.png"
    
    # Configure matplotlib for pretty inline plots
    #%matplotlib inline
    #%config InlineBackend.figure_format = 'retina'
    
    # Prepare the labels
    with open("imagenet-simple-labels.json") as f:
        labels = json.load(f)
    
    # First prepare the transformations: resize the image to what the model was trained on and convert it to a tensor
    data_transform = transforms.Compose(
        [
            transforms.Resize((224, 224)),
            transforms.ToTensor(),
            transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]),
        ]
    )
    # Load the image
    
    image = Image.open(test_image)
    plt.imshow(image), plt.xticks([]), plt.yticks([])
    
    # Now apply the transformation, expand the batch dimension, and send the image to the GPU
    # image = data_transform(image).unsqueeze(0).cuda()
    image = data_transform(image).unsqueeze(0)
    
    # Download the model if it's not there already. It will take a bit on the first run, after that it's fast
    model = models.resnet50(pretrained=True)
    # Send the model to the GPU
    # model.cuda()
    # Set layers such as dropout and batchnorm in evaluation mode
    model.eval()
    
    # Get the 1000-dimensional model output
    out = model(image)
    # Find the predicted class
    print("Predicted class is: {}".format(labels[out.argmax()]))
    
    """### Experiments:
    
    Study the code and the results obtained. Possibly add other images downloaded from the internet.
    
    Experiment with other pre-trained CNN models.
    
    ## Sequence 3: Transfer Learning
        
        
    For this work, we will use a pre-trained model (ResNet18) as a descriptor extractor and will refine the classification by training only the last fully connected layer of the network. Thus, the output layer of the pre-trained network will be replaced by a layer adapted to the new classes to be recognized which will be in our case ants and bees.
    Download and unzip in your working directory the dataset available at the address :
        
    https://download.pytorch.org/tutorial/hymenoptera_data.zip
        
    Execute the following code in order to display some images of the dataset.
    """
    
    import os
    
    import matplotlib.pyplot as plt
    import numpy as np
    import torch
    import torchvision
    from torchvision import datasets, transforms
    import torch.nn as nn
    import torch.optim as optim
    from torch.optim import lr_scheduler
    
    # Data augmentation and normalization for training
    # Just normalization for validation
    data_transforms = {
        "train": transforms.Compose(
            [
                transforms.RandomResizedCrop(
                    224
                ),  # ImageNet models were trained on 224x224 images
                transforms.RandomHorizontalFlip(),  # flip horizontally 50% of the time - increases train set variability
                transforms.ToTensor(),  # convert it to a PyTorch tensor
                transforms.Normalize(
                    [0.485, 0.456, 0.406], [0.229, 0.224, 0.225]
                ),  # ImageNet models expect this norm
            ]
        ),
        "val": transforms.Compose(
            [
                transforms.Resize(256),
                transforms.CenterCrop(224),
                transforms.ToTensor(),
                transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]),
            ]
        ),
    }
    
    data_dir = "hymenoptera_data"
    # Create train and validation datasets and loaders
    image_datasets = {
        x: datasets.ImageFolder(os.path.join(data_dir, x), data_transforms[x])
        for x in ["train", "val"]
    }
    dataloaders = {
        x: torch.utils.data.DataLoader(
            image_datasets[x], batch_size=4, shuffle=True, num_workers=0
        )
        for x in ["train", "val"]
    }
    dataset_sizes = {x: len(image_datasets[x]) for x in ["train", "val"]}
    class_names = image_datasets["train"].classes
    device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
    
    # Helper function for displaying images
    def imshow(inp, title=None):
        """Imshow for Tensor."""
        inp = inp.numpy().transpose((1, 2, 0))
        mean = np.array([0.485, 0.456, 0.406])
        std = np.array([0.229, 0.224, 0.225])
    
        # Un-normalize the images
        inp = std * inp + mean
        # Clip just in case
        inp = np.clip(inp, 0, 1)
        plt.imshow(inp)
        if title is not None:
            plt.title(title)
        plt.pause(0.001)  # pause a bit so that plots are updated
        plt.show()
    
    
    # Get a batch of training data
    inputs, classes = next(iter(dataloaders["train"]))
    
    # Make a grid from batch
    out = torchvision.utils.make_grid(inputs)
    
    imshow(out, title=[class_names[x] for x in classes])
    
    """Now, execute the following code which uses a pre-trained model ResNet18 having replaced the output layer for the ants/bees classification and performs the model training by only changing the weights of this output layer."""
    
    import copy
    import time
    
    
    def train_model(model, criterion, optimizer, scheduler, num_epochs=25):
        since = time.time()
    
        best_model_wts = copy.deepcopy(model.state_dict())
        best_acc = 0.0
    
        epoch_time = []  # we'll keep track of the time needed for each epoch
    
        for epoch in range(num_epochs):
            epoch_start = time.time()
            print("Epoch {}/{}".format(epoch + 1, num_epochs))
            print("-" * 10)
    
            # Each epoch has a training and validation phase
            for phase in ["train", "val"]:
                if phase == "train":
                    scheduler.step()
                    model.train()  # Set model to training mode
                else:
                    model.eval()  # Set model to evaluate mode
    
                running_loss = 0.0
                running_corrects = 0
    
                # Iterate over data.
                for inputs, labels in dataloaders[phase]:
                    inputs = inputs.to(device)
                    labels = labels.to(device)
    
                    # zero the parameter gradients
                    optimizer.zero_grad()
    
                    # Forward
                    # Track history if only in training phase
                    with torch.set_grad_enabled(phase == "train"):
                        outputs = model(inputs)
                        _, preds = torch.max(outputs, 1)
                        loss = criterion(outputs, labels)
    
                        # backward + optimize only if in training phase
                        if phase == "train":
                            loss.backward()
                            optimizer.step()
    
                    # Statistics
                    running_loss += loss.item() * inputs.size(0)
                    running_corrects += torch.sum(preds == labels.data)
    
                epoch_loss = running_loss / dataset_sizes[phase]
                epoch_acc = running_corrects.double() / dataset_sizes[phase]
    
                print("{} Loss: {:.4f} Acc: {:.4f}".format(phase, epoch_loss, epoch_acc))
    
                # Deep copy the model
                if phase == "val" and epoch_acc > best_acc:
                    best_acc = epoch_acc
                    best_model_wts = copy.deepcopy(model.state_dict())
    
            # Add the epoch time
            t_epoch = time.time() - epoch_start
            epoch_time.append(t_epoch)
            print()
    
        time_elapsed = time.time() - since
        print(
            "Training complete in {:.0f}m {:.0f}s".format(
                time_elapsed // 60, time_elapsed % 60
            )
        )
        print("Best val Acc: {:4f}".format(best_acc))
    
        # Load best model weights
        model.load_state_dict(best_model_wts)
        return model, epoch_time
    
    
    # Download a pre-trained ResNet18 model and freeze its weights
    model = torchvision.models.resnet18(pretrained=True)
    for param in model.parameters():
        param.requires_grad = False
    
    # Replace the final fully connected layer
    # Parameters of newly constructed modules have requires_grad=True by default
    num_ftrs = model.fc.in_features
    model.fc = nn.Linear(num_ftrs, 2)
    # Send the model to the GPU
    model = model.to(device)
    # Set the loss function
    criterion = nn.CrossEntropyLoss()
    
    # Observe that only the parameters of the final layer are being optimized
    optimizer_conv = optim.SGD(model.fc.parameters(), lr=0.001, momentum=0.9)
    exp_lr_scheduler = lr_scheduler.StepLR(optimizer_conv, step_size=7, gamma=0.1)
    model, epoch_time = train_model(
        model, criterion, optimizer_conv, exp_lr_scheduler, num_epochs=10
    )
    
    """### Experiments:
    
    Study the code and the results obtained.
    
    Modify the code and add an "eval_model" function to allow
    the evaluation of the model on a test set (different from the learning and validation sets used during the learning phase). Study the results obtained.
    
    Now modify the code to replace the current classification layer with a set of two layers using a "relu" activation function for the middle layer. Renew the experiments and study the results obtained.
    
    Experiment with other models and datasets.
    """