딥러닝 파이토치 교과서 (6-1: LeNet)

LeNet-5 는 합성곱 신경망이라는 개념을 최초로 얀 르쿤이 개발한 구조이다. LeNet-5 는 Convolution 과 Sub-sampling (혹은 풀링) 을 반복적으로 거치면서 마지막에 완전연결층에서 분류를 수행한다. 아래 그림과 같은 구조로 네트워크가 구성된다.

 

 

 

LeNet

 

 

 

이전에 사용한 개와 고양이 데이터셋을 다시 사용하여 LeNet 모델을 실습해보았다. 32 x 32 크기의 이미지에 Convolution 과 Max pooling 이 쌍으로 두 번 적용된 후 완전연결층을 거쳐 이미지가 분류되는 신경망인 것이다. 이번에도 우선 필요한 라이브러리들을 import 해주고 시작한다.

 

 

 

import torch
import torchvision
from torch.utils.data import DataLoader, Dataset
from torchvision import transforms
from torch.autograd import Variable
from torch import optim
import torch.nn as nn
import torch.nn.functional as F
import os
import cv2
from PIL import Image
from tqdm import tqdm_notebook as tqdm
import random
import matplotlib.pyplot as plt

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

 

 

 

우선 모델 학습에 필요한 데이터셋의 전처리를 위한 클래스를 작성한다.

 

 

 

class ImageTransform():    
    def __init__(self, resize, mean, std):
        self.data_transform = {
            'train': transforms.Compose([
                transforms.RandomResizedCrop(resize, scale=(0.5, 1.0)),
                transforms.RandomHorizontalFlip(),
                transforms.ToTensor(),
                transforms.Normalize(mean, std)
            ]),
            'val': transforms.Compose([
                transforms.Resize(256),
                transforms.CenterCrop(resize),
                transforms.ToTensor(),
                transforms.Normalize(mean, std)
            ])
        }
        
    def __call__(self, img, phase):
        return self.data_transform[phase](img)

 

 

 

이전에도 사용해보았듯이, torchvision 을 사용하면 이미지 전처리를 손쉽게 진행할 수 있다. 이번에는 전과 다르게 Compose 메서드 내에 Normalize 함수를 사용해준다. 전이학습에서 사용하는 사전 훈련된 모델들은 대개 ImageNet 데이터셋에서 훈련되었다. 사전 훈련된 모델을 사용하기 위해서는 ImageNet 데이터의 각 채널별 평균과 표준편차에 맞는 정규화를 진행해줄 필요가 있다고 한다. 참고로, class 내의 __call__ 함수는 클래스를 호출할 수 있도록 하는 메서드이다.

 

 

 

cat_directory = '/content/drive/MyDrive/Colab Notebooks/딥러닝파이토치/data/dogs-vs-cats/Cat'
dog_directory = '/content/drive/MyDrive/Colab Notebooks/딥러닝파이토치/data/dogs-vs-cats/Dog'

cat_images_filepaths = sorted([os.path.join(cat_directory, f) for f in os.listdir(cat_directory)])   
dog_images_filepaths = sorted([os.path.join(dog_directory, f) for f in os.listdir(dog_directory)])
images_filepaths = [*cat_images_filepaths, *dog_images_filepaths]    
correct_images_filepaths = [i for i in images_filepaths if cv2.imread(i) is not None]    

random.seed(42)    
random.shuffle(correct_images_filepaths)
train_images_filepaths = correct_images_filepaths[:400]    
val_images_filepaths = correct_images_filepaths[400:-10]  
test_images_filepaths = correct_images_filepaths[-10:]    
print(len(train_images_filepaths), len(val_images_filepaths), len(test_images_filepaths))

 

 

 

위 코드를 통해서 각각 Cat, Dog 폴더에 들어있는 이미지들을 train, val, test set 으로 분할하여 불러온다. 이번에도 이미지를 한 번 시각화를 통해 확인해본다.

 

 

 

def display_image_grid(images_filepaths, predicted_labels=(), cols=5):
    rows = len(images_filepaths) // cols
    figure, ax = plt.subplots(nrows=rows, ncols=cols, figsize=(12, 6))
    for i, image_filepath in enumerate(images_filepaths):
        image = cv2.imread(image_filepath)
        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        true_label = os.path.normpath(image_filepath).split(os.sep)[-2]
        predicted_label = predicted_labels[i] if predicted_labels else true_label
        color = "green" if true_label == predicted_label else "red"
        ax.ravel()[i].imshow(image)
        ax.ravel()[i].set_title(predicted_label, color=color)
        ax.ravel()[i].set_axis_off()
    plt.tight_layout()
    plt.show()
