|
from __future__ import print_function |
|
import keras |
|
from keras.datasets import mnist |
|
from keras.models import Sequential |
|
from keras.layers import Dense, Dropout, Flatten |
|
from keras.layers import Conv2D, MaxPooling2D |
|
from keras import backend as K |
|
from keras.callbacks import ModelCheckpoint, CSVLogger, EarlyStopping |
|
import os |
|
import numpy as np |
|
import math |
|
|
|
batch_size = 128 |
|
num_classes = 10 |
|
epochs = 5 |
|
|
|
log_file_path = r'E:\Landmark\mnist_training_log.log' |
|
model_checkpoint_path = r"E:\Landmark\\mnist.h5" |
|
model_save_path = r"E:\Landmark\\mnist_model_{}.hd5.h5" |
|
weights_filepath="E:\\Landmark\\mnist-weights-improvement-{epoch:02d}.hdf5" |
|
|
|
# input image dimensions |
|
img_rows, img_cols = 28, 28 |
|
|
|
# the data, split between train and test sets |
|
(x_train, y_train), (x_test, y_test) = mnist.load_data() |
|
|
|
if K.image_data_format() == 'channels_first': |
|
x_train = x_train.reshape(x_train.shape[0], 1, img_rows, img_cols) |
|
x_test = x_test.reshape(x_test.shape[0], 1, img_rows, img_cols) |
|
input_shape = (1, img_rows, img_cols) |
|
else: |
|
x_train = x_train.reshape(x_train.shape[0], img_rows, img_cols, 1) |
|
x_test = x_test.reshape(x_test.shape[0], img_rows, img_cols, 1) |
|
input_shape = (img_rows, img_cols, 1) |
|
|
|
x_train = x_train.astype('float32') |
|
x_test = x_test.astype('float32') |
|
x_train /= 255 |
|
x_test /= 255 |
|
print('x_train shape:', x_train.shape) |
|
print(x_train.shape[0], 'train samples') |
|
print(x_test.shape[0], 'test samples') |
|
|
|
# convert class vectors to binary class matrices |
|
y_train = keras.utils.to_categorical(y_train, num_classes) |
|
y_test = keras.utils.to_categorical(y_test, num_classes) |
|
|
|
#Data Batching |
|
class Generator(keras.utils.Sequence): |
|
# Class is a dataset wrapper for better training performance |
|
def __init__(self, x_set, y_set, batch_size, datacount): |
|
self.x = x_set |
|
self.y = y_set |
|
self.batch_size = batch_size |
|
self.indices = np.arange(self.x.shape[0]) |
|
self.idx = 0 |
|
self.datacount = datacount |
|
|
|
def __len__(self): |
|
print('length') |
|
print(math.ceil(self.datacount/ self.batch_size)) |
|
return math.ceil(self.datacount/ self.batch_size) |
|
|
|
def __getitem__(self, idx): |
|
print('idx') |
|
print(idx) |
|
i1 = idx*self.batch_size |
|
i2 = (idx+1)*self.batch_size |
|
print('Start-' + str(i1) + '- End-' + str(i2)) |
|
if(i2 > self.datacount): |
|
i2 = self.datacount |
|
batch_x = self.x[i1:i2] |
|
batch_y = self.y[i1:i2] |
|
return batch_x, batch_y |
|
|
|
def on_epoch_end(self): |
|
np.random.shuffle(self.indices) |
|
|
|
#Save Model after every Epoch |
|
#https://stackoverflow.com/questions/54323960/save-keras-model-at-specific-epochs |
|
class CustomSaver(keras.callbacks.Callback): |
|
def on_epoch_end(self, epoch, logs={}): |
|
#if epoch == 2: # or save after some epoch, each k-th epoch etc. |
|
self.model.save(model_save_path.format(epoch)) |
|
|
|
batch_size = 500 |
|
print('x_train') |
|
print(len(x_train)) |
|
print('x_test') |
|
print(len(x_test)) |
|
|
|
training_generator = Generator(x_train, y_train, batch_size, len(x_train)) |
|
validation_generator = Generator(x_test, y_test, batch_size, len(x_test)) |
|
|
|
from keras.models import load_model |
|
|
|
#Load and Continue Training |
|
# load weights if it exists |
|
if os.path.exists(model_checkpoint_path): |
|
print('Loading Definitions') |
|
model = load_model(model_checkpoint_path) |
|
else: |
|
model = Sequential() |
|
model.add(Conv2D(32, kernel_size=(3, 3), |
|
activation='relu', |
|
input_shape=input_shape)) |
|
model.add(Conv2D(64, (3, 3), activation='relu')) |
|
model.add(MaxPooling2D(pool_size=(2, 2))) |
|
model.add(Dropout(0.25)) |
|
model.add(Flatten()) |
|
model.add(Dense(128, activation='relu')) |
|
model.add(Dropout(0.5)) |
|
model.add(Dense(num_classes, activation='softmax')) |
|
model.compile(loss=keras.losses.categorical_crossentropy, |
|
optimizer=keras.optimizers.Adadelta(), |
|
metrics=['accuracy']) |
|
|
|
#Add Early Stop and Checkpoint |
|
early_stop = EarlyStopping(monitor='val_loss', patience=5, verbose=1) |
|
checkpoint = ModelCheckpoint(weights_filepath, monitor='val_loss', verbose=1, save_best_only=True, mode='auto') |
|
csv_logger = CSVLogger(log_file_path, append=False) |
|
saver = CustomSaver() |
|
callbacks_list = [checkpoint,early_stop,csv_logger,saver] |
|
|
|
model.fit_generator(training_generator, validation_data = validation_generator, epochs = 10, callbacks=callbacks_list) |
|
|
|
score = model.evaluate(x_test, y_test, verbose=0) |
|
print('Test loss:', score[0]) |
|
print('Test accuracy:', score[1]) |
|
model.save(model_save_path) |
|
|
|
import pandas as pd |
|
import matplotlib.pyplot as plt |
|
# Plot the Loss |
|
file_name = log_file_path |
|
df = pd.DataFrame.from_csv(file_name) |
|
print(df.head()) |
|
training_loss = df['loss'] |
|
test_loss = df['val_loss'] |
|
print(training_loss) |
|
print(test_loss) |
|
|
|
epoch_count = range(1, len(training_loss) + 1) |
|
plt.plot(epoch_count, training_loss, 'r--') |
|
plt.plot(epoch_count, test_loss, 'b-') |
|
plt.legend(['Training Loss', 'Test Loss']) |
|
plt.xlabel('Epoch') |
|
plt.ylabel('Loss') |
|
plt.show(); |
