Pytorch Resnet 예시코드(CNN, Resnet18, Cats and Dogs Dataset, Augmentation)

1. Data Set 준비하기

1.1 Data .zip 파일 다운받기

!wget --no-check-certificate \
https://storage.googleapis.com/mledu-datasets/cats_and_dogs_filtered.zip \
-O ./cats_and_dogs_filtered.zip

1.2 .zip 파일 압축 풀기

! unzip -q cats_and_dogs_filtered.zip -d ./

1.3 파일 구조

"cats and dogs filtered" 폴더 내부에서 "train", "validation" 폴더로 나누어진다.

각각 "cats", "dogs" 폴더를 가지며, 그 내부에는 해당하는 이미지들이 들어있다.

2. Data Preprocessing

2.1 이미지 경로&이름 리스트 생성하기

이미지 데이터는 수치형 데이터와는 다르게 한개의 csv 파일에 있는 것이 아닌, 계층적 폴더에 담겨있다.

따라서 우리는 이러한 이미지들의 경로&이름 정보를 가지는 하나의 리스트를 생성하여 관리한다.

from glob import glob

# * : 어떠한 값이든 들어올 수 있다
# ./cats_and_dogs_filtered/train/ 경로에 있는 모든 폴더(**)에 있는 g로 끝나는 파일들 모두
train_images = glob("./cats_and_dogs_filtered/train/**/*g")
# ./cats_and_dogs_filtered/validation/ 경로에 있는 모든 폴더(**)에 있는 g로 끝나는 파일들 모두
valid_images = glob("./cats_and_dogs_filtered/validation/**/*g")

#g로 끝나는 파일은 jpg, png 같은 이미지 파일만을 받아오기 위함이다.

# Train / Validation 개수 확인
print(f"train len : {len(train_images)}, validation len : {len(valid_images)}")
#=> train len : 2000, validation len : 1000

./cats_and_dogs_filtered/train/ 경로있는 cats와 dogs 폴더로부터 경로&이름 정보를 가져온 것을 확인할수 있다.

2.2 이미지 변환 함수 및 시각화 함수 생성

이미지를 처리하기 위해서 jpg 형태가 아닌 파이썬에서 사용가능한 PIL 형태로 변경해주어야한다.

load_images 함수는 이러한 형변환을 담당한다.

show_images 함수는 이미지의 확인 목적으로 시각화 해주는 함수이다.

import matplotlib.pyplot as plt
import PIL
%matplotlib inline

# 주어진 경로의 이미지를 PIL 객체로 변환해주는 함수
def load_images(image_paths):
    return [PIL.Image.open(image).convert('RGB') for image in image_paths]
    # 'RGB' 형식으로 변환도 해준다.

# 이미지, 이미지 라벨, fontsize를 입력받아 이미지 시각화 해주는 함수
def show_images(images, labels=None, fontsize=12):
    # 주어진 조건이 False일 경우에 메시지 출력
    assert len(images) > 1, "최소 2장의 이미지 경로값을 넣어주세요"
    
    # 여러개의 그림을 한줄로 출력하기 위한 subplots 선언
    # 1*{입력된 이미지 경로수} 차원의 subplots
    _, axes = plt.subplots(1, len(images), 
                           figsize=(2 * len(images), 20), 
                           subplot_kw={'xticks': [], 'yticks': []})

    # subplots의 각 칸에 이미지를 순차적으로 load 하기
    for i, ax in enumerate(axes.flat):
        ax.imshow(images[i])
        if labels is not None:
            ax.set_title(labels[i], fontsize=fontsize)

    plt.show()

load_images 함수를 사용하여 이미지 경로로부터 이미지 객체(PIL)를 만들고 이를 show_image에 넣어서 시각화 해본다.

image_path_list = train_images[:3] + train_images[-3:]
show_images(load_images(image_path_list))

2.3 이미지 라벨 생성

