Keras(케라스)로 CNN 모델 구성하기
반갑습니다. 이번 글에서는 Keras를 사용하여 모델을 구성해보고, 파일로 저장해보고, 평가(분류)까지 해보도록 하겠습니다.
Keras 자체가 파이썬 기반의 딥러닝 라이브러리이므로, 상당히 직관적인 코딩을 할 수 있다는 점이 매력적으로 다가오는 것 같습니다. 저의 경우 Tensorflow를 백엔드로 사용하고 있습니다.
지금부터는 코드를 하나하나 짚어가면서 진행해보도록 하겠습니다.
1. 데이터 셋(이미지) 가져오기
먼저 학습을 시키고 검증에 사용할 데이터로는 CIFAR-10 데이터셋을 사용하겠습니다. CIFAR-10 dataset의 경우, 32x32 사이즈의 6만 개의 이미지로 구성됩니다. 10개의 클래스로 분류되어 있으며, 각 클래스당 6000장의 이미지로 구성이 되어 있으며 실제로 훈련에 사용되는 것은 5000장, 검증에 사용되는 것은 1000장으로 구성이 되어 있습니다.
일반적으로 위의 데이터셋을 다운로드 하는 방법은 두 가지의 방법이 있습니다.
1. https://www.cs.toronto.edu/~kriz/cifar.html 에 방문하여 필요한 부분을 직접 다운로드 받는 경우
2. 케라스 상에서 데이터셋을 로딩하여 다운받는 경우
1번의 경우는 사이트에 직접 접속해서 파이썬 버전을 다운로드 받습니다. 파일을 다운로드하게 되면 tar 형식의 파일이 다운로드 됩니다.
이 파일의 경로를 맞춰줘야 하는데 저의 경우 C:\SPB_Data\.keras\datasets\cifar-10-batches-py 경로내에 존재하더군요. 아무래도 저는 Spyder 에디터의 Working directory가 SPB_Data로 설정이 되어있던 상태로 진행해서 그런 경우인 것 같습니다.
tar 압축을 해제해 .keras폴더를 찾아서 하위 경로를 맞춰주시면 됩니다. 위의 그림과 같이 cifar-10-batches-py 폴더에 파일을 전부 옮겨 주시면 됩니다.
2번의 경우는 간단하게 파이썬 코드를 작성해서 다운로드 받아볼 수 있습니다.
from keras.datasets import cifar10
(X_train, y_train), (X_test, y_test) = cifar10.load_data()
작성하고 실행시키면 Console창에서 다운로드 상황을 확인할 수 있습니다.
위의 코드에서는 cifar10을 import 시켜서 load_data()라는 함수를 사용하고 있습니다. 이 부분에 대해서 자세하게 보면 keras에서 함수로 다운받을 수 있게끔 만들어 준 것인데, cifar10.py 를 찾아서 보게되면 다음과 같이 구성되어 있습니다.
"""CIFAR10 small images classification dataset.
"""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from .cifar import load_batch
from ..utils.data_utils import get_file
from .. import backend as K
import numpy as np
import os
def load_data():
"""Loads CIFAR10 dataset.
# Returns
Tuple of Numpy arrays: `(x_train, y_train), (x_test, y_test)`.
"""
dirname = 'cifar-10-batches-py'
origin = 'https://www.cs.toronto.edu/~kriz/cifar-10-python.tar.gz'
path = get_file(dirname, origin=origin, untar=True)
num_train_samples = 50000
x_train = np.empty((num_train_samples, 3, 32, 32), dtype='uint8')
y_train = np.empty((num_train_samples,), dtype='uint8')
for i in range(1, 6):
fpath = os.path.join(path, 'data_batch_' + str(i))
(x_train[(i - 1) * 10000: i * 10000, :, :, :],
y_train[(i - 1) * 10000: i * 10000]) = load_batch(fpath)
fpath = os.path.join(path, 'test_batch')
x_test, y_test = load_batch(fpath)
y_train = np.reshape(y_train, (len(y_train), 1))
y_test = np.reshape(y_test, (len(y_test), 1))
if K.image_data_format() == 'channels_last':
x_train = x_train.transpose(0, 2, 3, 1)
x_test = x_test.transpose(0, 2, 3, 1)
return (x_train, y_train), (x_test, y_test)
load_data 라는 함수가 정의되어 있습니다. 위에서 말씀드렸던 페이지에서 데이터를 받아와 x_train, y_train x_test, y_test 에 데이터를 담는 형식입니다. 이 부분에서 이미지 데이터를 어떻게 담아내는지를 살짝 확인해볼 수 있는데요.
cifar10 데이터셋은 가로 세로가 32x32의 크기로 형성된 이미지입니다. 하지만 이러한 픽셀 데이터들을 학습시키기 위해서는 눈에 보이는 이미지를 그대로 넣는 것이 아닌 실제로는 연산을 하기 위해서 새롭게 정렬시켜줘야 한다는 점입니다.
위의 그림을 보게되면 2x2 사이즈의 1채널 이미지가 존재한다면, 각각의 픽셀값을 0~255의 숫자로 표현할 수 있습니다. 이렇게 표현되어진 숫자들을 왼쪽 상단의 픽셀값부터 순서대로 나열하여 배열에 저장하게 되는데 이러한 방법들이 cifar10 데이터에도 비슷하게 적용이 되어 x_train, x_test에 담겨있다고 생각하시면 될 것 같습니다.
cifar10 데이터셋에 담겨 있는 이미지들은 3채널이며(컬러) uint8(부호가 없는 8비트 : 0~255) 형식의 자료형으로 저장하게 되어있습니다. 따라서 이 부분을 출력해보면 다음과 같이 저장(메모리에 로드됨.)되어 있는 것을 확인할 수 있습니다.
2. 데이터 전처리 및 CNN 모델 생성
이제부터는 배열로 담아온 데이터들이 학습시킬 때 효율적으로 입력되게 끔 전처리를 해주어야 합니다.
# 정규화(dataset 전처리)
X_train = X_train.astype(float) / 255.0
X_test = X_test.astype(float) / 255.0
## 원-핫 인코딩
y_train = np_utils.to_categorical(y_train)
y_test = np_utils.to_categorical(y_test)
num_classes = y_test.shape[1]
X_train은 훈련에 사용될 데이터, X_test는 검증에 사용될 데이터가 배열로 담겨져 있습니다. 이러한 데이터들은 0~255의 unsigned int 8bit 형식으로 저장이 되어있는 상태입니다. Keras에서 모델에 대해 학습을 진행하기 위해서는 픽셀값들을 정규화 시켜주어야만 합니다. (0~255)이 들어있는 배열을 float32 형식으로 변환하고 기존데이터에서 표현될 수 있는 최대값인 255로 각각의 데이터를 나눠주면 됩니다. 즉, 정규화(Normalization)를 진행하면 0~255의 데이터가 0.0~1.0 사이의 실수값으로 재구성 됩니다. 이러한 방법은 각 픽셀들의 값이 세밀해지므로(표현할 수 있는 bit수가 늘어났으므로!), 특징점을 보다 더 세밀하게 추출할 수 있게 됩니다.
one-hot 인코딩 방식은 one-hot 벡터를 이용하여 대상 클래스의 인덱스 위치만 1로 표현하고 나머지는 0으로 채워놓는 방식입니다. 즉 y_train은 X_train의 데이터에 대해 각각의 대상 클래스를 분류해놓는 라벨(label, 레이블)이라고 생각하면 됩니다.
X_train의 첫번째 이미지 데이터가 bird 였다면 y_train의 첫번째 배열은 [0, 0, 1, 0, 0, 0, 0, 0, 0, 0] 가 됩니다. 이것을 다시 말하자면 X는 입력된 이미지 데이터, y는 클래스 라벨이 됩니다.
이 부분을 코드로 직접 확인하면 다음과 같습니다.
지금부터는 훈련에 사용될 모델을 생성하기 위한 작업을 진행하도록 하겠습니다.
# 모델 구성도
model = Sequential()
model.add(Conv2D(32, (3,3), input_shape=X_train.shape[1:], activation='relu', padding='same'))
model.add(Conv2D(64, (3,3), activation='relu', padding='same'))
model.add(MaxPooling2D(pool_size=(2,2)))
model.add(Dropout(0.25))
model.add(Flatten())
model.add(Dense(256, activation='relu'))
model.add(Dropout(0.25))
model.add(Dense(num_classes, activation='softmax'))
제가 생성한 모델의 구성도는 위의 소스와 같습니다. convolution layer가 두 개 있는 네트워크입니다.
몇 가지에 대해 간단하게 설명 드리자면,
Sequential() 은 모델을 구성할 때 계층 구조를 작성한 순서 그대로, 순차적으로 쌓아 생성할 수 있게 해주는 함수입니다.
Conv2D의 경우 각 픽셀에 대해 컨벌루션 연산을 수행하며, 각 인자의 순서대로 필터의 개수(특징점을 저장하는 공간), 필터의 크기, 입력 크기, 활성화 함수의 종류, 패딩 등으로 구성되어 있습니다. 좀 더 자세한 설명은 마지막에 첨부해놓겠습니다.
Dropout의 경우는 일부 연결된 노드를 제거해, 활성화된 노드를 좀 더 부각시킬 수 있는 방법입니다. 다양한 정보에서 특징점을 더 활성화 시킬 수 있고, 연산에서도 이득을 볼 수 있습니다.
다음으로는 구성해놓은 모델에 대해 컴파일을 진행합니다.
model.compile(loss='categorical_crossentropy', optimizer='adam', metrics=['accuracy'])
categorical_crossentropy의 경우 일반적으로 클래스가 3개 이상의 분류 모델을 생성할 때 주로 사용합니다.
adam은 경사 하강법으로, 학습률과 그래디언트(gradient)를 어느정도 고려해 나온 optimizer입니다.
평가 척도로는 accuracy 를 선택했습니다. 전체 데이터의 정확도를 말합니다.
다음으로는 구성된 모델을 학습시켜서 저장하는 소스입니다.
if not os.path.exists(model_dir):
os.mkdir(model_dir)
model_path = model_dir + '/cifar10_classification_test.h5'
checkpoint = ModelCheckpoint(filepath=model_path , monitor='val_loss', verbose=1, save_best_only=True)
early_stopping = EarlyStopping(monitor='val_loss', patience=8)
# 구성 해논 모델의 계층 구조를 간략적으로 보여주는 함수
model.summary()
history = model.fit(X_train, y_train, batch_size=64, epochs=80, validation_data=(X_test, y_test), callbacks=[checkpoint, early_stopping], shuffle=True)
h5는 모델이 저장되는 파일이며, fit() 메서드로 실제 학습을 진행합니다.
fit() 메서드에서는 모델 훈련에 관련하여 다양한 인자들을 입력할 수 있습니다. 첫 번째 인자는 학습에 사용될 정규화된 데이터(위에서는 X_train), 두 번째 인자는 첫 번째 인자의 라벨 데이터(y_train)로 모델에게 전달되는 학습용 전체 데이터를 입력해줍니다.
세 번째 인자로 표현되어 있는 batch_size의 경우는 1epochs 단위에서 데이터를 몇개씩 학습에 대해 진행할 것인지를 지정해줍니다. 예를들어 X_train의 데이터를 1만장을 입력하고 batch_size를 10으로 설정했다면 1epoch 에서 1만장을 학습시킬 때 10 20 30 40 50 ... 의 방식으로 10장씩 묶어서 진행하게 됩니다. 이 batch_size를 크게 잡을수록 프로세서 장비의 메모리를 더 많이 잡아먹게 되므로 항상 유의해서 설정해야 합니다.
추가적으로 몇 가지만 간단히 설명 드리자면,
epochs는 총 학습 횟수를 설정합니다. validation_data는 말 그대로 검증에 사용할 데이터를 입력합니다. callbacks 옵션은 여러가지가 있으나, 제가 작성하는 소스에서는 최고 정확도만을 모델에 저장하는 함수(ModelCheckpoint)와 patience 값에 따라 더 이상 정확도가 증가하지 않으면 멈추게 하는 함수(EarlyStopping)를 사용하였습니다.
아래는 .fit() 메서드의 목록을 보여주기 위해 추가해 놓겠습니다.
#fit method
Model.fit(
x=None,
y=None,
batch_size=None,
epochs=1,
verbose=1,
callbacks=None,
validation_split=0.0,
validation_data=None,
shuffle=True,
class_weight=None,
sample_weight=None,
initial_epoch=0,
steps_per_epoch=None,
validation_steps=None,
validation_batch_size=None,
validation_freq=1,
max_queue_size=10,
workers=1,
use_multiprocessing=False,
)
3. 생성한 모델 예측(분류)하기
이번 파트에서는 데이터를 얼마나 예측하고 있는지, 실제 이미지로 분류를 어떻게 해볼 수 있는지에 대해서 설명하겠습니다. Keras API 는 세션이 자동으로 생성되지 않기 때문에, 예측하는 것을 확인하기 위해서는 새롭게 소스를 작성할 필요가 있습니다.
검증 데이터로 사용할 X_test를 가져와서 확인해 보도록 하겠습니다.
import tensorflow as tf
import numpy as np
from keras.datasets import cifar10
from keras.utils import np_utils
import matplotlib.pyplot as plt
model = tf.keras.models.load_model('./model/cifar10_classification_test.h5')
model.summary()
(X_train, y_train), (X_test, y_test) = cifar10.load_data()
# 정규화(dataset 전처리)
X_train = X_train.astype(float) / 255.0
X_test = X_test.astype(float) / 255.0
## 원-핫 인코딩
y_train = np_utils.to_categorical(y_train)
y_test = np_utils.to_categorical(y_test)
num_classes = y_test.shape[1]
#predict는 input data가 numpy array나 list of array 형태만 가능하다.
prediction = model.predict(X_test)
np.set_printoptions(formatter={'float': lambda x: "{0:2.1f}".format(x)})
cnt = 0
#파일들이 있으면 해당 파일과 비교.
for i in prediction:
pre_ans = i.argmax() # 예측 레이블
print(i)
print(pre_ans)
pre_ans_str = ''
if pre_ans == 0: pre_ans_str = "airplane"
elif pre_ans == 1: pre_ans_str = "automobile"
elif pre_ans == 2: pre_ans_str = "bird"
elif pre_ans == 3: pre_ans_str = "cat"
elif pre_ans == 4: pre_ans_str = "deer"
elif pre_ans == 5: pre_ans_str = "dog"
elif pre_ans == 6: pre_ans_str = "frog"
elif pre_ans == 7: pre_ans_str = "horse"
elif pre_ans == 8: pre_ans_str = "ship"
else: pre_ans_str = "truck"
if i[0] >= 0.7:
print("해당 이미지는 "+pre_ans_str+"으로 추정됩니다.")
if i[1] >= 0.7:
print("해당 이미지는 "+pre_ans_str+"로 추정됩니다.")
if i[2] >= 0.7:
print("해당 이미지는 "+pre_ans_str+"로 추정됩니다.")
if i[3] >= 0.7:
print("해당 이미지는 "+pre_ans_str+"로 추정됩니다.")
if i[4] >= 0.7:
print("해당 이미지는 "+pre_ans_str+"로 추정됩니다.")
if i[5] >= 0.7:
print("해당 이미지는 "+pre_ans_str+"로 추정됩니다.")
if i[6] >= 0.7:
print("해당 이미지는 "+pre_ans_str+"로 추정됩니다.")
if i[7] >= 0.7:
print("해당 이미지는 "+pre_ans_str+"로 추정됩니다.")
if i[8] >= 0.7:
print("해당 이미지는 "+pre_ans_str+"으로 추정됩니다.")
if i[9] >= 0.7:
print("해당 이미지는 "+pre_ans_str+"으로 추정됩니다.")
위의 코드와 같이 로드한 데이터를 똑같이 정규화 시켜 model.predict()를 사용해 예측을 진행해 볼 수 있습니다.
코드를 수정하면 실제 가지고 있는 이미지 파일로도 확인해볼 수도 있습니다. 실제 가지고 있는 이미지 파일을 위에서 설명했던 방식처럼 배열로 나열한다면 가능합니다.
# 경로는 자신이 테스트해볼 파일의 경로로 바꿔주시면 됩니다!
caltech_dir = "D:/keras_cnn_cifar10/test_folder"
image_w = 32
image_h = 32
X = []
filenames = []
files = glob.glob(caltech_dir+"/*.*")
for i, f in enumerate(files):
img = Image.open(f)
img = img.convert("RGB")
img = img.resize((image_w, image_h))
data = np.asarray(img, dtype='float32') # 데이터를 하나의 index(=list)를 가진 배열로 재생성 해주는 함수.
filenames.append(f)
X.append(data) # append는 이어쓰기
위의 예측 소스에서는 0.7(=70%) 이상일 경우 출력을 해주게끔 진행했더니 전부 출력되진 않습니다(두 번째 데이터가 출력이 안되었네요!). 가지고 있는 이미지로 진행했다면 어떤 이미지였는지 직접 확인할 수 있었겠지만, 예측해볼 이미지 파일이 많거나, cifar10 데이터로 예측했을 때는 어떤 이미지가 정말 맞게 예측을 했는지는 하나하나 확인해 보기는 어려울 듯 합니다.
따라서 이런 부분을 시각화 하기 위해서는 다른 방법으로도 확인이 가능합니다. matplotlib.pyplot을 사용하면 소스를 쉽게 구성해볼 수 있습니다. 이 부분에 대해서는 다음에 시간이 나면 추가적으로 짜서 공개하겠습니다..ㅠ
4. 소스 코드
훈련을 진행하고 훈련된 모델을 파일로 저장하는 전체적인 소스코드는 아래와 같습니다.
import keras.backend.tensorflow_backend as K
from keras.datasets import cifar10
from keras.utils import np_utils
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Conv2D, MaxPooling2D, Dense, Flatten, Dropout
from keras.callbacks import EarlyStopping, ModelCheckpoint
import matplotlib.pyplot as plt
import os
(X_train, y_train), (X_test, y_test) = cifar10.load_data()
# 정규화(dataset 전처리)
X_train = X_train.astype(float) / 255.0
X_test = X_test.astype(float) / 255.0
## 원-핫 인코딩
y_train = np_utils.to_categorical(y_train)
y_test = np_utils.to_categorical(y_test)
num_classes = y_test.shape[1]
print("X_train data\n ", X_train)
print("y_train data\n ", y_train)
with K.tf_ops.device('/device:CPU:0'):
# Sequential은 모델의 계층을 순차적으로 쌓아 생성하는 방법을 말한다.
# Conv2D(컨볼루션 필터의 수, 컨볼루션 커널(행,열) 사이즈, padding(valid(input image > output image), same(입력 = 출력),
# 샘플 수를 제외한 입력 형태(행, 열 채널 수)), 입력 이미지 사이즈, 활성화 함수)
# MaxPooling은 풀링 사이즈에 맞춰 가장 큰 값을 추출함 (2,2)일 경우 입력 영상 크기에서 반으로 줄어듬.
model = Sequential()
model.add(Conv2D(32, (3,3), input_shape=X_train.shape[1:], activation='relu', padding='same'))
model.add(Conv2D(64, (3,3), activation='relu', padding='same'))
model.add(MaxPooling2D(pool_size=(2,2)))
model.add(Dropout(0.25))
# 전결합층(Fully-Conneected layer)에 전달하기 위해서 1차원 자료로 바꾸어 주는 함수
# Dense(출력 뉴런 수, 입력 뉴런 수, 활성화 함수(linear, relu, sigmoid, softmax)) 로 구성된다.
model.add(Flatten())
model.add(Dense(256, activation='relu'))
model.add(Dropout(0.25))
model.add(Dense(num_classes, activation='softmax'))
# model.compile(loss=카테고리가 3개 이상('categorical_crossentropy'), adam : 경사 하강법, accuracy : 평가 척도)
model.compile(loss='categorical_crossentropy', optimizer='adam', metrics=['accuracy'])
model_dir = './model'
if not os.path.exists(model_dir):
os.mkdir(model_dir)
model_path = model_dir + '/cifar10_classification_test.h5'
checkpoint = ModelCheckpoint(filepath=model_path , monitor='val_loss', verbose=1, save_best_only=True)
early_stopping = EarlyStopping(monitor='val_loss', patience=8)
# 구성 해논 모델의 계층 구조를 간략적으로 보여주는 함수
model.summary()
history = model.fit(X_train, y_train, batch_size=64, epochs=80, validation_data=(X_test, y_test), callbacks=[checkpoint, early_stopping], shuffle=True)
print("정확도 : %.4f" % (model.evaluate(X_test, y_test)[1]))
이렇게 간단하게 cnn모델을 구성해보고 예측해보는 방법을 진행해봤습니다. 두서 없이 작성하느라 도움이 될지 잘 모르겠습니다만.. 지속적으로 게시글을 수정하도록 하겠습니다.
2020/12/21
현재까지 확인해본 바로는 K.tf_ops.device('/device:CPU:0'): 부분에서 GPU로 변경했을시에 RuntimeError: /job:localhost/replica:0/task:0/device:GPU:0 unknown device. 로 런타임 에러가 발생하는 경우가 존재하는데, 이는 현재 해당 환경에서 GPU 사용이 제대로 인식되지 않을 가능성이 높습니다. 이는 CUDA, CuDNN을 설치하였을 경우에 Keras나 Tensorflow의 버전이 다르면 인식을 못하는 경우가 있는 듯 합니다(검증이 필요함). 따라서 자신이 설치한 환경의 버전을 확인하셔서 GPU를 사용하시면 될 듯 합니다.
K.tf_ops.device()를 사용하였으나 tf를 별도로 선언해서 사용하시면 tf.device()를 사용하여도 무방합니다.
'IT > AI' 카테고리의 다른 글
| Tensorflow2 Object Detection API 설치 및 환경 구성 (0) | 2020.11.24 |
|---|---|
| JETSON TX2 Yolo 설치 및 Onboard camera 사용하기 (0) | 2020.04.13 |
| Anaconda(아나콘다) 에서 Keras(케라스) 설치하기 (2) | 2020.03.18 |
