Pytorch 기본기 - 5

• 데이터 출처 : Kaggle

구글 코랩에서 실행하기

---

Kaggle simpsons data example

• 심슨에 나온 다양한 캐릭터를 분류하는 classification model 연습

Kaggle data 불러오기

# Kaggle json 넣기
!pip install kaggle
from google.colab import files
files.upload()

• Kaggle API 에서 얻은 kaggle.json 입력

# 데이터 다운로드 kaggle api
!mkdir -p ~/.kaggle
!cp kaggle.json ~/.kaggle/kaggle.json
!chmod 600 ~/.kaggle/kaggle.json
!ls -al ~/.kaggle/


!kaggle datasets download -d alexattia/the-simpsons-characters-dataset
!unzip -qq "the-simpsons-characters-dataset.zip"

사용할 패키지 불러오기

import torch
import torch.nn as nn
import torch.optim as optim
from torch.optim import lr_scheduler
from torch.autograd import Variable
import numpy as np
import torchvision
from torch.utils.data import DataLoader
from torchvision import datasets, models, transforms
import matplotlib.pyplot as plt
from google.colab.patches import cv2_imshow
import os
import shutil
import cv2
from tqdm import tqdm
import copy
from glob import glob
from torchvision.utils import make_grid
from torch.utils.data.sampler import SubsetRandomSampler

이미지 증분 및 train : val 분할

• class 당 이미지가 적어서 증분 사용

# train, val, test dataset 정의
def train_val(image_dir, ratio, transforms, batch):

  train_data = datasets.ImageFolder(image_dir, transforms)
  train_val_ratio = ratio

  # train, val 분할
  num_train = len(train_data)
  indices = list(range(num_train))
  np.random.shuffle(indices) # shuffle
  split = int(np.floor(train_val_ratio * num_train))
  train_idx, valid_idx = indices[split:], indices[:split]

  # batch sampler 정의
  train_sampler = SubsetRandomSampler(train_idx)
  valid_sampler = SubsetRandomSampler(valid_idx)

  # dataloader
  train_loader = DataLoader(train_data,
                            batch_size = batch,
                            sampler = train_sampler,
                            num_workers = 2)

  val_loader = DataLoader(train_data,
                            batch_size = batch,
                            sampler = train_sampler,
                            num_workers = 2)


  return train_loader, val_loader , train_data.classes

# 이미지 전처리 지정
data_transforms = transforms.Compose([
          transforms.Resize((256,256)),
          transforms.RandomHorizontalFlip(),
          transforms.ToTensor(),
          transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
      ])


train_loader, val_loader, class_name = train_val('/content/simpsons_dataset/simpsons_dataset', 0.2, data_transforms, 32)

데이터 시각화

def show_image(image):
  image_test = image.numpy().transpose((1,2,0))
  mean = np.array([0.485, 0.456, 0.406])
  std = np.array([0.229, 0.224, 0.225])
  image_test = std * image_test + mean
  image_test = np.clip(image_test, 0, 1)

  plt.imshow(image_test)
  # plt.title(label)
  plt.show()

# 배치별 데이터 가져오기
iterator = iter(train_loader)

# 배치 중에서 하나꺼내서 시각화
images, classes = next(iterator)
image = make_grid(images)

show_image(image)
print(f'class : {[class_name[i] for i in classes]}')

모델 설계

• 전이학습 기법을 사용 전체 layer를 학습에 사용
• resnet50 model 합성곱 신경망 미세조정 기법 사용

# 마지막 layer 변경
model_classification = torchvision.models.resnet50(pretrained=True)
num_fc = model_classification.fc.in_features
model_classification.fc = nn.Linear(num_fc, len(class_name))


# loss 지정
criterion = nn.CrossEntropyLoss()

# optimizer 설정
optimizer = optim.Adam(model_classification.parameters(), lr=0.001)

# gpu
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")

model = model_classification.to(device)

# epoch
num_epochs = 100
min_loss = int(1e9)

모델 훈련

• train 함수