해당 포스팅에서의 예측 목적은 "dogs"와 "cats"의 분류이다.

따라서 이미지에 대해서 라벨을 붙여줘야한다.

# 입력받은 경로에서 뒤에서 두번째 위치의 폴더 이름을 파싱한다.
def get_label(image_path):
    return image_path.split("/")[-2]

이미지 경로에 대해서 '/'로 split할경우 [-2]의 위치에는 label에 해당하는 폴더이름이 위치한다.

따라서 위의 get_label 함수로 이미지 별로 라벨링을 진행할 수 있다.

# get_label 함수를 모든 train_images,valid_images에 적용하기
train_labels = [get_label(image_path) for image_path in train_images]
valid_labels = [get_label(image_path) for image_path in valid_images]

2.4 이미지 데이터셋 Class 생성 및 변환

수치형 데이터처럼 어떠한 DataFrame에 넣어서 다룰 수 있지 않기 때문에 이미지에 맞는 데이터구조를 만들어준다.

import torch
from torch.utils.data import Dataset

class ImageDataset(Dataset):
    def __init__(self, image_paths, labels, transform=None):
        self.image_paths = image_paths
        self.labels = labels
        self.transform = transform

        self.classes = sorted(set(labels))
        self.class_to_index = {c: i for i, c in enumerate(self.classes)}
        self.onehot_labels = [self.class_to_index[label] for label in labels]
        self.onehot_vectors = torch.eye(len(self.classes))[self.onehot_labels]

    def __len__(self):
        return len(self.image_paths)

    def __getitem__(self, index):
        image = load_images([self.image_paths[index]])[0]
        if self.transform:
            image = self.transform(image)
        return image, self.onehot_vectors[index]

이때 몇가지 필드의 값 예시를 들자면 아래와 같다.

예측모델이 활용할 수 있는 형식의 데이터를 준비하는 과정이라고 볼수 있다.

torch.eye()는 단위벡터를 만들어주는 기능이지만,
추가 인덱싱을 통해서 onehot_vector를 만드는데 사용하였다.

하단에 간단한 torch.eye() 사용 예시를 첨부한다.

train/validation 데이터셋을 앞서 생성한 이미지 class에 넣어준다.

from torchvision import transforms

train_data = ImageDataset(train_images, train_labels,transform=transforms.ToTensor())

train_data.__getitem__(0)

0번째 이미지를 뽑아보면 RGB 형태의 벡터로 변환된 것을 확인할 수 있다.

라벨정보 또한 onehot vector 형태임을 확인 가능하다.

3. 모델 생성 및 학습

3.1 모델 생성

원래는 레이어를 일일이 쌓아야하지만, pythorch에서 제공하는 torchvision.models.resnet18을 사용하여 이 과정을 생략할 수 있다.

import torch
from torchvision.models import resnet18

