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

■ 순환 신경망을 만드는 방법을 보여준다.

▶ 예제 코드 (PY)


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"))

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

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

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"]

sourceNDArray = GetSourceNDArray(sequenceList, windowSize = 4)

trainInputNDArray         = sourceNDArray[:, 0:4]
trainCorrectOutputNDArray = sourceNDArray[:, 4  ]

maximumIndex = 13

trainInputNDArray = trainInputNDArray / float(maximumIndex)
trainInputNDArray = np.reshape(trainInputNDArray, (50, 4, 1))

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, 1), stateful = True))
model.add(layers.Dense(outputNodeCount, activation = "softmax"))

model.compile(loss = "categorical_crossentropy", optimizer = "adam", metrics = ["accuracy"])

print("모델 정의를 종료합니다.")

print("모델 학습을 시작합니다.")

epcohCount = 2000

history = LossHistory()

history.init()

for epochIndex in range(epcohCount):
    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
inputSequenceList  = [codeDictionary[item] / float(maximumIndex) for item in inputSequenceList]

for i in range(predictionCount):
    inputSequenceNDArray = np.array(inputSequenceList)
    inputSequenceNDArray = np.reshape(inputSequenceNDArray, (1, 4, 1)) # 샘플 수, 타입 스텝 수, 속성 수
    predictionNDArray = model.predict(inputSequenceNDArray)
    index = np.argmax(predictionNDArray)
    resultSequenceList.append(indexDictionary[index])
    inputSequenceList.append(index / float(maximumIndex))
    inputSequenceList.pop(0)

model.reset_states()

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

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

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

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"))

# 소스 ND 배열 구하기 함수를 정의한다.

def GetSourceNDArray(sourceList, windowSize):

targetList = []

for i in range(len(sourceList) - windowSize):

subsetList = sourceList[i:(i + windowSize + 1)]

targetList.append([codeDictionary[item] for item in subsetList])

return np.array(targetList)

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

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"]

sourceNDArray = GetSourceNDArray(sequenceList, windowSize = 4)

trainInputNDArray = sourceNDArray[:, 0:4]

trainCorrectOutputNDArray = sourceNDArray[:, 4 ]

maximumIndex = 13

trainInputNDArray = trainInputNDArray / float(maximumIndex)

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

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, 1), stateful = True))

model.add(layers.Dense(outputNodeCount, activation = "softmax"))

model.compile(loss = "categorical_crossentropy", optimizer = "adam", metrics = ["accuracy"])

print("모델 정의를 종료합니다.")

print("모델 학습을 시작합니다.")

epcohCount = 2000

history = LossHistory()

history.init()

for epochIndex in range(epcohCount):

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

inputSequenceList = [codeDictionary[item] / float(maximumIndex) for item in inputSequenceList]

for i in range(predictionCount):

inputSequenceNDArray = np.array(inputSequenceList)

inputSequenceNDArray = np.reshape(inputSequenceNDArray, (1, 4, 1)) # 샘플 수, 타입 스텝 수, 속성 수

predictionNDArray = model.predict(inputSequenceNDArray)

index = np.argmax(predictionNDArray)

resultSequenceList.append(indexDictionary[index])

inputSequenceList.append(index / float(maximumIndex))

inputSequenceList.pop(0)

model.reset_states()

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

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