CNN 신경망 Baseline

오랜만에 포스팅을 써보려고 한다.

최근 사내 딥러닝 관련 인증 시험도 있고 해서 전체적으로 신경망 구축에 대해 베이스라인을 몇개로 나눠서 진행해보겠다.

 

 

먼저 처음으로 로컬 파일의 폴더의 이미지 파일에 대한 라벨링 및 경로를 통해 data frame을 만드는 것부터 해서 

최종 예측결과까지 내는것을 할 것이고, 알렉스넷을 활용해서 하고자 한다.

 

 

 

CNN 신경망 baseline 순서 요약
1) glob 활용 image path (또는 image file name) array 만들기
2) 각 image path 별로 대응되는 image label 만들기
3) train/validation data set 나누기(train_test_split)
4) Keras sequence 상속받은 클래스 선언하여 , 이미지 사이즈 , 배치사이즈 등 세팅(sequence 내에서 이미지 -> Tensor 변환)
5) 신경망 모델 짜기
6) 학습시키고 , 예측하기 
* 이미지 사이즈가 작으면 sequence 없이 바로 for문으로 이미지-> Tensor 변환해도됨 , 메모리이슈 

 

1) glob 활용 image path (또는 image file name) array 만들기

이미지의 경로를 읽어들여와서 image path array 만들기 

import pandas as pd 
import numpy as np 

from glob import glob 

# glob 임포트하고 

PATH_TRAIN = "CIFAR-10-images-master/CIFAR-10-images-master/train"
PATH_TEST = "CIFAR-10-images-master/CIFAR-10-images-master/test"

# 위 경로는 개발환경에 따라 바뀔 수 있음

train = pd.DataFrame([])
# 빈 data frame 하나 먼저 만들고 

train["image_files"] = glob(PATH_TRAIN + "/*/*")

# "/*" 만 하면 PATH_TRAIN 하위 전부 
# "/*/*"만 하면 그 하위 폴더내의 전부를 가져오는 뜻
# "/*/*.csv" 원하는 확장자만 전부 가져올때 .png , .jpg , .csv로 가져올 수 있음

2) 각 image path 별로 대응되는 image label 만들기

이미지 경로별로 대응되는 라벨링을 만드는 단계는 쉬운것 같으면서도 어렵게 생각하면 괜히 꼬이는 부분들이 있다.

따라서 심플하게 접근하는것이 중요하다.

우선 대응되는 라벨링을 가져오고 해당 라벨에 대한 인덱싱을 추가로 해준다.

series.str.split(" ") 개념을 이해해야 한다.

판다스 seriese를 string으로 변환하고 " "사이 문자열로 해당 전체 문자를 쪼갭니다.

쪼갠 결과를 먼저 보고 몇번째를 선정할지 골라서 예를들어 첫번째라고한다면,

series.str.split("/").str[0] 으로 지정해줍니다. 이때 이 값은 다시 series가 되므로 

series.str.split("/").str[0].str.split("\\")로 한번 더 쪼개어서(윈도우 개발환경 기반)

series.str.split("/").str[0].str.split("\\").str[n]  , n번째를 쓰시면 경로에서 labeling만 가져올 수 있습니다 

# 윈도우 개발환경 
# train["image_labels"] 컬럼 추가 

train["image_labels"] = train["image_files"].str.split("/").str[i].str.split('\\').str[j]

# i , j는 하나씩 돌려보면서 판단

# 리눅스 개발환경 기반 - 클라우드(코랩 , 캐글)
train["image_labels"] = train["image_files"].str.split("/").str[i]

# i는 하나씩 돌려보면서 0부터~ 판단


### 개발환경은 돈표시 \\가 train["image_labels"] = train["image_files"].str.split("/")
### 에서 나타나면 윈도우입니다.

### 라벨링이 똑바로 들어가있으면 pd.factorize(train["image_labels"])[0]을 통해서 라벨인코딩

train["image_labels"] = pd.factorize(train["image_labels"])

 

 

 

3) train/validation data set 나누기(train_test_split)

train["image_files"] , train["image_labels"]는 series이므로 별도로 train_test_split하지 않고 , values로 array상태로 변환후에

데이터를 분리한다. * 단 이미지가 가벼울때는 바로 아래로 case(1)을 참조

 

from sklearn.model_selection import train_test_split

train_image_files = train["image_files"].values
train_image_labels = train["image_labels"].values


tr_path , val_path , tr_label , val_label = train_test_split(train_image_files , train_image_labels , test_size = 0.15 , random_state = 7777 , stratify = train_image_labels)


### label들은 one hot encoding 한다 

from tensorflow.keras.utils import to_categorical 

tr_label_ohe = to_categorical(tr_label)
val_label_ohe = to_categorical(val_label)

