A1_Part2.py

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

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/14bRzHBvLW8ZvhY-5SJMiQCQLmOD2ryft

# Assignment 1, Task 2

---
Authors: Chloe Tap, Evan Meltz, Giulia Rivetti (Group 36)
"""

# mount drive for file access
from google.colab import drive
drive.mount('/content/drive')


# import all relevant modules
import numpy as np
import tensorflow as tf
import keras
import pandas as pd
from keras.models import Sequential
from keras.layers import Conv2D, MaxPooling2D, Flatten, Dense, BatchNormalization, Dropout
from sklearn.model_selection import train_test_split
import matplotlib.pyplot as plt
from sklearn.utils import shuffle
from sklearn.preprocessing import StandardScaler
scaler = StandardScaler()
from __future__ import print_function
from keras import backend as K

# define path to image and label files needed (in Drive)
path_images = '/content/drive/MyDrive/IDL files/images_reduced.npy'
path_labels = '/content/drive/MyDrive/IDL files/labels_reduced.npy'

# load data into numpy arrays
labels = np.load(path_labels)
data = np.load(path_images)


#save data shape (x,x) pixels and max pixel size (to normalize)
data_shape = data.shape[1:3]

# see what a given image looks like
plt.imshow(data[100],cmap='gray')

# SHUFFLE DATA
# Create an array of indices and shuffle them
indices = np.arange(data.shape[0])
np.random.shuffle(indices)

# Use the shuffled indices to shuffle both data and labels
data = data[indices] # shuffled data
labels = labels[indices] # shuffled labels

# create a dataframe that stores labels for hours and mins from input data
# will be used later for changing labels in each model
label_df = pd.DataFrame(labels,columns=['hours','mins'])

"""## Classification model

First we need to create label categories for specified time intervals which can be altered as needed. We do this for 30 minutes intervals to begin with and then reduce the interval once a functioning model is found
"""

# Labels for classification model

def create_categories(interval_min,hours=np.arange(0,12), start_min=0, end_min=60):
  '''
  This function adds a new column to label_df which defines the classification label
  (depending on interval, 1/10/30 seconds) for better distinction of categories.
  This means we don't have to worry about defining specific category names,
  since this is a classification problem we just need a way to define class labels
  '''
  # split into sub-categories for given minute interval - interval_min
  mins = np.arange(start_min,end_min,interval_min)
  # keep track of class label, starting at index 0 and continues up to class (total_time/interval)-1
  c = 0
  for h in hours:
    for m in mins:
      # create a mask for the given hour and min range for each category
      mask = (label_df['hours']==h)&(label_df['mins'].between(m,m+30))
      # create new column 'class' to assign class label
      label_df.loc[mask,'class'] = c
      # increase class label for next iteration
      c += 1
  # extract class labels to dataframe for model
  labels = label_df['class'].values
  return labels.astype(int)

# define interval
interval = 1 #in minutes
# create categories
class_labels = create_categories(interval)

# CLASSIFICATION MODEL

'''Trains a simple classification CNN on the Clocks dataset.'''

batch_size = 128
# number of unique class labels = number of classification categories/classes
num_classes = len(label_df['class'].unique())
epochs = 5

# input image dimensions - from data_shape defined previously
img_rows, img_cols = data_shape

# the data, split between train and test sets
x_train, x_test, y_train, y_test = train_test_split(data,class_labels,train_size=0.8,test_size=0.2)

# reshape images
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)

# normalize pixels into range [0,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)

# model layers
model = Sequential()
model.add(Conv2D(32, kernel_size=(3,3), activation='relu',input_shape=input_shape))
model.add(MaxPooling2D(pool_size=(2, 2)))
model.add(BatchNormalization(synchronized=True))
model.add(Conv2D(64, (3, 3), activation='relu'))
model.add(MaxPooling2D(pool_size=(2, 2)))
model.add(BatchNormalization(synchronized=True))
model.add(Flatten())
model.add(Dense(128, activation='relu'))
model.add(Dense(64,activation='relu'))
model.add(Dense(num_classes, activation='softmax'))
model.summary()

# compile model
model.compile(loss=keras.losses.categorical_crossentropy,
              optimizer=keras.optimizers.Adam(),
              metrics=['accuracy'])

# fit the model
history = model.fit(x_train, y_train,
          batch_size=batch_size,
          epochs=epochs,
          verbose=1,
          validation_data=(x_test, y_test))

# evaluate how well it does on test data
score = model.evaluate(x_test, y_test, verbose=0)
# get test predictions

y_pred = model.predict(x_test)

#show results of test data from model predictions
print('Test loss:', score[0])
print('Test accuracy:', score[1])

# summarize history for CNN classification model accuracy

x= np.arange(1,epochs+1)
fig,ax=plt.subplots(figsize=(8,5))
ax.plot(x,np.array(history.history['accuracy'])*100)
ax.plot(x,np.array(history.history['val_accuracy'])*100)
ax.set_title('CNN Model Accuracy for %i minute interval classes' %(interval))
ax.grid(lw=0.2)
ax.set_ylabel('Accuracy (%)')
ax.set_xlabel('Epoch')
# # only used for all 720 classes
# ax.set_ylim(0,2)
ax.set_xticks(x)
#ax.set_yticks(np.arange(0,101,10))
ax.legend(['training', 'test'], loc='upper left')
plt.show()
# fig.savefig(path + 'classcnn_accuracy_' + str(interval) + 'min.pdf')

# summarize history for CNN classification model loss
fig1,ax1 = plt.subplots(figsize=(8,5))
ax1.plot(x,history.history['loss'])
ax1.plot(x,history.history['val_loss'])
ax1.set_title('CNN Model Loss for %i minute interval classes' %(interval))
ax1.set_ylabel('Loss')
ax1.set_xlabel('Epoch')
ax1.legend(['training', 'test'], loc='upper left')
ax1.grid(lw=0.2)
ax1.set_xticks(x)
plt.show()
# fig1.savefig(path + 'classcnn_test_loss_' + str(interval) + 'min.pdf')

"""## Regression Model

