deep learning/computer vision
ResNet-101 Classification 실습 연습
문과 열등생
2024. 2. 11. 23:53
Computer Vision에서의 classification model들에 대한 outline을 검토하면서, 대표적인 모델인 ResNet-101을 활용하여 CIFAR10 데이터셋에 대한 기본적인 classification 모델을 실습해보고자 한다.
PyTorch를 활용하여 Classfication 학습 및 테스트 모델을 만들어보면서,
- 모델 구축에 사용되는 PyTorch 문법을 연습해보고,
- Classification Model의 전체적인 Architecture를 학습하고,
- 주요 Hyper-Parameter들에 대한 조정을 통한 성능 변화 여부 검토,
- 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)이 가능
- Zero-centered input
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)