ResNet-101 Classification 실습 연습

Computer Vision에서의 classification model들에 대한 outline을 검토하면서, 대표적인 모델인 ResNet-101을 활용하여 CIFAR10 데이터셋에 대한 기본적인 classification 모델을 실습해보고자 한다.

 

PyTorch를 활용하여 Classfication 학습 및 테스트 모델을 만들어보면서,

1. 모델 구축에 사용되는 PyTorch 문법을 연습해보고,
2. Classification Model의 전체적인 Architecture를 학습하고,
3. 주요 Hyper-Parameter들에 대한 조정을 통한 성능 변화 여부 검토,
4. Loss Function 및 Optimizer에 대한 분석을 진행해보려고 한다.

 

Classification Model Architecture는 크게 6단계로 구분하여 진행하였는데,

① 주요 Module importing, ② Data Transform, ③ Dataset 및 DataLoader 생성, ④ Model 생성 및 Hyper-parameters 설정 (중간중간 설정하였다), ⑤ Model Train과 Validation 진행, 마지막으로 ⑥ Custom Dataset을 활용한 Test를 진행해보았다.

 

코드 구현의 각 부문에서 헷갈리거나 잘 몰랐던 내용들은 주석을 통해 각각 표시해두었다

Module Import

import os, sys, glob, csv, cv2, tqdm
from typing import Tuple, List, Dict
import numpy as np
from PIL import Image   # 이미지 처리를 위한 라이브러리
import matplotlib.pyplot as plt

# torch library
#   : Tensor와 같은 다차원 배열을 조작하는 기본 기능
#   : 신경망 모델의 정의와 학습, 최적화 알고리즘을 포함한 PyTorch의 핵심 모듈
import torch
from torch import Tensor
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import Dataset, DataLoader
from torch.utils.data.sampler import SubsetRandomSampler
    # PyTorch의 데이터 로딩 및 처리를 위한 클래스 및 함수

# torchvision library
#   : PyTorch에서 이미지 및 비디오 데이터셋 및 전처리를 위한 라이브러리
#   : CV 작업에 특화 (이미지 분류 모델들의 사전 학습된 가중치도 있음)
import torchvision
from torchvision import transforms, models
import torch.optim as optim

 

Data Transform

# 데이터 전처리 형식 규정
train_transform = transforms.Compose(
    [
        transforms.RandomHorizontalFlip(),  # 훈련에 있어 Segmentation 기법 사용
        transforms.ToTensor(),
        transforms.Normalize(mean=(0.5, 0.5, 0.5), std=(0.5, 0.5, 0.5))
    ]
)
valid_transform = transforms.Compose(
    [
        transforms.ToTensor(),
        transforms.Normalize(mean=(0.5, 0.5, 0.5), std=(0.5, 0.5, 0.5))
    ]
)

