multiscale_feature_fusion_msr.py

# -*- coding: utf-8 -*-
"""multiscale_feature_fusion.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/1O7fQwMQiKjh6ctFqYmALx5o9znGVw5yM
"""

from google.colab import drive
drive.mount('/content/drive')

import numpy as np
import matplotlib.pyplot as plt
import os
import cv2

if not os.path.exists('/content/checkpoint'):
  os.mkdir('/content/checkpoint')

import keras
from keras.callbacks import ModelCheckpoint, LearningRateScheduler
from keras.regularizers import l1, l2, l1_l2
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Activation, Flatten, Conv2D, MaxPooling2D, Dropout
from tensorflow.keras.preprocessing.image import ImageDataGenerator

model_DJD = Sequential() 

model_DJD.add(Conv2D(32, (7, 7), input_shape=(112, 112, 3), activation = 'relu', kernel_regularizer = l1(0.001)))
model_DJD.add(MaxPooling2D(pool_size=(2, 2)))

model_DJD.add(Conv2D(64, (5, 5),activation = 'relu', kernel_regularizer = l1(0.001)))
model_DJD.add(MaxPooling2D(pool_size=(2, 2)))

model_DJD.add(Conv2D(128, (3, 3),activation = 'relu'))
model_DJD.add(MaxPooling2D(pool_size=(2, 2)))

model_DJD.add(Conv2D(256, (3, 3),activation = 'relu'))
model_DJD.add(MaxPooling2D(pool_size=(2, 2)))

model_DJD.add(Conv2D(512, (2, 2),activation = 'relu', kernel_regularizer = l2(0.01)))
model_DJD.add(MaxPooling2D(pool_size=(2, 2)))

model_DJD.add(Flatten())


model_DSM = Sequential() 

model_DSM.add(Conv2D(32, (7, 7), input_shape=(112, 112, 1), activation = 'relu', kernel_regularizer = l1(0.001)))
model_DSM.add(MaxPooling2D(pool_size=(2, 2)))

model_DSM.add(Conv2D(64, (5, 5),activation = 'relu', kernel_regularizer = l1(0.001)))
model_DSM.add(MaxPooling2D(pool_size=(2, 2)))

model_DSM.add(Conv2D(128, (3, 3),activation = 'relu'))
model_DSM.add(MaxPooling2D(pool_size=(2, 2)))

model_DSM.add(Conv2D(256, (3, 3),activation = 'relu'))
model_DSM.add(MaxPooling2D(pool_size=(2, 2)))

model_DSM.add(Conv2D(512, (2, 2),activation = 'relu', kernel_regularizer = l2(0.01)))
model_DSM.add(MaxPooling2D(pool_size=(2, 2)))

model_DSM.add(Flatten())


model = tf.keras.layers.Concatenate()([model_DJD.output,model_DSM.output])

x = Dropout(0.4)(model)
x = Dense(128, activation = 'relu')(x)
x = Dropout(0.2)(x)
output = Dense(20,activation='softmax')(x)

final_model = keras.Model(inputs=[model_DJD.input,model_DSM.input], outputs=[output])

opt = keras.optimizers.Adam(learning_rate = 0.0001)
final_model.compile(loss="categorical_crossentropy", optimizer=opt, metrics=["accuracy"])


filepath='/content/checkpoint/checkpoint-{val_accuracy:.4f}-{val_loss:.2f}-{epoch:02d}.hdf5'
checkpoint = ModelCheckpoint(filepath, monitor='val_accuracy', verbose=1, save_best_only=True, mode='max')

def generate_generator_multiple(generator, dir1, dir2, batch_size, image_size):
  gen_DJD = generator.flow_from_directory(dir1,
                                          target_size = (image_size,image_size),
                                          batch_size = batch_size,
                                          color_mode = 'rgb',
                                          class_mode="categorical",
                                          shuffle = False)
  gen_DSM = generator.flow_from_directory(dir2,
                                          target_size = (image_size,image_size),
                                          batch_size = batch_size,
                                          color_mode = 'grayscale',
                                          class_mode = 'categorical',
                                          shuffle = False)
  while True:
    DJD = gen_DJD.next()
    DSM = gen_DSM.next()
    yield [DJD[0],DSM[0]], DJD[1]

    # 0 - images
    # 1 - labels

batch_size = 64

train_imgen = ImageDataGenerator()

