In [ ]:
import torch
import torchvision
import torchvision.transforms as transforms
import torch.nn as nn
import torch.optim as optim
import numpy as np
import matplotlib.pyplot as plt
from torch.utils.data import random_split, Dataset
Load data with pytorch
In [ ]:
# Set up config variables
config = {
‘data_path’: ‘./data’, # directory path of dataset,
}
In [ ]:
# Import MNIST digit dataset
def importMnistDataset(root, train, transform=[]):
# it will try to download the dataset if dataset is not found under the root directory
return torchvision.datasets.MNIST(root=root, train=train, download=True, transform=transform)
# ImportMnistDataset may fail to download because of HTTP error (happens a lot recently).
# We can manually download the datasets with the following comments,
# If data_path value is changed in config, update it here as well.
!wget -nc www.di.ens.fr/~lelarge/MNIST.tar.gz -P ./tmp
!mkdir -p ./data/
!tar -zxvf ./tmp/MNIST.tar.gz -C ./data/
In [ ]:
# After download finished, run the code below.
# Test dataset contains 10,000 images
# Train dataset contains 60,000 images
train_set = importMnistDataset(root=config[‘data_path’], train=True)
test_set = importMnistDataset(root=config[‘data_path’], train=False)
print(train_set)
print(test_set)
del test_set, train_set
In [ ]:
# To add noise to images, we can take advantage of the transform module.
# We have defined custom transform class Noise to add noise during image transformation
class Noise(object):
# Here we create a noisy version of the data set
# The way that we do it is we go over all the pixels of
# each of the data points; then with probability p we multiply
# the value of that pixel by 0 (making it essentially black).
# Otherwise (with probability 1-p) we multiply the value of that
# pixel by 1 (essentially keeping the pixel untouched)
# drop_probability is basically the probability of dropping a pixel (p in the above)
# This is how we create the noisy data set.
# Convert image_set to a numpy array
def __init__(self, drop_probability=0):
self.drop_probability = drop_probability
def __call__(self, tensor):
n = torch.from_numpy(np.random.choice([0, 1], size=tensor.size(), p=[self.drop_probability, 1-self.drop_probability]))
return tensor * n
def __repr__(self):
return self.__class__.__name__ + ‘(drop_probability={0})’.format(self.drop_probability)
# Return transform function to convert an image into a tensor.
# Add noise if drop_probability is provided.
def generateTransform(drop_probability):
if drop_probability is not None and drop_probability > 0:
trans_noise = transforms.Compose([
transforms.ToTensor(),
Noise(drop_probability)
])
return trans_noise
else:
return transforms.Compose([transforms.ToTensor()])
In the following we print some information about the data set. Then we will visualize some noisy digits.
In [ ]:
# Load training dataset and add noise
train_set = importMnistDataset(config[‘data_path’], True, generateTransform(drop_probability=0.7))
# Add dataset to pytorch DataLoader with mini-batch size 64 s.t. each batch contains 64 images.
train_loader = torch.utils.data.DataLoader(train_set, batch_size=64, shuffle=False)
# Get first batch of images and labels
train_image_batch, classe_set = iter(train_loader).next()
print(f’train_loader contains {len(train_loader)} batches of data.’)
print(f’train_image_batch has shape {train_image_batch.shape},’)
print(‘where 64 is the number of images in a batch, 1 is the number of image channels (1 for grayscale image),\
28X28 stands for WxH (width and height of a single image).’)
In [ ]:
def show_gray_digits(image_set, row=2, col=3):
# Here we visualize some of the data points in the data set.
# Create a large figure, to be filled with multiple subplots.
# Since image_set is a tensor variable, we transform it to a numpy type variable.
image_set = image_set.detach().numpy()
for i in range(row*col):
# define subplot
plt.subplot(row, col, i+1)
# plot raw pixel data
plt.imshow(image_set[i,0], cmap=plt.get_cmap(‘gray’))
# show the figure
plt.show()
In [ ]:
# Display noised images and their corresponding labels.
show_gray_digits(train_image_batch, 2, 3)
print(classe_set[0:6])
del train_image_batch, classe_set, train_set, train_loader
Let us formally define functions to load data. The goal is to train a network that, given a noisy image, recovers the original image. Therefore, each training point consists of the input (noisy image) and the expected output (true image). Also, we will use only a small portion of the training data to make the task more challenging.
In [ ]:
def load_data(path, drop_probability, split_ratio, batch_size):
# Import MNIST train and test datasets.
# Import train set without adding noise.
train_set = importMnistDataset(path, True, generateTransform(0))
# Use only the first 800 points for training
train_set = torch.utils.data.Subset(train_set, list(range(1, 800)))
# Import train set and add noise.
train_set_noise = importMnistDataset(path, True, generateTransform(drop_probability))
# Use only the first 800 points for training
train_set_noise = torch.utils.data.Subset(train_set_noise, list(range(1, 800)))
# Load the whole test dataset
test_set = importMnistDataset(path, False, generateTransform(0))
test_set_noise = importMnistDataset(path, False, generateTransform(drop_probability))
# Create a new dataset storing image pairs,
# an item in the dataset is a pair of images (original and noised).
train_set = PairDataset(train_set, train_set_noise)
test_set = PairDataset(test_set, test_set_noise)
# Further split train dataset into training and validation datasets.
# Use validation dataset to get early feedback without peeking test dataset.
train_set, val_set = split_dataSet(train_set, split_ratio)
# Generate train, validation and test dataloaders,
# Dataloader is used to loop through data batches.
train_loader = torch.utils.data.DataLoader(train_set, batch_size, shuffle=True)
val_loader = torch.utils.data.DataLoader(val_set, batch_size, shuffle=False)
test_loader = torch.utils.data.DataLoader(test_set, batch_size, shuffle=False)
return train_loader, val_loader, test_loader
In [ ]:
# Partitions a dataset into two sets based on the split ratio.
# e.g. 0.8 means 80% data in the first set and 20% in the second.
# This will be used to split the training data into “training” and “validation” sets
def split_dataSet(dataset, split_ratio):
a_size = int(split_ratio * len(dataset))
b_size = len(dataset) – a_size
a_set, b_set = random_split(dataset, [a_size, b_size])
return a_set, b_set
# When get an item from the dataset, it returns a pair of data.
# In our case, it returns image and corresponding noised image.
class PairDataset(Dataset):
def __init__(self, dataset_origin, dataset_noisy):
self.dataset_origin = dataset_origin
self.dataset_noisy = dataset_noisy
def __getitem__(self, index):
x1 = self.dataset_origin[index]
x2 = self.dataset_noisy[index]
return x1, x2
def __len__(self):
return len(self.dataset_origin)
Let us load all the data.
In [ ]:
# Load train, validation and test data.
train_loader, val_loader, test_loader = load_data(
path=config[‘data_path’], drop_probability=0.7, split_ratio=0.5, batch_size=64)
print(f’train_loader has {len(train_loader)} batches’)
print(f’val_loader has {len(val_loader)} batches’)
print(f’test_loader has {len(test_loader)} batches’)
In [ ]:
# print(np.ceil(800*0.5/64)) #7
# print(np.ceil(800*(1-0.5)/64)) #7
# print(np.ceil(10000/64)) #157
Let us visualize images in each dataloader.
In [ ]:
# Get a batch of images from each dataloader
train_image_batch, train_noise_image_batch = iter(train_loader).next()
val_image_batch, val_noise_image_batch = iter(val_loader).next()
test_image_batch, test_noise_image_batch = iter(test_loader).next()
# Show the first 3 image pairs
print(‘train’)
show_gray_digits(train_image_batch[0], row=1, col=3)
show_gray_digits(train_noise_image_batch[0], 1, 3)
print(‘validation’)
show_gray_digits(val_image_batch[0], 1, 3)
show_gray_digits(val_noise_image_batch[0], 1, 3)
print(‘test’)
show_gray_digits(test_image_batch[0], 1, 3)
show_gray_digits(test_noise_image_batch[0], 1, 3)
del train_loader, val_loader, test_loader, train_image_batch, val_image_batch, test_image_batch, train_noise_image_batch, val_noise_image_batch, test_noise_image_batch
Define our model
In [ ]:
class MyNeuralNetRegressor(nn.Module):
# We create a neural network with 1 hidden layers with 1000 neurons,
# and an output layer with 784 neurons (to match the dimension of our images!)
def __init__(self):
super(MyNeuralNetRegressor, self).__init__()
# These are the layers. fc1 is the first feed-forward/fully connedted layer etc
# Here we are just defining layers but the way that they will be put together is determined in the forward function.
self.fc1 = nn.Linear(28*28, 1000) # input dimension is 784 (28*28 is the number of pixels) and the layer has 1000 neurons
self.fc2 = nn.Linear(1000, 28*28) # input dimension is 1000 (from previous layer) and the layer has 784 neurons/
# We use 784 neurons so the output dimension matches the input dimension.
# This defines the forward pass of the network; the input is passed through each layer, and the nonlinearity is applied.
def forward(self, x):
x = x.view(-1, 28*28) # We ‘flatten’ the image to be of dimensions (#images, 784) instead of (#images, 1, 28, 28)
x = torch.relu(self.fc1(x)) # Pass the input through linear layer and apply ReLU to outputs.
x = torch.relu(self.fc2(x)) # This is the output layer (again, linear + ReLU)
x = x.view(x.shape[0],1,28,28) # ‘unflatten’ output data from dimensions (#images, 784) to (#images, 1, 28, 28)
return x
Main function
In [ ]:
# We can use the free GPUs offered by Colab
if torch.cuda.is_available():
device = torch.device(‘cuda’) #GPU if available (cuda represents GPU)
else:
device = torch.device(‘cpu’)
# Verifying which device is being used GPU vs. CPU
print(device)
In [ ]:
# A placeholder function.
# Implement the function when work on the Task1 Q1.
def plot_eval_results(train_report, test_loss):
print(‘!to be implemented’)
# Net is the NN model, e.g. MyNeuralNetRegressor.
# Criterion is the loss function we use.
# Epochs is the number of times that model will be trained with the whole dataset,
# it controls the outter loop of the training process.
# Learning rate is the step size of optimization algorithm (stochastic gradient descent e.g.).
# We may fail to learn if learning rate is too big or too small.
# You may try play with different rate values, e.g. 1, 0.1, 0.001, 0.0001.
def main(config, net, criterion, drop_probability=0.7, epochs=10, batch_size=64, split_ratio=0.5, learning_rate=0.05):
# Here the task is to “denoise” images. In particular,
# the input of the regression algorithm is a noisy
# picture of a digit, and the regressor should reconstruct
# the original image of the digit. To do the training,
# we create a data set where the inputs are noisy
# digits and the outputs are the original images.
# The learning method is then tested on new unseen
# noisy digits. This is done using the test set.
#create a fresh model
NeuralNet = net()
# Define optimizer for learning weights(parameters).
# Adam and SGD are two commonly used optimizers.
# We use Adam here but you can update it to SGD.
optimizer = optim.Adam(NeuralNet.parameters(), lr=learning_rate)
train_loader, val_loader, test_loader = load_data(path=config[‘data_path’],
drop_probability=drop_probability, split_ratio=split_ratio, batch_size=batch_size)
print(f’Successfully loaded data’)
# Train the network.
# Return the network with updated weights(parameters).
# train_report contains information for analyzing train/validation loss.
NeuralNet, train_report = train(train_loader, val_loader, NeuralNet, epochs, criterion, optimizer, config[‘model_path’])
print(‘Training finished’)
# Load the best model.
NeuralNet.load_state_dict(torch.load(config[‘model_path’]))
# Validate with test dataset and return test loss.
test_loss = test(test_loader, NeuralNet, criterion)
print(f’Test loss is {test_loss} for drop_probability={drop_probability}, epochs={epochs}, batch_size={batch_size}, split_ratio={split_ratio}, learning_rate={learning_rate}’)
# Visualize the train/validation/test loss.
plot_eval_results(train_report, test_loss)
return NeuralNet
Train model
In [ ]:
# This is our training function.
# train_loader contains training image pairs.
# val_loader stores validation image pairs.
def train(train_loader, val_loader, model, epochs, loss_function, optimizer, model_path, print_loss=True):
best_loss = None
report = []
# Loop over the entire dataset #epochs times.
for epoch in range(epochs):
print(f”epoch {epoch}”)
# Set model to training mode.
model.train()
# Average loss between a pair of denoised image and original image.
train_loss = 0.0
val_loss = 0.0
report_single={
train_loss: None,
val_loss: None
}
for train_image_batch, train_noise_image_batch in iter(train_loader):
# Move model and data to GPU.
model = model.to(device)
train_noise_image_batch = train_noise_image_batch[0].to(device)
train_image_batch = train_image_batch[0].to(device)
# Below, we update weights(parameters).
# We pass the inputs through the model, compute the loss, and backpropogate the error.
# Pytorch optimizer accumulates gradient values.
# Zero out the gradient before backpropogate.
optimizer.zero_grad()
# Forward pass. Compute output.
denoised_images = model(train_noise_image_batch)
loss = loss_function(denoised_images, train_image_batch)
# Backward pass. Compute gradients.
loss.backward()
# Optimize.
optimizer.step()
# Add the average loss of the batch to the total loss.
train_loss += loss.item()
# Calculate the average training loss for the whole training dataset.
train_loss = train_loss/len(train_loader)
report_single[‘train_loss’] = train_loss
if print_loss == True:
print(f”train_loss={train_loss}”)
# Validate model with validation set.
# Disable gradiant since we are not updating weights during validation.
with torch.no_grad():
for val_image_batch, val_noise_image_batch in iter(val_loader):
val_image_batch = val_image_batch[0].to(device)
val_noise_image_batch = val_noise_image_batch[0].to(device)
denoised_images = model(val_noise_image_batch)
loss = loss_function(denoised_images, val_image_batch)
val_loss += loss.item()
# Calculate the average loss for the whole validation dataset.
val_loss = val_loss/len(val_loader)
report_single[‘val_loss’] = val_loss
if print_loss == True:
print(f”val_loss={val_loss}”)
# If validation loss is better than the current best loss,
# save it as the best loss and save the model as the best model.
if best_loss is None or (val_loss < best_loss):
best_loss = val_loss
torch.save(model.state_dict(), model_path)
report.append(report_single)
# Return the model and report.
# The model is the last updated model but may not be the best model saved.
return model, report
Test model
In [ ]:
# Run test against test dataset.
def test(dataloader, model, loss_function):
# This is our test function. We pass all the testing data
# through the Network and compute the loss.
test_loss = 0.0
model.to(device)
# Set model to eval mode.
model.eval()
for image_batch, noise_image_batch in iter(dataloader):
# Move data to GPU
image_batch = image_batch[0].to(device)
noise_image_batch = noise_image_batch[0].to(device)
denoised_images = model(noise_image_batch)
loss = loss_function(denoised_images, image_batch)
test_loss += loss.item()
test_loss = test_loss/len(dataloader)
return test_loss
Run our model
In [ ]:
# Set up config variables
config = {
'data_path': './data', # directory of dataset,
'model_path': './checkpoints/best_model_1.pt' # directory saving the best model
}
# create folder checkpoints to save model
!mkdir -p checkpoints
In [ ]:
# Set up objective function (we use L1 loss or absolute loss function function)
criterion = nn.L1Loss()
# Train and test model
model = main(config, net=MyNeuralNetRegressor, criterion=criterion, epochs=50,
learning_rate=0.05, drop_probability=0.7, batch_size=64)
Task 1
Q1 (25 points). In this question we want to draw a graph that shows the training loss and the validation loss throughout the traning. Finish the implementation of functin plot_eval_results based on the following requirements:
1. You can use plot function from matplotlib library.
2. X-axis is the epochs number
3. Y-axis corresponds loss.
4. Draw two graphs for train loss and validation loss on the same plot.
5. Draw a horizontal line for test loss (it is basically the test loss at the epoch with the best validation loss)
After the implementation, run main(config, criterion=nn.L1Loss(), net=MyNeuralNetRegressor, epochs=50, drop_probability=0.7, learning_rate=0.05, batch_size=64).
Include the graph in your report. Justify the trend of the graphs.
An example graph is shown below.

