<jupyter_start><jupyter_text><jupyter_code>import os
import shutil
import cv2
import random
import numpy as np
import matplotlib.pyplot as plt
import tensorflow as tf
import tensorflow.keras.layers as layers
from tensorflow.keras.preprocessing.image import ImageDataGenerator
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.callbacks import ModelCheckpoint
from sklearn.metrics import confusion_matrix
from tensorflow.python.ops.numpy_ops import np_config
from tqdm import tqdm<jupyter_output><empty_output><jupyter_text>Data loading and preprocessing<jupyter_code>input_data_path='/kaggle/input/cell-images-for-detecting-malaria/cell_images'
labels=os.listdir(input_data_path)
print(labels)
labels.pop()
print(labels)
def get_image_path():
    image_path_list,name_list=[],[]
    for label in tqdm(labels):
        folder_path=os.path.join(input_data_path,label)
        image_set=random.sample(os.listdir(folder_path),1)
        for file in image_set:
            file_path=os.path.join(folder_path,file)
            image_path_list.append(file_path)
            name_list.append(label)
    return image_path_list,name_list
image_path_list,name_list=get_image_path()
def plot_images(image_path_list,name_list,row,col):
    fig=plt.figure(figsize=(16,16))
    for i in range(len(name_list)):
        fig.add_subplot(row,col,i+1)
        plt.title(name_list[i])
        plt.axis('off')
        plt.imshow(cv2.imread(image_path_list[i]))
    plt.tight_layout()
    plt.show()
plot_images(
    image_path_list=image_path_list,
    name_list=name_list,
    row=1,col=2
)
data_dir='/kaggle/data'
train_path=os.path.join(data_dir,'train')
valid_path=os.path.join(data_dir,'valid')
test_path=os.path.join(data_dir,'test')
def create_dir():
    if not os.path.isdir(data_dir):
        os.mkdir(data_dir)
        os.mkdir(train_path)
        os.mkdir(valid_path)
        os.mkdir(test_path)
        for label in tqdm(labels):
            os.mkdir(os.path.join(train_path,label))
            os.mkdir(os.path.join(valid_path,label))
            os.mkdir(os.path.join(test_path,label))
create_dir()
def check_dir():
    print(f'{data_dir}: {os.path.isdir(data_dir)}')
    print(f'{train_path}: {os.path.isdir(train_path)}')
    print(f'{valid_path}: {os.path.isdir(valid_path)}')
    print(f'{test_path}: {os.path.isdir(test_path)}')
    for label in labels:
        print(f'{os.path.join(train_path,label)}: {os.path.isdir(os.path.join(train_path,label))}')
        print(f'{os.path.join(valid_path,label)}: {os.path.isdir(os.path.join(valid_path,label))}')
        print(f'{os.path.join(test_path,label)}: {os.path.isdir(os.path.join(test_path,label))}')
check_dir()
def load_images(n,mode,src_dir,dest_dir):
    for label in labels:
        src_folder=os.path.join(src_dir,label)
        dest_folder=os.path.join(dest_dir,label)
        image_set=random.sample(os.listdir(src_folder),n)
        if mode=='train':
            print(f'Loading the training images for {label}')
        elif mode=='valid':
            print(f'Loading the validation images for {label}')
        elif mode=='test':
            print(f'Loading the testing images for {label}')
        else:
            print('Invalid mode of operation')
            return
        for file in tqdm(image_set):
            file_path=os.path.join(src_folder,file)
            shutil.copy(file_path,dest_folder)
load_images(
    n=7000,mode='train',
    src_dir=input_data_path,
    dest_dir=train_path
)
load_images(
    n=3000,mode='valid',
    src_dir=input_data_path,
    dest_dir=valid_path
)
load_images(
    n=2000,mode='test',
    src_dir=input_data_path,
    dest_dir=test_path
)
target_size=(64,64)
batch_size=32
epochs=40
datagen=ImageDataGenerator(
    rescale=1./255,
    preprocessing_function=tf.keras.applications.vgg16.preprocess_input
)

