smd_neural_networks_torch.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "2a42f6dc",
   "metadata": {},
   "source": [
    "# Short Introduction to Neural Networks and Deep Learning with Pytorch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2d86167c",
   "metadata": {},
   "outputs": [],
   "source": [
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "\n",
    "from tqdm.auto import tqdm\n",
    "\n",
    "import torch\n",
    "from torch import nn\n",
    "from torch.utils.data import DataLoader\n",
    "from torchvision import datasets\n",
    "from torchvision.transforms import ToTensor, Resize, Compose"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1ead697a",
   "metadata": {},
   "outputs": [],
   "source": [
    "%matplotlib widget"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "53b553b9",
   "metadata": {},
   "source": [
    "# How to define a Neural Network Architecture in Torch\n",
    "\n",
    "To declare a new Network architecture, we create a new class inheriting from `torch.nn.Model`.\n",
    "\n",
    "The simplest way to declare a Network architecture is to declare the sequence of layers using `torch.nn.Sequential`\n",
    "in `__init__` and we have to implement the `forward` pass. The rest is taken care of by torch (gradients, backword propagation, ...) automagically.\n",
    "\n",
    "Torch builds a computational graph, that can be executed (on different devices) and transformed (e.g. calculate the gradient)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ff06fc7b",
   "metadata": {},
   "outputs": [],
   "source": [
    "class FullyConnected(nn.Module):\n",
    "    \n",
    "    def __init__(self):\n",
    "        super().__init__()\n",
    "        \n",
    "        self.fc = nn.Sequential(\n",
    "            nn.Linear(8 * 8 * 1, 128),\n",
    "            nn.ReLU(),\n",
    "            nn.Linear(128, 128),\n",
    "            nn.ReLU(),\n",
    "            nn.Linear(128, 10),\n",
    "            nn.Softmax(dim=1)\n",
    "        )\n",
    "        self.flatten = nn.Flatten()\n",
    "        \n",
    "    def forward(self, x):\n",
    "        # x.shape = (batchsize, 1, 8, 8)\n",
    "        x = self.flatten(x)\n",
    "        x = self.fc(x)\n",
    "        return x"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "17f6b9e2",
   "metadata": {},
   "outputs": [],
   "source": [
    "net = FullyConnected()\n",
    "net"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f842e09a",
   "metadata": {},
   "source": [
    "Now we are building a more flexible model, where we can pass some options:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "479c7554",
   "metadata": {},
   "outputs": [],
   "source": [
    "class FullyConnected(nn.Module):\n",
    "    def __init__(self, input_size, n_classes, dropout=0.25, n_hidden=256):\n",
    "        super().__init__()\n",
    "        self.flatten = nn.Flatten()\n",
    "        \n",
    "        self.fc_stack = nn.Sequential(\n",
    "            nn.BatchNorm1d(input_size),\n",
    "            \n",
    "            # First Hidden Layer\n",
    "            nn.Linear(input_size, n_hidden),\n",
    "            nn.BatchNorm1d(n_hidden),\n",
    "            nn.Dropout(dropout),\n",
    "            nn.LeakyReLU(),\n",
    "            \n",
    "            # Second Hidden Layer\n",
    "            nn.Linear(n_hidden, n_hidden),\n",
    "            nn.BatchNorm1d(n_hidden),\n",
    "            nn.Dropout(dropout),\n",
    "            nn.LeakyReLU(),\n",
    "            \n",
    "            # Output Layer\n",
    "            nn.Linear(n_hidden, n_classes),\n",
    "            nn.Softmax(dim=1),\n",
    "        )\n",
    "\n",
    "    def forward(self, x):\n",
    "        x = self.flatten(x)\n",
    "        x = self.fc_stack(x)\n",
    "        return x\n",
    "    \n",
    "net = FullyConnected(input_size=3 * 50 * 50, n_classes=2)\n",
    "net"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5f24277b",
   "metadata": {},
   "source": [
    "# Training\n",
    "\n",
    "Unfortunately, training the network is not as simple as calling `fit` like in sklearn.\n",
    "Torch is a very flexible framework, and we have to decide for the data loader, loss function, the optimizer, the model, device and how we evaluate the performance on the test data set.\n",
    "\n",
    "In the end, we are going to write our own `fit` function, to make it simpler."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "435aef2a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# device = \"cpu\"\n",
    "# uncomment to use GPU if available\n",
    "# CPU offers better debugging\n",
    "DEVICE = \"cuda\" if torch.cuda.is_available() else \"cpu\"\n",
    "print(\"Using {} device\".format(DEVICE))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1a43b6fa",
   "metadata": {},
   "outputs": [],
   "source": [
    "def train(dataloader, model, loss_fn, optimizer, device=DEVICE):    \n",
    "    model = model.to(device)\n",
    "    model.train()\n",
    "    \n",
    "    # losses = []\n",
    "    for X, y in dataloader:\n",
    "        X, y = X.to(device), y.to(device)\n",
    "               \n",
    "        # Compute prediction error\n",
    "        pred = model(X)\n",
    "        loss = loss_fn(pred, y)\n",
    "       \n",
    "        # Backpropagation\n",
    "        optimizer.zero_grad()\n",
    "        loss.backward()\n",
    "        optimizer.step()\n",
    "        \n",
    "    #     # store loss for plotting\n",
    "    #     losses.append(loss.item())\n",
    "    # return losses\n",
    "\n",
    "            \n",
    "def test(dataloader, model, loss_fn, device=DEVICE):\n",
    "    model = model.to(device)\n",
    "    test_losses = []\n",
    "    with torch.no_grad():\n",
    "        model.eval()\n",
    "        for X, y in dataloader:\n",
    "            X, y = X.to(device), y.to(device)\n",
    "            pred = model(X)\n",
    "            test_losses.append(loss_fn(pred, y).item())\n",
    "\n",
    "    return test_losses\n",
    "\n",
    "\n",
    "def fit_one_epoch(train_dataloader, test_dataloader, model, loss_fn, optimizer, device=DEVICE):\n",
    "    train(train_dataloader, model, loss_fn, optimizer, device)\n",
    "\n",
    "    train_losses = test(train_dataloader, model, loss_fn, device)\n",
    "    test_losses = test(test_dataloader, model, loss_fn, device)\n",
    "    return train_losses, test_losses\n",
    "\n",
    "\n",
    "def accuracy(dataloader, model, device=DEVICE):\n",
    "    correct = 0\n",
    "    total = 0\n",
    "    model = model.to(device)\n",
    "    with torch.no_grad():\n",
    "        model.eval()\n",
    "        for X, y in dataloader:\n",
    "            X, y = X.to(device), y.to(device)\n",
    "            pred = model(X)\n",
    "            total += len(y)\n",
    "            correct += (pred.argmax(1) == y).type(torch.float).sum().item()\n",
    "    return correct / total\n",
    "\n",
    "\n",
    "def predictions(dataloader, model, device=DEVICE):\n",
    "    predictions = []\n",
    "    truth = []\n",
    "    model = model.to(device)\n",
    "    with torch.no_grad():\n",
    "        model.eval()\n",
    "        for X, y in dataloader:\n",
    "            X, y = X.to(device), y.to(device)\n",
    "            predictions.append(model(X).argmax(1))\n",
    "            truth.append(y)\n",
    "    return torch.cat(predictions), torch.cat(truth)\n",
    "\n",
    "\n",
    "def report_accuracy(test_dataloader, train_dataloader, model):\n",
    "    accuracy_test = accuracy(test_dataloader, model)\n",
    "    accuracy_train = accuracy(train_dataloader, model)\n",
    "    print(f'Accuracy: train={accuracy_train:5.1%}, test={accuracy_test:5.1%}')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "91cfebec",
   "metadata": {},
   "source": [
    "# MNIST"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4dca0824",
   "metadata": {},
   "outputs": [],
   "source": [
    "mnist_train = datasets.MNIST(\n",
    "    root=\"data\",\n",
    "    train=True,\n",
    "    transform=Compose([Resize((16, 16)), ToTensor()]),\n",
    "    download=True,\n",
    ")\n",
    "\n",
    "mnist_test = datasets.MNIST(\n",
    "    root=\"data\",\n",
    "    train=False,\n",
    "    transform=Compose([Resize((16, 16)), ToTensor()]),\n",
    "    download=True,\n",
    "\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b5f71853",
   "metadata": {},
   "outputs": [],
   "source": [
    "batch_size = 64\n",
    "\n",
    "train_dataloader = DataLoader(mnist_train, batch_size=batch_size, shuffle=True)\n",
    "test_dataloader = DataLoader(mnist_test, batch_size=batch_size, shuffle=True)\n",
    "\n",
    "\n",
    "# get first batch\n",
    "X, y = next(iter(test_dataloader))\n",
    "\n",
    "print(\"Shape of X: \", X.shape)\n",
    "print(\"Shape of y: \", y.shape)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c5bed474",
   "metadata": {},
   "outputs": [],
   "source": [
    "fig, axs = plt.subplots(2, 5, figsize=(9, 3), constrained_layout=True)\n",
    "\n",
    "for i, ax in enumerate(axs.flat):\n",
    "    ax.imshow(X[i, 0], cmap='gray_r')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "87f11b66",
   "metadata": {},
   "outputs": [],
   "source": [
    "model = FullyConnected(\n",
    "    input_size=X[0].shape.numel(),\n",
    "    n_classes=len(mnist_train.classes),\n",
    "    n_hidden=256,\n",
    "    dropout=0.25,\n",
    ")\n",
    "\n",
    "loss_fn = nn.CrossEntropyLoss()\n",
    "optimizer = torch.optim.AdamW(model.parameters())\n",
    "\n",
    "\n",
    "epochs = 20\n",
    "test_losses = []\n",
    "train_losses = []\n",
    "report_accuracy(test_dataloader, train_dataloader, model)\n",
    "for t in tqdm(range(epochs)):\n",
    "    epoch_loss_train, epoch_loss_test = fit_one_epoch(\n",
    "        train_dataloader, test_dataloader, model, loss_fn, optimizer\n",
    "    )\n",
    "    train_losses.append(epoch_loss_train)\n",
    "    test_losses.append(epoch_loss_test)\n",
    "    report_accuracy(test_dataloader, train_dataloader, model)\n",
    "    \n",
    "print(\"Done!\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4f50fc2e",
   "metadata": {},
   "outputs": [],
   "source": [
    "def plot_losses(train_losses, test_losses):\n",
    "    plt.figure()\n",
    "    \n",
    "    for i, (label, losses) in enumerate(zip((\"Train\", \"Test\"), (train_losses, test_losses))):\n",
    "        losses = np.array(losses)\n",
    "       \n",
    "        x = np.linspace(0, len(losses), losses.size)\n",
    "        plt.plot(x, losses.ravel(), label=f'Loss {label}', color=f'C{i}', alpha=0.5)\n",
    "                \n",
    "        mean_loss = losses.mean(axis=1)\n",
    "        x = np.arange(1, len(mean_loss) + 1)\n",
    "        plt.plot(x, mean_loss, label=f'Mean Epoch Loss {label}', color=f'C{i}', zorder=3, marker='.')\n",
    "        \n",
    "    plt.xlabel('Epoch')\n",
    "    plt.legend()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5bf4e9ae",
   "metadata": {},
   "outputs": [],
   "source": [
    "plot_losses(train_losses, test_losses)\n",
    "# plt.yscale('log')\n",
    "plt.grid()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6a6fee71",
   "metadata": {},
   "source": [
    "# CIFAR-10"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "91821c11",
   "metadata": {},
   "outputs": [],
   "source": [
    "cifar10_train = datasets.CIFAR10(\n",
    "    root=\"data\",\n",
    "    train=True,\n",
    "    transform=ToTensor(),\n",
    "    download=True,\n",
    ")\n",
    "\n",
    "cifar10_test = datasets.CIFAR10(\n",
    "    root=\"data\",\n",
    "    train=False,\n",
    "    transform=ToTensor(),\n",
    "    download=True,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "763c9611",
   "metadata": {},
   "outputs": [],
   "source": [
    "batch_size = 64\n",
    "\n",
    "train_dataloader = DataLoader(cifar10_train, batch_size=batch_size)\n",
    "test_dataloader = DataLoader(cifar10_test, batch_size=batch_size)\n",
    "\n",
    "# get first batch\n",
    "X, y = next(iter(test_dataloader))\n",
    "\n",
    "print(\"Shape of X: \", X.shape)\n",
    "print(\"Shape of y: \", y.shape)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fc54c5a2",
   "metadata": {},
   "outputs": [],
   "source": [
    "fig, axs = plt.subplots(4, 4, figsize=(9, 9), constrained_layout=True)\n",
    "\n",
    "for idx, ax in enumerate(axs.flat):\n",
    "    img = np.swapaxes(np.asarray(X[idx + 16]), 1, 2).T\n",
    "    \n",
    "\n",
    "    ax.set_title(cifar10_train.classes[y[idx + 16]])\n",
    "    ax.imshow(img)\n",
    "    ax.set_axis_off()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "42d95256",
   "metadata": {},
   "outputs": [],
   "source": [
    "model = FullyConnected(\n",
    "    input_size=X[0].shape.numel(),\n",
    "    n_classes=len(cifar10_train.classes),\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e30310f4",
   "metadata": {},
   "outputs": [],
   "source": [
    "loss_fn = nn.CrossEntropyLoss()\n",
    "optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)\n",
    "\n",
    "\n",
    "test_losses = []\n",
    "train_losses = []\n",
    "\n",
    "\n",
    "report_accuracy(test_dataloader, train_dataloader, model)\n",
    "\n",
    "epochs = 15\n",
    "for t in tqdm(range(epochs)):\n",
    "    epoch_loss_train, epoch_loss_test = fit_one_epoch(\n",
    "        train_dataloader, test_dataloader, model, loss_fn, optimizer\n",
    "    )\n",
    "    train_losses.append(epoch_loss_train)\n",
    "    test_losses.append(epoch_loss_test)\n",
    "    report_accuracy(test_dataloader, train_dataloader, model)\n",
    "    \n",
    "print(\"Done!\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5310d293",
   "metadata": {},
   "outputs": [],
   "source": [
    "plot_losses(train_losses, test_losses)\n",
    "plt.yscale('log')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7d0f93bc",
   "metadata": {},
   "source": [
    "We do not get much better than 50 % with a fully connected network.\n",
    "\n",
    "Let's try a deep learning network with convolutional layers. The architecture follows the one proposed here:\n",
    "https://arxiv.org/abs/1409.1556\n",
    "\n",
    "> Very Deep Convolutional Networks for Large-Scale Image Recognition  \n",
    "> Karen Simonyan, Andrew Zisserman\n",
    "\n",
    "> In this work we investigate the effect of the convolutional network depth on its accuracy in the large-scale image recognition setting. Our main contribution is a thorough evaluation of networks of increasing depth using an architecture with very small (3x3) convolution filters, which shows that a significant improvement on the prior-art configurations can be achieved by pushing the depth to 16-19 weight layers. These findings were the basis of our ImageNet Challenge 2014 submission, where our team secured the first and the second places in the localisation and classification tracks respectively. We also show that our representations generalise well to other datasets, where they achieve state-of-the-art results. We have made our two best-performing ConvNet models publicly available to facilitate further research on the use of deep visual representations in computer vision. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cc77cdf1",
   "metadata": {},
   "outputs": [],
   "source": [
    "class ConvolutionalNetwork(nn.Module):\n",
    "    def __init__(self):\n",
    "        super().__init__()\n",
    "        \n",
    "        self.conv_stack = nn.Sequential(\n",
    "            # 1st stack of conv layers\n",
    "            nn.Conv2d(3, 32, kernel_size=(3, 3), padding='same'),\n",
    "            nn.Conv2d(32, 32, kernel_size=(3, 3), padding='same'),\n",
    "            nn.BatchNorm2d(32),\n",
    "            nn.LeakyReLU(),\n",
    "            nn.MaxPool2d(kernel_size=2, stride=2),\n",
    "            nn.Dropout(0.25),\n",
    "            \n",
    "            # 2nd stack\n",
    "            nn.Conv2d(32, 64, kernel_size=(3, 3), padding='same'),\n",
    "            nn.Conv2d(64, 64, kernel_size=(3, 3), padding='same'),\n",
    "            nn.BatchNorm2d(64),\n",
    "            nn.LeakyReLU(),\n",
    "            nn.MaxPool2d(kernel_size=2, stride=2),\n",
    "            nn.Dropout(0.25),\n",
    "            \n",
    "            # 3rd stack\n",
    "            nn.Conv2d(64, 128, kernel_size=(3, 3), padding='same'),\n",
    "            nn.Conv2d(128, 128, kernel_size=(3, 3), padding='same'),\n",
    "            nn.BatchNorm2d(128),\n",
    "            nn.LeakyReLU(),\n",
    "            nn.MaxPool2d(kernel_size=2, stride=2),\n",
    "            nn.Dropout(0.25),\n",
    "        )\n",
    "        \n",
    "        self.flatten = nn.Flatten()\n",
    "        \n",
    "        self.linear_relu_stack = nn.Sequential(\n",
    "            nn.Linear(128 * 4 * 4, 128),\n",
    "            nn.Dropout(0.25),\n",
    "            nn.LeakyReLU(),\n",
    "            nn.Linear(128, 10),\n",
    "            nn.Softmax(dim=1),\n",
    "        )\n",
    "\n",
    "    def forward(self, x):\n",
    "        x = self.conv_stack(x)\n",
    "        x = self.flatten(x)\n",
    "        x = self.linear_relu_stack(x)\n",
    "        return x"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6cedbcc4",
   "metadata": {},
   "outputs": [],
   "source": [
    "model = ConvolutionalNetwork()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4388223d",
   "metadata": {},
   "outputs": [],
   "source": [
    "loss_fn = nn.CrossEntropyLoss()\n",
    "optimizer = torch.optim.AdamW(model.parameters())\n",
    "\n",
    "test_losses = []\n",
    "train_losses = []"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "787fde22",
   "metadata": {},
   "outputs": [],
   "source": [
    "epochs = 25\n",
    "\n",
    "\n",
    "report_accuracy(test_dataloader, train_dataloader, model)\n",
    "\n",
    "for t in tqdm(range(epochs)):\n",
    "    epoch_loss_train, epoch_loss_test = fit_one_epoch(\n",
    "        train_dataloader, test_dataloader, model, loss_fn, optimizer\n",
    "    )\n",
    "    train_losses.append(epoch_loss_train)\n",
    "    test_losses.append(epoch_loss_test)\n",
    "    report_accuracy(test_dataloader, train_dataloader, model)\n",
    "    \n",
    "print(\"Done!\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c820f23d",
   "metadata": {},
   "outputs": [],
   "source": [
    "plot_losses(train_losses=train_losses, test_losses=test_losses)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b221de1f",
   "metadata": {},
   "outputs": [],
   "source": [
    "prediction, y = predictions(test_dataloader, model)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0f1da51d",
   "metadata": {},
   "outputs": [],
   "source": [
    "from sklearn.metrics import confusion_matrix\n",
    "\n",
    "\n",
    "classes = ('plane', 'car', 'bird', 'cat',\n",
    "           'deer', 'dog', 'frog', 'horse', 'ship', 'truck')\n",
    "\n",
    "matrix = confusion_matrix(y.cpu().numpy(), prediction.cpu().numpy())\n",
    "matrix = np.divide(matrix, matrix.sum(axis = 1))\n",
    "\n",
    "with plt.rc_context({\"figure.constrained_layout.use\": False}):\n",
    "    plt.matshow(matrix)\n",
    "    plt.xticks(np.arange(len(classes)), classes, rotation=90)\n",
    "    plt.yticks(np.arange(len(classes)), classes)\n",
    "    plt.colorbar()\n",
    "    \n",
    "None"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}