image_classification.py

# -*- coding: utf-8 -*-
"""CI-HW3.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/13xB1OxpOmgYbgzw_ZV2QRYs0bYm5CDWO
"""

import tensorflow as tf 
from tensorflow import keras
from tensorflow.keras import layers, activations
from tensorflow.keras.preprocessing.image import ImageDataGenerator


from tensorflow.keras.layers import Input, Conv2D, MaxPooling2D, ZeroPadding2D,\
     Flatten, BatchNormalization, AveragePooling2D, Dense, Activation, Add , Dropout
from tensorflow.keras.models import Sequential
from tensorflow.keras.models import Model
from tensorflow.keras import activations
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.callbacks import EarlyStopping
from tensorflow.keras.regularizers import l2
from tensorflow.keras.utils import get_file

import pathlib
import matplotlib.pyplot as plt

data_directory = '/root/.keras/datasets/Images'
dataset_url = "http://vision.stanford.edu/aditya86/ImageNetDogs/images.tar"
data_directory = get_file('Images', origin=dataset_url, untar = True)
data_directory = pathlib.Path(data_directory)
print(data_directory)

img_height = 192
img_width = 192
batch_size = 32

# img_height = 224
# img_width = 224
# batch_size = 64

train_ds = tf.keras.preprocessing.image_dataset_from_directory(
  data_directory,
  validation_split=0.2,
  subset="training",
  shuffle = True,
  seed=123,
  image_size=(img_height, img_width),
  batch_size=batch_size)

val_ds = tf.keras.preprocessing.image_dataset_from_directory(
  data_directory,
  validation_split=0.2,
  subset="validation",
  shuffle = False,
  seed=123,
  image_size=(img_height, img_width),
  batch_size=batch_size)


cls_number = len(train_ds.class_names)

AUTOTUNE = tf.data.experimental.AUTOTUNE

train_ds = train_ds.cache().shuffle(1000).prefetch(buffer_size=AUTOTUNE)
val_ds = val_ds.cache().prefetch(buffer_size=AUTOTUNE)

def first_model():
  
  model = Sequential([
  layers.experimental.preprocessing.Rescaling(1./255, input_shape=(img_height, img_width, 3)),

  layers.Conv2D(16, 3, padding='same', activation='relu'),
  layers.MaxPooling2D(),
  layers.Conv2D(32, 3, padding='same', activation='relu'),
  layers.MaxPooling2D(),
  layers.Conv2D(64, 3, padding='same', activation='relu'),
  layers.MaxPooling2D(),
  layers.Flatten(),
  layers.Dense(128, activation='relu'),
  layers.Dense(cls_number)
  ])
  return model

def res_conv(x, s, filters):
  '''
  here the input size changes''' 
  x_skip = x
  f1, f2 = filters

  # first block
  x = Conv2D(f1, kernel_size=(1, 1), strides=(s, s), padding='valid', kernel_regularizer=l2(0.001))(x)
  # when s = 2 then it is like downsizing the feature map
  x = BatchNormalization()(x)
  x = Activation(activations.relu)(x)

  # second block
  x = Conv2D(f1, kernel_size=(3, 3), strides=(1, 1), padding='same', kernel_regularizer=l2(0.001))(x)
  x = BatchNormalization()(x)
  x = Activation(activations.relu)(x)

  #third block
  x = Conv2D(f2, kernel_size=(1, 1), strides=(1, 1), padding='valid', kernel_regularizer=l2(0.001))(x)
  x = BatchNormalization()(x)

  # shortcut 
  x_skip = Conv2D(f2, kernel_size=(1, 1), strides=(s, s), padding='valid', kernel_regularizer=l2(0.001))(x_skip)
  x_skip = BatchNormalization()(x_skip)

  # add 
  x = Add()([x, x_skip])
  x = Activation(activations.relu)(x)

  return x

def res_identity(x, filters): 
  #renet block where dimension doesnot change.
  #The skip connection is just simple identity conncection
  #we will have 3 blocks and then input will be added

  x_skip = x # this will be used for addition with the residual block 
  f1, f2 = filters

  #first block 
  x = Conv2D(f1, kernel_size=(1, 1), strides=(1, 1), padding='valid', kernel_regularizer=l2(0.001))(x)
  x = BatchNormalization()(x)
  x = Activation(activations.relu)(x)

  #second block # bottleneck (but size kept same with padding)
  x = Conv2D(f1, kernel_size=(3, 3), strides=(1, 1), padding='same', kernel_regularizer=l2(0.001))(x)
  x = BatchNormalization()(x)
  x = Activation(activations.relu)(x)

  # third block activation used after adding the input
  x = Conv2D(f2, kernel_size=(1, 1), strides=(1, 1), padding='valid', kernel_regularizer=l2(0.001))(x)
  x = BatchNormalization()(x)
  # x = Activation(activations.relu)(x)

  # add the input 
  x = Add()([x, x_skip])
  x = Activation(activations.relu)(x)

  return x