train_data=ImageDataGenerator(
    rescale=1./255,
    preprocessing_function=tf.keras.applications.vgg16.preprocess_input,
    horizontal_flip=True,vertical_flip=True
).flow_from_directory(
    directory=train_path,target_size=target_size,
    classes=labels,batch_size=batch_size
)

valid_data=datagen.flow_from_directory(
    directory=valid_path,target_size=target_size,
    classes=labels,batch_size=batch_size
)

test_data=datagen.flow_from_directory(
    directory=test_path,target_size=target_size,
    classes=labels,batch_size=batch_size,shuffle=False
)<jupyter_output><empty_output><jupyter_text>GoogleNet-Resnet<jupyter_code>class MCDropout(layers.Dropout):
    def call(self,inputs):
        return super().call(inputs,training=True)
lrn_layer=layers.Lambda(
    lambda x: tf.nn.local_response_normalization(
        input=x,depth_radius=2,bias=1,
        alpha=0.00002,beta=0.75
    )
)
class ResnetLayer(layers.Layer):
    def __init__(self,filters,kernel_size=3,strides=1,n_conv=4,**kwargs):
        super().__init__(**kwargs)
        self.resnet_layer=[]
        for _ in range(n_conv):
            self.resnet_layer.append(
                layers.Conv2D(
                    filters=filters,kernel_size=kernel_size,
                    strides=strides,padding='same',activation='relu'
                )
            )
            self.resnet_layer.append(layers.BatchNormalization())
    def call(self,inputs):
        out=inputs
        for resnet_layer in self.resnet_layer:
            out=resnet_layer(out)
        out=layers.Concatenate()([out,inputs])
        return tf.keras.activations.relu(out)
class InceptionModule(layers.Layer):
    def __init__(self,filters_list,**kwargs):
        super().__init__(**kwargs)
        self.layer_list=[]
        for i in range(len(filters_list)):
            if i==0:
                google_net_layer=tf.keras.models.Sequential([
                    layers.Conv2D(
                        filters=filters_list[i],kernel_size=1,
                        strides=1,padding='same',activation='relu'
                    ),
                    layers.BatchNormalization()
                ])
            elif i!=len(filters_list)-1:
                google_net_layer=tf.keras.models.Sequential([
                    layers.Conv2D(
                        filters=filters_list[i],kernel_size=1,
                        strides=1,padding='same',activation='relu'
                    ),
                    layers.BatchNormalization(),
                    layers.Conv2D(
                        filters=int(filters_list[i])/2,kernel_size=2*i-1,
                        strides=1,padding='same',activation='relu'
                    ),
                    layers.BatchNormalization()
                ])
            else:
                google_net_layer=tf.keras.models.Sequential([
                    layers.MaxPool2D(
                        pool_size=3,strides=1,
                        padding='same'
                    ),
                    layers.Conv2D(
                        filters=filters_list[i],kernel_size=1,
                        strides=1,padding='same',activation='relu'
                    ),
                    layers.BatchNormalization()
                ])
            self.layer_list.append(google_net_layer)
    def call(self,inputs):
        output_list=[]
        for google_net_path in self.layer_list:
            out=google_net_path(inputs)
            output_list.append(out)
        output=layers.Concatenate()(output_list)
        return tf.keras.activations.relu(output)
def build_model_googlenet_resnet(target_size,n_conv=4,rate=0.45):
    model=[]
    steps=int(np.log2(target_size[0]))
    model.append(layers.Input(shape=(*target_size,3)))
    filters=8
    cnt=0
    for _ in range(steps):
        if cnt==0:
            model.append(InceptionModule(
                filters_list=[int(filters/2),filters,filters*2,int(filters/2)]
                )
            )
        else:
            model.append(ResnetLayer(filters=filters,n_conv=n_conv))
        model.append(lrn_layer)
        model.append(layers.AvgPool2D(
            pool_size=2,
            strides=2
        ))
        filters*=2
        cnt=(cnt+1)%2
    model+=[
        layers.Flatten(),
        MCDropout(rate),
        layers.Dense(units=4096,activation='relu'),
        MCDropout(rate),
        layers.Dense(units=4096,activation='relu'),
        layers.Dense(units=len(labels),activation='softmax'),
    ]
    return tf.keras.models.Sequential(model)