First we need to convert our labels into the float format hr.min, where min is in a fraction of 60

Compute this for all data labels!!
"""

# Labels regression model

# use label_df dataframe defined previously to convert hours and mins into float format

# first normalize mins (divide by 60)
label_df['min_norm'] = label_df['mins'].apply(lambda x: x/60)
# create hr.min by adding respective columns
label_df['time'] = label_df['hours'] + label_df['min_norm']

# save new labels into array for model input
time_labels = label_df['time'].values

# common sense accuracy function to input into regression model
def common_sense_accuracy(y_true, y_pred):
  abs_diff = tf.math.abs(y_true - y_pred)
  csa = abs_diff * 60
  return tf.reduce_mean(csa)

# Regression model

# Normalize the data
from sklearn.preprocessing import StandardScaler
scaler = StandardScaler()
# Normalized data
data = scaler.fit_transform(data.reshape(-1, data.shape[-1])).reshape(data.shape)

# Split data
x_train, x_test, y_train, y_test = train_test_split(data, time_labels, test_size=0.2, random_state=42)
x_valid, x_train = x_train[:2000], x_train[2000:]
y_valid, y_train = y_train[:2000], y_train[2000:]

num_epochs = 50
batch_size = 32

# Create a CNN regression model
model = Sequential()

model.add(Conv2D(32, (3, 3), activation='relu', input_shape=(75, 75, 1)))
model.add(MaxPooling2D((2, 2)))
model.add(BatchNormalization())

model.add(Conv2D(64, (3, 3), activation='relu'))
model.add(MaxPooling2D((2, 2)))
model.add(BatchNormalization())

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

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

model.add(Flatten())
model.add(Dense(128, activation='relu'))
model.add(Dense(1, activation='linear'))  # Output layer for regression


# Compile the model
model.compile(loss='mae', optimizer=tf.keras.optimizers.AdamW(learning_rate=0.001),
              metrics= common_sense_accuracy)
model.summary()

# Train the model
history = model.fit(x_train, y_train, batch_size=batch_size,
          epochs=num_epochs, validation_data=(x_test, y_test))


# evaluate the model
score_train = model.evaluate(x_train, y_train, verbose=0)
score_valid = model.evaluate(x_valid, y_valid, verbose=0)
score_test = model.evaluate(x_test, y_test, verbose=0)

print("MAE: ")
print("Train: ", score_train[0])
print("Valid: ", score_valid[0])
print("Test:  ", score_test[0])
print("CSA: ")
print("Train: ", score_train[1])
print("Valid: ", score_valid[1])
print("Test:  ", score_test[1])


# plot custom_mae during training
plt.title('Common sense accuracy')
plt.plot(history.history['common_sense_accuracy'], label='train')
plt.plot(history.history['val_common_sense_accuracy'], label='valid')
plt.legend()
plt.show()

"""## Multi-head Model

