[PYTHON/KERAS] 순환 신경망 만들기 (나비야) : 상태 유지, 입력 속성 2개

import keras.callbacks as callbacks
import keras.models as models
import keras.layers as layers
import keras.utils as utils
import matplotlib.pyplot as pp
import numpy as np

np.random.seed(5)

# 손실 이력 클래스를 정의한다.
class LossHistory(callbacks.Callback):
    def init(self):
        self.lossList = []
        
    def on_epoch_end(self, batch, logDictionary = {}):
        self.lossList.append(logDictionary.get("loss"))

# 특징 리스트 구하기 함수를 정의한다.
def GetFeatureList(code):
    featureList = []
    featureList.append(scaleDictionary[code[0]] / float(maximumScale))
    featureList.append(lengthDictionary[code[1]])
    return featureList

# 소스 ND 배열 구하기 함수를 정의한다.
def GetSourceNDArray(sourceList, windowSize):
    targetInputList         = []
    targetCorrectOutputList = []
    for i in range(len(sourceList) - windowSize):
        subsetList = sourceList[i:(i + windowSize + 1)]
        for j in range(len(subsetList) - 1):
            featureList = GetFeatureList(subsetList[j])
            targetInputList.append(featureList)
        targetCorrectOutputList.append([codeDictionary[subsetList[windowSize]]])
    return np.array(targetInputList), np.array(targetCorrectOutputList)

print("데이터 로드를 시작합니다.")

maximumScale = 6.0

scaleDictionary  = {"c" : 0, "d" : 1, "e" : 2, "f" : 3, "g" : 4, "a" : 5, "b" : 6}

lengthDictionary = {"4" : 0, "8" : 1}

codeDictionary = {"c4" : 0, "d4" : 1, "e4" : 2, "f4" : 3 , "g4" : 4 , "a4" : 5 , "b4" : 6,
                  "c8" : 7, "d8" : 8, "e8" : 9, "f8" : 10, "g8" : 11, "a8" : 12, "b8" : 13}

indexDictionary = {0 : "c4", 1 : "d4", 2 : "e4", 3  : "f4", 4  : "g4", 5  : "a4", 6  : "b4",
                   7 : "c8", 8 : "d8", 9 : "e8", 10 : "f8", 11 : "g8", 12 : "a8", 13 : "b8"}

sequenceList = ["g8", "e8", "e4", "f8", "d8", "d4", "c8", "d8", "e8", "f8", "g8", "g8", "g4",
                "g8", "e8", "e8", "e8", "f8", "d8", "d4", "c8", "e8", "g8", "g8", "e8", "e8", "e4",
                "d8", "d8", "d8", "d8", "d8", "e8", "f4", "e8", "e8", "e8", "e8", "e8", "f8", "g4",
                "g8", "e8", "e4", "f8", "d8", "d4", "c8", "e8", "g8", "g8", "e8", "e8", "e4"]

trainInputNDArray, trainCorrectOutputNDArray = GetSourceNDArray(sequenceList, windowSize = 4)

trainInputNDArray = np.reshape(trainInputNDArray, (50, 4, 2))

trainCorrectOutputNDArray = utils.np_utils.to_categorical(trainCorrectOutputNDArray)

outputNodeCount = trainCorrectOutputNDArray.shape[1]

print("데이터 로드를 종료합니다.")
 
print("모델 정의를 시작합니다.")

model = models.Sequential()

model.add(layers.LSTM(128, batch_input_shape = (1, 4, 2), stateful = True))
model.add(layers.Dense(outputNodeCount, activation = "softmax"))
    
model.compile(loss = "categorical_crossentropy", optimizer = "adam", metrics = ["accuracy"])

print("모델 정의를 종료합니다.")
 
print("모델 학습을 시작합니다.")

epochCount = 2000

history = LossHistory()

history.init()

for epochIndex in range(epochCount):
    print("epoch : " + str(epochIndex))
    model.fit(trainInputNDArray, trainCorrectOutputNDArray, epochs = 1, batch_size = 1, verbose = 2, shuffle = False, callbacks = [history])
    model.reset_states()
    
pp.plot(history.lossList)

pp.ylabel("loss")
pp.xlabel("epoch")
pp.legend(["train"], loc = "upper left")

pp.show()

print("모델 학습을 종료합니다.")
 
print("모델 평가를 시작합니다.")

evaluationList = model.evaluate(trainInputNDArray, trainCorrectOutputNDArray, batch_size = 1)

print("%s : %.2f%%" %(model.metrics_names[1], evaluationList[1] * 100))

model.reset_states()

print("모델 평가를 종료합니다.")
 
print("모델 사용을 시작합니다.")

predictionCount = 50

print("한 스텝 예측을 시작합니다.")

resultSequenceList = ["g8", "e8", "e4", "f8"]

predictionNDArray = model.predict(trainInputNDArray, batch_size = 1)

for i in range(predictionCount):
    index = np.argmax(predictionNDArray[i])
    resultSequenceList.append(indexDictionary[index])

model.reset_states()

print("한 스텝 예측 : ", resultSequenceList)

print("한 스텝 예측을 종료합니다.")

print("곡 전체 예측을 시작합니다.")

inputSequenceList  = ["g8", "e8", "e4", "f8"]
resultSequenceList = inputSequenceList

inputSequenceFeatureList = []

for inputSequenceItem in inputSequenceList:
    featureList = GetFeatureList(inputSequenceItem)
    inputSequenceFeatureList.append(featureList)

for i in range(predictionCount):
    inputSequenceNDArray = np.array(inputSequenceFeatureList)
    inputSequenceNDArray = np.reshape(inputSequenceNDArray, (1, 4, 2)) # 샘플 수, 타임 스텝 수, 속성 수
    predictionNDArray = model.predict(inputSequenceNDArray)
    index = np.argmax(predictionNDArray)
    resultSequenceList.append(indexDictionary[index])
    featureList = GetFeatureList(indexDictionary[index])
    inputSequenceFeatureList.append(featureList)
    inputSequenceFeatureList.pop(0)

model.reset_states()
    
print("곡 전체 예측 : ", resultSequenceList)

print("곡 전체 예측을 종료합니다.")