inputgenerator = generate_generator_multiple(generator = train_imgen,
                                             dir1 = '/content/drive/My Drive/M.Tech NITK/Minor Project/MinorProject/DJD_MSR_IDG/subject_odd',
                                             dir2 = '/content/drive/My Drive/M.Tech NITK/Minor Project/MinorProject/DSM_MSR_IDG/subject_odd',
                                             batch_size = batch_size,
                                             image_size = 112)

test_imgen = ImageDataGenerator()

testgenerator = generate_generator_multiple(generator = test_imgen,
                                             dir1 = '/content/drive/My Drive/M.Tech NITK/Minor Project/MinorProject/DJD_MSR_IDG/subject_even',
                                             dir2 = '/content/drive/My Drive/M.Tech NITK/Minor Project/MinorProject/DSM_MSR_IDG/subject_even',
                                             batch_size = batch_size,
                                             image_size = 112)

STEP_SIZE_TRAIN = 292 // batch_size
STEP_SIZE_VALID = 275 // batch_size

# clear checkpoint folder
folder = '/content/checkpoint/'
for filename in os.listdir(folder):
    file_path = os.path.join(folder, filename)
    try:
        if os.path.isfile(file_path) or os.path.islink(file_path):
            os.unlink(file_path)
        elif os.path.isdir(file_path):
            shutil.rmtree(file_path)
    except Exception as e:
        print('Failed to delete %s. Reason: %s' % (file_path, e))

history = final_model.fit(inputgenerator, 
                          epochs = 2000,
                          steps_per_epoch = STEP_SIZE_TRAIN+1,
                          validation_data = testgenerator,
                          validation_steps = STEP_SIZE_VALID+1, 
                          verbose = 2,
                          callbacks = [checkpoint], 
                          shuffle = False)

# get model with best val accuracy
import glob
list_of_files = glob.glob('/content/checkpoint/*') 
latest_file = max(list_of_files, key=os.path.getctime)
print(latest_file)

print("Model Evaluate...")
new_model = tf.keras.models.load_model(latest_file)
# loss, accuracy = new_model.evaluate(X_test,y_test)
loss, accuracy = new_model.evaluate(testgenerator,steps = STEP_SIZE_VALID+1)
print("Loss: ",loss)
print("Tesing Accuracy: ", accuracy)

plt.plot(history.history['accuracy'][:1000])
plt.plot(history.history['val_accuracy'][:1000])
plt.title('Model Accuracy for MSRAction3D Dataset')
plt.ylabel('Acuuracy')
plt.xlabel('Epochs')
plt.legend(['Train','Test'], loc='upper left')
plt.show()
plt.plot(history.history['loss'][:1000])
plt.plot(history.history['val_loss'][:1000])
plt.title('Model Loss for MSRAction3D Dataset')
plt.ylabel('Loss')
plt.xlabel('Epochs')
plt.legend(['Train','Test'], loc='upper right')
plt.show()

dot_img_file = 'model_1.png'
tf.keras.utils.plot_model(final_model, to_file=dot_img_file, show_shapes=True)

y_pred = new_model.predict(testgenerator,steps = STEP_SIZE_VALID+1)

#for getting classes
gen = ImageDataGenerator()
testing = gen.flow_from_directory( '/content/drive/My Drive/M.Tech NITK/Minor Project/MinorProject/DJD_MSR_IDG/subject_even',
                                          target_size = (112,112),
                                          batch_size = 64,
                                          color_mode = 'rgb',
                                          class_mode="categorical",
                                          shuffle = False)

def check_accuracy(y_pred,label):
  pred = y_pred.argmax(axis=1)
  res = 0
  for i,j in zip(pred,label):
    if i == j:
      res += 1
  return (res/len(label))

print(check_accuracy(y_pred,testing.classes))
np.save('y_pred_2_1',y_pred)

y_pred.shape

import pandas as pd
h_df = pd.DataFrame(history.history)
h_df.to_csv('history.csv')

y_pred1 = y_pred.argmax(axis=1)

matrix = tf.math.confusion_matrix(testing.classes, y_pred1)

labels = ['High-arm-wave',
          'Horizontal-arm-wave',
          'Hammer',
          'Hand-catch',
          'Forward-punch',
          'High-throw',
          'Draw-x',
          'Draw-tick',
          'Draw-circle',
          'Hand-clap',
          'Two-hand-wave',
          'Side-boxing',
          'Bend',
          'Forward-kick',
          'Side-kick',
          'Jogging',
          'Tennis-swing',
          'Tennis-serve',
          'Golf-swing',
          'Pick-up-and-throw']    