class CatDogModel(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.model = resnet18(weights=None, num_classes=len(set(train_labels)))

    # 순전파, 입력벡터(X)를 넣어서 예측 결과를 받는 매서드
    def forward(self, x):
        return self.model(x)

    # 예측 결과를 받는 매서드
    # argmax는 가장 높은 수를 1로 변환하고 나머지를 0으로 변환하는 기법
    # [8,12,30] => [0,0,1]
    def predict(self, x):
        return torch.argmax(self.forward(x), dim=1)

    # 예측확률 결과를 받는 매서드
    # softmax는 모든 수의 합을 1로 맞추는 기법
    # [8,12,30] => [0.16,0.24,0.6]
    def predict_proba(self, x):
        return torch.softmax(self.forward(x), dim=1)

from torchinfo import summary

model = CatDogModel() # 모델 생성
summary(model, input_size=(1, 3, 224, 224)) # 모델 요약정보 생성

3.2 디바이스 지정

Pytorch에서는 학습에서 사용할 디바이스를 지정해줘야한다.

Tensorflow2에서는 기본으로 gpu가 잡혀있어 지정이 필요없지만, pytorch에서는 지정을 해주어야한다.

# gpu(cuda:0)가 사용가능하다면 지정하고 아니면 cpu 지정하는 코드
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
print(device) #=> cuda:0

# cuda:0이 나오면 gpu가 설정되었다는 의미

3.3 학습 함수 구현 및 학습

Tensorflow2에서는 .fit으로 간단히 수행이 가능했지만 pytorch에서는 아래와 같이 학습함수를 구현해주어야한다.

다른 방법으로는 Tensorflow의 keras와 같이 pytorch에도 pytorch lightning을 사용해 좀더 간편하게 수행 가능하다.

import time
import os
from tqdm.auto import trange

def train_all_in_one(train_dataset, valid_dataset):
    
    # 모델 선언 및 디바이스 지정(gpu)
    # Pytorch는 모델이 사용할 디바이스를 지정해줘야한다.
    model = CatDogModel().to(device)
    
    # 최적화 함수 및 손실함수 지정
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
    loss_fn = torch.nn.CrossEntropyLoss()
    
    # Train/valid 데이터 DataLoader에 넣기
    # DataLoader는 학습 및 평가시에 batch 단위로 데이터를 준비해주는 기능
    # num_workers : 병렬작업할 쓰레드 개수
    train_loader = torch.utils.data.DataLoader(
        train_dataset, batch_size=32, shuffle=True, num_workers=16)
    valid_loader = torch.utils.data.DataLoader(
        valid_dataset, batch_size=32, shuffle=False, num_workers=16)
    
    # early_stop기법 구현을 위한 변수
    # early stop : validation set에 대한 성능향상이 이루어지지 않으면 학습을 중단하는 기법
    early_stop_max_round = 20 # validation set에 대한 성능이 개선되기까지의 dead line
    # 최고성능 갱신 후 20 epoch안에 다시 최고성능 갱신을 해야한다
    early_stop_now_round = early_stop_max_round # 성능 갱신까지 남은 기회
    
    # 성능 출력을 위한 함수
    highest_accuracy = 0.0 # 최고 정확도
    lowest_loss = 1000.0 # 최저 loss
    
    print("#### Start Learning ####")
    
    # 반복학습 구간
    # trange(201)은 range(201)과 같은나, 진행상황을 하단에 게이지바로 보여줌
    for epoch in trange(201):
        start_time = time.time() # epoch 시작시간 저장
        
        ##### Train part #####
        model.train() # 모델을 train 모드로 설정
        train_correct = 0 # epoch에서 맞춘 데이터 개수
        train_total = 0 # epoch에서 사용된 데이터 개수
        
        for batch in train_loader: # 데이터로더에서 batch별로 받아오기
            images, labels = batch # X, Y로 나누어 받기
            images = images.to(device) # 처리할 디바이스 지정
            labels = labels.to(device) # 처리할 디바이스 지정
            
            optimizer.zero_grad() # 최적화함수 초기화
            predictions = model(images) # batch별로 예측하기
            train_loss = loss_fn(predictions, labels) # 예측결과로 loss 계산
            train_loss.backward() # 각 hidden layer의 node들에게 역전파
            optimizer.step() # 역전파 내용 기준 node의 weight 수정
            
            # 예측 결과가 정답인 것 개수 세기
            train_correct += (predictions.argmax(dim=1) ==
                              labels.argmax(dim=1)).sum().item()
            # 사용된 데이터 개수 세기
            train_total += len(labels)
            # 맞춘개수/데이터개수
            train_accuracy = train_correct / train_total
        
        ##### Validation part #####
        model.eval() # 모델을 eval 모드로 설정 (weight 갱신이 일어나지 않음)
        with torch.no_grad():
            valid_correct = 0
            valid_total = 0
            losses = []
            for batch in valid_loader:
                images, labels = batch
                images = images.to(device)
                labels = labels.to(device)

                predictions = model(images)
                valid_loss = loss_fn(predictions, labels)
                losses.append(valid_loss.item())

                valid_correct += (predictions.argmax(dim=1) ==
                            labels.argmax(dim=1)).sum().item()
                valid_total += len(labels)
                
        valid_accuracy = valid_correct / valid_total
        valid_loss = sum(losses) / len(losses)
        
        # 기존 최고점수보다 높을 경우 갱신(관찰용)
        if valid_accuracy > highest_accuracy:
            highest_accuracy = valid_accuracy
        
        # 최저 loss일경우
        if valid_loss < lowest_loss:
            lowest_loss = valid_loss # 최저 loss 갱신
            early_stop_now_round = early_stop_max_round # now_round를 max값으로 초기화
            print("Get a new best loss: {:.4f}".format(valid_loss))
            torch.save(model.state_dict(), "./md_da_model_temp.ckpt") # 해당 모델 파라미터 저장
        # lowest_loss보다 현재 loss가 높다면 early_stop_now_round를 1 감소
        # 0이되면 학습을 멈추기 위해서 카운트 중
        elif valid_loss > lowest_loss:
            early_stop_now_round -= 1
            
        # 1epoch가 종료되었으므로 종료시간 측정
        end_time = time.time()
        
        # 1epoch에 대한 정보 출력
        print(
            f"Epoch {epoch}\t"
            f"time: {end_time - start_time:.3f}s\t"
            f"train-acc: {train_accuracy:.3f}\t"
            f"valid-acc: {valid_accuracy:.3f}\t"
            f"highest-acc: {highest_accuracy:.3f}\t"
            f"valid-loss: {valid_loss:.3f}\t"
            f"lowest-loss: {lowest_loss:.3f}"
        )
        
        # early_stop_now_round가 0이라면 학습 중단
        if early_stop_now_round == 0:
            break
            
    print("#### Finish Learning ####")
    
    # 저장되어 있는 가장 좋았던 epoch의 모델 정보 불러와 덮어쓰기
    model.load_state_dict(torch.load("./md_da_model_temp.ckpt"))
    # 중간저장 파일 삭제
    os.remove("./md_da_model_temp.ckpt")
    
    # 모델 반환
    return model

만들어진 함수를 통해서 학습을 수행한다.

# 이미지에 적용해줄 전처리 기법 지정
raw_transform = transforms.Compose([
    transforms.Resize((224, 224)), # 사이즈를 224*224로 맞춤
    transforms.ToTensor(), # 이미지 형태를 Tensor로 변환
])

# train/validation 데이터셋을 ImageDataset 객체에 저장
raw_train_dataset = ImageDataset(train_images, train_labels, transform=raw_transform)
raw_valid_dataset = ImageDataset(valid_images, valid_labels, transform=raw_transform)

# 학습함수 호출
raw_model = train_all_in_one(raw_train_dataset, raw_valid_dataset)

200 epoch를 돌게 입력되었으나, early stopping으로 인해서 27epoch에서 학습이 종료되었다.

게이지바가 빨간색인 이유는 중단되었기 때문이다.(원래는 파란색임)

 

3.4 테스트 함수 구현

원래는 데이터를 Train, Validation, Test 세가지를 준비해야하지만,

본 포스팅에서는 Test 데이터를 준비하지 못한 관계로 Validation dataset을 Test에도 사용한다.

import random

def test_all_in_one(model, test_dataset):
    test_loader = torch.utils.data.DataLoader(
        test_dataset, batch_size=32, shuffle=False, num_workers=16)

    correct_images = []
    correct_labels = []
    correct_predictions = []

    incorrect_images = []
    incorrect_labels = []
    incorrect_predictions = []

    correct = 0
    total = 0

    model.eval()
    with torch.no_grad():
        for batch in test_loader:
            images, labels = batch
            images = images.to(device)
            labels = labels.to(device)

            predictions = model(images)
            correct += (predictions.argmax(dim=1) ==
                        labels.argmax(dim=1)).sum().item()
            total += len(labels)

            for i in range(len(images)):
                if predictions.argmax(dim=1)[i] == labels.argmax(dim=1)[i]:
                    correct_images.append(images[i].cpu())
                    correct_labels.append(labels[i].cpu())
                    correct_predictions.append(predictions[i].cpu())
                else:
                    incorrect_images.append(images[i].cpu())
                    incorrect_labels.append(labels[i].cpu())
                    incorrect_predictions.append(predictions[i].cpu())

    accuracy = correct / total
    print(f"Accuracy: {accuracy:.3f}")

    # Convert images to PIL images
    correct_images = [transforms.ToPILImage()(image) for image in correct_images]
    incorrect_images = [transforms.ToPILImage()(image) for image in incorrect_images]

    # Convert labels to class names
    correct_labels = [test_dataset.classes[label.argmax()] for label in correct_labels]
    incorrect_labels = [test_dataset.classes[label.argmax()] for label in incorrect_labels]

    # Convert predictions to class names
    correct_predictions = [test_dataset.classes[prediction.argmax()] for prediction in correct_predictions]
    incorrect_predictions = [test_dataset.classes[prediction.argmax()] for prediction in incorrect_predictions]

    # Shuffle the images, labels, and predictions
    c = list(zip(correct_images, correct_labels, correct_predictions))
    random.shuffle(c)
    correct_images, correct_labels, correct_predictions = zip(*c)

    c = list(zip(incorrect_images, incorrect_labels, incorrect_predictions))
    random.shuffle(c)
    incorrect_images, incorrect_labels, incorrect_predictions = zip(*c)

    # Print 10 correct predictions
    print("Correct predictions:")
    show_images(correct_images[:10], correct_labels[:10])

    # Print 10 incorrect predictions
    print("Incorrect predictions:")
    show_images(incorrect_images[:10], [f"{prediction}\n({label})" for label, prediction in zip(incorrect_labels[:10], incorrect_predictions[:10])])

만들어진 함수를 통해서 테스트를 수행한다.

raw_test_dataset = ImageDataset(valid_images, valid_labels, transform=raw_transform)

test_all_in_one(raw_model, raw_test_dataset)

아래 그림을 통해서 66%의 정확도를 보임을 알수 있다.

4. Augmentation 기법을 통한 이미지 증강 및 모델 성능 개선

augmentation은 원본이미지에 어떠한 기법을 가하여 변형된 이미지를 만들어내는 기법이다.

이는 모델이 노이즈에 좀더 강건하게 해주며 데이터 수를 증가시키는 효과를 가져온다.

해당 포스팅에서는 간단히 사용 예시만 다루어본다.

# 기존에 resize 및 totensor만 있는 transforms에 여러개의 augmentation 기법을 추가하였다.
torch_transform = transforms.Compose([
    transforms.RandomHorizontalFlip(),
    transforms.RandomAffine(30),
    transforms.RandomRotation(12),
    transforms.RandomResizedCrop(224),
    transforms.AugMix(),
    transforms.Resize((224, 224)),
    transforms.ToTensor(),
])

# 앞서 구현한 학습 및 테스트 함수를 사용해준다.
torch_transform_train_dataset = ImageDataset(
    train_images, train_labels, transform=torch_transform)
torch_transform_valid_dataset = ImageDataset(
    valid_images, valid_labels, transform=torch_transform)

torch_transform_model = train_all_in_one(
    torch_transform_train_dataset, torch_transform_valid_dataset)

위에서와는 다르게 200 epoch가 모두 돌아갔다.

이는 지속적으로 best socore가 갱신되었으며 모델이 아직 underfitting임을 의미한다.

또한 성능이 85%로 이전보다 상당히 개선되었음을 알수 있다.