병든 잎사귀 식별 경진대회

 

MUSTHAVE <머신러닝 • 딥러닝 문제해결 전략> 12장을 실습한 내용입니다. 

---

1. 경진대회 이해

미국의 사과 과수원들이 병원균과 해충에 피해를 입지 않으려면 질병을 조기에 발견해야 합니다. 

시기를 놓치면 피해가 눈덩이처럼 불어나게 됩니다. 그런데 사람이 일일이 확인하려면 시간과 비용이 많이 듭니다. 그래서 주어진 여러 잎사귀 사진을 이용해 건강한 잎사귀와 병든 잎사귀를 구별하는 딥러닝 모델을 구축해보겠습니다. 

 

본 경진대회는 다중분류문제로 타깃값이 4개입니다. 잎사귀가 각 타깃값일 확률을 예측하면 됩니다.

(예를 들어 사진을 보고 건강한 잎사귀일 확률 0.6, 여러 질병에 걸린 잎사귀일 확률 0.2, 녹병에 걸린 잎사귀일 확률 0.1, 붉은 곰팡이병에 걸린 잎사귀일 확률 0.1처럼 예측해야합니다. )

 

• healthy: 건강한 잎사귀
• multiple_diseases: 여러 질병에 걸린 잎사귀 
• rust: 녹병에 걸린 잎사귀
• scab: 붉은곰팡이병에 걸린 잎사귀

 

주어지는 데이터는 jpg형식의 이미지데이터와 이미지데이터 파일명, 타깃값 등이 있습니다. 

 

• train.csv: 훈련 이미지 데이터 ID(이미지 파일명)와 타깃값
• test.csv: 테스트 이미지 데이터ID(이미지 파일명)
• images: 훈련/테스트 이미지 데이터가 들어있는 디렉터리 (JPG)
• sample_submission.csv: 샘플 제출 양식

 

2. 탐색적 데이터 분석

csv 데이터를 둘러보고, 타깃값 분포를 알아본 뒤 실제 이미지를 출력해봅시다.

 

① 데이터 둘러보기

import pandas as pd

# 데이터 경로
data_path = '/kaggle/input/plant-pathology-2020-fgvc7/'

train = pd.read_csv(data_path + 'train.csv')
test = pd.read_csv(data_path + 'test.csv')
submission = pd.read_csv(data_path + 'sample_submission.csv')

train.shape, test.shape

타깃값이 4개라서 훈련데이터의 열개수가 4개 더 많을 뿐, 훈련/테스트 데이터의 개수 모두 1821개로 개수가 같습니다.

 

train.head()

image_id는 훈련데이터의 파일명을 나타내고, 훈련데이터에는 잎사귀 상태를 4개 열로 나눠 원-핫 인코딩 형태로 기록해두었다는 것을 알게 되었습니다. 

 

test.head()

테스트 데이터에는 타깃값이 없기 때문에 image_id만 존재한다는 것을 확인했습니다. 

 

submission.head()

제출 샘플 파일은 타깃값 4개의 값이 모두 0.25로 각 확률을 25%로 일괄 기재해두었다는 것을 확인할 수 있었습니다. 

 

② 데이터 시각화

csv 파일의 데이터를 시각화하고 타깃값 분포를 알아보고, 이어서 이미지 데이터를 실제로 출력해 보겠습니다. 

 

타깃값 분포

# 데이터를 타깃값별로 추출
healthy = train.loc[train['healthy']==1]
multiple_diseases = train.loc[train['multiple_diseases']==1]
rust = train.loc[train['rust']==1]
scab = train.loc[train['scab']==1]

각 변수에 타깃값의 데이터를 할당시켰습니다. 이 변수를 활용해 타깃값 분포를 파이그래프로 그려봅시다. 

import matplotlib as mpl
import matplotlib.pyplot as plt
%matplotlib inline

mpl.rc('font', size=15)
plt.figure(figsize=(7, 7))