def resnet50():

  input_im = Input(shape=(img_height, img_width, 3)) # cifar 10 images size

  x = layers.experimental.preprocessing.Rescaling(1./255)(input_im)
  x = ZeroPadding2D(padding=(3, 3))(x)

  # 1st stage
  # here we perform maxpooling, see the figure above

  x = Conv2D(64, kernel_size=(7, 7), strides=(2, 2))(x)
  x = BatchNormalization()(x)
  x = Activation(activations.relu)(x)
  x = MaxPooling2D((3, 3), strides=(2, 2))(x)

  #2nd stage 
  # frm here on only conv block and identity block, no pooling

  x = res_conv(x, s=1, filters=(64, 256))
  x = res_identity(x, filters=(64, 256))
  x = res_identity(x, filters=(64, 256))

  # 3rd stage

  x = res_conv(x, s=2, filters=(128, 512))
  x = res_identity(x, filters=(128, 512))
  x = res_identity(x, filters=(128, 512))
  x = res_identity(x, filters=(128, 512))

  # 4th stage

  x = res_conv(x, s=2, filters=(256, 1024))
  x = res_identity(x, filters=(256, 1024))
  x = res_identity(x, filters=(256, 1024))
  x = res_identity(x, filters=(256, 1024))
  x = res_identity(x, filters=(256, 1024))
  x = res_identity(x, filters=(256, 1024))

  # 5th stage

  x = res_conv(x, s=2, filters=(512, 2048))
  x = res_identity(x, filters=(512, 2048))
  x = res_identity(x, filters=(512, 2048))

  # ends with average pooling and dense connection

  x = AveragePooling2D((2, 2), padding='same')(x)

  x = Flatten()(x)
  x = Dense(120, activation='softmax', kernel_initializer='he_normal')(x) #multi-class

  # define the model 

  model = Model(inputs=input_im, outputs=x, name='Resnet50')

  return model

def second_model():
  
  model = Sequential([
  layers.experimental.preprocessing.Rescaling(1./255, input_shape=(img_height, img_width, 3)),

  layers.experimental.preprocessing.RandomFlip("horizontal_and_vertical"),
  # layers.experimental.preprocessing.RandomRotation(0.2),
  # layers.experimental.preprocessing.RandomZoom(.5, .2),

  layers.Conv2D(16, 3, padding='same', activation='relu'),
  layers.MaxPooling2D(),
  layers.Conv2D(32, 3, padding='same', activation='relu'),
  layers.MaxPooling2D(),
  layers.Conv2D(64, 3, padding='same', activation='relu'),
  layers.MaxPooling2D(),
  layers.Flatten(),
  layers.Dense(128, activation='relu'),
  layers.Dense(cls_number)
  ])
  return model

def mobile_net():
  input_im = (img_height, img_width, 3) # cifar 10 images size
  
  base_model = tf.keras.applications.MobileNetV2(input_shape= input_im, include_top=False, weights='imagenet')
  base_model.trainable = False

  model = Sequential([
    layers.experimental.preprocessing.Rescaling(1./255, input_shape = input_im),
    # layers.experimental.preprocessing.RandomFlip('horizontal'),
    # layers.experimental.preprocessing.RandomRotation(0.1),
    base_model,
    tf.keras.layers.GlobalAveragePooling2D(),
    tf.keras.layers.Dense(120, activation='softmax')
  ])

  return model

def xception():
  input_im = (img_height, img_width, 3) # cifar 10 images size
  
  base_model = tf.keras.applications.Xception(input_shape= input_im, include_top=False, weights='imagenet')
  base_model.trainable = False

  model = Sequential([
    layers.experimental.preprocessing.Rescaling(1./255, input_shape = input_im),
    # layers.experimental.preprocessing.RandomFlip('horizontal'),
    # layers.experimental.preprocessing.RandomRotation(0.1),
    base_model,
    tf.keras.layers.GlobalAveragePooling2D(),
    tf.keras.layers.Dense(120, activation='softmax')
  ])

  return model

def vgg():
  input_im = (img_height, img_width, 3) # cifar 10 images size
  
  base_model = tf.keras.applications.VGG19(input_shape= input_im, include_top=False, weights='imagenet')
  base_model.trainable = False

  model = Sequential([
    layers.experimental.preprocessing.Rescaling(1./255, input_shape = input_im),
    # layers.experimental.preprocessing.RandomFlip('horizontal'),
    # layers.experimental.preprocessing.RandomRotation(0.1),
    base_model,
    tf.keras.layers.GlobalAveragePooling2D(),
    tf.keras.layers.Dense(120, activation='softmax')
  ])

  return model