display_image_grid(test_images_filepaths)

 

 

result

 

 

 

위에서 사용된 os.path.normpath 는 경로명을 정규화한다. 그 후 split(os.sep) 함수는 경로를 / 혹은 \ 을 기준으로 분할할 때 사용한다. 

 

 

 

이제 위 데이터를 DataSet, DataLoader 을 활용하여 훈련에 사용할 수 있게끔 한다.

 

 

 

size = 224
mean = (0.485, 0.456, 0.406)
std = (0.229, 0.224, 0.225)
batch_size = 32

class DogvsCatDataset(Dataset):    
    def __init__(self, file_list, transform=None, phase='train'):    
        self.file_list = file_list
        self.transform = transform
        self.phase = phase
        
    def __len__(self):
        return len(self.file_list)
    
    def __getitem__(self, idx):       
        img_path = self.file_list[idx]
        img = Image.open(img_path)        
        img_transformed = self.transform(img, self.phase)
        
        label = img_path.split('/')[-1].split('.')[0]
        if label == 'dog':
            label = 1
        elif label == 'cat':
            label = 0
        return img_transformed, label
        
train_dataset = DogvsCatDataset(train_images_filepaths, transform=ImageTransform(size, mean, std), phase='train')
val_dataset = DogvsCatDataset(val_images_filepaths, transform=ImageTransform(size, mean, std), phase='val')

train_dataloader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
val_dataloader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False)
dataloader_dict = {'train': train_dataloader, 'val': val_dataloader}

batch_iterator = iter(train_dataloader)
inputs, label = next(batch_iterator)
print(inputs.size())
print(label)

 

 

result

 

 

 

이제 준비가 다 되었으니 모델을 정의해준다. 아래 코드는 LeNet 모델이다.

 

 

 

class LeNet(nn.Module):
    def __init__(self):
        super(LeNet, self).__init__()
        self.cnn1 = nn.Conv2d(in_channels=3, out_channels=16, kernel_size=5, stride=1, padding=0) 
        self.relu1 = nn.ReLU() 
        self.maxpool1 = nn.MaxPool2d(kernel_size=2) 
        self.cnn2 = nn.Conv2d(in_channels=16, out_channels=32, kernel_size=5, stride=1, padding=0) 
        self.relu2 = nn.ReLU() # activation
        self.maxpool2 = nn.MaxPool2d(kernel_size=2)         
        self.fc1 = nn.Linear(32*53*53, 512) 
        self.relu5 = nn.ReLU()         
        self.fc2 = nn.Linear(512, 2) 
        self.output = nn.Softmax(dim=1)        
    
    def forward(self, x):
        out = self.cnn1(x) 
        out = self.relu1(out)
        out = self.maxpool1(out)
        out = self.cnn2(out) 
        out = self.relu2(out) 
        out = self.maxpool2(out) 
        out = out.view(out.size(0), -1) 
        out = self.fc1(out) 
        out = self.fc2(out)                    
        out = self.output(out)
        return out

 

 

 

처음 그림으로 확인해보았던 것처럼 Convolution, Maxpool 을 두 번 거치고 완전연결층으로 연결되는 것을 확인할 수 있다. pip install torchsummary 를 하여 모델에 대한 정보를 아래와 같이 확인해볼 수 있다.

 

 

 

model = LeNet().to(device)

from torchsummary import summary
summary(model, input_size=(3, 224, 224))

 

 

result

 

 

 

이제 손실함수, 최적화함수를 만들어준 후 훈련을 진행해주면 되겠다. 아래 코드로 진행해준다.

 

 

 

import time

optimizer = optim.SGD(model.parameters(), lr=0.001, momentum=0.9)
criterion = nn.CrossEntropyLoss()
model = model.to(device)
criterion = criterion.to(device)