4) Keras sequence 상속받은 클래스 선언하여 , 이미지 사이즈 , 배치사이즈 등 세팅
(sequence 내에서 이미지 -> Tensor 변환) ,

*단 이미지가 작고 용량이 가벼운 경우는 sequence없이 그냥 바로 이미지 -> Tensor 변환 

 

### 이미지 가벼울때 - case (1) 
IMAGE_SIZE = 227
import cv2 
image_array_list = [] ## 빈 리스트하나 만들고 

for i in range(len(train_image_files)):
	image = cv2.cvtColor(cv2.imread(train_image_files[i]) , cv2.COLOR_BGR2RGB) 
    image = image / 255.0 ## scaling 
    image = cv2.resize(image , (IMAGE_SIZE , IMAGE_SIZE))
    image_array_list.append(image)
    
image_array = np.array(image_array_list)

### 이미지 -> Tensor 변환 완료

from sklearn.model_selection import train_test_split

tr_image_array , val_image_array , tr_label , val_label = train_test_split(image_array , train_image_labels , test_size = 0.15 , random_state = 7777 , stratify = train_image_labels)

### train ,validation 데이터셋 분할 완료 

from tensorflow.keras.utils import to_categorical

tr_label_ohe = to_categorical(tr_label)
val_label_ohe = to_categorical(val_label)

### 모델 선언 

from tensorflow.keras.layers import Input, Dense , Flatten , Conv2D , AveragePooling2D , MaxPooling2D , Dropout
from tensorflow.keras.models import Model 
from tensorflow.keras.optimizers import Adam 


from tensorflow.keras.layers import experimental, Activation , BatchNormalization ,MaxPooling2D , ZeroPadding2D

def Alex_net():
    input_tensor = Input(shape = (IMAGE_SIZE , IMAGE_SIZE ,3 ))
    x = Conv2D(filters = 96 , kernel_size = 11 , strides  = 4 , padding = 'valid')(input_tensor)
    x = BatchNormalization()(x)
    x = Activation('relu')(x)
    
    x = MaxPooling2D((3,3) , 2)(x)
    
    x = Conv2D(filters = 256 , kernel_size  = 5 , strides = 1, activation = 'relu' , padding = 'valid')(x)
    x = ZeroPadding2D(padding = 2)(x)
    x = BatchNormalization()(x)
    x = Activation('relu')(x)
    
    x = Conv2D(filters = 384 , kernel_size = 3 , strides = 1 , activation = 'relu' , padding = 'same')(x)
    x = BatchNormalization()(x)
    x = Activation('relu')(x)
    
    x = Conv2D(filters = 384 , kernel_size = 3 , strides = 1 , activation = 'relu' , padding = 'same')(x)
    x = BatchNormalization()(x)
    x = Activation('relu')(x)
    
    x = Conv2D(filters = 256 , kernel_size = 3 , strides = 1 , activation = 'relu' , padding = 'same')(x)
    x = BatchNormalization()(x)
    x = Activation('relu')(x)
    
    x = MaxPooling2D((3,3) , 2)(x)
    
    x = Dropout(0.5)(x)
    x = Flatten()(x)
    x = Dense(4096 , activation = 'relu')(x)
    x = Dense(4096 , activation = 'relu')(x)
    
    output = Dense(10 , activation = 'softmax')(x)
    
    model = Model(inputs = input_tensor , outputs = output)
    return model


### 모델 컴파일


model = Alex_net()
model.compile(optimizer = Adam(0.001) , loss = "categorical_crossentropy" , metrics = 'accuracy')
model.summary()


### 학습 (epoch ,batch 는 조절하세요!)

history = model.fit(tr_image_array , tr_label_ohe , batch_size = 64 , epochs = 2 ,validation_data= (val_image , val_labels))

위의 경우 메모리 셧다운이 될수 있으므로 작은 이미지 용량에서만 사용 , 용량이 클 경우 아래와 같이 sequence 활용 

### 이미지 무거울때 sequence 상속 
### sequence 상속받으면 배치사이즈 단위로 이미지파일을 가져와서 변환하고 전처리하고 학습시킨다.

from tensorflow.keras.utils import Sequence

BATCH_SIZE = 64 
IMAGE_SIZE = 227 ### 이미지 사이즈는 모델이 원하는 사이즈로 , alexnet = 227 , vgg = 224 


