NeuralNetwork_Tutorial.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Simple Neural Network tutorial\n",
    "\n",
    "This is an example of how neural networks can be used to categorize images.  The test images are the MNIST collection of handwritten digits, and we will train a network to recognized hand-written digits.  Other sets of data are available.\n",
    "\n",
    "### Import packages and dataset\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "%matplotlib inline\n",
    "import numpy as np\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.nn.functional as F\n",
    "import torch.optim as optim\n",
    "from torch.optim.lr_scheduler import StepLR\n",
    "from torchvision import datasets\n",
    "from torchvision.transforms import v2 as transforms\n",
    "import matplotlib.pyplot as plt\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Tell PyTorch to run on the cpu.  Set up a random number generator."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "use_cuda = torch.cuda.is_available()\n",
    "use_mps = torch.backends.mps.is_available()\n",
    "\n",
    "if use_cuda:\n",
    "    device = torch.device('cuda')\n",
    "    print('Running on Nvidia GPU (cuda)')\n",
    "elif use_mps:\n",
    "    device = torch.device('mps')\n",
    "    print('Running on Apple Silicon GPU device (mps)')\n",
    "else:\n",
    "    device = torch.device('cpu')\n",
    "    print('Running on CPU')\n",
    "    \n",
    "rng = np.random.default_rng()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Select and look at the dataset\n",
    "\n",
    "We start with the standard **MNIST** dataset of handwritten digits.\n",
    "\n",
    "Harder data sets are **CIFAR10** and **CIFAR100**, tiny color images with 10 or 100 categories.  \n",
    "\n",
    "An interesting data set is **Imagenette**, it exists in a 160px and 320px version.  Even the 160px version is too hard for this simple network but may work for more complicated neural network architectures.\n",
    "\n",
    "Complication: the images are of different size, and our network requires that they are the same.  A transformation can be applied to scale all images to same size.  This dataset also takes different parameters to split into training and dataset, see the documentation at https://pytorch.org/vision/stable/datasets.html\n",
    "\n",
    "Data scientists always start by visualizing their data - avoid surprises later."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Transformation for the MNIST and CIFAR datasets: Convert to tensor of float32\n",
    "transform=transforms.Compose([\n",
    "        transforms.ToImage(),   # Converts PIL images to tensors\n",
    "        transforms.ToDtype(torch.float32, scale=True),\n",
    "        ])\n",
    "\n",
    "# Transformation for the Imagenette dataset: Convert and rescale to 160 x 160\n",
    "transform_imagenette=transforms.Compose([\n",
    "        transforms.ToImage(),   # Converts PIL images to tensors\n",
    "        transforms.Resize((160,160)),\n",
    "        transforms.ToDtype(torch.float32, scale=True),\n",
    "        ])\n",
    "\n",
    "#transform = transforms.ToTensor()\n",
    "\n",
    "dataset_train = datasets.MNIST('data', train=True, download=True, transform=transform)\n",
    "dataset_test = datasets.MNIST('data', train=False, transform=transform)\n",
    "#dataset_train = datasets.CIFAR100('data', train=True, download=True, transform=transform)\n",
    "#dataset_test = datasets.CIFAR100('data', train=False, transform=transform)\n",
    "#dataset_train = datasets.Imagenette('data', split='train', size='160px', download=True, transform=transform_imagenette)\n",
    "#dataset_test = datasets.Imagenette('data', split='val', size='160px', transform=transform_imagenette)\n",
    "\n",
    "ntrain = len(dataset_train)\n",
    "ntest = len(dataset_test)\n",
    "imgsize = dataset_test[0][0].shape\n",
    "print(imgsize)\n",
    "num_categories = len(dataset_train.classes)\n",
    "print(\"Training set: {} images of shape {}\".format(ntrain, list(imgsize)))\n",
    "print(\"Pixels in image\", np.prod(imgsize))\n",
    "print(\"Number of categories:\", num_categories)\n",
    "print(\"Categories:\", dataset_train.classes)\n",
    "print(\"Training set identifies as\")\n",
    "print(dataset_train)\n",
    "print(\"Test set identifies as\")\n",
    "print(dataset_test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Sanity check of data types\n",
    "print(dataset_test[0][0].shape, dataset_test[0][0].dtype)\n",
    "print(dataset_test[0][0].min(), dataset_test[0][0].max())\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "nplot = 10\n",
    "rnd = rng.integers(len(dataset_test), size=(nplot,))\n",
    "fig, axes = plt.subplots(1, nplot, figsize=(12,12/nplot))\n",
    "for i in range(nplot):\n",
    "    image, label = dataset_test[rnd[i]]\n",
    "    print(image.shape)\n",
    "    # Shape (1, X, Y) if BW, (3, X, Y) if color\n",
    "    if image.shape[0] == 1:\n",
    "        image = image[0]  # New shape (X, Y)\n",
    "    elif image.shape[0] in (3, 4):\n",
    "        image = torch.moveaxis(image, 0, -1)    # New shape (X, Y, 3)\n",
    "    axes[i].imshow(image, cmap='gray')\n",
    "    axes[i].axis('off')\n",
    "    axes[i].set_title(label)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define a function creating the neural network\n",
    "Make multiple functions to be able to experiment with multiple architectures."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "def make_minimalnet(categories, imshape, hiddenneurons=200):\n",
    "    imsize = np.prod(imshape)\n",
    "    return nn.Sequential(\n",
    "        nn.Flatten(1),\n",
    "        nn.Linear(imsize, hiddenneurons),\n",
    "        nn.Sigmoid(),\n",
    "        nn.Linear(hiddenneurons, categories),\n",
    "        nn.LogSoftmax(dim=1)\n",
    "    )\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define training and test functions\n",
    "\n",
    "This is more or less boilerplate code, but you may want to modify it a little for other applications."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "lossfunc = F.nll_loss   # Pick loss function for both functions below.\n",
    "\n",
    "def train(model, device, train_loader, optimizer, epoch, verbose=False):\n",
    "    \"\"\"Train model for a single epoch.\n",
    "    \n",
    "    Parameters:\n",
    "    model:         PyTorch model to train.\n",
    "    device:        Device used for training (i.e. cpu, gpu, ...)\n",
    "    train_loader:  Data loader\n",
    "    optimizer:     The optimizer used in the training.\n",
    "    epoch:         Epoch number\n",
    "    \"\"\"\n",
    "    model.train()  # Put model in training mode\n",
    "    for batch_idx, (data, target) in enumerate(train_loader):\n",
    "        data, target = data.to(device), target.to(device)\n",
    "        optimizer.zero_grad()\n",
    "        output = model(data)\n",
    "        loss = lossfunc(output, target)\n",
    "        loss.backward()\n",
    "        optimizer.step()\n",
    "        # Print progress\n",
    "        if verbose:\n",
    "            ndone = batch_idx * len(data)\n",
    "            ntot = len(train_loader.dataset)\n",
    "            percent = 100 * ndone / ntot\n",
    "            print(f'Train epoch: {epoch} [{ndone}/{ntot} ({percent:.0f}%)]\\tLoss: {loss.item():.6f}')\n",
    "\n",
    "def test(model, device, test_loader, epoch):\n",
    "    \"\"\"Evaluate a model.\n",
    "    \n",
    "    It does not return anything, but prints the test performance.\n",
    "\n",
    "    Parameters:\n",
    "    model:       PyTorch model\n",
    "    device:      Device used for training / evaluating.\n",
    "    test_loader: Data loader.\n",
    "    \"\"\"\n",
    "    model.eval()  # Put model in evaluation mode.\n",
    "    test_loss = 0\n",
    "    correct = 0\n",
    "    with torch.no_grad():\n",
    "        for data, target in test_loader:\n",
    "            data, target = data.to(device), target.to(device)\n",
    "            output = model(data)\n",
    "            test_loss += lossfunc(output, target, reduction='sum').item()  # sum up batch loss\n",
    "            pred = output.argmax(dim=1, keepdim=True)  # get the index of the max log-probability\n",
    "            correct += pred.eq(target.view_as(pred)).sum().item()\n",
    "\n",
    "    test_loss /= len(test_loader.dataset)\n",
    "\n",
    "    ndata = len(test_loader.dataset)\n",
    "    percent = 100. * correct / ndata\n",
    "    print(f'Test set, epoch {epoch}: Average loss: {test_loss:.4f}, Accuracy: {correct}/{ndata} ({percent:.0f}%)\\n')\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Do the training.\n",
    "\n",
    "Define\n",
    "* Data loaders\n",
    "* model\n",
    "* optimizer  (adjusts weights in model)\n",
    "* scheduler  (adjusts learning rate in optimizer)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_loader = torch.utils.data.DataLoader(dataset_train, batch_size=64)\n",
    "test_loader = torch.utils.data.DataLoader(dataset_test, batch_size=1000)\n",
    "\n",
    "model = make_minimalnet(num_categories, imgsize).to(device)\n",
    "#optimizer = optim.Adadelta(model.parameters(), lr=0.1)\n",
    "optimizer = optim.Adam(model.parameters(), lr=1e-3)\n",
    "# Decay learning rate exponentially\n",
    "scheduler = StepLR(optimizer, step_size=1, gamma=0.7)\n",
    "\n",
    "def count_parameters(model):\n",
    "    \"Returns number of trainable parameters.\"\n",
    "    return sum(p.numel() for p in model.parameters() if p.requires_grad)\n",
    "\n",
    "print(model)\n",
    "print('Number of trainable parameters:', count_parameters(model))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "nepochs = 10\n",
    "for epoch in range(nepochs):\n",
    "    train(model, device, train_loader, optimizer, epoch)\n",
    "    test(model, device, test_loader, epoch)\n",
    "    scheduler.step()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Show some of the failures\n",
    "\n",
    "We run the network on the test data again, this time keeping the results and looking at them.\n",
    "\n",
    "We need to run the model on the input.  The variable ``dataset_test`` is a sequence of tuples of (image, label).  We need to use a DataLoader to get it converted to the right types and splitting it into images and labels.  We are lazy here and use ``test_loader`` and just get the first batch from it, so we only look at the first 1000 datapoints, which should be fine."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "images, labels = next(iter(test_loader))\n",
    "print(images.shape, images.dtype, labels.shape, labels.dtype)\n",
    "with torch.no_grad():   # Don't waste memory and time on the gradients!\n",
    "    predictions = model(images.to(device))\n",
    "print(predictions.shape, predictions.dtype)\n",
    "\n",
    "# Convert to numpy arrays for further processing.\n",
    "predictions = predictions.cpu().detach().numpy()\n",
    "labels = labels.cpu().detach().numpy()\n",
    "print(predictions.shape, predictions.dtype)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The output is a matrix of *logarithms* of probabilities.  Each row contains the probabilities that the image belongs to each of the ten classes.  Let us look at the first five.\n",
    "\n",
    "Let us also take the exponentials to get the actual probabilities, and check that the sums are 1.0\n",
    "\n",
    "This kind of sanity checks are important when working with complex libraries like PyTorch, at least until you are very familiar with what they are doing."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with np.printoptions(precision=4, suppress=True):\n",
    "    print(predictions[:5])\n",
    "    print(np.exp(predictions[:5]))\n",
    "    print(np.exp(predictions[:5]).sum(axis=1))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Convert this into predictions: find the most probable class for each image."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "predicted_class = np.argmax(predictions, axis=1)\n",
    "print(predicted_class.shape)\n",
    "\n",
    "n_error = (labels != predicted_class).sum()\n",
    "print(f\"There are {n_error} errors out of {len(predicted_class)}.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Find the failures, show the first five."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "failures = np.argsort(labels != predicted_class)[-n_error:]\n",
    "\n",
    "nshow = 5\n",
    "fig, axes = plt.subplots(1, nshow, figsize=(12,12/nshow))\n",
    "\n",
    "classes = dataset_test.classes\n",
    "for i, f in enumerate(failures[:nshow]):\n",
    "    img = images[f]   # Shape (1, X, Y) if BW, (3, X, Y) if color\n",
    "    if img.shape[0] == 1:\n",
    "        img = img[0]  # New shape (X, Y)\n",
    "    elif img.shape[0] in (3, 4):\n",
    "        img = torch.moveaxis(img, 0, -1)    # New shape (X, Y, 3)\n",
    "    axes[i].imshow(img,cmap='gray')\n",
    "    axes[i].axis('off')\n",
    "    prd = predicted_class[f]\n",
    "    prob = np.exp(predictions[f, prd])\n",
    "    title = f'''{f}\n",
    "Predicted: \n",
    "{prd} \"{classes[prd]}\"\n",
    "Correct class: \n",
    "{labels[f]} \"{classes[labels[f]]}\"\n",
    "Probability {100*prob:.1f}%'''\n",
    "    axes[i].set_title(title)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with np.printoptions(precision=4, suppress=True):\n",
    "    print(np.exp(predictions[619]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataset_test.classes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "venv",
   "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.9.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}