model_djd_utd.py

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

Automatically generated by Colaboratory.

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

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

def get_data(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] * 27
      label[i] = 1
      return label
    else:
      i = int(action_label[1:]) - 1
      label = [0] * 27
      label[i] = 1
      return label
  from random import shuffle
  from tqdm import tqdm

  training_data = []

  def create_training_data():
    for img in tqdm(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))
        class_num = action_label_img(img)
        training_data.append([new_array, class_num])
      except Exception as e:
        pass
    shuffle(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

X_train,y_train = get_data("/content/drive/My Drive/M.Tech NITK/Minor Project/MinorProject/MJD_UTD/subject_odd")

print(X_train.shape,y_train.shape)

X_test,y_test = get_data("/content/drive/My Drive/M.Tech NITK/Minor Project/MinorProject/MJD_UTD/subject_even")

print(X_test.shape,y_test.shape)

import keras
from keras.callbacks import ModelCheckpoint, EarlyStopping, LearningRateScheduler
from keras.regularizers import l1, l2, l1_l2
import tensorflow as tf
from tensorflow.keras.models import Sequential
from sklearn.model_selection import KFold
from tensorflow.keras.layers import Dense, Dropout, Activation, Flatten, Conv2D, MaxPooling2D
from sklearn.model_selection import train_test_split

model = Sequential()

model.add(Conv2D(32, (7, 7), input_shape=X_train.shape[1:], activation = 'relu',kernel_regularizer = l1(0.001)))
model.add(MaxPooling2D(pool_size=(2, 2)))

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

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

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

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

model.add(Flatten())

model.add(Dropout(0.3))
model.add(Dense(128))
model.add(Activation('relu'))

model.add(Dropout(0.2))
model.add(Dense(27))
model.add(Activation('softmax'))

opt = keras.optimizers.Adam(learning_rate=0.0001)
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 scheduler(epoch, lr):
  if lr % 500 != 0 and lr <= 0.0001:
    return lr 
  else:
    return lr * 0.01

lr_scheduler = LearningRateScheduler(scheduler)

# 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 = model.fit(X_train, y_train, epochs=2000, batch_size=64, validation_data=(X_test, y_test), verbose=0,callbacks=[checkpoint])

# 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 = model.evaluate(X_test,y_test)
print("loss: ",loss)
print("Tesing Accuracy: ", accuracy)

plt.plot(history.history['accuracy'])
plt.plot(history.history['val_accuracy'])
plt.title('Model Accuracy for UTD MHAD Dataset')
plt.ylabel('Acuuracy')
plt.xlabel('Epoch')
plt.legend(['train','test'], loc='upper left')
plt.show()

plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])
plt.title('Model Loss for UTD MHAD Dataset')
plt.ylabel('Loss')
plt.xlabel('Epoch')
plt.legend(['train','test'], loc='upper right')
plt.show()

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

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

y_pred = model.predict(X_test)
matrix = tf.math.confusion_matrix(y_test.argmax(axis=1), y_pred.argmax(axis=1))

labels = ['Right arm swipe to the left', 
         'Right arm swipe to the right', 
         'Right hand wave', 
         'Two hand front clap', 
         'Right arm throw', 
         'Cross arms in the chest', 
         'Basketball shoot', 
         'Right hand draw x', 
         'Right hand draw circle (clockwise)', 
         'Right hand draw circle (counter clockwise)', 
         'Draw triangle', 
         'Bowling (right hand)', 
         'Front boxing', 
         'Baseball swing from right', 
         'Tennis right hand forehand swing', 
         'Arm curl (two arms)', 
         'Tennis serve', 
         'Two hand push', 
         'Right hand knock on door', 
         'Right hand catch an object',
         'Right hand pick up and throw', 
         'Jogging in place', 
         'Walking in place', 
         'Sit to stand', 
         'Stand to sit', 
         'Forward lunge (left foot forward)',
         'Squat (two arms stretch out)']

def plot(cm):
  plt.imshow(cm,interpolation = 'nearest', cmap = plt.cm.Blues)
  plt.title("Confusion Matrix for UTD MHAD Dataset")
  plt.colorbar(fraction=0.0455)
  tick_marks = np.arange(len(labels))
  plt.xticks(tick_marks, labels, rotation = 90)
  plt.yticks(tick_marks, labels)

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