class data_set(Sequence):
    
    def __init__(self , image_filenames , labels , batch_size = BATCH_SIZE , augmentor = None , pre_func = None , shuffle = None):
        
        self.image_filenames = image_filenames 
        self.labels = labels
        self.batch_size = BATCH_SIZE 
        self.augmentor = augmentor 
        self.pre_func = pre_func 
        self.shuffle = shuffle
        
    def __len__(self):
        return int(np.ceil(len(self.labels)/self.batch_size))
    
    def __getitem__(self, index):
        
        image_name_batch = self.image_filenames[index * self.batch_size : (index+1)* self.batch_size]
        label_batch = self.labels[index * self.batch_size : (index+1) * self.batch_size]
        
        image_batch = np.zeros((image_name_batch.shape[0],IMAGE_SIZE , IMAGE_SIZE , 3))
        
        for image_index in range(image_name_batch.shape[0]):
            image = cv2.cvtColor(cv2.imread(image_name_batch[image_index]) , cv2.COLOR_BGR2RGB)
            image = image/ 255.0
            image = cv2.resize(image , (IMAGE_SIZE , IMAGE_SIZE))
        
            if self.augmentor is not None:
                image = self.augmentor(image = image)["image"]
            if self.pre_func is not None:
                image = self.pre_func(image)
            
            image_batch[image_index] = image
            
        return image_batch , label_batch
    
    
    def on_epoch_end(self):
        if(self.shuffle):
            self.image_filenames, self.labels = sklearn.utils.shuffle(self.image_filenames , self.labels)
        else:
            pass

위의 선언한 Sequence를 활용해서 train 데이터 / validation 데이터의 인스턴스를 생성함 , 생성한 인스턴스를 활용해서
모델에 학습을 시킨다.

tr_ds = data_set(tr_path , tr_label_ohe)
val_ds = data_set(val_path , val_label_ohe)

tr_path , val_path , tr_label_ohe , val_label_ohe 는 이미지 작을 경우 baseline 위의 train_test_split  및

to_categorical한 결과이다. 

 

5) 신경망 모델 짜기

위의 이미지 작은 경우에서 선언한 alex_net() 함수를 활용한 모델 구조 참조

 

6) 학습 및 예측하기 (Local 환경의 경우 GPU 세팅 포함)

모델 컴파일 / 학습시킨 후 결과 확인 및 예측하기 

import tensorflow as tf 
import cv2
# GPU 세팅 , 클라우드 환경에선 별도 GPU 세팅방법에 맞게 하면됨 

gpus = tf.config.experimental.list_physical_devices('GPU')
if gpus:
  try:
    tf.config.experimental.set_visible_devices(gpus[0], 'GPU')
  except RuntimeError as e:
    print(e)
    
 


# 모델 학습 
history = model.fit(tr_ds , epochs = 10 , steps_per_epoch= int(np.ceil(tr_path.shape[0]/BATCH_SIZE)) , validation_data= val_ds , validation_steps=int(np.ceil(val_path.shape[0]/BATCH_SIZE)) )

 

위와 같이 학습을 종료한 후 , 테스트데이터셋을(우린 테스트 데이터의 Label은 모를경우 예측 결과를 내야함) 

변환한 후에 모델로 예측한 결과를 제출하면 된다. 이때 최종 결과물은 label indexing = "airplane" 과 같은 문자열일지 , 

0,1,2,3,4 의 encoding 값일지는 문제를 보고 판단하면 된다. 하기와 같이 간단히 진행한다.

 

단 테스트 데이터가 만약 너무 큰 경우엔 이 다음 포스팅에서 별도의 TEST 데이터만을 위한 Sequence를 만들어서 
해당 Sequence를 상속받은 인스턴스로 예측을 해야한다. (이 부분은 복잡하므로 다음 포스팅에 공유)

### TEST DATA도 Train 과 동일하게 불러옴

PATH_TEST = 'test data 경로' 

glob(PATH_TEST + "/*/*")    # *를 바꿔가며 이미지 경로 .jpg 단위확장자까지 잘들어가는지 확인 후

test = pd.DataFrame([])

test["image_files"] = glob(PATH_TEST + "/*/*")   

test_image_path = test["image_files"].values 


# test_image_array 빈 리스트 하나 선언

test_image_array = []

for i in range(len(test_image_path)):
	image = cv2.cvtCOLOR(cv2.imread(test_image_path[i]) , cv2.COLOR_BGR2RGB)
    image = image / 255.0 # scaling , train과 동일하게 
    image = cv2.resize(image, (IMAGE_SIZE , IMAGE_SIZE) ## resize , 원하는 사이즈로 
    test_image_array.append(image) 
    
test_image_array = np.array(test_image_array) 



# 예측 (predict)

pred = model.predict(test_image_array)

# 각 label에 대한 probabilities 로 결과 산출됨  

result = [ np.argmax(value) for value in pred ] # for 문 활용 argmax로 가장 높은 라벨 뽑아냄

# 최종 결과 제출이 0,1,2,3,4 등의 라벨인코딩값이면 이걸 제출

# 라벨의 문자열 값이면 아래 한번 더 변환 

categories = pd.factorize(train["image_labels"])[1] # 0이면 라벨인코딩 , 1이면 인덱스(문자열값임)

# 위와 같이 categories로 받아서 

final_pred = categories[result]  

# 최종 결과값으로 제출 끝!