add testing notebook for convolutional neural network

notebooks/2.0-hfk-convolutional_testing.ipynb

@@ -0,0 +1,203 @@
+{
+ "cells": [
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "## Testing FocusDataSet"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 1,
+ "metadata": {},
+ "outputs": [
+ {
+ "name": "stdout",
+ "output_type": "stream",
+ "text": [
+ "2450\n"
+ ]
+ },
+ {
+ "data": {
+ "text/plain": [
+ "{'image': array([[[211, 185, 62],\n",
+ " [216, 192, 68],\n",
+ " [223, 198, 79],\n",
+ " ...,\n",
+ " [214, 190, 64],\n",
+ " [222, 199, 71],\n",
+ " [224, 201, 73]],\n",
+ " \n",
+ " [[218, 192, 69],\n",
+ " [223, 197, 74],\n",
+ " [229, 205, 83],\n",
+ " ...,\n",
+ " [216, 193, 65],\n",
+ " [225, 202, 74],\n",
+ " [226, 203, 75]],\n",
+ " \n",
+ " [[223, 198, 72],\n",
+ " [228, 202, 79],\n",
+ " [234, 210, 88],\n",
+ " ...,\n",
+ " [220, 197, 69],\n",
+ " [228, 205, 77],\n",
+ " [226, 203, 73]],\n",
+ " \n",
+ " ...,\n",
+ " \n",
+ " [[157, 138, 17],\n",
+ " [163, 145, 21],\n",
+ " [178, 157, 32],\n",
+ " ...,\n",
+ " [166, 169, 40],\n",
+ " [170, 173, 42],\n",
+ " [176, 179, 46]],\n",
+ " \n",
+ " [[145, 126, 5],\n",
+ " [155, 137, 13],\n",
+ " [177, 156, 31],\n",
+ " ...,\n",
+ " [156, 158, 31],\n",
+ " [166, 169, 40],\n",
+ " [175, 178, 47]],\n",
+ " \n",
+ " [[147, 128, 7],\n",
+ " [159, 141, 17],\n",
+ " [181, 160, 35],\n",
+ " ...,\n",
+ " [149, 151, 24],\n",
+ " [162, 164, 37],\n",
+ " [172, 175, 46]]], dtype=uint8),\n",
+ " 'focus_value': tensor(0.5450)}"
+ ]
+ },
+ "execution_count": 1,
+ "metadata": {},
+ "output_type": "execute_result"
+ }
+ ],
+ "source": [
+ "from importlib.machinery import SourceFileLoader\n",
+ "\n",
+ "focus_datamodule = SourceFileLoader(\"focus_datamodule\", \"../src/datamodules/focus_datamodule.py\").load_module()\n",
+ "from focus_datamodule import FocusDataSet\n",
+ "\n",
+ "ds = FocusDataSet(\"../data/focus150/metadata.csv\", \"../data/focus150/\")\n",
+ "\n",
+ "counter = 0\n",
+ "for d in ds:\n",
+ " counter += 1\n",
+ "\n",
+ "print(counter)\n",
+ "\n",
+ "d"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 81,
+ "metadata": {},
+ "outputs": [
+ {
+ "name": "stdout",
+ "output_type": "stream",
+ "text": [
+ "14\n"
+ ]
+ },
+ {
+ "data": {
+ "text/plain": [
+ "torch.Size([64, 1])"
+ ]
+ },
+ "execution_count": 81,
+ "metadata": {},
+ "output_type": "execute_result"
+ }
+ ],
+ "source": [
+ "from focus_datamodule import FocusDataModule\n",
+ "\n",
+ "datamodule = FocusDataModule(data_dir=\"../data/focus150\", csv_file=\"../data/focus150/metadata.csv\")\n",
+ "datamodule.setup()\n",
+ "\n",
+ "for data in datamodule.test_dataloader():\n",
+ " break\n",
+ "\n",
+ "len(data[\"focus_value\"])\n",
+ "\n",
+ "# https://pytorch.org/tutorials/beginner/blitz/cifar10_tutorial.html\n",
+ "import torch\n",
+ "import torch.nn as nn\n",
+ "import torch.nn.functional as F\n",
+ "\n",
+ "class Net(nn.Module):\n",
+ " def __init__(self):\n",
+ " super().__init__()\n",
+ " pool_size = 3\n",
+ " \n",
+ " conv1_size = 5\n",
+ " conv1_out = 6\n",
+ " conv2_size = 5\n",
+ " conv2_out = 16\n",
+ " size_img = 150\n",
+ "\n",
+ " size_img -= conv1_size - 1\n",
+ " size_img = int( (size_img) / pool_size)\n",
+ " size_img -= conv2_size - 1\n",
+ " size_img = int(size_img / pool_size)\n",
+ "\n",
+ " print(size_img)\n",
+ "\n",
+ " self.model = nn.Sequential(\n",
+ " nn.Conv2d(3, conv1_out, conv1_size),\n",
+ " nn.MaxPool2d(pool_size, pool_size),\n",
+ " nn.Conv2d(conv1_out, conv2_out, conv2_size),\n",
+ " nn.MaxPool2d(pool_size, pool_size),\n",
+ " nn.Flatten(),\n",
+ " nn.Linear(conv2_out * size_img * size_img, 120), # 16 * 34 * 34 or [64, 16, 15, 15]\n",
+ " nn.Linear(120, 84),\n",
+ " nn.Linear(84, 1)\n",
+ " )\n",
+ "\n",
+ " def forward(self, x):\n",
+ " x = self.model(x)\n",
+ " return x\n",
+ "\n",
+ "\n",
+ "net = Net()\n",
+ "\n",
+ "net(data[\"image\"]).shape"
+ ]
+ }
+ ],
+ "metadata": {
+ "interpreter": {
+ "hash": "f9f85f796d01129d0dd105a088854619f454435301f6ffec2fea96ecbd9be4ac"
+ },
+ "kernelspec": {
+ "display_name": "Python 3.9.7 64-bit",
+ "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.8.10"
+ },
+ "orig_nbformat": 4
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}

src/models/focus_conv_module.py

@@ -0,0 +1,181 @@
+from typing import Any, List
+
+import torch
+import torch.nn.functional as F
+from torch import nn
+from pytorch_lightning import LightningModule
+from torchmetrics import MaxMetric, MeanAbsoluteError, MinMetric
+from torchmetrics.classification.accuracy import Accuracy
+
+
+class SimpleConvNet(nn.Module):
+ def __init__(self):
+ super().__init__()
+ self.conv1 = nn.Conv2d(3, 6, 5)
+ self.pool = nn.MaxPool2d(2, 2)
+ self.conv2 = nn.Conv2d(6, 16, 5)
+ self.pool = nn.MaxPool2d(2, 2)
+ self.conv3 = 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 = torch.flatten(x, 1) # flatten all dimensions except batch
+ x = F.relu(self.fc1(x))
+ x = F.relu(self.fc2(x))
+ x = self.fc3(x)
+ return x
+
+
+class SimpleDenseNet(nn.Module):
+ def __init__(self, hparams: dict):
+ super().__init__()
+
+ self.model = nn.Sequential(
+ nn.Linear(hparams["input_size"], hparams["lin1_size"]),
+ nn.BatchNorm1d(hparams["lin1_size"]),
+ nn.ReLU(),
+ nn.Linear(hparams["lin1_size"], hparams["lin2_size"]),
+ nn.BatchNorm1d(hparams["lin2_size"]),
+ nn.ReLU(),
+ nn.Linear(hparams["lin2_size"], hparams["lin3_size"]),
+ nn.BatchNorm1d(hparams["lin3_size"]),
+ nn.ReLU(),
+ nn.Linear(hparams["lin3_size"], hparams["output_size"]),
+ )
+
+ def forward(self, x):
+ batch_size, channels, width, height = x.size()
+
+ # (batch, 1, width, height) -> (batch, 1*width*height)
+ x = x.view(batch_size, -1)
+
+ return self.model(x)
+
+
+class FocusLitModule(LightningModule):
+ """
+ Example of LightningModule for MNIST classification.
+
+ A LightningModule organizes your PyTorch code into 5 sections:
+ - Computations (init).
+ - Train loop (training_step)
+ - Validation loop (validation_step)
+ - Test loop (test_step)
+ - Optimizers (configure_optimizers)
+
+ Read the docs:
+ https://pytorch-lightning.readthedocs.io/en/latest/common/lightning_module.html
+ """
+
+ def __init__(
+ self,
+ input_size: int = 75 * 75 * 3,
+ lin1_size: int = 256,
+ lin2_size: int = 256,
+ lin3_size: int = 256,
+ output_size: int = 1,
+ lr: float = 0.001,
+ weight_decay: float = 0.0005,
+ ):
+ super().__init__()
+
+ # this line allows to access init params with 'self.hparams' attribute
+ # it also ensures init params will be stored in ckpt
+ self.save_hyperparameters(logger=False)
+
+ self.model = SimpleDenseNet(hparams=self.hparams)
+
+ # loss function
+ self.criterion = torch.nn.L1Loss()
+
+ # use separate metric instance for train, val and test step
+ # to ensure a proper reduction over the epoch
+ self.train_mae = MeanAbsoluteError()
+ self.val_mae = MeanAbsoluteError()
+ self.test_mae = MeanAbsoluteError()
+
+ # for logging best so far validation accuracy
+ self.val_mae_best = MinMetric()
+
+ def forward(self, x: torch.Tensor):
+ return self.model(x)
+
+ def step(self, batch: Any):
+ x = batch["image"]
+ y = batch["focus_value"]
+ logits = self.forward(x)
+ loss = self.criterion(logits, y)
+ preds = torch.squeeze(logits)
+ return loss, preds, y
+
+ def training_step(self, batch: Any, batch_idx: int):
+ loss, preds, targets = self.step(batch)
+
+ # log train metrics
+ mae = self.train_mae(preds, targets)
+ self.log("train/loss", loss, on_step=False, on_epoch=True, prog_bar=False)
+ self.log("train/mae", mae, on_step=False, on_epoch=True, prog_bar=True)
+
+ # we can return here dict with any tensors
+ # and then read it in some callback or in `training_epoch_end()`` below
+ # remember to always return loss from `training_step()` or else backpropagation will fail!
+ return {"loss": loss, "preds": preds, "targets": targets}
+
+ def training_epoch_end(self, outputs: List[Any]):
+ # `outputs` is a list of dicts returned from `training_step()`
+ pass
+
+ def validation_step(self, batch: Any, batch_idx: int):
+ loss, preds, targets = self.step(batch)
+
+ # log val metrics
+ mae = self.val_mae(preds, targets)
+ self.log("val/loss", loss, on_step=False, on_epoch=True, prog_bar=False)
+ self.log("val/mae", mae, on_step=False, on_epoch=True, prog_bar=True)
+
+ return {"loss": loss, "preds": preds, "targets": targets}
+
+ def validation_epoch_end(self, outputs: List[Any]):
+ mae = self.val_mae.compute() # get val accuracy from current epoch
+ self.val_mae_best.update(mae)
+ self.log(
+ "val/mae_best", self.val_mae_best.compute(), on_epoch=True, prog_bar=True
+ )
+
+ def test_step(self, batch: Any, batch_idx: int):
+ loss, preds, targets = self.step(batch)
+
+ # log test metrics
+ mae = self.test_mae(preds, targets)
+ self.log("test/loss", loss, on_step=False, on_epoch=True)
+ self.log("test/mae", mae, on_step=False, on_epoch=True)
+
+ return {"loss": loss, "preds": preds, "targets": targets}
+
+ def test_epoch_end(self, outputs: List[Any]):
+ print(outputs)
+ pass
+
+ def on_epoch_end(self):
+ # reset metrics at the end of every epoch
+ self.train_mae.reset()
+ self.test_mae.reset()
+ self.val_mae.reset()
+
+ def configure_optimizers(self):
+ """Choose what optimizers and learning-rate schedulers.
+
+ Normally you'd need one. But in the case of GANs or similar you might have multiple.
+
+ See examples here:
+ https://pytorch-lightning.readthedocs.io/en/latest/common/lightning_module.html#configure-optimizers
+ """
+ return torch.optim.Adam(
+ params=self.parameters(),
+ lr=self.hparams.lr,
+ weight_decay=self.hparams.weight_decay,
+ )