def model_train(model, data_loader, loss_fn, optimizer, device):

    model.train()

    # loss와 accuracy 계산을 위한 임시 변수 입니다. 0으로 초기화합니다.
    running_loss = 0
    corr = 0

    # mini-batch 학습을 시작합니다.
    for img, lbl in tqdm(data_loader):
        # image, label 데이터를 device에 올립니다.
        img, lbl = img.to(device), lbl.to(device)

        # 누적 Gradient를 초기화
        optimizer.zero_grad()

        output = model(img)

        # 손실 계산
        loss = loss_fn(output, lbl)

        # 미분 값을 계산
        loss.backward()

        # 계산된 Gradient를 업데이트
        optimizer.step()

        # output의 max(dim=1)은 max probability와 max index를 반환
        # max probability는 무시하고, max index는 pred에 저장하여 label 값과 대조하여 정확도를 계산
        _, pred = output.max(dim=1)

        # 정확히 맞춘 label의 합계를 계산
        corr += pred.eq(lbl).sum().item()

        # 평균 손실(loss) *  배치사이즈(batch size) =  배치의 전체 loss가 계산
        # 이를 누적한 뒤 Epoch 종료시 전체 데이터셋의 개수로 나누어 평균 loss를 산출
        running_loss += loss.item() * img.size(0)

    # 누적된 정답수를 전체 개수로 나누어 주면 정확도 계산
    acc = corr / len(data_loader.dataset)

    # 평균 손실(loss)과 정확도를 반환
    return running_loss / len(data_loader.dataset), acc

• evaluate 함수

def model_evaluate(model, data_loader, loss_fn, device):
    # model.eval()은 모델을 평가모드로 설정
    model.eval()

    # Gradient 업데이트 방지 
    with torch.no_grad():

        corr = 0
        running_loss = 0


        for img, lbl in tqdm(data_loader):
            img, lbl = img.to(device), lbl.to(device)
            output = model(img)
            _, pred = output.max(dim=1)
            corr += torch.sum(pred.eq(lbl)).item()
            running_loss += loss_fn(output, lbl).item() * img.size(0)

        acc = corr / len(data_loader.dataset)

        return running_loss / len(data_loader.dataset), acc

• 훈련 과정
  • train -> eval -> best_model save 순서

# Epoch 별 훈련 및 검증을 수행합니다.
for epoch in range(num_epochs):
    # 훈련 손실과 정확도를 반환 
    train_loss, train_acc = model_train(model, train_loader, criterion, optimizer, device)

    # 검증 손실과 검증 정확도 반환
    val_loss, val_acc = model_evaluate(model, val_loader, criterion, device)   

    # val_loss 가 개선되었다면 min_loss를 갱신, model의 가중치(weights)를 저장
    if val_loss < min_loss:
        print(f'model loss 갱신 {min_loss:.5f} -> {val_loss:.5f}')
        min_loss = val_loss
        torch.save(model.state_dict(), 'best_model.pth')

    # Epoch 별 결과를 출력
    print(f'epoch {epoch+1:02d}, loss: {train_loss:.5f}, acc: {train_acc:.5f}, val_loss: {val_loss:.5f}, val_accuracy: {val_acc:.5f}')

테스트 데이터로 검증

• 훈련이 끝난 최고 score 모델 불러오기

checkpoint =  torch.load('best_model.pth')

model_classification = torchvision.models.resnet50(pretrained=True)
num_fc = model_classification.fc.in_features
model_classification.fc = nn.Linear(num_fc, len(class_name))
model = model_classification.to(device)

model.load_state_dict(checkpoint)

• test 데이터 label, data 분리

test_data = glob('./kaggle_simpson_testset/kaggle_simpson_testset/*.jpg')

os.mkdir('test_data')

for file_name in test_data:
  sep = file_name.split('kaggle_simpson_testset/')[-1].split('_')

  name = '_'.join(sep[:-1])
  number = sep[-1]

  if os.path.exists(f'./test_data/{name}') == False:
    os.mkdir(f'./test_data/{name}')


  shutil.move(file_name, f'./test_data/{name}/{number}')

• test loader 설정

test = datasets.ImageFolder('./test_data', data_transforms)
ans_class = test.classes
test_loader = DataLoader(test,
                         shuffle=True,
                          batch_size = 1,
                          num_workers = 2)

• 결과 확인

# 배치별 데이터 가져오기
iterator = iter(test_loader)
images, classes = next(iterator)


with torch.no_grad():
  model.eval()
  outputs = model(images.to(device))
  _, preds  = torch.max(outputs, dim=1)


  image = make_grid(images)
  show_image(image)
  print()
  print(f'pred_class : {class_name[preds[0].item()]}')
  print(f'answer : {ans_class[classes.item()]}')