def plot(cm):
    matrix_norm = np.asarray(cm)
    matrix_norm = matrix_norm.astype('float') / matrix_norm.sum(axis = 1)
    plt.figure(figsize=(10,10))
    plt.imshow(matrix_norm,interpolation = 'nearest', cmap = plt.cm.Blues)
    plt.title("Confusion Matrix for MSRAction3D Dataset")
    plt.colorbar(fraction=0.0455)
    tick_marks = np.arange(len(labels))
    plt.xticks(tick_marks, labels, rotation = 90)
    plt.yticks(tick_marks, labels)

plot(matrix)

def get_data_DJD(DATADIR):
  CATEGORIES = ["a01", "a02", "a03", "a04", "a05", "a06", "a07", "a08", "a09", "a10", "a11", "a12", "a13", "a14", "a15", "a16", "a17", "a18", "a19", "a20"]

  for img in os.listdir(DATADIR):
    img_array = cv2.imread(os.path.join(DATADIR, img), cv2.IMREAD_COLOR)
    plt.imshow(img_array)
    plt.show()
    break

  IMG_SIZE = 112

  new_array = cv2.resize(img_array, (IMG_SIZE, IMG_SIZE))
  plt.imshow(new_array, cmap = 'gray')
  plt.show()
  # a01_s01_e01
  def action_label_img(img):
    action_label = img.split('_')[-4]
    if action_label[1] == '0':
      i = int(action_label[2:]) - 1
      label = [0] * 20
      label[i] = 1
      return label
    else:
      i = int(action_label[1:]) - 1
      label = [0] * 20
      label[i] = 1
      return label
  from random import shuffle
  from tqdm import tqdm

  training_data = []

  def create_training_data():
    for img in tqdm(sorted(os.listdir(DATADIR))):
      try:
        img_array = cv2.imread(os.path.join(DATADIR, img), cv2.IMREAD_COLOR)
        new_array = cv2.resize(img_array, (IMG_SIZE, IMG_SIZE))
        #category = img.split('_')[-4]
        #class_num = CATEGORIES.index(category)
        class_num = action_label_img(img)
        training_data.append([new_array, class_num])
      except Exception as e:
        pass
    # shuffle(training_data)
    # np.save('train_data.npy', training_data)
  create_training_data()
  X = []
  Y = []

  for features, label in training_data:
    X.append(features)
    Y.append(label)
  
  X = np.array(X).reshape(-1, IMG_SIZE, IMG_SIZE, 3)
  Y = np.array(Y)
  return X,Y

def get_data_DSM(DATADIR):
  CATEGORIES = ["a01", "a02", "a03", "a04", "a05", "a06", "a07", "a08", "a09", "a10", "a11", "a12", "a13", "a14", "a15", "a16", "a17", "a18", "a19", "a20"]

  for img in os.listdir(DATADIR):
    img_array = cv2.imread(os.path.join(DATADIR, img), cv2.IMREAD_GRAYSCALE)
    plt.imshow(img_array)
    plt.show()
    break

  IMG_SIZE = 112

  new_array = cv2.resize(img_array, (IMG_SIZE, IMG_SIZE))
  plt.imshow(new_array, cmap = 'gray')
  plt.show()
  # a01_s01_e01
  def action_label_img(img):
    action_label = img.split('_')[-4]
    if action_label[1] == '0':
      i = int(action_label[2:]) - 1
      label = [0] * 20
      label[i] = 1
      return label
    else:
      i = int(action_label[1:]) - 1
      label = [0] * 20
      label[i] = 1
      return label
  from random import shuffle
  from tqdm import tqdm

  training_data = []

  def create_training_data():
    for img in tqdm(sorted(os.listdir(DATADIR))):
      try:
        img_array = cv2.imread(os.path.join(DATADIR, img), cv2.IMREAD_GRAYSCALE)
        new_array = cv2.resize(img_array, (IMG_SIZE, IMG_SIZE))
        #category = img.split('_')[-4]
        #class_num = CATEGORIES.index(category)
        class_num = action_label_img(img)
        training_data.append([new_array, class_num])
      except Exception as e:
        pass
    # shuffle(training_data)
    # np.save('train_data.npy', training_data)
  create_training_data()
  X = []
  Y = []

  for features, label in training_data:
    X.append(features)
    Y.append(label)
  
  X = np.array(X).reshape(-1, IMG_SIZE, IMG_SIZE, 1)
  Y = np.array(Y)
  return X,Y