model=build_model_googlenet_resnet(target_size=target_size)
model.summary()
ckpt_path1='/kaggle/model_checkpoint1'
if not os.path.isdir(ckpt_path1):
    os.mkdir(ckpt_path1)
model_checkpoint=ModelCheckpoint(
    filepath=ckpt_path1,monitor='val_accuracy',
    save_best_only=True,save_weights_only=True,
    mode='max'
)
model.compile(optimizer=Adam(learning_rate=0.0003),loss='categorical_crossentropy',metrics=['accuracy'])
history1=model.fit(
    x=train_data,batch_size=batch_size,epochs=epochs,
    callbacks=[model_checkpoint],validation_data=valid_data
)
def accuracy(data,steps=20):
    y_prob_stack=np.stack([np.array(model.predict(data)) for _ in tqdm(range(steps))])
    y_prob=np.mean(y_prob_stack,axis=0)
    cm=confusion_matrix(y_true=data.classes,y_pred=np.argmax(y_prob,axis=-1))
    acc=cm.trace()/cm.sum()
    return acc*100
print(f'The accuracy of the model on test set is {accuracy(test_data)}')
model.load_weights(ckpt_path1)
print(f'The accuracy of the model with MCDropout on test set is {accuracy(test_data)}')
def plot_history(history):
    val_loss=history.history['val_loss']
    train_loss=history.history['loss']
    plt.figure()
    plt.xlabel('Epochs')
    plt.ylabel('Loss')
    plt.plot(train_loss,'bo--')
    plt.plot(val_loss,'ro--')
    plt.title('Loss vs Epochs')
    plt.legend(['Train','Valid'])
    plt.tight_layout()
    plt.show()
plot_history(history1)
tf.keras.backend.clear_session()<jupyter_output><empty_output><jupyter_text>Bottleneck Resnet<jupyter_code>class BottleNeckLayer(tf.keras.models.Model):
    def __init__(self,out_channels,n_conv=3,ratio=2,kernel_size=3,strides=1):
        super().__init__()
        self.model=[]
        intermediate_filters=int(out_channels/ratio)
        self.model.append(layers.Conv2D(filters=intermediate_filters,kernel_size=1,strides=1,activation='relu',padding='same'))
        for i in range(n_conv):
            self.model.append(layers.Conv2D(filters=intermediate_filters,kernel_size=kernel_size,strides=strides,activation='relu',padding='same'))
            self.model.append(layers.BatchNormalization())
        self.model.append(layers.Conv2D(filters=out_channels,kernel_size=1,strides=1,activation='relu',padding='same'))
        self.model=tf.keras.models.Sequential(self.model)
    def call(self,inputs):
        out=self.model(inputs)
        out=layers.Concatenate()([inputs,out])
        return tf.keras.activations.relu(out)
def build_model_bottleneck_resnet(size=target_size[0],n_conv=4,ratio=2):
    steps=int(np.log2(size))
    model=[]
    out_channels=16
    model.append(layers.Input(shape=(size,size,3)))
    model.append(layers.Conv2D(filters=out_channels,kernel_size=1,strides=1,activation='relu',padding='same'))
    for i in range(steps):
        model.append(BottleNeckLayer(out_channels=out_channels,n_conv=n_conv,ratio=ratio))
        model.append(lrn_layer)
        model.append(layers.AvgPool2D(pool_size=2,strides=2))
        out_channels*=2
    model+=[
        layers.Flatten(),
        layers.Dense(units=4096,activation='relu'),
        layers.Dense(units=4096,activation='relu'),
        layers.Dense(units=len(labels),activation='softmax')
    ]
    return tf.keras.models.Sequential(model)
model=build_model_bottleneck_resnet(size=target_size[0])
model.summary()
ckpt_path2='/kaggle/model_checkpoint2'
if not os.path.isdir(ckpt_path2):
    os.mkdir(ckpt_path2)