• ToTensor() 메서드를 통해 0과 1사이의 정규분포를 갖는 텐서 값으로의 변환과 스케일링을 동시에 수행하지만, Normalize() 메서드를 통해 -1과 1사이로 스케일링을 추가로 진행
• Normalize()를 통해 [0, 1]의 범위값이 아닌 [-1, 1]로 스케일링 했을 때의 주요 장점은 다음과 같다.
  • Zero-centered input
    • 데이터가 평균이 0인 중앙에 위치하게 됨에 따라, 학습 과정에서 가중치 업데이트가 안정적으로 이루어짐 (이 장점은 [0, 1]로의 스케일링을 통해서도 얻을 수 있는 장점
  • 기울기 편향 극복
    • 0을 기준으로 좌우 대칭인 구조(Symmetry)를 갖고 있으며, 역전파 결과 output이 음수값이 될 수 있어 이동 편향에 따른 기울기 편향을 상대적으로 극복할 수 있음 ([0, 1]의 경우 항상 양수값을 가짐으로, 이동 편향이 발생)
  • 표현 가능한 attribute 갯수가 더 다양
    • [0, 1]의 범위를 [-1, 1]로 확장하여 표현할 수 있다는 점에서 더 다양한 attribute를 표현할 수 있고, 이는 연쇄적인 gradient의 계산에 있어 vanishing 문제를 극복할 수 있음
    • mode collapse 문제의 상대적 해결이 가능
    • 이를 통한 강건한 학습(Robustness)이 가능

 

Dataset 및 DataLoader 생성

Dataset Downloading

# Hyper-Parameter
batch_size = 32
val_size = 0.2

# Download CIFAR10 and define Train and Validation Dataset

train_dataset = torchvision.datasets.CIFAR10(root=r'.\data',
                                             train=True, download=True,
                                             transform=train_transform)
                                            # 50000의 (3, 32, 32) 데이터와 labels
valid_dataset = torchvision.datasets.CIFAR10(root=r'.\data',
                                             train=True, download=True,
                                             transform=valid_transform)

# 클래스 정의 : 10개의 클래스
classes = ('plane', 'car', 'bird', 'cat', 'deer', 'dog', 'frog', 'horse', 'ship', 'truck')

 

Train DataLoader와 Validation DataLoader의 생성

# validation data를 만들기 위해, split 기준선 만드는 작업
num_train = len(train_dataset)
indices = list(range(num_train))    # train_dataset의 인덱스를 만들기 : 50000개의 인덱스 리스트
split = int(np.floor(val_size * num_train))
	# np.floor : 배열의 각 요소를 내림하여 새로운 배열을 생성
train_idx, val_idx = indices[split:], indices[:split]

train_sampler = SubsetRandomSampler(train_idx)
valid_sampler = SubsetRandomSampler(val_idx)
    # SubsetRandomSampler : Samples elements randomly from a given list of indices

# DataLoader의 역할 : Download한 데이터를 불러오는 역할로, batch size, workers, shuffle 여부 등을 지정하여 load하는 역할
train_loader = DataLoader(train_dataset, batch_size=batch_size,
                                           sampler=train_sampler, num_workers=2)
valid_loader = DataLoader(valid_dataset, batch_size=batch_size,
                                           sampler=valid_sampler, num_workers=2)
    # num_workers : 데이터를 로드할 때 사용할 병렬로드(worker)의 개수를 지정 (데이터 로드 속도 향상 가능)

• SubsetRandomSampler의 역할은 DataLoader에서 data를 batch 단위로 loading할 때 각 sampler에 할당된 index를 무작위적으로 sampling하기 위한 것으로, shuffle의 역할을 하는 것과 동일
• 위 코드의 문제점은, 
  • train_idx와 val_idx가 이미 편향적으로 추출되고 있다는 점에서, overfitting이 발생할 여지가 있다는 것이다.
  • 이를 해결하기 위해서는 다음과 같은 코드를 추가하는 것이 좋아 보인다.

from sklearn.model_selection import train_test_split
train_idx, val_idx = train_test_split(indices, test_size=val_size, random_state=42)
train_sampler = SubsetRandomSampler(train_idx)
val_sampler = SubsetRandomSampler(val_idx)

 

Test DataLoader 생성

# Test Data set
test_dataset = torchvision.datasets.CIFAR10(root=r'.\data',
                                            train=False, download=True,
                                            transform=valid_transform)
test_loader = DataLoader(test_dataset, batch_size=batch_size,
                                          shuffle=False, num_workers=2)

 

데이터를 시각화하여 보고 싶을 때는 다음과 같이 시각화해볼 수 있을 것 같다.

# 이미지 데이터 시각화
def imshow(img):
    img = img / 2 + 0.5
        # 모델 학습을 위해 normalize한 데이터를 복원하기 위해 unnormalize 작업 수행
    npimg = img.numpy()
    plt.imshow(np.transpose(npimg, (1, 2, 0)))
        # (1, 2, 0)의 순서로 차원을 변경
        # PyTorch의 경우 (channel, width, height)로 되어있기 때문에 이를 변경경
    plt.show()

# 학습 이미지 얻기
dataiter = iter(train_loader)
    # train_loader에서 iterator 생성 (iterator로의 변환)
images, labels = next(dataiter)
    # 다음 미니배치를 가져오는 함수
    
# 이미지 출력 : 하나의 grid로 이미지 출력
imshow(torchvision.utils.make_grid(images))
# 라벨 프린트
print(' '.join(f'{classes[labels[j]]:5s}' for j in range(batch_size)))

 

Custom Dataset 및 DataLoader 생성

# Custom Dataset Class 만들기 - test할 때 사용
class CUSTOMDataset(Dataset):
    def __init__(self, mode: str = 'test', transforms: transforms = None):
            # mode : str = 'test' 데이터셋의 모드를 지정하는 것으로, CUSTOMDataset은 test 목적으로 사용되는 데이터임을 명시
            # 이때 transform은 별도로 명시하지 않는 한 데이터 변환을 수행하지 않음을 의미
        self.mode = mode
        self.transforms = transforms
        self.images = []    # 데이터셋의 이미지를 저장할 리스트
        self.labels = []    # 데이터셋의 라벨을 저장할 리스트
        
        for folder in os.listdir(r".\data\custom_dataset\custom_dataset"):
            files = os.path.join(r".\datacustom_dataset\custom_dataset", folder)
            if os.path.isdir(files):
                files_path = os.listdir(files)
                for file in files_path:
                    if file == '._.DS_Store':
                        continue
                    self.images.append(os.path.join(files, file))
                    self.labels.append(classes.index(folder))
                
    def __len__(self) :
        return len(self.labels)
    
    def __getitem__(self, index: int) -> Tuple[Tensor]:
        image = image.open(self.images[index]).convert('RBG')
        if self.transfroms is not None:
            image = self.transfroms(image)
        image = np.array(image)
        label = self.labels[index]
        return image, label
        
# Custom Dataset 형성과 Dataloader 생성
custom_dataset = CUSTOMDataset('test', transforms=valid_transform)
custom_loader = DataLoader(dataset=custom_dataset, batch_size=batch_size,
                           shuffle=False, num_workers=2)

• Custom dataset의 경우 다운로드를 통해 확보된 dataset과는 별개로, 추가적인 test를 진행하기 위해 생성

 

Model Creation

# PyTorch를 활용한 모델 로드 : ResNet-101 모델 사용
model = torch.hub.load('pytorch/vision:v0.10.0', 'resnet101', pretrained=True)
model

• torch.hub.load() 함수를 사용하면 특정 도메인의 다양한 모델을 사용할 수 있다.
• 이번 실습에서는 ResNet-101을 사용하며 ImageNet에서 사용된 학습 가중치를 활용한다.
• ResNet-101 모델의 output shape은 (batch_size, 1000)으로 나오는데, 이는 1개의 mini_batch 내 data_points들에 대해서 1000개의 features를 갖고 있으며, 각 feature들은 해당 feature가 나올 확률분포의 값을 나타내고 있다.
• model 입력시 전체 layer의 갯수와 kernel의 크기, 배치 순서, input size 등 layers의 전반적인 정보들이 나오기 때문에 자세한 architecture를 파악할 수 있다.

# model의 hyper-parameter 정의
learning_rate = 0.001
momentum = 0.9
epochs = 10
best_acc = 0.0

# 목적함수와 optimizer 정의
criterion = nn.CrossEntropyLoss()
optimizer = optim.SGD(model.parameters(), lr = learning_rate, momentum=momentum)

• ADAM을 활용하면 성능이 더 좋게 나올 것으로 예상되나, 가장 기본적인 SGD를 적용해보았다. 
• 목적 손실함수를 CrossEntropyLoss로 사용한 이유는..? 조금더 공부해보고 채우려고 한다.

Train, Validation, Test 과정 Creation

# train function : forward and backward
def train(epoch):
    train_loss = 0.0    # 초기 loss 값
    model.train()
    for i, data in enumerate(tqdm.tqdm(train_loader), 0):
        inputs, labels = data[0], data[1]
        
        # parameter gradients를 초기화 : 새로운 batch에 대한 gradient를 계산하기 전에 계산된 gradient를 초기화
        optimizer.zero_grad()
        
        ## forward propagation step : 연산 수행
        # 입력에 대한 model의 출력 생성
        outputs = model(inputs)
        
        # 손실함수 계산 및 gradient update
        loss = criterion(outputs, labels)
        loss.backward()     # 역전파 실시
        optimizer.step()    # 각 step 별로 최적화 알고리즘 실시 - parameter update
        train_loss += loss.item()
            # 각 에폭에서 계산된 loss를 .item() method를 통해 python scalar 값으로 반환
            # epoch 단위로 loss를 계산하여 누적하는 방식
        
    return train_loss

# validation function : forward
def valid():    # validation은 train과 달리 반복학습을 시키는 것이 아니므로 epoch이 불필요
    val_loss = 0.0
    val_accuracy = 0.0
    
    # 해당 context에 포함된 블록에서는 gradient 계산 혹은 tensor update가 진행되지 않음
    # gradient와 tensor update가 불필요한 경우에 with 구문과 함께 사용
    with torch.no_grad():
        model.eval()
        for i, data in enumerate(tqdm.tqdm(valid_loader), 0):
            inputs, labels = data[0], data[1]
            
            ## forward propagation step : 연산 수행
            # 입력에 대한 model의 출력 생성
            outputs = model(inputs)
            
            # 손실함수 계산
            loss = criterion(outputs, labels)
            val_loss += loss.item()
            
            # accuracy 계산을 위한 예측 라벨
            # output = values, indices (최대값과 해당 index를 반환 / dim = 최대값을 찾을 차원)
            # model을 통과한 output은 (batch_size, 1000)의 shape를 갖고 있고,
            # torch.max()는 1번째 index에 해당하는 datapoints 들에 대해서 가장 큰 값의 index를 반환
            _, predicted = torch.max(input=outputs, dim=1)
            
            # 예측 값과 정답값이 같은 횟수를 모두 더해 val_accuracy 변수에 할당
            val_accuracy += (predicted == labels).sum().item()
    
    return val_loss, val_accuracy

# test function : forward
def test(test_loader):
    correct, total = 0, 0
    correct_class = {classname : 0 for classname in classes}
    total_class = {classname : 0 for classname in classes}
    
    model.eval()
    with torch.no_grad():
        for i, data in enumerate(tqdm.tqdm(test_loader), 0):
            inputs, labels = data[0], data[1]
            
            outputs = model(inputs)
            
            _, predicted = torch.max(input=outputs, dim=1)
            
            total += labels.size(0)
                # .size() : 텐서의 크기를 나타내는 method
                # ()안의 수는 텐서의 크기 index를 입력하는 것
            correct += (predicted == labels).sum().item()
            
            for label, prediction in zip(labels, predicted):
                if label == prediction:
                    correct_class[classes[label]] += 1
                total_class[classes[label]] += 1
    
    print(f"Accuracy of the network on the 10000 test images: {100 * correct // total} %")
    
    for classname, correct_count in correct_class.items():
        if total_class[classname] == 0:
            continue
        accuracy = 100 * float(correct_count) / total_class[classname]
        print(f"Accuracy for class : {classname:5s} is {accuracy:.1f} %")

 

Model Execution : Train and Validation

# 모델 저장 경로 정의
model_path = r'.\cifar_resnet101.pth'
for epoch in range(epochs):
  # 학습 메소드 실행
  train_loss = train(epoch)
  print(f'[{epoch + 1}] loss: {train_loss / len(train_loader):.3f}')
  # 검증 메소드 실행
  val_loss, val_acc = valid()
  vaild_acc = val_acc / (len(valid_loader)*batch_size)
  print(f'[{epoch + 1}] loss: {val_loss / len(valid_loader):.3f} acc: {vaild_acc:.3f}')
  # 정확도가 기존 베스트를 갱신할 경우 모델 저장
  if vaild_acc >= best_acc:
    best_acc = vaild_acc
    torch.save(model.state_dict(), model_path)
print('Finished Training')

• 결과는 아래와 같은 형식으로 나온다

# evaluate custom dataset
model.load_state_dict(torch.load(model_path))
    # torch.load(model_path) : 지정된 경로에 저장된 모델의 상태 사전을 불러오기
    # model.load_state_dict : 모델에 불러온 상태 사전을 적용
test(custom_loader)

# evaluate test dataset
model.load_state_dict(torch.load(model_path))
test(test_loader)