label = ['healthy', 'multiple diseases', 'rust', 'scab'] # 타깃값 레이블
# 타깃값 분포 파이 그래프
plt.pie([len(healthy), len(multiple_diseases), len(rust), len(scab)], 
        labels=label, 
        autopct='%.1f%%');

비율이 rust > scap > healthy > multiple diseases 순으로 나머지 타깃값에 비해 multiple diseases가 상대적으로 적다는 것을 확인할 수 있습니다.

비율 차이가 크기 때문에 훈련데이터와 검증데이터로 나눌 때 타깃값 비율에 맞게 나누는 것이 좋겠습니다. 

 

이미지 출력

이미지를 출력하기 앞서, 이미지 ID를 전달받아 화면에 이미지를 출력하는 show_image() 함수를 먼저 정의해보겠습니다.

import matplotlib.gridspec as gridspec
import cv2 # OpenCV 라이브러리

def show_image(img_ids, rows=2, cols=3): 
    assert len(img_ids) <= rows*cols # 이미지가 행/열 개수보다 많으면 오류 발생

    plt.figure(figsize=(15, 8))          # 전체 Figure 크기 설정 #*
    grid = gridspec.GridSpec(rows, cols) # 서브플롯 배치

    # 이미지 출력
    for idx, img_id in enumerate(img_ids):
        img_path = f'{data_path}/images/{img_id}.jpg'  # 이미지 파일 경로 
        image = cv2.imread(img_path)                   # 이미지 파일 읽기 
        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) # 이미지 색상 보정 
        ax = plt.subplot(grid[idx])
        ax.imshow(image) # 이미지 출력

# 각 타깃값별 image_id를 구하는 코드 (마지막 6개)
num_of_imgs = 6 #*
last_healthy_img_ids = healthy['image_id'][-num_of_imgs:]
last_multiple_diseases_img_ids = multiple_diseases['image_id'][-num_of_imgs:]
last_rust_img_ids = rust['image_id'][-num_of_imgs:]
last_scab_img_ids = scab['image_id'][-num_of_imgs:]

이미지 id를 알아냈으니 show_image() 함수로 잎사귀 이미지를 출력해봅시다. 

show_image(last_healthy_img_ids) # 건강한 잎사귀 출력

show_image(last_multiple_diseases_img_ids) # 여러 질병에 걸린 잎사귀 출력

show_image(last_rust_img_ids) # 녹병에 걸린 잎사귀 출력

show_image(last_scab_img_ids) # 붉은곰팡이병에 걸린 잎사귀

더 많은 이미지를 보고싶은 경우 #*표시된 num_of_imgs값과 figsize 크기를 수정한 후 

show_image 호출 시 rows 파라미터까지 지정해주면 됩니다. (예시: show_image(last_healthy_img_ids, rows=4, cols=3) ) 

 

 

---

3. 베이스라인 모델 

사전 훈련된 모델 EfficientNet을 활용해 전이학습을 진행하고, 모델 훈련과 성능검증을 동시에 진행하면서 훈련을 반복할 것입니다. 이번에는 예측 단계에서 성능을 개선할 수 있는 방법을 활용하여 유용한 성능 향상 기법들을 실습해보겠습니다. 

 

① 시드값 고정 및 GPU 장비 설정

머신러닝 경진대회에서는 없었지만 딥러닝과 파이토치의 특성으로 인해 추가된 단계입니다. 

시드값 고정

import torch # 파이토치 
import random
import numpy as np
import os

# 시드값 고정. 다시 실행해도 같은 값을 얻기 위해서 실시한다. 
seed = 50
os.environ['PYTHONHASHSEED'] = str(seed)
random.seed(seed)
np.random.seed(seed)
torch.manual_seed(seed)
torch.cuda.manual_seed(seed)
torch.cuda.manual_seed_all(seed)
torch.backends.cudnn.deterministic = True
torch.backends.cudnn.benchmark = False
torch.backends.cudnn.enabled = False

 

GPU 장비 설정

