Pytorch Tutorial 5

The reason use the NN is inner kernel of logistic regression is still linear, to avoid the linear relationship, the NN can use activation function, for instance ReLU.

In this case, we use ReLu as our activation function to predict the image, and it can be found that the accuracy is far better than LR, shows more abilities.

from os import path, mkdir
from random import randint

import torch
import numpy as np
import torchvision
from matplotlib import pyplot as plt
from torchvision.datasets import MNIST
from torchvision.transforms import ToTensor
from torch.utils.data.sampler import SubsetRandomSampler
from torch.utils.data.dataloader import DataLoader
import torch.nn.functional as F
import torch.nn as nn

dataset = MNIST(root="./data", download=True, transform=ToTensor())
test_dataset = MNIST(root='./data', train=False, transform=ToTensor())

def split_indices(n, rate):
    # create number of validation set
    n_val = int(n * rate)
    # create shuffled index from 0-n, with no repeat
    idxs = np.random.permutation(n)
    # retuen (n_val,last) index and (first n_val) index
    # i.e. training index and validation index
    return idxs[n_val:], idxs[:n_val]

train_indices, val_indices = split_indices(len(dataset), 0.2)

batch_size = 100
train_sampler = SubsetRandomSampler(train_indices)
train_loder = DataLoader(dataset,
                         batch_size,
                         sampler=train_sampler)

val_sampler = SubsetRandomSampler(val_indices)
val_loder = DataLoader(dataset,
                       batch_size,
                       sampler=val_sampler)

input_size = 28 * 28
num_classes = 10

class MnistModel(nn.Module):

    def __init__(self, in_size, hidden_size, out_size):
        super().__init__()

        self.linear1 = nn.Linear(in_size, hidden_size)

        self.linear2 = nn.Linear(hidden_size, out_size)

    def forward(self, xb):
        # flatten
        xb = xb.view(xb.size(0), -1)
        # xb = xb.reshape(xb.size(0), -1)
        return self.linear2(F.relu(self.linear1(xb)))

# for t in model.parameters():
#     print(t.shape)

# for img, labels in train_loder:
#     outputs = model(img)
#     loss = F.cross_entropy(outputs, labels)
#     break

def get_device():
    if torch.cuda.is_available():
        return torch.device('cuda')
    else:
        return torch.device('cpu')

def to_device(data, device):
    if isinstance(data, (list, tuple)):
        return [to_device(x, device) for x in data]
    return data.to(device, non_blocking=True)

# for img, label in train_loder:
#     print(img.shape)
#     img = to_device(img, device)
#     print(img.device)
#     break

class DeviceDataLoder():
    def __init__(self, dl, device):
        self.dl = dl
        self.device = device

    def __iter__(self):
        # lazy load here
        # instead of load data into device each time, instead, load each batch
        for b in self.dl:
            yield to_device(b, self.device)

    def __len__(self):
        return len(self.dl)

# use DeviceDataLoader as warpper
train_dl = DeviceDataLoder(train_loder, get_device())
valid_dl = DeviceDataLoder(val_loder, get_device())

def loss_batch(model, loss_func, xb, yb, opt=None, metric=None):
    preds = model(xb)

    loss = loss_func(preds, yb)

    if opt is not None:
        loss.backward()
        opt.step()
        opt.zero_grad()

    metric_result = None
    if metric is not None:
        metric_result = metric(preds, yb)

    return loss.item(), len(xb), metric_result

def evaluate(model, loss_func, valid_dl, metric=None):
    with torch.no_grad():
        results = [loss_batch(model, loss_func, xb, yb, metric=metric)
                   for xb, yb in valid_dl]

        # separate the lists
        loss, nums, metric = zip(*results)
        total = np.sum(nums)
        avg_loss = np.sum(np.multiply(loss, nums)) / total
        avg_metric = None
        if metric is not None:
            avg_metric = np.sum(np.multiply(metric, nums)) / total
    return avg_loss, total, avg_metric

def fit(epochs, lr, model, loss_func, train_dl, valid_dl, opt_fn=None, metric=None):
    if opt_fn is None:
        opt_fn = torch.optim.SGD
    opt = opt_fn(model.parameters(), lr=lr)
    loss_history = []
    metric_history = []

    for epoch in range(epochs):
        for xb, yb in train_dl:
            loss_batch(model, loss_func, xb, yb, opt)
        result = evaluate(model, loss_func, valid_dl, metric)
        val_loss, total, val_metric = result

        loss_history.append(val_loss)
        metric_history.append(val_metric)

        if metric is not None:
            print(f'Epoch [{epoch + 1}/{epochs}], Loss: {val_loss:.4f}, Metric: {val_metric:.4f}')
        else:
            print(f'Epoch [{epoch + 1}/{epochs}], Loss: {val_loss:.4f}')

    return loss_history, metric_history

def accuracy(output, label):
    _, preds = torch.max(output, dim=1)
    return torch.sum(label == preds).item() / len(preds)

model = MnistModel(input_size, 32, num_classes)
to_device(model, get_device())

if path.exists('./tutorial5/mnist-logistic.pth'):
    model.load_state_dict(torch.load('./tutorial5/mnist-logistic.pth'))

else:
    loss_history, metric_history = fit(5, 0.5, model, F.cross_entropy,
                                       train_dl,
                                       valid_dl,
                                       opt_fn=torch.optim.SGD,
                                       metric=accuracy)
    # it will save the weight and bias for this model
    # new dir
    mkdir('./tutorial5')
    torch.save(model.state_dict(), './tutorial5/mnist-logistic.pth')

def prediction_img(img, model):
    xb = img.unsqueeze(0)
    yb = model(xb)
    _, preds = torch.max(yb, dim=1)
    return preds[0].item()

for i in range(10):
    img, label = test_dataset[randint(0, len(test_dataset) - 1)]
    img_np = np.array(img)
    plt.imshow(img_np.squeeze(), cmap='gray')
    plt.show()
    print(prediction_img(img, model))