[PYTHON/TENSORFLOW] 컨볼루션 신경망 만들기 (얼굴 감정 판정)

■ 얼굴 감정을 판정하는 컨볼루션 신경망을 만드는 방법을 보여준다.

 

※ 첨부 소스 코드를 특정 폴더에 압축을 풀고, EmotionDetector.py를 실행해 데이터를 학습하고, TestEmotion.py를 실행해 샘플 이미지를 테스트 한다.

 

SourceCode.zip

9.91MB

SourceCode.z01

19.53MB

▶ EmotionHelper.py

import numpy as np
import os
import pandas as pd
import six.moves.cPickle as pickle

IMAGE_SIZE         = 48
OUTPUT_COUNT       = 7
VALIDATION_PERCENT = 0.1

np.random.seed(0)

class EmotionResult:
    def __init__(self):
        self.anger    = 0
        self.disgust  = 0
        self.fear     = 0
        self.happy    = 0
        self.sad      = 0
        self.surprise = 0
        self.neutral  = 0

    def Evaluate(self, label):
        if (0 == label): self.anger    = self.anger    + 1
        if (1 == label): self.disgust  = self.disgust  + 1
        if (2 == label): self.fear     = self.fear     + 1
        if (3 == label): self.happy    = self.happy    + 1
        if (4 == label): self.sad      = self.sad      + 1
        if (5 == label): self.surprise = self.surprise + 1
        if (6 == label): self.neutral  = self.neutral  + 1

    def DisplayResult(self, totalScore):
        print("분노 = " + str((self.anger    / float(totalScore)) * 100) + "%")
        print("혐오 = " + str((self.disgust  / float(totalScore)) * 100) + "%")
        print("공포 = " + str((self.fear     / float(totalScore)) * 100) + "%")
        print("행복 = " + str((self.happy    / float(totalScore)) * 100) + "%")
        print("슬픔 = " + str((self.sad      / float(totalScore)) * 100) + "%")
        print("놀람 = " + str((self.surprise / float(totalScore)) * 100) + "%")
        print("중립 = " + str((self.neutral  / float(totalScore)) * 100) + "%")

def ReadData(sourceDirectoryPath, force = False):
    def CreateOneHotLabel(x):
        labelNDArray = np.zeros((1, OUTPUT_COUNT), dtype = np.float32)
        labelNDArray[:, int(x)] = 1
        return labelNDArray

    pickleFilePath = os.path.join(sourceDirectoryPath, "EmotionDetector.pickle")

    if force or not os.path.exists(pickleFilePath):
        trainFilePath = os.path.join(sourceDirectoryPath, "train.csv")

        dataFrame = pd.read_csv(trainFilePath)

        dataFrame["Pixels"] = dataFrame["Pixels"].apply(lambda x: np.fromstring(x, sep = " ") / 255.0)

        dataFrame = dataFrame.dropna()

        print("train.csv 파일을 읽고 있습니다...")

        trainImageNDArray = np.vstack(dataFrame["Pixels"]).reshape(-1, IMAGE_SIZE, IMAGE_SIZE, 1)

        print(trainImageNDArray.shape)

        trainLabelNDArray = np.array(list(map(CreateOneHotLabel, dataFrame["Emotion"].values))).reshape(-1, OUTPUT_COUNT)

        print(trainLabelNDArray.shape)

        permutations = np.random.permutation(trainImageNDArray.shape[0])

        trainImageNDArray = trainImageNDArray[permutations]
        trainLabelNDArray = trainLabelNDArray[permutations]

        validationIndex = int(trainImageNDArray.shape[0] * VALIDATION_PERCENT)

        validImageNDArray = trainImageNDArray[:validationIndex]
        validLabelNDArray = trainLabelNDArray[:validationIndex]

        trainImageNDArray = trainImageNDArray[validationIndex:]
        trainLabelNDArray = trainLabelNDArray[validationIndex:]

        print("test.csv 파일을 읽고 있습니다...")

        testFilePath = os.path.join(sourceDirectoryPath, "test.csv")

        dataFrame = pd.read_csv(testFilePath)

        dataFrame["Pixels"] = dataFrame["Pixels"].apply(lambda x: np.fromstring(x, sep = " ") / 255.0)

        dataFrame = dataFrame.dropna()

        testImageNDArray = np.vstack(dataFrame['Pixels']).reshape(-1, IMAGE_SIZE, IMAGE_SIZE, 1)

        with open(pickleFilePath, "wb") as file:
            try:
                print("파일을 직렬화하고 있습니다...")

                saveDictionary = {
                    "trainImageNDArray" : trainImageNDArray,
                    "trainLabelNDArray" : trainLabelNDArray,
                    "validImageNDArray" : validImageNDArray,
                    "validLabelNDArray" : validLabelNDArray,
                    "testImageNDArray"  : testImageNDArray,
                }

                pickle.dump(saveDictionary, file, pickle.HIGHEST_PROTOCOL)
            except:
                print("파일을 직렬화할 수 없습니다.")

    with open(pickleFilePath, "rb") as file:
        saveDictionary = pickle.load(file)

        trainImageNDArray = saveDictionary["trainImageNDArray"]
        trainLabelNDArray = saveDictionary["trainLabelNDArray"]
        validImageNDArray = saveDictionary["validImageNDArray"]
        validLabelNDArray = saveDictionary["validLabelNDArray"]
        testImageNDArray  = saveDictionary["testImageNDArray" ]

    return trainImageNDArray, trainLabelNDArray, validImageNDArray, validLabelNDArray, testImageNDArray

 

