import numpy as np
from numpy import asarray, array
from numpy import savetxt, save, load
import pandas as pd
import pdb
import time
from sklearn.datasets import load_digits, fetch_openml
from sklearn.linear_model import Ridge
from sklearn.metrics import classification_report, confusion_matrix, plot_confusion_matrix
import matplotlib.pyplot as plt
from sklearn.model_selection import train_test_split, validation_curve, cross_val_score
from sklearn.utils import check_random_state
from IPython.core.debugger import set_trace
import warnings
warnings.filterwarnings('ignore')
%matplotlib inline
import h5py
from matplotlib.image import imread
from PIL import Image as im
from scipy import ndimage
from skimage import filters
import math
import tensorflow as tf
import matplotlib.pyplot as plt
from PIL import Image
##### open the file and extract to the test and train arrays #####
im_shape_org = (64, 64)
ds = []
with h5py.File('/content/drive/MyDrive/Colab Notebooks/Datasets/MNIST_synthetic.h5', 'r') as f:
for key in f.keys():
ds.append(f[key][()])
print(key) # using this command to see the keys in the dataset
ds_X_tst_org, ds_X_trn_org, ds_y_trn_org = ds[0], ds[1], ds[2]
ds_tst_smpl_nr = ds_X_tst_org.shape[0]
ds_trn_smpl_nr = ds_X_trn_org.shape[0]
print(ds_X_tst_org.shape, ds_X_trn_org.shape, ds_y_trn_org.shape) #check the shape of dataset
def image_split_object(ds_arr, digit_width = 12, display_samples=False):
##### initialize some variables #####
im_shape_obj = (14,14)
smpl_nr = ds_arr.shape[0]
cls_nr = 5
critical_ind = np.array([])
img_index = 0
dgt_width = digit_width
#######################################
##### Change all the elements to a zero array of shape im_shape #####
img_zero = np.zeros(im_shape_obj)
ds_new = np.full([smpl_nr, cls_nr], None)
for rw in range(smpl_nr):
for col in range(cls_nr):
ds_new[rw,col] = img_zero
#######################################################################
##### split each image to the objects and creat a new dataset #####
for img_index in range(smpl_nr): #range(smpl_nr):
## upload the image from dataset ##
img_curr = np. array(ds_arr[img_index])
img_curr = img_curr.reshape(im_shape_org)
## plot the current sample ##
if img_index == 0 and display_samples==True: # just the 1st image for test
print("digits width:", dgt_width)
plt.imshow(img_curr)
plt.title('The first image to split.', fontsize=8)
plt.show()
###############################
#####################################
## crop the image and remove zero margin ##
zero_cols = np.argwhere(np.all(img_curr[:, :] == 0, axis=0))
nzero_cols = np.argwhere(np.any(img_curr[:, :] != 0, axis=0))
zero_rows = np.argwhere(np.all(img_curr[:, :] == 0, axis=1))
nzero_rows = np.argwhere(np.any(img_curr[:, :] != 0, axis=1))
img_curr = img_curr[nzero_rows.min()-1:nzero_rows.max()+1, nzero_cols.min()-1:nzero_cols.min()-1+(cls_nr*dgt_width)]
## plot the current sample ##
if img_index == 0 and display_samples==True: # just the 1st image for test
plt.imshow(img_curr)
plt.title('The image after crop.', fontsize=8)
plt.show()
###############################
#############################################
## extract the image size and possible number of objects ##
frm_size = img_curr.shape[1]
dgt_nr = math.ceil(frm_size/dgt_width)
#############################################################
## insert the objects in the new dataset and remove redundant objects ##
obj_ind = 0
obj_last_ind = 5
for j in range(dgt_nr):
if j == 0:
start_col = 0
else:
start_col = (j*dgt_width)-2
end_col = start_col + dgt_width
tmp_im = img_curr[0:img_curr.shape[0], start_col:end_col]
## plot the current sample ##
if img_index == 0 and display_samples==True: # just the 1st image for test
plt.imshow(tmp_im)
plt.title('Digit after splitting', fontsize=8)
plt.show()
###############################
if np.count_nonzero(tmp_im) > 0.1*tmp_im.size: # remove redundant objects with mostly (0.1*tmp_im.size) zero pixels
ds_tmp = np.zeros(im_shape_obj)
ds_tmp[:tmp_im.shape[0],:tmp_im.shape[1]] = tmp_im #[:tmp_im.shape[0],:tmp_im.shape[1]]
ds_new[img_index, obj_ind] = ds_tmp
obj_ind += 1
##########################################################################
##### compare the num of found objects to y_trn #####
# ind_cls_10 = np.where(ds_y_trn_org[img_index]==10)
# y_trim = np.delete(ds_y_trn_org[img_index], ind_cls_10)
# if obj_ind != y_trim.shape[0]:
# np.append(critical_ind, img_index) # critical_ind.append(img_index)
# print("size diff: NOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOO")
#######################################################
return ds_new, critical_ind
def grid_show(images_array, rows=3, columns=5, first_ind = 0):
##### display the loaded file to check the new dataset #####
num_row = rows
num_col = columns
num_of_images = num_row*num_col # data.shape[0]
# plot images
fig, axes = plt.subplots(num_row, num_col, figsize=(10,10))
for i in range(num_of_images):
image = images_array[i+first_ind]
ax = axes[i//num_col, i%num_col]
ax.imshow(image, cmap='gray')
# ax.set_title('Label: {}'.format(ds_y_trn_fltn[i]))
plt.tight_layout()
plt.show()
##############################################################
##### split the objects #####
ds_X_tst_splt, X_tst_crit_ind = image_split_object(ds_X_tst_org, digit_width=12)
ds_X_tst_splt_fltn = np.ndarray.flatten(ds_X_tst_splt)
ds_X_trn_splt, X_trn_crit_ind = image_split_object(ds_X_trn_org, digit_width=12)
ds_X_trn_splt_fltn = np.ndarray.flatten(ds_X_trn_splt)
ds_y_trn_fltn = np.ndarray.flatten(ds_y_trn_org)
###############################
for i in range(3):
image_split_object(ds_X_trn_org[0:1], digit_width=11+i, display_samples=True)
##### Pad the images and change from 14*14 to 18*18 #####
if ds_X_tst_splt_fltn[0].shape == (14,14):
for p in range(ds_X_trn_splt_fltn.shape[0]):
ds_X_trn_splt_fltn[p] = np.pad(ds_X_trn_splt_fltn[p], (2,2), 'constant', constant_values = 0)
for p in range(ds_X_tst_splt_fltn.shape[0]):
ds_X_tst_splt_fltn[p] = np.pad(ds_X_tst_splt_fltn[p], (2,2), 'constant', constant_values = 0)
###########################################################
##### do the conversion to convert from (x,) to (x,y)!!! #####
ds_X_tst_splt_fltn = np.asarray(list(ds_X_tst_splt_fltn))
ds_X_trn_splt_fltn = np.asarray(list(ds_X_trn_splt_fltn))
################################################################
##### save the new datasets to files #####
from numpy import save, load
save('/content/drive/MyDrive/Colab Notebooks/Datasets/ds_X_tst_splt_fltn-v02.npy', ds_X_tst_splt_fltn)
save('/content/drive/MyDrive/Colab Notebooks/Datasets/ds_X_trn_splt_fltn-v02.npy', ds_X_trn_splt_fltn)
save('/content/drive/MyDrive/Colab Notebooks/Datasets/ds_y_trn_fltn-v02.npy', ds_y_trn_fltn)
############################################
##### load the .npy file #####
data_trn = load('/content/drive/MyDrive/Colab Notebooks/Datasets/ds_X_trn_splt_fltn-v02.npy', allow_pickle=True)
data_tst = load('/content/drive/MyDrive/Colab Notebooks/Datasets/ds_X_tst_splt_fltn-v02.npy', allow_pickle=True)
print("Training data shape:", data_trn.shape)
print("Test data shape:", data_tst.shape)
################################
##### display the loaded file to check the new dataset #####
grid_show(data_trn)
##############################################################
import numpy as np
from numpy import asarray, array
from numpy import savetxt, save, load
import pandas as pd
import tensorflow as tf
import pdb
import time
from sklearn.datasets import load_digits, fetch_openml
from sklearn.linear_model import Ridge
from sklearn.metrics import classification_report, confusion_matrix, plot_confusion_matrix
import matplotlib.pyplot as plt
from sklearn.model_selection import train_test_split, validation_curve, cross_val_score
from sklearn.naive_bayes import GaussianNB
from sklearn.neighbors import KNeighborsClassifier
from sklearn.tree import DecisionTreeClassifier
from sklearn.utils import check_random_state
from IPython.core.debugger import set_trace
import warnings
warnings.filterwarnings('ignore')
%matplotlib inline
import h5py
from matplotlib.image import imread
from PIL import Image
from scipy import ndimage
from skimage import filters
##### load the .npy file #####
X_trn_arr = load('/content/drive/MyDrive/Colab Notebooks/Datasets/ds_X_trn_splt_fltn-v02.npy', allow_pickle=True)
X_tst_arr = load('/content/drive/MyDrive/Colab Notebooks/Datasets/ds_X_tst_splt_fltn-v02.npy', allow_pickle=True)
################################
# smooth the image (to remove small objects)
def image_blur(im_inp, blur_radius, disp_image=False):
blr_rds = blur_radius
im_blr = ndimage.gaussian_filter(im_inp, blr_rds)
if disp_image == True:
plt.imshow(im_blr)
plt.show()
return im_blr
def image_mask(im_input, threshold):
if threshold == 1:
thrsh = filters.threshold_otsu(im_input)
elif threshold == 2:
thrsh = filters.threshold_isodata(im_input)
elif threshold == 3:
thrsh = filters.threshold_sauvola(im_input)
else:
thrsh = threshold
###
im_mask = im_input > threshold
###
return im_mask
ds = X_trn_arr[36:37]
ds_tmp = ds
ds_tmp_cln = ds_tmp # initialize the output dataset
dgt_size = (18,18)
im_mult_obj = np.zeros(ds_tmp.shape[0]) # initialize an array to store the indeces of multi object images.
for i in range(ds_tmp.shape[0]):
if np.any(ds_tmp[i] != 0): # empty images can not be blured
print(i)
plt.imshow(ds_tmp[i])
plt.title('The original image after splitting.', fontsize=8)
plt.show()
for j in [0.2, 0.6, 1]:
ds_tmp_blr = image_blur(ds_tmp[i], j)
plt.imshow(ds_tmp_blr)
plt.title("The image blur radiation:%1.1f"%j , fontsize=8)
plt.show()
ds_tmp_blr_msk = image_mask(ds_tmp_blr, 1)
plt.imshow(ds_tmp_blr_msk)
plt.title('The image masked', fontsize=8)
plt.show()
ds_num = 0
ds_cln = [0, 0]
for ds in [X_tst_arr, X_trn_arr]:
ds_tmp = ds
ds_tmp_cln = ds_tmp # initialize the output dataset
dgt_size = (18,18)
im_mult_obj = np.zeros(ds_tmp.shape[0]) # initialize an array to store the indeces of multi object images.
for i in range(ds_tmp.shape[0]):
if np.any(ds_tmp[i] != 0): # empty images can not be blured
ds_tmp_blr = image_blur(ds_tmp[i], 0.2)
ds_tmp_blr_msk = image_mask(ds_tmp_blr, 1)
labeling_structure = [[1, 1, 1],[1, 1, 1],[1, 1, 1]]
ds_tmp_lbld, nr_objects = ndimage.label(ds_tmp_blr_msk, structure=labeling_structure)
obj_slice = (ndimage.find_objects(ds_tmp_lbld))
if nr_objects>1:
for z in range(nr_objects):
obj_tmp = ds_tmp[i][obj_slice[z]]
obj_size_tmp = np.count_nonzero(obj_tmp)
if obj_size_tmp<10:
ds_tmp_cln[i][obj_slice[z]] = 0
ds_cln[ds_num] = ds_tmp_cln
ds_num +=1
# save('ds_X_trn_cln.npy', ds_tmp_cln)
save('/content/drive/MyDrive/Colab Notebooks/Datasets/ds_X_tst_cln-v02.npy', ds_cln[0])
save('/content/drive/MyDrive/Colab Notebooks/Datasets/ds_X_trn_cln-v02.npy', ds_cln[1])
X_trn_splt_org = load('/content/drive/MyDrive/Colab Notebooks/Datasets/ds_X_trn_splt_fltn-v02.npy')
X_tst_splt_org = load('/content/drive/MyDrive/Colab Notebooks/Datasets/ds_X_tst_splt_fltn-v02.npy')
X_tst_splt_cln = load('/content/drive/MyDrive/Colab Notebooks/Datasets/ds_X_tst_cln-v02.npy')
X_trn_splt_cln = load('/content/drive/MyDrive/Colab Notebooks/Datasets/ds_X_trn_cln-v02.npy')
y_trn_fltn = load('/content/drive/MyDrive/Colab Notebooks/Datasets/ds_y_trn_fltn-v02.npy')
##### display the loaded file to check the new dataset #####
plt.imshow(X_trn_splt_org[1])
plt.show()
plt.imshow(X_trn_splt_cln[1])
plt.show()
##############################################################
import tensorflow as tf
import numpy as np
from numpy import mean
from numpy import std
from numpy import load
from matplotlib import pyplot
from sklearn.model_selection import KFold, train_test_split
from keras.datasets import mnist
from keras.utils import to_categorical
from keras.models import Sequential
from keras.layers import Conv2D
from keras.layers import MaxPooling2D
from keras.layers import Dense
from keras.layers import Flatten
from keras.layers import Dropout
from keras.optimizers import Adam, SGD
from keras.constraints import maxnorm
from keras.layers import BatchNormalization
from keras.callbacks import ModelCheckpoint, ReduceLROnPlateau
from keras.preprocessing.image import ImageDataGenerator
from sklearn.model_selection import GridSearchCV
from keras.wrappers.scikit_learn import KerasClassifier
from keras.utils import plot_model
from google.colab import drive
drive.mount('/content/gdrive/', force_remount=True)
def load_our_dataset():
X_train = load('gdrive/MyDrive/ds_X_trn_cln.npy', allow_pickle=True)
X_train= X_train.reshape((X_train.shape[0], 14, 14, 1))
X_train = tf.keras.utils.normalize(X_train, axis=1)
X_test = load('gdrive/MyDrive/ds_X_tst_cln.npy', allow_pickle=True)
X_test= X_test.reshape((X_test.shape[0], 14, 14, 1))
X_test = tf.keras.utils.normalize(X_test, axis=1)
Y_train = load('gdrive/MyDrive/ds_y_trn_fltn.npy', allow_pickle=True)
Y_train = to_categorical(Y_train)
return X_train, Y_train, X_test
def build_cnn(lrx=0.001, momentumx=0.9, epochsx=22):
model = Sequential()
model.add(Conv2D(32, (3, 3), activation='relu', padding='same', kernel_initializer='lecun_uniform', input_shape=(14, 14, 1)))
model.add(BatchNormalization())
model.add(Conv2D(64, (3, 3), activation='relu', padding='same', kernel_initializer='lecun_uniform'))
model.add(MaxPooling2D((2, 2)))
model.add(Flatten())
model.add(Dropout(0.2))
model.add(Dense(120, activation='relu', kernel_initializer='lecun_uniform'))
model.add(Dropout(0.16))
model.add(Dense(55, activation='relu', kernel_initializer='lecun_uniform'))
model.add(Dense(11, activation='softmax'))
opt = Adam(lr=lrx)
model.compile(optimizer=opt ,loss='categorical_crossentropy', metrics=['accuracy'])
return model
def cross_validation(dataX, dataY, n_folds=5, epochsx=22, batch_sizex=32, lrx=0.01, momentumx=0.9):
scores, histories = list(), list()
kfold = KFold(n_folds, shuffle=True, random_state=1)
for train_index, test_index in kfold.split(dataX):
cnn = build_cnn(lrx, momentumx, epochsx)
X_train, Y_train, X_test, Y_test = dataX[train_index], dataY[train_index], dataX[test_index], dataY[test_index]
learning_rate_reduction = ReduceLROnPlateau(monitor='val_loss', patience=3, verbose=1, factor=0.5, min_lr=0.000001)
history = cnn.fit(X_train, Y_train , epochs = epochsx, validation_data= (X_test, Y_test),callbacks=[checkpoint, learning_rate_reduction], verbose=1)
_, acc = cnn.evaluate(X_test, Y_test, verbose=0)
print('> %.3f' % (acc * 100.0))
scores.append(acc)
histories.append(history)
return scores, histories
def cnn_performance(scores, histories):
pyplot.figure(figsize=(10,8))
for i in range(len(histories)):
pyplot.subplot(2, 1, 1)
pyplot.title('Cross Entropy Loss')
pyplot.plot(histories[i].history['loss'], color='blue', label='train')
pyplot.plot(histories[i].history['val_loss'], color='orange', label='test')
pyplot.subplot(2, 1, 2)
pyplot.title('Classification Accuracy')
pyplot.plot(histories[i].history['accuracy'], color='blue', label='train')
pyplot.plot(histories[i].history['val_accuracy'], color='orange', label='test')
pyplot.show()
print('Accuracy: mean=%.3f std=%.3f, n=%d' % (mean(scores)*100, std(scores)*100, len(scores)))
pyplot.boxplot(scores)
pyplot.show()
def run_test_cnn():
X_train, Y_train, X_test = load_our_dataset()
scores, histories = cross_validation(X_train, Y_train,lrx=0.001, epochsx=30, batch_sizex=60)
cnn_performance(scores, histories)
checkpoint = ModelCheckpoint('best_model_improved.h5', # model filename
monitor='val_loss', # quantity to monitor
verbose=0, # verbosity - 0 or 1
save_best_only= True, # The latest best model will not be overwritten
mode='auto')
run_test_cnn()
def cnn_predict():
_, _, X_test = load_our_dataset()
cnn = tf.keras.models.load_model('best_model_improved.h5')
predictions=np.argmax(cnn.predict([X_test]), axis=-1)
predictions= concat_output(predictions)
np.savetxt("final_predictions1.csv", predictions, delimiter=",", fmt='%s')
cnn_predict()
