13_ResNet/resnet.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ResNet이란?\n",
    "\n",
    "![resnet](./images/resnet-layers.png)\n",
    "\n",
    "ResNet(Residual Network)는 딥러닝에서 사용되는 컨볼루션 신경망(CNN) 구조이다.\n",
    "\n",
    "2015년 마이크로소프트 연구팀에 의해 개발되었으며, 깊은 신경망을 효율적으로 학습시키기 위해 \"잔차 학습(Residual Learning)\" 개념을 도입했다. \n",
    "\n",
    "이 아이디어는 신경망의 층을 거쳐가는 동안 신호가 약화되거나 왜곡되는 것을 방지하기 위해, 입력을 층의 출력에 직접 추가한 것이다."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ResNet의 구조\n",
    "\n",
    "![resnet](./images/residual.png)\n",
    "\n",
    "ResNet의 핵심 구조는 \"잔차 블록(Residual Block)\"이다. \n",
    "\n",
    "이 블록은 입력을 블록의 출력에 더하는 스킵 연결(skip connection)을 포함한다. \n",
    "\n",
    "이를 통해 네트워크는 학습해야 할 목표 함수를 보다 쉽게 최적화할 수 있다. \n",
    "\n",
    "![resnet](./images/resnet.png)\n",
    "\n",
    "ResNet은 깊이에 따라 여러 버전이 있으며, ResNet-34, ResNet-50, ResNet-101, ResNet-152 등이 일반적이다. \n",
    "\n",
    "여기서 숫자는 네트워크에 있는 층의 수를 나타낸다.\n",
    "\n",
    "![resnet](./images/resnet-network.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Import"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "C:\\Users\\pinb\\AppData\\Local\\Packages\\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\\LocalCache\\local-packages\\Python310\\site-packages\\tqdm\\auto.py:22: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html\n",
      "  from .autonotebook import tqdm as notebook_tqdm\n"
     ]
    }
   ],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "import torch.nn.init as init\n",
    "\n",
    "import torchvision\n",
    "import torchvision.datasets as datasets\n",
    "import torchvision.transforms as transforms\n",
    "\n",
    "from torch.utils.data import DataLoader\n",
    "\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "import tqdm\n",
    "from tqdm.auto import trange\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 하이퍼 파라미터"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "batch_size = 50\n",
    "learning_rate = 0.0002\n",
    "num_epoch = 100"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 데이터셋 구성\n",
    "\n",
    "ResNet 학습을 위한 데이터셋 구성으로 CIFAR10를 사용한다."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Files already downloaded and verified\n",
      "Files already downloaded and verified\n"
     ]
    }
   ],
   "source": [
    "transform = transforms.Compose([transforms.ToTensor(), transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))])\n",
    "\n",
    "# define dataset\n",
    "cifar10_train = datasets.CIFAR10(root=\"./Data/\", train=True, transform=transform, target_transform=None, download=True)\n",
    "cifar10_test = datasets.CIFAR10(root=\"./Data/\", train=False, transform=transform, target_transform=None, download=True)\n",
    "\n",
    "# define loader\n",
    "train_loader = DataLoader(cifar10_train,batch_size=batch_size, shuffle=True, num_workers=2, drop_last=True)\n",
    "test_loader = DataLoader(cifar10_test,batch_size=batch_size, shuffle=False, num_workers=2, drop_last=True)\n",
    "\n",
    "# define classes\n",
    "classes = ('plane', 'car', 'bird', 'cat', 'deer', 'dog', 'frog', 'horse', 'ship', 'truck')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Basic Module"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "def conv_block_1(in_dim,out_dim, activation,stride=1):\n",
    "    model = nn.Sequential(\n",
    "        nn.Conv2d(in_dim,out_dim, kernel_size=1, stride=stride),\n",
    "        nn.BatchNorm2d(out_dim),\n",
    "        activation,\n",
    "    )\n",
    "    return model\n",
    "\n",
    "\n",
    "def conv_block_3(in_dim,out_dim, activation, stride=1):\n",
    "    model = nn.Sequential(\n",
    "        nn.Conv2d(in_dim,out_dim, kernel_size=3, stride=stride, padding=1),\n",
    "        nn.BatchNorm2d(out_dim),\n",
    "        activation,\n",
    "    )\n",
    "    return model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Bottleneck Module"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "class BottleNeck(nn.Module):\n",
    "    def __init__(self,in_dim,mid_dim,out_dim,activation,down=False):\n",
    "        super(BottleNeck,self).__init__()\n",
    "        self.down=down\n",
    "        \n",
    "        # 특성지도의 크기가 감소하는 경우\n",
    "        if self.down:\n",
    "            self.layer = nn.Sequential(\n",
    "              conv_block_1(in_dim,mid_dim,activation,stride=2),\n",
    "              conv_block_3(mid_dim,mid_dim,activation,stride=1),\n",
    "              conv_block_1(mid_dim,out_dim,activation,stride=1),\n",
    "            )\n",
    "            \n",
    "            # 특성지도 크기 + 채널을 맞춰주는 부분\n",
    "            self.downsample = nn.Conv2d(in_dim,out_dim,kernel_size=1,stride=2)\n",
    "            \n",
    "        # 특성지도의 크기가 그대로인 경우\n",
    "        else:\n",
    "            self.layer = nn.Sequential(\n",
    "                conv_block_1(in_dim,mid_dim,activation,stride=1),\n",
    "                conv_block_3(mid_dim,mid_dim,activation,stride=1),\n",
    "                conv_block_1(mid_dim,out_dim,activation,stride=1),\n",
    "            )\n",
    "            \n",
    "        # 채널을 맞춰주는 부분\n",
    "        self.dim_equalizer = nn.Conv2d(in_dim,out_dim,kernel_size=1)\n",
    "                  \n",
    "    def forward(self,x):\n",
    "        if self.down:\n",
    "            downsample = self.downsample(x)\n",
    "            out = self.layer(x)\n",
    "            out = out + downsample\n",
    "        else:\n",
    "            out = self.layer(x)\n",
    "            if x.size() is not out.size():\n",
    "                x = self.dim_equalizer(x)\n",
    "            out = out + x\n",
    "        return out"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ResNet-50 Network"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 50-layer\n",
    "class ResNet(nn.Module):\n",
    "\n",
    "    def __init__(self, base_dim, num_classes=10):\n",
    "        super(ResNet, self).__init__()\n",
    "        self.activation = nn.ReLU()\n",
    "        self.layer_1 = nn.Sequential(\n",
    "            nn.Conv2d(3,base_dim,7,2,3),\n",
    "            nn.ReLU(),\n",
    "            nn.MaxPool2d(3,2,1),\n",
    "        )\n",
    "        self.layer_2 = nn.Sequential(\n",
    "            BottleNeck(base_dim,base_dim,base_dim*4,self.activation),\n",
    "            BottleNeck(base_dim*4,base_dim,base_dim*4,self.activation),\n",
    "            BottleNeck(base_dim*4,base_dim,base_dim*4,self.activation,down=True),\n",
    "        )   \n",
    "        self.layer_3 = nn.Sequential(\n",
    "            BottleNeck(base_dim*4,base_dim*2,base_dim*8,self.activation),\n",
    "            BottleNeck(base_dim*8,base_dim*2,base_dim*8,self.activation),\n",
    "            BottleNeck(base_dim*8,base_dim*2,base_dim*8,self.activation),\n",
    "            BottleNeck(base_dim*8,base_dim*2,base_dim*8,self.activation,down=True),\n",
    "        )\n",
    "        self.layer_4 = nn.Sequential(\n",
    "            BottleNeck(base_dim*8,base_dim*4,base_dim*16,self.activation),\n",
    "            BottleNeck(base_dim*16,base_dim*4,base_dim*16,self.activation),\n",
    "            BottleNeck(base_dim*16,base_dim*4,base_dim*16,self.activation),            \n",
    "            BottleNeck(base_dim*16,base_dim*4,base_dim*16,self.activation),\n",
    "            BottleNeck(base_dim*16,base_dim*4,base_dim*16,self.activation),\n",
    "            BottleNeck(base_dim*16,base_dim*4,base_dim*16,self.activation,down=True),\n",
    "        )\n",
    "        self.layer_5 = nn.Sequential(\n",
    "            BottleNeck(base_dim*16,base_dim*8,base_dim*32,self.activation),\n",
    "            BottleNeck(base_dim*32,base_dim*8,base_dim*32,self.activation),\n",
    "            BottleNeck(base_dim*32,base_dim*8,base_dim*32,self.activation),\n",
    "        )\n",
    "        self.avgpool = nn.AvgPool2d(1,1) \n",
    "        self.fc_layer = nn.Linear(base_dim*32,num_classes)\n",
    "        \n",
    "    def forward(self, x):\n",
    "        out = self.layer_1(x)\n",
    "        out = self.layer_2(out)\n",
    "        out = self.layer_3(out)\n",
    "        out = self.layer_4(out)\n",
    "        out = self.layer_5(out)\n",
    "        out = self.avgpool(out)\n",
    "        out = out.view(batch_size,-1)\n",
    "        out = self.fc_layer(out)\n",
    "        \n",
    "        return out"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 학습"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "  0%|          | 0/100 [00:00<?, ?it/s]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Epoch [0/100] Train Loss: 2.0092\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "  2%|▏         | 2/100 [02:47<2:17:32, 84.21s/it]"
     ]
    }
   ],
   "source": [
    "device = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\n",
    "model = ResNet(base_dim=64).to(device)\n",
    "loss_func = nn.CrossEntropyLoss()\n",
    "optimizer = optim.Adam(model.parameters(), lr=learning_rate)\n",
    "\n",
    "for i in trange(num_epoch):\n",
    "    model.train()  # 모델을 학습 모드로 설정\n",
    "    train_loss = 0.0\n",
    "    for j, [image, label] in enumerate(train_loader):\n",
    "        x = image.to(device)\n",
    "        y_ = label.to(device)\n",
    "\n",
    "        optimizer.zero_grad()\n",
    "        output = model(x)\n",
    "        loss = loss_func(output, y_)\n",
    "        loss.backward()\n",
    "        optimizer.step()\n",
    "\n",
    "        train_loss += loss.item()\n",
    "    \n",
    "    train_loss /= len(train_loader)\n",
    "\n",
    "    if i % 10 == 0:\n",
    "        print(f\"Epoch [{i}/{num_epoch}] Train Loss: {train_loss:.4f}\")\n",
    "        torch.save(model.state_dict(), f'model_epoch_{i}.pth')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 결과 이미지 표시"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import matplotlib.pyplot as plt\n",
    "\n",
    "# 테스트 이미지 로드 및 변환\n",
    "test_image = Image.open('test.jpg')\n",
    "test_image = transform(test_image).unsqueeze(0)  # 차원 추가\n",
    "\n",
    "# 예측 수행\n",
    "model.eval()\n",
    "with torch.no_grad():\n",
    "    output = model(test_image.to(device))\n",
    "\n",
    "# 결과 시각화\n",
    "plt.imshow(test_image.squeeze().permute(1, 2, 0))  # 차원 변경 및 시각화\n",
    "plt.title(\"Predicted Label\")\n",
    "plt.show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import matplotlib.pyplot as plt\n",
    "\n",
    "# 모델을 평가 모드로 설정\n",
    "model.eval()\n",
    "\n",
    "# 테스트 데이터셋의 첫 번째 배치를 가져옴\n",
    "images, labels = next(iter(test_loader))\n",
    "images, labels = images.to(device), labels.to(device)\n",
    "\n",
    "# 모델 예측\n",
    "with torch.no_grad():\n",
    "    outputs = model(images)\n",
    "\n",
    "# 예측 결과 처리\n",
    "_, predicted = torch.max(outputs, 1)\n",
    "\n",
    "# 이미지 출력 설정\n",
    "fig, axs = plt.subplots(1, len(images), figsize=(15, 3))\n",
    "\n",
    "for i, img in enumerate(images.cpu()):\n",
    "    img = img.numpy().transpose((1, 2, 0))\n",
    "    axs[i].imshow(img)\n",
    "    axs[i].set_title(f'Label: {labels[i].item()}, Predict: {predicted[i].item()}')\n",
    "    axs[i].axis('off')\n",
    "\n",
    "plt.show()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.10.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}