def train_model(model, dataloader_dict, criterion, optimizer, num_epoch):    
    since = time.time()
    best_acc = 0.0
    
    for epoch in range(num_epoch):
        print('Epoch {}/{}'.format(epoch + 1, num_epoch))
        print('-'*20)
        
        for phase in ['train', 'val']:           
            if phase == 'train':
                model.train()
            else:
                model.eval()
                
            epoch_loss = 0.0
            epoch_corrects = 0
            
            for inputs, labels in tqdm(dataloader_dict[phase]):
                inputs = inputs.to(device)
                labels = labels.to(device)
                optimizer.zero_grad()
                
                with torch.set_grad_enabled(phase == 'train'):
                    outputs = model(inputs)
                    _, preds = torch.max(outputs, 1)
                    loss = criterion(outputs, labels)
                    
                    if phase == 'train':
                        loss.backward()
                        optimizer.step()
                        
                    epoch_loss += loss.item() * inputs.size(0)
                    epoch_corrects += torch.sum(preds == labels.data)
                    
            epoch_loss = epoch_loss / len(dataloader_dict[phase].dataset)
            epoch_acc = epoch_corrects.double() / len(dataloader_dict[phase].dataset)
            
            print('{} Loss: {:.4f} Acc: {:.4f}'.format(phase, epoch_loss, epoch_acc))
            
            if phase == 'val' and epoch_acc > best_acc:
                best_acc = epoch_acc
                best_model_wts = model.state_dict()
                
    time_elapsed = time.time() - since
    print('Training complete in {:.0f}m {:.0f}s'.format(
        time_elapsed // 60, time_elapsed % 60))
    print('Best val Acc: {:4f}'.format(best_acc))
    return model

num_epoch = 10
model = train_model(model, dataloader_dict, criterion, optimizer, num_epoch)

 

 

 

위 코드를 보면 오차와 입력을 곱하는 부분이 있다. (epoch_loss += loss.item() * inputs.size(0)) 이를 위해서는 손실함수의 reduction 이라는 파라미터를 이해할 필요가 있다고 한다. CrossEntropyLoss 에서 reduction 인자를 따로 지정해주지 않으면 기본값인 mean 으로 들어가게 된다고 한다. mean 은 정답과 예측 값의 오차를 구한 후 그 값들의 평균을 구한다. 즉, 손실 함수 특성상 전체 오차를 배치 크기로 나눔으로써 평균을 반환하기 때문에 epoch_loss 를 계산하는 동안 loss.item() 과 inpu.size(0) 을 곱해줘야 하는 것이다. 훈련 결과는 아래와 같다.

 

 

 

result

 

 

 

62% 정도의 높지 않은 정확도를 보이고 있다. 이제 훈련 데이터셋과 테스트 데이터셋을 모델에 적용하여 정확도를 측정해본다. 측정 결과는 데이터프레임에 담아 CSV 파일로 저장한다. 여기에서는 테스트셋을 이용하기에 model.eval() 메서드를 사용한다.

 

 

 

import pandas as pd

id_list = []
pred_list = []
_id=0
with torch.no_grad():
    for test_path in tqdm(test_images_filepaths):
        img = Image.open(test_path)
        _id =test_path.split('/')[-1].split('.')[1]
        transform = ImageTransform(size, mean, std)
        img = transform(img, phase='val')
        img = img.unsqueeze(0)
        img = img.to(device)

        model.eval()
        outputs = model(img)
        preds = F.softmax(outputs, dim=1)[:, 1].tolist()        
        id_list.append(_id)
        pred_list.append(preds[0])
       
res = pd.DataFrame({
    'id': id_list,
    'label': pred_list
})

res.sort_values(by='id', inplace=True)
res.reset_index(drop=True, inplace=True)

res.to_csv('./LesNet.csv', index=False)

 

 

result

 

 

 

위에서 사용된 F.softmax 함수는 지정된 차원을 따라 텐서의 개별 값들이 (0, 1) 범위에 있고 합계가 1이 되도록 크기를 다시 조정한다. 마지막으로 예측 결과를 시각화해본다.

 

 

 

class_ = classes = {0:'cat', 1:'dog'}
def display_image_grid(images_filepaths, predicted_labels=(), cols=5):
    rows = len(images_filepaths) // cols
    figure, ax = plt.subplots(nrows=rows, ncols=cols, figsize=(12, 6))
    for i, image_filepath in enumerate(images_filepaths):
        image = cv2.imread(image_filepath)
        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        
        a = random.choice(res['id'].values)    
        label = res.loc[res['id'] == a, 'label'].values[0]
        if label > 0.5:
            label = 1
        else:
            label = 0
        ax.ravel()[i].imshow(image)
        ax.ravel()[i].set_title(class_[label])
        ax.ravel()[i].set_axis_off()
    plt.tight_layout()
    plt.show()
display_image_grid(test_images_filepaths)

 

 

result

 

 

 

결과만 보아도 예측이 잘 되지 않고 있음을 확인할 수 있다. 전이학습, 더 많은 데이터 활용, 더 많은 에폭 등 다양한 방법을 활용하여 정확성을 높일 필요가 있을 것이다.

 

 

 

 

 

 

 

Reference:

딥러닝 파이토치 교과서