labels = ['High-arm-wave',
          'Horizontal-arm-wave',
          'Hammer',
          'Hand-catch',
          'Forward-punch',
          'High-throw',
          'Draw-x',
          'Draw-tick',
          'Draw-circle',
          'Hand-clap',
          'Two-hand-wave',
          'Side-boxing',
          'Bend',
          'Forward-kick',
          'Side-kick',
          'Jogging',
          'Tennis-swing',
          'Tennis-serve',
          'Golf-swing',
          'Pick-up-and-throw']    
def plot(cm):
  plt.imshow(cm,interpolation = 'nearest', cmap = plt.cm.Blues)
  plt.title("Confusion Matrix for MSRAction3D Dataset")
  plt.colorbar(fraction=0.0455)
  tick_marks = np.arange(len(labels))
  plt.xticks(tick_marks, labels, rotation = 90)
  plt.yticks(tick_marks, labels)

model_1 = tf.keras.models.load_model('/content/DSM.hdf5')

model_2 = tf.keras.models.load_model(latest_file)

model_3 = tf.keras.models.load_model('/content/DJD.hdf5')

X_DJD, Y_DJD = get_data_DJD('/content/drive/My Drive/M.Tech NITK/Minor Project/MinorProject/DJD_MSR/subject_even')

X_DSM, Y_DSM = get_data_DSM('/content/drive/My Drive/M.Tech NITK/Minor Project/MinorProject/DSM_MSR/subject_even')

y_pred_2 = model_2.predict(testgenerator,steps = 5)
# y_pred_2 = np.load('/content/y_pred_MFF.npy')

testing.class_indices

ans2 = check_accuracy(y_pred_2,testing.classes)
print(ans2)
np.save('y_pred_2_3',y_pred_2)

print(check_accuracy(y_pred_2,Y_DJD.argmax(axis = 1)))
np.save('y_pred_2_2',y_pred_2)

y_pred_1 = model_1.predict(X_DSM)
np.save('y_pred_1',y_pred_1)
ans1 = check_accuracy(y_pred_1,Y_DJD.argmax(axis = 1))
print(ans1)

y_pred_3 = model_3.predict(X_DJD)
np.save('y_pred_3',y_pred_3)
ans3 = check_accuracy(y_pred_3,Y_DSM.argmax(axis = 1))
print(ans2)

fusion_1 = y_pred_1 * y_pred_3
ans4 = check_accuracy(fusion_1,Y_DSM.argmax(axis = 1))
print(ans3)

fusion_2 = y_pred_1 * y_pred_2
ans5 = check_accuracy(fusion_2,Y_DSM.argmax(axis = 1))
print(ans4)

fusion_3 = y_pred_2
ans6 = check_accuracy(fusion_3,Y_DSM.argmax(axis = 1))
print(ans5)

fusion_4 = y_pred_1 * y_pred_2 * y_pred_3
ans7 = check_accuracy(fusion_4,Y_DSM.argmax(axis = 1))
print(ans6)

fusion_5 = fusion_3 * fusion_4
ans8 = check_accuracy(fusion_5,Y_DSM.argmax(axis = 1))
print(ans7)

ans = max(ans1,ans2,ans3,ans4,ans5,ans6,ans7,ans8)

if ans == ans1:
	action = y_pred_1
elif ans == ans2:
	action = y_pred_2
elif ans == ans3:
	action = y_pred_3
elif ans == ans4:
	action = fusion_1
elif ans == ans5:
	action = fusion_2
elif ans == ans6:
	action = fusion_3
elif ans == ans7:
	action = fusion_4
elif ans == ans8:
	action = fusion_5

np.save('action',action)
print(check_accuracy(action,Y_DSM.argmax(axis = 1)))

matrix = tf.math.confusion_matrix(Y_DSM.argmax(axis = 1), action.argmax(axis=1))

matrix_norm = np.asarray(matrix)
matrix_norm = matrix_norm.astype('float') / matrix_norm.sum(axis = 1)
plt.figure(figsize=(10,10))
plot(matrix_norm)
plt.savefig('cm.png')

np.save('confusion',matrix)