model_checkpoint=ModelCheckpoint(
    filepath=ckpt_path2,monitor='val_accuracy',
    save_best_only=True,save_weights_only=True,
    mode='max'
)
model.compile(optimizer=Adam(learning_rate=0.0003),loss='categorical_crossentropy',metrics=['accuracy'])
history2=model.fit(
    x=train_data,batch_size=batch_size,epochs=epochs,
    callbacks=[model_checkpoint],validation_data=valid_data
)
print(f'The accuracy of the model on test set is {accuracy(test_data)}')
model.load_weights(ckpt_path2)
print(f'The accuracy of the model with MCDropout on test set is {accuracy(test_data)}')
plot_history(history2)
tf.keras.backend.clear_session()<jupyter_output><empty_output><jupyter_text>Resnet<jupyter_code>def build_model_resnet(target_size,n_conv=4,rate=0.45):
    model=[]
    steps=int(np.log2(target_size[0]))
    model.append(layers.Input(shape=(*target_size,3)))
    filters=8
    for _ in range(steps):
        model.append(ResnetLayer(filters=filters,n_conv=n_conv))
        model.append(lrn_layer)
        model.append(layers.AvgPool2D(
            pool_size=2,
            strides=2
        ))
        filters*=2
    model+=[
        layers.Flatten(),
        MCDropout(rate),
        layers.Dense(units=4096,activation='relu'),
        MCDropout(rate),
        layers.Dense(units=4096,activation='relu'),
        layers.Dense(units=len(labels),activation='softmax'),
    ]
    return tf.keras.models.Sequential(model)
model=build_model_resnet(target_size=target_size)
model.summary()
ckpt_path3='/kaggle/model_checkpoint3'
if not os.path.isdir(ckpt_path3):
    os.mkdir(ckpt_path3)
model_checkpoint=ModelCheckpoint(
    filepath=ckpt_path3,monitor='val_accuracy',
    save_best_only=True,save_weights_only=True,
    mode='max'
)
model.compile(optimizer=Adam(learning_rate=0.0003),loss='categorical_crossentropy',metrics=['accuracy'])
history3=model.fit(
    x=train_data,batch_size=batch_size,epochs=epochs,
    callbacks=[model_checkpoint],validation_data=valid_data
)
print(f'The accuracy of the model on test set is {accuracy(test_data)}')
model.load_weights(ckpt_path3)
print(f'The accuracy of the model with MCDropout on test set is {accuracy(test_data)}')
plot_history(history3)
tf.keras.backend.clear_session()<jupyter_output><empty_output><jupyter_text>SE-Resnet<jupyter_code>np_config.enable_numpy_behavior()
class ConvStack(layers.Layer):
    def __init__(self,filters,n_conv=4,kernel_size=3,strides=1,**kwargs):
        super().__init__(**kwargs)
        self.resnet_layers=[]
        for _ in range(n_conv):
            self.resnet_layers.append(layers.Conv2D(
                filters=filters,kernel_size=kernel_size,
                strides=1,padding='same',activation='relu'
            ))
            self.resnet_layers.append(layers.BatchNormalization())
    def call(self,inputs):
        output=inputs
        for resnet_layer in self.resnet_layers:
            output=resnet_layer(output)
        return output
class SE_block(layers.Layer):
    def __init__(self,output_channels,**kwargs):
        super().__init__(**kwargs)
        self.block=[]
        self.block.append(layers.GlobalAveragePooling2D())
        if output_channels<128:
            self.block.append(layers.Dense(units=int(output_channels)/2,activation='relu'))
        else:
            self.block.append(layers.Dense(units=int(output_channels)/16,activation='relu'))
        self.block.append(layers.Dense(units=output_channels,activation='sigmoid'))
        self.out_channels=output_channels
    def call(self,inputs):
        output=inputs
        for block_layer in self.block:
            output=block_layer(output)
        outputs=output.reshape(inputs.shape[0],1,1,self.out_channels)
        outputs=layers.Multiply()([inputs,outputs])
        return outputs