def alexnet():
    model = keras.models.Sequential([
    keras.layers.Conv2D(filters=96, kernel_size=(11,11), strides=(4,4), activation='relu', input_shape=(img_height, img_width, 3)),
    keras.layers.BatchNormalization(),
    keras.layers.MaxPool2D(pool_size=(2,2), strides=(2,2)),
    keras.layers.Conv2D(filters=256, kernel_size=(5,5), strides=(1,1), activation='relu', padding="same"),
    keras.layers.BatchNormalization(),
    keras.layers.MaxPool2D(pool_size=(2,2), strides=(2,2)),
    keras.layers.Conv2D(filters=384, kernel_size=(3,3), strides=(1,1), activation='relu', padding="same"),
    keras.layers.BatchNormalization(),
    keras.layers.Conv2D(filters=384, kernel_size=(1,1), strides=(1,1), activation='relu', padding="same"),
    keras.layers.BatchNormalization(),
    keras.layers.Conv2D(filters=256, kernel_size=(1,1), strides=(1,1), activation='relu', padding="same"),
    keras.layers.BatchNormalization(),
    keras.layers.MaxPool2D(pool_size=(2,2), strides=(2,2)),
    keras.layers.Flatten(),
    keras.layers.Dense(4096, activation='relu'),
    keras.layers.Dropout(0.5),
    keras.layers.Dense(4096, activation='relu'),
    keras.layers.Dropout(0.5),
    keras.layers.Dense(cls_number, activation='softmax')
    ])
    return model

def result(model, epoche):
  #Report 
  model.summary()
  model.compile(optimizer='adam',
            loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
            metrics=['accuracy'])
  
  # tf.keras.optimizers.Adam(
  #     learning_rate=10, beta_1=0.9, beta_2=0.999, epsilon=1e-07, amsgrad=False,
  #     name='Adam'
  # )

  #Train
  epochs=epoche
  history = model.fit(
    train_ds,
    validation_data=val_ds,
    epochs=epochs
  )

  # Graph
  acc = history.history['accuracy']
  val_acc = history.history['val_accuracy']

  loss = history.history['loss']
  val_loss = history.history['val_loss']

  epochs_range = range(epochs)

  plt.figure(figsize=(8, 8))
  plt.subplot(1, 2, 1)
  plt.plot(epochs_range, acc, label='Training Accuracy')
  plt.plot(epochs_range, val_acc, label='Validation Accuracy')
  plt.legend(loc='lower right')
  plt.title('Training and Validation Accuracy')

  plt.subplot(1, 2, 2)
  plt.plot(epochs_range, loss, label='Training Loss')
  plt.plot(epochs_range, val_loss, label='Validation Loss')
  plt.legend(loc='upper right')
  plt.title('Training and Validation Loss')
  plt.show()

#Usbaility = High
model = first_model()
result(model, 50)

#Usbaility = High
model = second_model()
result(model, 50)
# model.save_weights("weights.h5")

#Usbaility = Low
model = resnet50()

#Transfer Learning
WEIGHTS_PATH = ('https://github.com/fchollet/deep-learning-models/'
                'releases/download/v0.2/'
                'resnet50_weights_tf_dim_ordering_tf_kernels.h5')
# WEIGHTS_PATH_NO_TOP = ('https://github.com/fchollet/deep-learning-models/'
#                        'releases/download/v0.2/'
#                        'resnet50_weights_tf_dim_ordering_tf_kernels_notop.h5')

weight_directory = get_file(fname='Weights', origin=WEIGHTS_PATH)
model.load_weights(weight_directory, by_name=True)
result(model, 10)

#Usbaility = Very Low
model = alexnet()
result(model, 10)

model = mobile_net()
result(model, 10)

model = xception()
result(model, 10)

model = vgg()
result(model, 10)

# aug = ImageDataGenerator(
# 	# rotation_range=20,
# 	# zoom_range=0.15,
# 	# width_shift_range=0.2,
# 	# height_shift_range=0.2,
# 	shear_range=0.15,
# 	# horizontal_flip=True,
#   # validation_split=0.2,
# 	# fill_mode="nearest")
# )
# train_generator = aug.flow_from_directory(data_directory,
#   subset="training",
#   shuffle = True,
#   seed=123,
#   target_size=(img_height, img_width),
#   batch_size=64)

# print(train_generator.image_shape)
# print(train_generator)

# validation_generator = aug.flow_from_directory(data_directory,
#   subset="validation",
#   shuffle = False,
#   seed=123,
#   target_size=(img_height, img_width),
#   batch_size=batch_size)

# print(validation_generator.image_shape)
# print(validation_generator)
 
# # model.compile(optimizer='adam',
# #             loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
# #             metrics=['accuracy'])
# model.compile(loss='categorical_crossentropy',optimizer='adam',metrics=['accuracy'])

# e = 10
# model.fit(
#     train_generator,
#     # steps_per_epoch = train_generator.samples // batch_size,
#     validation_data = validation_generator, 
#     # validation_steps = validation_generator.samples // batch_size,
#     epochs = 10)