▶ EmotionDetector.py

import tensorflow as tf
import numpy as np
import time

import EmotionHelper

dataDirectoryPath = "data/"
logDirectoryPath  = "log/"

BATCH_SIZE     = 128
LEARNING_RATIO = 0.001
EPOCH_COUNT    = 1000
REGULARIZATION = 0.001
IMAGE_SIZE     = 48
OUTPUT_COUNT   = 7

def GetNextBatchData(sourceImageNDArray, sourceLabelNDArray, epoch):
    offset = (epoch * BATCH_SIZE) % (sourceImageNDArray.shape[0] - BATCH_SIZE)
    batchInputNDArray       = sourceImageNDArray[offset:offset + BATCH_SIZE]
    batchRightOutputNDArray = sourceLabelNDArray[offset:offset + BATCH_SIZE]
    return batchInputNDArray, batchRightOutputNDArray

# 학습 데이터 등을 로드한다.
trainImageNDArray, trainLabelNDArray, validImageNDArray, validLabelNDArray, testImageNDArray = EmotionHelper.ReadData(dataDirectoryPath)

print("학습 데이터 크기   : %s" % trainImageNDArray.shape[0])
print('검증 데이터 크기   : %s' % validImageNDArray.shape[0])
print("테스트 데이터 크기 : %s" % testImageNDArray.shape[0])