오른쪽 SETTING > Accelerator에서 GPU로 바꾸고, 다음 코드를 실행합니다. 

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

device

 

 

② 데이터 준비

import pandas as pd

# 데이터 경로
data_path = '/kaggle/input/plant-pathology-2020-fgvc7/'

train = pd.read_csv(data_path + 'train.csv')
test = pd.read_csv(data_path + 'test.csv')
submission = pd.read_csv(data_path + 'sample_submission.csv')

 

훈련데이터, 검증 데이터 분리

전체 훈련 데이터인 train을 훈련데이터와 검증 데이터로 분리하겠습니다. 

from sklearn.model_selection import train_test_split

# 훈련 데이터, 검증 데이터 분리
train, valid = train_test_split(train, 
                                test_size=0.1,
                                stratify=train[['healthy', 'multiple_diseases', 'rust', 'scab']],
                                random_state=50)

타깃값이 고루 분포되기 위해 stratify 파라미터에 타깃값 4개를 전달했습니다. 

 

데이터셋 클래스 정의

import cv2
from torch.utils.data import Dataset # 데이터 생성을 위한 클래스
import numpy as np

class ImageDataset(Dataset):
    # 초기화 메서드(생성자)
    def __init__(self, df, img_dir='./', transform=None, is_test=False):
        super().__init__() # 상속받은 Dataset의 __init__() 메서드 호출
        self.df = df
        self.img_dir = img_dir
        self.transform = transform
        self.is_test = is_test
    
    # 데이터셋 크기 반환 메서드 
    def __len__(self):
        return len(self.df)
    
    # 인덱스(idx)에 해당하는 데이터 반환 메서드
    def __getitem__(self, idx):
        img_id = self.df.iloc[idx, 0]             # 이미지 ID
        img_path = self.img_dir + img_id + '.jpg' # 이미지 파일 경로
        image = cv2.imread(img_path)              # 이미지 파일 읽기
        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) # 이미지 색상 보정
        # 이미지 변환 
        if self.transform is not None:
            image = self.transform(image=image)['image']
        # 테스트 데이터면 이미지 데이터만 반환, 그렇지 않으면 타깃값도 반환 
        if self.is_test:
            return image # 테스트용일 때
        else:
            # 타깃값 4개 중 가장 큰 값의 인덱스 
            label = np.argmax(self.df.iloc[idx, 1:5]) 
            return image, label # 훈련/검증용일 때

 

이미지 변환기 정의 

데이터 증강용 이미지 변환기를 정의해보겠습니다. albumentation가 제공하는 이미지 변환기를 사용해보겠습니다. 

 

먼저 albumentation 모듈을 임포트하고, 

이미지 변환기 정의
# 이미지 변환을 위한 모듈
import albumentations as A
from albumentations.pytorch import ToTensorV2

 

훈련 데이터용 변환기부터 정의한 뒤, 검증 및 테스트 데이터용을 정의하겠습니다. 

 

훈련데이터용 변환기는 여러 변환기를 Compose() 메서드로 묶어 사용하는 구조입니다. 

# 훈련 데이터용 변환기
transform_train = A.Compose([
    A.Resize(450, 650),       # 이미지 크기 조절 
    A.RandomBrightnessContrast(brightness_limit=0.2, # 밝기 대비 조절
                               contrast_limit=0.2, p=0.3),
    A.VerticalFlip(p=0.2),    # 상하 대칭 변환
    A.HorizontalFlip(p=0.5),  # 좌우 대칭 변환 
    A.ShiftScaleRotate(       # 이동, 스케일링, 회전 변환
        shift_limit=0.1,
        scale_limit=0.2,
        rotate_limit=30, p=0.3),
    A.OneOf([A.Emboss(p=1),   # 양각화, 날카로움, 블러 효과
             A.Sharpen(p=1),
             A.Blur(p=1)], p=0.3),
    A.PiecewiseAffine(p=0.3), # 어파인 변환 
    A.Normalize(),            # 정규화 변환 
    ToTensorV2()              # 텐서로 변환
])

 

