I’m working on Malaria detection with TensorFlow and I tried to solve the problem by looking at the documentation and other people who have similar problems and similar solutions. I actively use Git and you can look at the python script from the link below with .py extension, this is my first time with the error and learning computer vision all along, so I appreciate all the help and feedback (python script at malaria_detection.py)
You can find the code in from my GitHub link or
import tensorflow as tf
import numpy as np
import matplotlib.pyplot as plt
import tensorflow_datasets as tfds
from keras.models import Model
from keras.layers import Conv2D, MaxPool2D, Dense, Flatten, InputLayer, \
BatchNormalization, Input, Layer, Dropout, RandomFlip, RandomRotation,\
Resizing, Rescaling
from keras.optimizers import Adam
from keras.losses import BinaryCrossentropy
from keras.metrics import BinaryAccuracy, FalseNegatives, FalsePositives, \
TrueNegatives, TruePositives, Precision, Recall, AUC
from sklearn.metrics import confusion_matrix, roc_curve
import seaborn as sns
from keras.callbacks import Callback, CSVLogger, EarlyStopping, LearningRateScheduler, \
ModelCheckpoint, ReduceLROnPlateau
from keras.regularizers import L2
from keras import Sequential
import cv2
dataset, dataset_info = tfds.load('malaria',
with_info=True,
as_supervised=True,
shuffle_files=True,
split=['train'])
print(dataset)
print(dataset_info)
def splits(dataset, TRAIN_RATIO, VAL_RATIO, TEST_RATIO):
DATASET_SIZE = len(dataset)
train_dataset = dataset.take(int(TRAIN_RATIO * DATASET_SIZE))
val_test_dataset = dataset.skip(int(TRAIN_RATIO * DATASET_SIZE))
val_dataset = val_test_dataset.take(int(VAL_RATIO * DATASET_SIZE))
test_dataset = val_test_dataset.skip(int(VAL_RATIO * DATASET_SIZE))
return train_dataset, val_dataset, test_dataset
TRAIN_RATIO = 0.6
VAL_RATIO = 0.2
TEST_RATIO = 0.2
#dataset = tf.data.Dataset.range(10)
train_dataset, val_dataset, test_dataset = splits(dataset[0],
TRAIN_RATIO,
VAL_RATIO,
TEST_RATIO)
print(list(train_dataset.take(1).as_numpy_iterator()),
list(val_dataset.take(1).as_numpy_iterator()),
list(test_dataset.take(1).as_numpy_iterator()))
for data in dataset[0].take(4):
print(data)
for i, (image, label) in enumerate(train_dataset.take(16)):
ax = plt.subplot(4, 4, i + 1)
plt.imshow(image)
plt.title(dataset_info.features['label'].int2str(label))
plt.axis('off')
plt.show()
"""for i, (image, label) in enumerate(train_dataset.take(2)):
plt.subplot(1, 4, 2*i + 1)
plt.imshow(image)
plt.subplot(1, 4, 2 * i + 2)
plt.imshow(tf.image.adjust_saturation(image, 0.3))
plt.title(dataset_info.features['label'].int2str(label))
plt.axis('off')
plt.show()"""
print(dataset_info.features['label'].int2str(1))
def visualize(original, augmented):
plt.subplot(1, 2, 1)
plt.imshow(original)
plt.subplot(1, 2, 2)
plt.imshow(augmented)
original_image, label = next(iter(train_dataset))
augmented_image = tf.image.adjust_saturation(image, saturation_factor = 0.3)
visualize(original_image, augmented_image)
IM_SIZE = 224
def resize_rescale(image, label):
return tf.image.resize(image, (IM_SIZE, IM_SIZE))/255.0, label
#tf.keras.layer resizing and rescaling
resize_rescale_layers = Sequential([
Resizing(IM_SIZE, IM_SIZE),
Rescaling(1.0/255),
])
#tf.image augment
def augment(image, label):
image, label = resize_rescale(image, label)
image = tf.image.rot90(image)
#image = tf.image.adjust_saturation(image, saturation_factor = 0.3)
image = tf.image.flip_left_right(image)
return image, label
class RotNinety(Layer):
def __init__(self):
super().__init__()
def call(self, image):
return tf.image.rot90(image)
#tf.keras.layers augment
augment_layers = Sequential([
RotNinety(),
RandomFlip(mode='horizontal',),
])
def augment_layer(image, label):
return augment_layers(resize_rescale_layers(image), training = True), label
#test_dataset = test_dataset.map(resize_rescale_layers)
#print(test_dataset)
#for image, label in train_dataset.take(1):
#print(image, label)
BATCH_SIZE = 32
train_dataset = (
train_dataset.shuffle(buffer_size = 8, reshuffle_each_iteration = True)
#.map(augment_layer)
.batch(1).
prefetch(tf.data.AUTOTUNE)
)
val_dataset = (
val_dataset
.shuffle(buffer_size = 8,reshuffle_each_iteration = True)
#.map(resize_rescale_layers)
.batch(1)
.prefetch(tf.data.AUTOTUNE)
)
print(train_dataset)
print(val_dataset)
IM_SIZE = 224
dropout_rate = 0.3
regularization_rate = 0.01
lenet_model = Sequential([
InputLayer(input_shape=(None, None, 3)),
resize_rescale_layers,
augment_layers,
Conv2D(filters = 6,
kernel_size = 3,
strides=1,
padding='valid',
activation= 'relu',
), #kernel_regularizer = L2(regularization_rate)
BatchNormalization(),
MaxPool2D(pool_size=2, strides=2),
#Dropout(rate = dropout_rate),
Conv2D(filters = 16,
kernel_size = 3,
strides=1,
padding='valid',
activation= 'relu',
), #kernel_regularizer= L2(regularization_rate)
BatchNormalization(),
MaxPool2D(pool_size=2, strides=2),
Flatten(),
Dense(100, activation= "relu", kernel_regularizer= L2(regularization_rate)),
BatchNormalization(),
Dropout(rate = dropout_rate),
Dense(10, activation= "relu", kernel_regularizer= L2(regularization_rate)),
BatchNormalization(),
Dense(1, activation= "sigmoid", kernel_regularizer= L2(regularization_rate)),
])
print(lenet_model.summary())
class LossCallback(Callback):
def on_epoch_end(self, epoch, logs):
print("\n For Epoch Number {} the model has a loss of {} ".format(epoch+1, logs["loss"]))
def on_batch_end(self, batch, logs):
print("\n For Batch Number {} the model has a loss of {} ".format(batch+1, logs))
csv_callback = CSVLogger(
'logs.csv', separator=',', append=True
)
es_callback = EarlyStopping(
monitor='val_loss',
min_delta=0,
patience=2,
verbose=1,
mode='auto',
baseline=None,
restore_best_weights=False
)
def scheduler(epoch, lr):
if epoch <= 3:
return lr
else:
return lr * tf.math.exp(-0.1)
scheduler_callback = LearningRateScheduler(scheduler, verbose=1)
checkpoint_callback = ModelCheckpoint(
'checkpoints/',
monitor = 'val_loss',
verbose = 0,
save_best_only = False,
save_weights_only = True,
mode = 'auto',
save_freq='epoch',
options=None,
initial_value_threshold=None,
)
plateau_callback = ReduceLROnPlateau(
monitor='val_accuracy', factor=0.1, patience=5, verbose=1
)
metrics = [TruePositives(name='tp'),FalsePositives(name='fp'),
TrueNegatives(name='tn'), FalseNegatives(name='fn'),
BinaryAccuracy(name='accuracy'), Precision(name='precision'),
Recall(name='recall'), AUC(name='auc')]
lenet_model.compile(optimizer=Adam(learning_rate=0.01),
loss=BinaryCrossentropy(),
metrics=metrics,
run_eagerly=False)
history = lenet_model.fit(train_dataset,
validation_data=val_dataset,
epochs = 1,
verbose = 1,) #callbacks= [plateau_callback]
image = cv2.imread('cell1.png')
print(image.shape)
image = tf.expand_dims(image, axis = 0)
print(image.shape)
print(lenet_model.predict(image))
plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])
plt.title('Model loss')
plt.ylabel('loss')
plt.xlabel('epoch')
plt.legend(['train_loss', 'val_loss'])
plt.show()
print(lenet_model.predict(train_dataset.take(1)).shape)
test_dataset = test_dataset.batch(1)
print(lenet_model.evaluate(test_dataset))
def parasite_or_not(x):
if(x<0.5):
return str('P')
else:
return str('U')
labels = []
inp = []
for x, y in test_dataset.as_numpy_iterator():
labels.append(y)
inp.append(x)
labels = np.array([i[0] for i in labels])
predicted = lenet_model.predict(np.array(inp)[:,0,...])
threshold = 0.6265
cm = confusion_matrix(labels, predicted > threshold)
print(cm)
plt.figure(figsize=(8, 8))
sns.heatmap(cm, annot=True,)
plt.title('Confusion matrix - {}'.format(threshold))
plt.ylabel('Actual')
plt.xlabel('Predicted')
plt.show()
fp, tp, thresholds = roc_curve(labels, predicted)
print(len(fp), len(tp), len(thresholds))
fp, tp, thresholds = roc_curve(labels, predicted)
plt.plot(fp, tp)
plt.xlabel("False Positive rate")
plt.ylabel("True Positive rate")
plt.grid()
skip = 20
for i in range(0, len(thresholds), skip):
plt.text(fp[i], tp[i], thresholds[i])
plt.show()
print(parasite_or_not(lenet_model.predict(test_dataset.take(1))[0][0]))
for i, (image, label) in enumerate(test_dataset.take(9)):
ax = plt.subplot(3, 3, i + 1)
plt.imshow(image[0])
plt.title(str(parasite_or_not(label.numpy()[0])) + ":" +
str(parasite_or_not(lenet_model.predict(image)[0][0])))
plt.axis('off')
plt.show()
#Functional API
func_input = Input(shape=(IM_SIZE, IM_SIZE, 3), name="Input Image")
x = Conv2D(filters = 6,
kernel_size = 3,
strides=1,
padding='valid',
activation= 'relu')(func_input)
x = BatchNormalization()(x)
x = MaxPool2D(pool_size=2, strides=2)(x)
x = Conv2D(filters = 16,
kernel_size = 3,
strides=1,
padding='valid',
activation= 'relu')(x)
x = BatchNormalization()(x)
output = MaxPool2D(pool_size=2, strides=2)(x)
x = Flatten()(x)
x = Dense(100, activation= "relu")(x)
x = BatchNormalization()(x)
x = Dense(10, activation= "relu")(x)
x = BatchNormalization()(x)
feature_extractor_seq_model = tf.keras.Sequential([
InputLayer(input_shape=(IM_SIZE, IM_SIZE, 3)),
Conv2D(filters=6,
kernel_size=3,
strides=1,
padding='valid',
activation='relu'),
BatchNormalization(),
MaxPool2D(pool_size=2, strides=2),
Conv2D(filters=16,
kernel_size=3,
strides=1,
padding='valid',
activation='relu'),
BatchNormalization(),
MaxPool2D(pool_size=2, strides=2),
])
print(feature_extractor_seq_model.summary())
func_output = Dense(1, activation= "sigmoid")(x)
lenet_model = Model(func_input, func_output, name = "Lenet Model")
print(lenet_model.summary())
#Model Subclassing
class FeatureExtractor(Layer):
def __init__(self, filters, kernel_size, strides, padding, activation,
pool_size):
super(FeatureExtractor, self).__init__()
self.conv_1 = Conv2D(filters = filters,
kernel_size = kernel_size,
strides = strides,
padding = padding,
activation = activation)
self.batch_1 = BatchNormalization()
self.pool_1 = MaxPool2D(pool_size=pool_size, strides=2*strides)
self.conv_2 = Conv2D(filters = filters*2,
kernel_size = kernel_size,
strides = strides,
padding = padding,
activation = activation)
self.batch_2 = BatchNormalization()
self.pool_2 = MaxPool2D(pool_size=pool_size, strides=2*strides)
def call(self, x, training):
x = self.conv_1(x)
x = self.batch_1(x)
x = self.pool_1(x)
x = self.conv_2(x)
x = self.batch_2(x)
x = self.pool_2(x)
return x
feature_sub_classed = FeatureExtractor(8, 3, 1, "valid", "relu", 2)
func_input = Input(shape=(IM_SIZE, IM_SIZE, 3), name="Input Image")
x = feature_sub_classed(func_input)
x = Flatten()(x)
x = Dense(100, activation="relu")(x)
x = BatchNormalization()(x)
x = Dense(10, activation="relu")(x)
x = BatchNormalization()(x)
func_output = Dense(1, activation="sigmoid")(x)
lenet_model_func = Model(func_input, func_output, name="Lenet_Model")
print(lenet_model_func.summary())
class LenetModel(Model):
def __init__(self):
super(LenetModel, self).__init__()
self.feature_extractor = FeatureExtractor(8, 3, 1, "valid", "relu", 2)
self.flatten = Flatten()
self.dense_1 = Dense(100, activation="relu")
self.batch_1 = BatchNormalization()
self.dense_2 = Dense(10, activation="relu")
self.batch_2 = BatchNormalization()
self.dense_3 = Dense(1, activation="sigmoid")
def call(self, x, training):
x = self.feature_extractor(x)
x = self.flatten(x)
x = self.dense_1(x)
x = self.batch_1(x)
x = self.dense_2(x)
x = self.batch_2(x)
x = self.dense_3(x)
return x
lenet_sub_classed = LenetModel()
lenet_sub_classed(tf.zeros([1, 224, 224, 3]))
print(lenet_sub_classed.summary())
class NeuralearnDense(Layer):
def __init__(self, output_units, activation):
super(NeuralearnDense, self).__init__()
self.output_units = output_units
self.activation = activation
def build(self, input_features_shape):
self.w = self.add_weight(shape=(input_features_shape[-1], self.output_units), initializer="random_normal",
trainable=True)
self.b = self.add_weight(shape=(self.output_units,), initializer="random_normal", trainable=True)
def call(self, input_features):
pre_output = tf.matmul(input_features, self.w) + self.b
if (self.activation == "relu"):
return tf.nn.relu(pre_output)
elif (self.activation == "sigmoid"):
return tf.math.sigmoid(pre_output)
else:
return pre_output
IM_SIZE = 224
lenet_custom_model = tf.keras.Sequential([
InputLayer(input_shape=(IM_SIZE, IM_SIZE, 3)),
Conv2D(filters = 6,
kernel_size = 3,
strides=1,
padding='valid',
activation= 'relu'),
BatchNormalization(),
MaxPool2D(pool_size=2, strides=2),
Conv2D(filters = 16,
kernel_size = 3,
strides=1,
padding='valid',
activation= 'relu'),
BatchNormalization(),
MaxPool2D(pool_size=2, strides=2),
Flatten(),
NeuralearnDense(100, activation= "relu"),
BatchNormalization(),
NeuralearnDense(10, activation= "relu"),
BatchNormalization(),
NeuralearnDense(1, activation= "sigmoid"),
])
print(lenet_custom_model.summary())
lenet_custom_model.compile(optimizer=Adam(learning_rate=0.01),
loss=BinaryCrossentropy(),
metrics='accuracy')
history = lenet_custom_model.fit(train_dataset,
validation_data=val_dataset,
epochs=3,
verbose=1)
Output:
Traceback (most recent call last):
File “/Users/atatekeli/PycharmProjects/comp-vision-projects/tensorflow-comp-vision/Malaria Detection/malaria_detection.py”, line 278, in
else:
ValueError: setting an array element with a sequence. The requested array has an inhomogeneous shape after 2 dimensions. The detected shape was (5513, 1) + inhomogeneous part.
https://github.com/Killpit/comp-vision-projects/tree/main/tensorflow-comp-vision/Malaria%20Detection