# 가중치를 정의한다.
convolutionLayer1WeightVariable     = tf.get_variable("convolutionLayer1Weight"    , initializer = tf.truncated_normal([5, 5, 1, 32], stddev = 0.02))
convolutionLayer2WeightVariable     = tf.get_variable("convolutionLayer2Weight"    , initializer = tf.truncated_normal([3, 3, 32, 64], stddev = 0.02))
fullyConntectedLayer1WeightVariable = tf.get_variable("fullyConntectedLayer1Weight", initializer = tf.truncated_normal([(IMAGE_SIZE // 4) * (IMAGE_SIZE // 4) * 64, 256], stddev = 0.02))
outputLayerWeightVariable           = tf.get_variable("outputLayerWeight"          , initializer = tf.truncated_normal([256, OUTPUT_COUNT], stddev = 0.02))

# 바이어스를 정의한다.
convolutionLayer1BiasVariable    = tf.get_variable("convolutionLayer1Bias"   , initializer = tf.constant(0.0, shape = [32]))
convolutionLayer2BiasVariable    = tf.get_variable("convolutionLayer2Bias"   , initializer = tf.constant(0.0, shape = [64]))
fullyConnectedLayer1BiasVariable = tf.get_variable("fullyConnectedLayer1Bias", initializer = tf.constant(0.0, shape = [256]))
outputLayerBiasVariable          = tf.get_variable("outputLayerBias"         , initializer = tf.constant(0.0, shape = [OUTPUT_COUNT]))

# 모델을 정의한다.
globalStepVariable = tf.Variable(0, trainable = False)
inputTensor        = tf.placeholder(tf.float32, [None, IMAGE_SIZE, IMAGE_SIZE, 1], name = "input")
rightOutputTensor  = tf.placeholder(tf.float32, [None, OUTPUT_COUNT])

# 컨볼루션 레이어 #1을 정의한다.
tf.summary.histogram("convolutionLayer1Weight", convolutionLayer1WeightVariable)
tf.summary.histogram("convolutionLayer1Bias"  , convolutionLayer1BiasVariable  )

convolutionLayer1OutputTensor        = tf.nn.conv2d(inputTensor, convolutionLayer1WeightVariable, strides = [1, 1, 1, 1], padding = "SAME")
convolutionLayer1OutputTensorBias    = tf.nn.bias_add(convolutionLayer1OutputTensor, convolutionLayer1BiasVariable)
convolutionLayer1OutputTensorReLU    = tf.nn.relu(convolutionLayer1OutputTensorBias)
convolutionLayer1OutputTensorMaxPool = tf.nn.max_pool(convolutionLayer1OutputTensorReLU, ksize = [1, 2, 2, 1], strides = [1, 2, 2, 1], padding = "SAME")

tf.add_to_collection("loss", tf.nn.l2_loss(convolutionLayer1WeightVariable))
tf.add_to_collection("loss", tf.nn.l2_loss(convolutionLayer1BiasVariable  ))

# 컨볼루션 레이어 #2를 정의한다.
tf.summary.histogram("convolutionLayer2Weight", convolutionLayer2WeightVariable)
tf.summary.histogram("convolutionLayer2Bias"  , convolutionLayer2BiasVariable  )

convolutionLayer2OutputTensor        = tf.nn.conv2d(convolutionLayer1OutputTensorMaxPool, convolutionLayer2WeightVariable, strides = [1, 1, 1, 1], padding = "SAME")
convolutionLayer2OutputTensorBias    = tf.nn.bias_add(convolutionLayer2OutputTensor, convolutionLayer2BiasVariable)
convolutionLayer2OutputTensorReLU    = tf.nn.relu(convolutionLayer2OutputTensorBias)
convolutionLayer2OutputTensorMaxPool = tf.nn.max_pool(convolutionLayer2OutputTensorReLU, ksize = [1, 2, 2, 1], strides = [1, 2, 2, 1], padding = "SAME")

tf.add_to_collection("loss", tf.nn.l2_loss(convolutionLayer2WeightVariable))
tf.add_to_collection("loss", tf.nn.l2_loss(convolutionLayer2BiasVariable  ))

# 전 연결층 레이어 #1을 정의한다.
tf.summary.histogram("fullyConntectedLayer1Weight", fullyConntectedLayer1WeightVariable)
tf.summary.histogram("fullyConnectedLayer1Bias"   , fullyConnectedLayer1BiasVariable   )

fullyConnectedLayer1OutputTensor        = tf.reshape(convolutionLayer2OutputTensorMaxPool, [-1, (IMAGE_SIZE // 4) * (IMAGE_SIZE // 4) * 64])
fullyConnectedLayer1OutputTensorReLU    = tf.nn.relu(tf.matmul(fullyConnectedLayer1OutputTensor, fullyConntectedLayer1WeightVariable) + fullyConnectedLayer1BiasVariable)
fullyConnectedLayer1OutputTensorDropout = tf.nn.dropout(fullyConnectedLayer1OutputTensorReLU, 0.5)

# 출력 레이어를 정의한다.
tf.summary.histogram("outputLayerWeight", outputLayerWeightVariable)
tf.summary.histogram("outputLayerBias"  , outputLayerBiasVariable  )

outputLayerTensor = tf.matmul(fullyConnectedLayer1OutputTensorDropout, outputLayerWeightVariable) + outputLayerBiasVariable

outputTensor = tf.nn.softmax(outputLayerTensor, name = "output")

# 비용 함수를 정의한다.
costTensor = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits_v2(logits = outputLayerTensor, labels = rightOutputTensor))

tf.summary.scalar("cost", costTensor)

regularizationLossTensor = tf.add_n(tf.get_collection("loss"))

tf.summary.scalar("regularizationLoss", regularizationLossTensor)

lossTensor = costTensor + REGULARIZATION * regularizationLossTensor

# 옵티마이저를 정의한다.
optimizerOperation = tf.train.AdamOptimizer(LEARNING_RATIO).minimize(lossTensor, global_step = globalStepVariable)

# 데이터를 학습한다.
summaryTensorp = tf.summary.merge_all()

with tf.Session() as session:
    session.run(tf.global_variables_initializer())
    fileWriter = tf.summary.FileWriter(logDirectoryPath, session.graph)
    saver = tf.train.Saver()
    checkpointStatue = tf.train.get_checkpoint_state(logDirectoryPath)
    if checkpointStatue and checkpointStatue.model_checkpoint_path:
        saver.restore(session, checkpointStatue.model_checkpoint_path)
        print("모델이 복구되었습니다...")
    for epoch in range(EPOCH_COUNT + 1):
        batchInputNDArray, batchRightOutputNDArray = GetNextBatchData(trainImageNDArray, trainLabelNDArray, epoch)
        session.run(optimizerOperation, feed_dict = {inputTensor : batchInputNDArray, rightOutputTensor : batchRightOutputNDArray})
        if epoch % 10 == 0:
            loss, summary = session.run([lossTensor, summaryTensorp], feed_dict = {inputTensor : batchInputNDArray, rightOutputTensor : batchRightOutputNDArray})
            fileWriter.add_summary(summary, global_step = epoch)
            print("훈련 손실 : %f" % loss)
        if epoch % 100 == 0:
            loss = session.run(lossTensor, feed_dict = {inputTensor : validImageNDArray, rightOutputTensor : validLabelNDArray})
            print("%s 검증 손실 : %f" % (time.strftime("%X", time.localtime(time.time())), loss))
            saver.save(session, logDirectoryPath + "model.ckpt", global_step = epoch)

 

▶ TestEmotion.py

import matplotlib.image as img
import matplotlib.pyplot as pp
import numpy as np
import tensorflow as tf

import EmotionHelper

# 소스 이미지를 로드합니다.
sourceImageNDArray    = img.imread("source.jpg")
grayscaleImageNDArray = np.dot(sourceImageNDArray[...,:3], [0.299, 0.587, 0.114])
inputImageNDArray     = np.resize(grayscaleImageNDArray, (1, 48, 48, 1))

pp.imshow(grayscaleImageNDArray, cmap = pp.get_cmap("gray"))
pp.show()

# 신경망 모델 데이터를 로드합니다.
session = tf.InteractiveSession()

saver = tf.train.import_meta_graph("log/model.ckpt-1000.meta")

saver.restore(session, "log/model.ckpt-1000")

tf.get_default_graph().as_graph_def()

inputTensor  = session.graph.get_tensor_by_name("input:0" )
outputTensor = session.graph.get_tensor_by_name("output:0")

# 얼굴 감정을 평가합니다.
emotionResult   = EmotionHelper.EmotionResult()
evaluationCount = 1000

for i in range(0, evaluationCount):
    outputNDArray = session.run(outputTensor, feed_dict = {inputTensor : inputImageNDArray})
    output = session.run(tf.argmax(outputNDArray, 1))
    output = output[0]
    output = int(output)
    emotionResult.Evaluate(output)

emotionResult.DisplayResult(evaluationCount)