class SEResnet(layers.Layer):
    def __init__(self,filters,kernel_size=3,strides=1,n_conv=4,**kwargs):
        super().__init__(**kwargs)
        self.hidden_layers=[]
        self.hidden_layers.append(ConvStack(
            filters=filters,n_conv=n_conv,
            kernel_size=kernel_size,strides=strides
        ))
        self.hidden_layers.append(SE_block(output_channels=filters))
    def call(self,inputs):
        output=inputs
        for hidden_layer in self.hidden_layers:
            output=hidden_layer(output)
        output=layers.Concatenate()([output,inputs])
        return tf.keras.activations.relu(output)
def build_model_se_net(n_conv=4,rate=0.45,target_size=target_size):
    steps=int(np.log2(int(target_size[0])))
    inputs=layers.Input(shape=(*target_size,3),batch_size=batch_size)
    filters=8
    model=SEResnet(filters=filters,n_conv=n_conv)(inputs)
    model=layers.AvgPool2D(pool_size=2,strides=2)(model)
    model=layers.Lambda(lrn_layer)(model)
    for _ in tqdm(range(steps-1)):
        filters*=2
        model=SEResnet(filters=filters,n_conv=n_conv)(model)
        model=layers.AvgPool2D(pool_size=2,strides=2)(model)
        model=layers.Lambda(lrn_layer)(model)
    model=MCDropout(rate)(model)
    model=layers.Conv2D(
        filters=4096,kernel_size=1,
        padding='valid',strides=1
    )(model)
    model=MCDropout(rate)(model)
    model=layers.Conv2D(
        filters=4096,kernel_size=1,
        padding='valid',strides=1
    )(model)
    model=layers.Flatten()(model)
    model=layers.Dense(units=len(labels),activation='softmax')(model)
    model=tf.keras.models.Model(inputs=inputs,outputs=model)
    model.compile(optimizer=Adam(learning_rate=0.0003),loss='categorical_crossentropy',metrics=['accuracy'])
    return model
model=build_model_se_net(target_size=target_size)
model.summary()
ckpt_path4='/kaggle/ckpt'
if not os.path.isdir(ckpt_path4):
    os.mkdir(ckpt_path4)
model_checkpoint=ModelCheckpoint(
    filepath=ckpt_path4,monitor='val_accuracy',
    save_best_only=True,save_weights_only=True,
    mode='max'
)
tf.config.run_functions_eagerly(True)
history4=model.fit(
    x=train_data,batch_size=batch_size,epochs=epochs,
    validation_data=valid_data,callbacks=[model_checkpoint]
)
print(f'The accuracy of the model on test set is {accuracy(test_data)}')
model.load_weights(ckpt_path4)
print(f'The accuracy of the model with MCDropout on test set is {accuracy(test_data)}')
plot_history(history4)
tf.keras.backend.clear_session()
googlenet_resnet_val_loss=history1.history['val_loss']
bottleneck_resnet_val_loss=history2.history['val_loss']
resnet_val_loss=history3.history['val_loss']
se_resnet_val_loss=history4.history['val_loss']
plt.figure()
plt.plot(googlenet_resnet_val_loss,'bo--')
plt.plot(bottleneck_resnet_val_loss,'ro--')
plt.plot(resnet_val_loss,'go--')
plt.plot(se_resnet_val_loss,'co--')
plt.title('Model validation loss vs Epochs')
plt.xlabel('Epoch')
plt.ylabel('Validation Loss')
plt.legend(['Googlenet Resnet','Bottleneck Resnet','Resnet','SE-resnet'])
plt.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12761945806173572, 'fraction_numerical': 0.01438345095081614, 'mean_word_length': 3.7710100231303008, 'pattern_counts': {'":': 0, '<': 26, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '154641', 'license': 'MIT', 'n_tokens_mistral': 6472, 'n_tokens_neox': 6299, 'n_words': 702, 'path': 'project_comparing_models_on_malaria_classification.ipynb', 'repo_name': 'danielsoy/ALOCC-CVPR2018', 'size': '18563'}