검증 및 테스트 데이터용 변환기는 필수적인 변환기(이미지 크기조절, 정규화, 텐서변환)만 적용해 정의하겠습니다. 

# 검증 및 테스트 데이터용 변환기
transform_test = A.Compose([
    A.Resize(450, 650), # 이미지 크기 조절 
    A.Normalize(),      # 정규화 변환
    ToTensorV2()        # 텐서로 변환
])

 

데이터셋 및 데이터 로더 생성

데이터셋부터 정의해보겠습니다.

img_dir = '/kaggle/input/plant-pathology-2020-fgvc7/images/'

dataset_train = ImageDataset(train, img_dir=img_dir, transform=transform_train)
dataset_valid = ImageDataset(valid, img_dir=img_dir, transform=transform_test)

 

모델 훈련 시간이 꽤 걸리기 때문에 멀티프로세싱을 활용해보겠습니다. 

멀티프로세싱을 사용하려면 데이터 로더의 시드값을 고정해야합니다. 

 

먼저, seed_worker()를 정의하고 제너레이터를 생성합니다. 

def seed_worker(worker_id):
    worker_seed = torch.initial_seed() % 2**32
    np.random.seed(worker_seed)
    random.seed(worker_seed)
    
g = torch.Generator()
g.manual_seed(0)

데이터 로더도 이어서 생성합니다. 

def seed_worker(worker_id):
    worker_seed = torch.initial_seed() % 2**32
    np.random.seed(worker_seed)
    random.seed(worker_seed)
    
g = torch.Generator()
g.manual_seed(0)

 

③ 모델 생성

사전 훈련된 EfficientNet 모델을 전이학습시키는 방식으로 진행해보겠습니다. 

 

사전 훈련 모델이란 이미 한 분야에서 훈련을 마친 모델을 일컬으며,

전이학습이란 사전 훈련 모델을 유사한 다른 영역에서 재훈련시키는 기법입니다.

 

EfficientNet 모델 생성 

!pip install efficientnet-pytorch==0.7.1

은 2019년 개발된 CNN모델로, 캐글환경에는 설치되어있지 않아 별도로 설치해야합니다.

다음 코드로 설치를 진행합니다. 

!pip install efficientnet-pytorch==0.7.1

EfficientNet은 efficientNet-b0~7까지 종류가 여러 가지인데 숫자가 높아질수록 성능이 좋아집니다. efficientNet-b7을 사용해 실습해보겠습니다. 

 

먼저 efficientnet-pytorch 모듈의 EfficientNet 모델을 임포트합니다. 

from efficientnet_pytorch import EfficientNet # EfficientNet 모델

# 사전 훈련된 efficientnet-b7 모델 불러오기
model = EfficientNet.from_pretrained('efficientnet-b7', num_classes=4) 

model = model.to(device) # 장비 할당

 

④ 모델 훈련 및 성능 검증

손실 함수와 옵티마이저 설정

CrossEntrophyLoss()를 사용해 손실함수부터 정의해보겠습니다. 

import torch.nn as nn # 신경망 모듈

# 손실 함수
criterion = nn.CrossEntropyLoss()

 

Adam W라는 옵티마이저를 사용해서 옵티마이저를 정의합니다. 

 

Adam W는 Adam에 가중치 감쇠를 추가로 적용해서 일반화 성능이 더 우수한 옵티마이저입니다. 

가중치 감쇠란 가중치를 작게 조절하는 규제기법으로 과대적합을 억제해줍니다. 

# 옵티마이저
optimizer = torch.optim.AdamW(model.parameters(), lr=0.00006, weight_decay=0.0001)

 

훈련 및 성능 검증

이번 실습에서는 매 에폭마다 검증해보겠습니다. 

이 방법은 더 오래 걸리지만 과대적합 없이 훈련이 잘 되고 있는지 확인할 수 있다는 장점이 있습니다. 

