from keras.models import Sequential, Model
from keras.layers import Activation, Dropout, Flatten, Dense, Input, Reshape
from keras.preprocessing.image import ImageDataGenerator, array_to_img, img_to_array, load_img
from img_exp import ImgExp
import numpy as np
import matplotlib.pyplot as plt
from random import randint
from keras.callbacks import ModelCheckpoint, CSVLogger, EarlyStopping
import time
from keras.models import load_model
import os
from util import *
from data_management import load_data
import sys
from multi_AEmodel import *
class AEExp(ImgExp):
"""
Frame based autoencoder experiment. Images are reconstructed individually, with no temporal information used.
All params are attributes, and are initialized in ImgExp parent class.
"""
def __init__(self, model = None, img_width = 64, img_height = 64, \
model_name = 'None',
batch_size = 32,\
model_type = None, \
pre_load = None, initial_epoch = 0, \
epochs = 1, hor_flip = None,\
zoom_range = 0, dset = 'Thermal'):
ImgExp.__init__(self, model = model, img_width = img_width,\
img_height = img_height, model_name = model_name, \
batch_size = batch_size, \
model_type = model_type, \
pre_load = pre_load, initial_epoch = initial_epoch,\
epochs = epochs, hor_flip = hor_flip,\
zoom_range = zoom_range, dset = dset)
def play_frames_with_reconstructions(self, to_save = None):
"""
Plays frames of test_data with reconstuction. Saves animation to to_save
Params:
str to_save: if not None, saves animation to './to_save.mp4'
"""
preds = self.model.predict(self.test_data.reshape(len(self.test_data),64,64,1))
print(np.amax(preds[0]), np.amin(preds[0]))
ani = animate_fall_detect(self.model, self.test_data, self.img_width, self.img_height)
if to_save != None:
ani.save('{}.mp4'.format(to_save))
ani.event_source.stop()
del ani
plt.close()
def train(self, sample_weight=None, verbose = 2):
"""
trains the autoencoder model on data loaded from load_train_data. This data is non-sequential; that is,
frames are reconstructed one by one. Reconstruction error (MSE) is minimized. Checkpoints and logs are saved to
'./Checkpoints/dset/'
'./logs/dset/'
Model is saved as per save_exp method in parent class
"""
print(self.model.summary())
model_name = self.model_name
base_cp = './Checkpoints/{}'.format(self.dset)
base_logs = './logs/{}'.format(self.dset)
if not os.path.isdir(base_cp):
os.makedirs(base_cp)
if not os.path.isdir(base_logs):
os.makedirs(base_logs)
checkpointer = ModelCheckpoint( filepath = base_cp + '/' + model_name + '-' + \
'{epoch:03d}-{loss:.3f}.hdf5', period = 100, verbose =1)
early_stopper = EarlyStopping(patience=5, verbose = 1, monitor = 'loss', min_delta = 1e-5)
timestamp = time.time()
csv_logger = CSVLogger(base_logs + '/' + model_name + '-' + 'training-' + \
str(timestamp) + '.log')
callbacks_list = [csv_logger, checkpointer]
self.train_data = self.train_data.reshape(len(self.train_data) ,self.img_width, self.img_height, 1)
print('training on data of shape {}, with model {}, with hor_flip {}'.format(self.train_data.shape, self.model_name, self.hor_flip))
if self.hor_flip == True:
datagen = ImageDataGenerator(horizontal_flip= self.hor_flip, zoom_range = self.zoom_range)
self.model.fit_generator(datagen.flow(self.train_data, self.train_data,\
batch_size = self.batch_size), steps_per_epoch=len(self.train_data) / self.batch_size, \
epochs=self.epochs, callbacks = callbacks_list, verbose = verbose, initial_epoch = self.initial_epoch)
else:
self.model.fit(self.train_data, self.train_data,
epochs = self.epochs,
batch_size= self.batch_size,
shuffle=True,
callbacks = callbacks_list, verbose = verbose, initial_epoch = self.initial_epoch
)
self.save_exp()
def test(self, raw = False, animate = False):
get_stats_for_all_vids(experiment = self,\
metric = 'AUC', dset = self.dset, raw = raw, animate = animate)
def get_MSE(self, test_data):
'''
Gets mean squared error between test_data array of images, and their reconstructions from self.model
Params:
ndarray test_data: Data consiting of frames, ie. of dimension (samples, img_height, img_width)
Returns:
ndarray of MSE scores, one for each frame in test_data, ie dimensions (samples,1).
'''
if self.model_type == 'conv' or 1: #If using flow(no reason not to) then all take same shape TODO remove condition?
test_data = test_data.reshape(len(test_data), self.img_width, self.img_height, 1)
else:
test_data = test_data.reshape((len(test_data), np.prod(test_data.shape[1:])))
decoded_imgs = self.model.predict(test_data)
RE = MSE(test_data, decoded_imgs)
return RE
def get_features(self, layer_name, train_or_test = 'test'):
from keras.models import Model
model = self.model # create the original model
if train_or_test == 'test':
data = self.test_data.reshape(len(self.test_data), self.img_width, self.img_height, 1)
else:
data = self.train_data.reshape(len(self.train_data), self.img_width, self.img_height, 1)
#layer_name = 'my_layer'
intermediate_layer_model = Model(inputs=model.input,
outputs=model.get_layer(layer_name).output)
intermediate_output = intermediate_layer_model.predict(data)
#print(intermediate_output)
return intermediate_output
def set_train_data(self, raw = False):
"""
sets train_data attribute to ADL data correpsonding to the dataset for this experiment
"""
split_by_vid_or_class = 'Split_by_class'
vid_class = 'NonFall'
data = load_data(split_by_vid_or_class = split_by_vid_or_class, raw = raw,\
img_width = self.img_width, img_height = self.img_height, vid_class = vid_class, dset = self.dset)
self.train_data = data
Datasets
Models