Here we create a model with multiple output ''heads'' for hours and minutes separately. We apply a classification head for hours (0-10) and a regression head for minutes. The labels used here are the original labels which are split using a function named extract_hours_mins
"""

# Multi-head model

NUM_EPOCHS = 50
BATCH_SIZE = 128

# Split data
x_train, x_test, y_train, y_test = train_test_split(data, labels, train_size=0.8, test_size=0.2, random_state=42)

# reshape images to have a single color channel and reduce to [0,1]
X_train = x_train.reshape((x_train.shape[0], 75, 75, 1))/255
X_test = x_test.reshape((x_test.shape[0], 75, 75, 1))/255

# extract labels for train and test - split into hours and minutes
def extract_hours_mins(data):
  # Split labels into hours and minutes
  hours_labels = data[:, 0]  # Extract hours
  minutes_labels = data[:, 1]  # Extract minutes
  # convert hours to required categorical format as in classification model
  hours_labels = keras.utils.to_categorical(hours_labels, num_classes=12)
  return hours_labels, minutes_labels

# create label arrays for hours and minutes of train and test data
# index 0 is hours, index 1 is mins
label_train = extract_hours_mins(y_train)
label_test = extract_hours_mins(y_test)

# Define the model architecture
image_input = keras.layers.Input(shape=(75, 75, 1), name='image_input')  # Assuming grayscale images

# CNN layers for multi head model
x = keras.layers.Conv2D(32, (3, 3), activation='relu')(image_input)
x = keras.layers.MaxPooling2D((2, 2))(x)
x = keras.layers.BatchNormalization()(x)

x = keras.layers.Conv2D(64, (3, 3), activation='relu')(x)
x = keras.layers.MaxPooling2D((2, 2))(x)
x = keras.layers.BatchNormalization()(x)

x = keras.layers.Flatten()(x)
x = keras.layers.Dense(128, activation='relu')(x)

# Output head for hours (multi-class classification)
hours_output = keras.layers.Dense(12, activation='softmax', name='hours_output')(x)  # Assuming 12 hours in a clock

# Output head for minutes (regression)
minutes_output = keras.layers.Dense(1, activation='linear', name='minutes_output')(x)

# Create a model with multiple outputs - hours and minutes
model = keras.models.Model(inputs=image_input, outputs=[hours_output, minutes_output])
model.summary()

# Define losses for the two output heads
losses = {
    'hours_output': 'categorical_crossentropy',  # Use an appropriate loss for multi-class classification
    'minutes_output': 'mean_squared_error'  # Use mean squared error for regression
}

# Compile the model
model.compile(optimizer='adam', loss=losses, metrics={'hours_output': 'accuracy', 'minutes_output': 'mae'})

# Train the model
history = model.fit(X_train, {'hours_output': label_train[0], 'minutes_output': label_train[1]}, epochs=NUM_EPOCHS, batch_size=BATCH_SIZE,validation_data=(X_test, {'hours_output': label_test[0], 'minutes_output': label_test[1]}))

# Make predictions
predictions = model.predict(X_test)

# Separate the predictions for hours and minutes
predicted_hours = predictions[0]
predicted_minutes = predictions[1]

print("\nPredicted hours:")
print(predicted_hours)

print("\nPredicted minutes:")
print(predicted_minutes)

def mh_common_sense_accuracy(pred_hrs, pred_mins, true_test):
  # Get the predicted class for each sample, extract hour class from index with highest value (0-11)
  predicted_hour = np.argmax(pred_hrs, axis=1)
  # ensure minutes output is a flat array (no nested arrays)
  flat_pred_mins = np.array([element for innerList in pred_mins for element in innerList])
  # find total time IN MINS for both predictions and true values
  total_pred = predicted_hour * 60 + flat_pred_mins
  # same for true time - from input
  total_true = true_test[:,0]*60 + true_test[:,1]
  # find the time difference
  diff = np.abs(total_true - total_pred)
  # ensure it is the smallest difference (in case time rolls over 11:59 and back to 0:00)
  csa = np.minimum(diff,720-diff)
  # take average from overall csa array
  avg_csa = np.mean(csa)
  return csa, avg_csa

# evaluate common sense accuracy
common_sense_mh, mean_csa = mh_common_sense_accuracy(np.array(predicted_hours),np.array(predicted_minutes),y_test)

# visualize the common sense accuracy across all test data (histogram distribution)
fig_hist,ax_hist = plt.subplots(figsize=(7,5))
ax_hist.hist(common_sense_mh,bins=100)
ax_hist.set_title('Common Sense Accuracy for Multi-head model')
ax_hist.set_xlabel('Common Sense Accuracy (mins)')
ax_hist.set_ylabel('Count')
ax_hist.grid(alpha=0.2)
ax_hist.set_xticks(np.arange(0,361,30))
#fig_hist.savefig('mh_csa_model.pdf')

print('Average common sense accuracy for multi-head model is', round(mean_csa,2), 'mins')

# summarize history for hour accuracy across epochs
import matplotlib.pyplot as plt
x= np.arange(1,NUM_EPOCHS+1,1)
fig2,ax2=plt.subplots(figsize=(8,5))
#convert accuracy to percentage for easy reading
ax2.plot(x,np.array(history.history['hours_output_accuracy'])*100)
ax2.plot(x,np.array(history.history['val_hours_output_accuracy'])*100)
ax2.set_title('Multi Head Model: Hours Classification Accuracy')
ax2.grid(lw=0.2)
ax2.set_ylabel('Accuracy (%)')
ax2.set_xlabel('Epoch')
ax2.set_yticks(np.arange(0,101,10))
ax2.legend(['training', 'test'], loc='upper left')
plt.show()

# summarize history for minutes mae across epochs
fig3,ax3 = plt.subplots(figsize=(8,5))
ax3.plot(x,history.history['minutes_output_mae'])
ax3.plot(x,history.history['val_minutes_output_mae'])
ax3.set_title('Multi Head Model: Minutes Regression MAE')
ax3.set_ylabel('Mean absolute error (MAE)')
ax3.set_xlabel('Epoch')
ax3.legend(['training', 'test'], loc='upper left')
ax3.grid(lw=0.2)
plt.show()