from sklearn.metrics import roc_auc_score # ROC AUC 점수 계산 함수
from tqdm.notebook import tqdm # 진행률 표시 막대 

epochs = 5

# 총 에폭만큼 반복
for epoch in range(epochs):
    # == [ 훈련 ] ==============================================
    model.train()        # 모델을 훈련 상태로 설정 
    epoch_train_loss = 0 # 에폭별 손실값 초기화 (훈련 데이터용)
    
    # '반복 횟수'만큼 반복 
    for images, labels in tqdm(loader_train):
        # 이미지, 레이블(타깃값) 데이터 미니배치를 장비에 할당 
        images = images.to(device)
        labels = labels.to(device)
        
        # 옵티마이저 내 기울기 초기화
        optimizer.zero_grad()
        # 순전파 : 이미지 데이터를 신경망 모델의 입력값으로 사용해 출력값 계산
        outputs = model(images)
        # 손실 함수를 활용해 outputs와 labels의 손실값 계산
        loss = criterion(outputs, labels)
        # 현재 배치에서의 손실 추가 (훈련 데이터용)
        epoch_train_loss += loss.item() 
        loss.backward() # 역전파 수행
        optimizer.step() # 가중치 갱신
    # 훈련 데이터 손실값 출력
    print(f'에폭 [{epoch+1}/{epochs}] - 훈련 데이터 손실값 : {epoch_train_loss/len(loader_train):.4f}')
    
    # == [ 검증 ] ==============================================
    model.eval()          # 모델을 평가 상태로 설정 
    epoch_valid_loss = 0  # 에폭별 손실값 초기화 (검증 데이터용)
    preds_list = []       # 예측 확률값 저장용 리스트 초기화 
    true_onehot_list = [] # 실제 타깃값 저장용 리스트 초기화 
    
    with torch.no_grad(): # 기울기 계산 비활성화
        # 미니배치 단위로 검증
        for images, labels in loader_valid:
            images = images.to(device)
            labels = labels.to(device)
            
            outputs = model(images)
            loss = criterion(outputs, labels)
            epoch_valid_loss += loss.item()
            
            preds = torch.softmax(outputs.cpu(), dim=1).numpy() # 예측 확률값
            # 실제값 (원-핫 인코딩 형식)
            true_onehot = torch.eye(4)[labels].cpu().numpy()  
            # 예측 확률값과 실제값 저장
            preds_list.extend(preds)
            true_onehot_list.extend(true_onehot)
    # 검증 데이터 손실값 및 ROC AUC 점수 출력 
    print(f'에폭 [{epoch+1}/{epochs}] - 검증 데이터 손실값 : {epoch_valid_loss/len(loader_valid):.4f} / 검증 데이터 ROC AUC : {roc_auc_score(true_onehot_list, preds_list):.4f}')

 

 

⑤ 예측 및 결과 제출 

예측을 하기 위해 테스트용 데이터 셋과 데이터 로더를 생성합니다.

dataset_test = ImageDataset(test, img_dir=img_dir, 
                            transform=transform_test, is_test=True)
loader_test = DataLoader(dataset_test, batch_size=batch_size, 
                         shuffle=False, worker_init_fn=seed_worker,
                         generator=g, num_workers=2)

 

예측

테스트 데이터를 활용해 타깃확률을 예측해보겠습니다. 

model.eval() # 모델을 평가 상태로 설정 

preds = np.zeros((len(test), 4)) # 예측값 저장용 배열 초기화

with torch.no_grad():
    for i, images in enumerate(loader_test):
        images = images.to(device)
        outputs = model(images)
        # 타깃 예측 확률 
        preds_part = torch.softmax(outputs.cpu(), dim=1).squeeze().numpy()
        preds[i*batch_size:(i+1)*batch_size] += preds_part

 

결과 제출

submission[['healthy', 'multiple_diseases', 'rust', 'scab']] = preds
submission.to_csv('submission.csv', index=False)