In [ ]:
def plot_eval_results(train_report, test_loss):
# Add your code here
In [ ]:
# Run your model training.
criterion = nn.L1Loss()
model = main(config, criterion=criterion, net=MyNeuralNetRegressor, epochs=50, drop_probability=0.7, learning_rate=0.05, batch_size=64)
Q2 (10 points). How many parameters does MyNeuralNetRegressor have?
Task 2
Convolutional Neural Network(CNN) have been quite successful in the field of image processing. Let us implement a basic CNN.
We can replace fully connected layers in MyNeuralNetRegressor with convolutional layers. A convolutional layer can be implemented in pytorch like
nn.Conv2d(in_channels=, out_channels=, kernel_size=, padding=, stride=)
Q1 (25 points). Finish implementing MyCNN based on the description. Train the network with the same configuration, i.e.,
main(config, MyCNN, criterion=nn.L1Loss(), drop_probability=0.7, learning_rate=0.05, epochs=50, batch_size=64)
Draw the loss graph for both train, validation, and test (like the previous part) for the implemented CNN. Include the graph in your report and compare the results with the previous part (and justify them).
In [ ]:
class MyCNN(nn.Module):
# We create a 2 layers CNN.
# First conv is configured with filters(kernels) size 3, padding 1, and stride 1.
# The first conv's inputs are images in dimension 28*28*1 (W*H*1).
# Its output is in dimension 28*28*10 (W1*H1*num_feature_maps), where
# W1=(W-Filter+2*Padding)/Stride+1 => 28=(28-3+2)/1+1.
# H1=(H-Filter+2*Padding)/Stride+1 => 28=(28-3+2)/1+1.
# Apply ReLU to the first conv output.
# Pass the result as input to the second conv layer.
# The second conv is configured with padding 1 and stride 1.
# You will need to figure out the filter size for the second conv.
# Apply ReLU to the second conv output.
# The final outputs are denoised images (dimension 28*28*1).
def __init__(self):
super(MyCNN, self).__init__()
self.num_input_channels = 1 # number of input image channel,
# 1 for grayscale images, 3 for color images
self.num_feature_maps = 10 # number of feature maps
self.num_output_channels = 1 # number of output image channel
self.kernel_size = 3
self.padding = 1
self.stride = 1
# Add your conv layers below
def forward(self, x):
# Add your code here
In [ ]:
# Run your model training.
model = main(config, MyCNN, criterion=nn.L1Loss(), drop_probability=0.7, learning_rate=0.05, epochs=50, batch_size=64)
Q2 (20 points). How many parameters does MyCNN have?
Q3 (10 points). Denoise 10 images using MyNeuralNetworkRegressor and MyCNN separately. Display the denoised images and compare the results.
Task 3
Q1 (30 points). In this part the goal is to improve the performance of MyCNN. We still want to use the same drop_probability, same training/validation size, and the same loss function. However, you can change the architecture and/or parameters of the network.
main(config, MySuperNN, drop_probability=0.7, criterion=, learning_rate=, epochs=, batch_size=)
Include the changes that you made in your report. Explain the intuition behind your modifications.
Also, draw the graphs for train/test/validation similar to Task 1. How much improvement did you get in your test loss compared to MyCNN?