<jupyter_start><jupyter_text>--- **Pneumonia Classification using RESNET**---<jupyter_code>import cv2
import os
import random
from PIL import Image, ImageOps
import numpy as np
from keras import layers
from keras.layers import Input, Add, Dense, Activation, ZeroPadding2D, BatchNormalization, Flatten, Conv2D, AveragePooling2D, MaxPooling2D, GlobalMaxPooling2D
from keras.models import Model, load_model
import keras.preprocessing as preprocessing
from keras.utils import layer_utils
from keras.utils.data_utils import get_file
from keras import regularizers
from keras.applications.imagenet_utils import preprocess_input
import pydot
from IPython.display import SVG
from keras.utils.vis_utils import model_to_dot
from keras.utils import plot_model
from keras.initializers import glorot_uniform
from keras.metrics import Recall
from keras.losses import BinaryCrossentropy
from keras import optimizers 
from sklearn.preprocessing import LabelEncoder
from sklearn.preprocessing import OneHotEncoder
import tensorflow as tf
import scipy.misc
from matplotlib.pyplot import imshow
import matplotlib.pyplot as plt
%matplotlib inline

import keras.backend as K
K.set_image_data_format('channels_last')<jupyter_output><empty_output><jupyter_text>**1. Initializing a TPU system**<jupyter_code>resolver = tf.distribute.cluster_resolver.TPUClusterResolver(tpu='')
tf.config.experimental_connect_to_cluster(resolver)
tf.tpu.experimental.initialize_tpu_system(resolver)
print("All devices: ", tf.config.list_logical_devices('TPU'))
try:
 device_name = os.environ['COLAB_TPU_ADDR']
 TPU_ADDRESS = 'grpc://' + device_name
 print('Found TPU at: {}'.format(TPU_ADDRESS))
except KeyError:
 print('TPU not found')<jupyter_output>Found TPU at: grpc://10.51.205.106:8470<jupyter_text>**2. Importing data**<jupyter_code>from google.colab import drive
drive.mount('/content/gdrive')
def input_fn(x,y,batch_size=0,use_batch=True):
    # Convert the inputs to a Tensorflow dataset.
    dataset = tf.data.Dataset.from_tensor_slices((x,y))

    # Shuffle, repeat, and batch the examples.
    dataset = dataset.cache()
    dataset = dataset.shuffle(1000, reshuffle_each_iteration=True)
    dataset = dataset.repeat()

    if use_batch:
      dataset = dataset.batch(batch_size, drop_remainder=True)
      

    return dataset

def input_fn_1(x,batch_size=0,use_batch=True):
    # Convert the inputs to a Tensorflow dataset.
    dataset = tf.data.Dataset.from_tensor_slices((x))

    # Shuffle, repeat, and batch the examples.
    dataset = dataset.cache()
    dataset = dataset.shuffle(1000, reshuffle_each_iteration=True)
    dataset = dataset.repeat()

    if use_batch:
      dataset = dataset.batch(batch_size, drop_remainder=True)
      

    return dataset

def load_data(data_dir, sample_size, use_sample_size, img_dim=64):
  
  ds = []

  label_dir = os.listdir(data_dir)
  
  for label in label_dir:

    if not(label == '.DS_Store'):
      label_dir_path = os.path.join(data_dir,label)
      filenames = os.listdir(label_dir_path)
      imgs_path = [os.path.join(label_dir_path, img_name) for img_name in filenames if img_name.endswith('.jpeg')]

      if use_sample_size: 
        imgs_path = imgs_path[:sample_size]

      print('Number of images of type {}:'.format(label),len(imgs_path))
     
      for img_path in imgs_path:
        img = Image.open(img_path)
        img = ImageOps.grayscale(img)
        img = np.array(img)
        img = np.reshape(img,(img_dim,img_dim,1))
        img = img.astype('float32')
        img = img/225
        ds.append([img,label])
  
  random.seed(1234)
  random.shuffle(ds)

  return ds

def load_tensorflow_ds(train_val=True, sample_size=255, use_sample_size=True, img_dim=64, train_batch_size=120, val_batch_size=8, test_batch_size=16):
  if train_val:
    train_ds = load_data('/content/gdrive/MyDrive/chest_xray/Resized_train_64_all',sample_size, use_sample_size, img_dim)
    labels = []
    images = []
    for x in range(len(train_ds)):
      labels.append(train_ds[x][1])
      images.append(train_ds[x][0])
    
    train_size = 5160
    X_train = images[:train_size]
    y_train = labels[:train_size]

    X_val = images[train_size:]
    y_val = labels[train_size:]

    # One-hot-encoding y_train
    label_encoder = LabelEncoder()
    integer_encoded = label_encoder.fit_transform(y_train)
    onehot_encoder = OneHotEncoder(sparse=False)
    integer_encoded = integer_encoded.reshape(len(integer_encoded), 1)
    y_train = onehot_encoder.fit_transform(integer_encoded)

    # One-hot-encoding y_val
    integer_encoded = label_encoder.fit_transform(y_val)
    integer_encoded = integer_encoded.reshape(len(integer_encoded), 1)
    y_val = onehot_encoder.fit_transform(integer_encoded)

    print('Loading train Tensorflow dataset')
    train_ds = input_fn(X_train, y_train, train_batch_size)
    print('Loading val Tensorflow dataset')
    val_ds = input_fn(X_val, y_val, val_batch_size)

    train_val_list = [len(X_train),np.array(X_train).shape,np.array(y_train).shape]

    return train_ds,val_ds,train_val_list
  
  else:
    test_ds = load_data('/content/gdrive/MyDrive/chest_xray/Resized_test_64',sample_size, use_sample_size, img_dim)
    labels = []
    images = []

    for x in range(len(test_ds)):
      labels.append(test_ds[x][1])
      images.append(test_ds[x][0])
    X_test = images
    y_test = labels

    # One-hot-encoding y_test
    label_encoder = LabelEncoder()
    integer_encoded = label_encoder.fit_transform(y_test)

    onehot_encoder = OneHotEncoder(sparse=False)
    integer_encoded = integer_encoded.reshape(len(integer_encoded), 1)
    y_test = onehot_encoder.fit_transform(integer_encoded)

    print('Loading test Tensorflow dataset')
    test_ds = input_fn(X_test, y_test, test_batch_size)

    test_list = [len(X_test),np.array(X_test).shape,np.array(y_test).shape]
    
    return test_ds,test_list,X_test,y_test<jupyter_output><empty_output><jupyter_text>Loading train, val, and test datasets<jupyter_code>print('\033[1mLoading train & val ds\033[0m')
train_ds,val_ds, train_val_list = load_tensorflow_ds(use_sample_size=False)
print('\033[1mLoafing test ds\033[0m')
test_ds,test_list,X_test,y_test = load_tensorflow_ds(train_val=False,use_sample_size=False)
x_test = input_fn_1(X_test,624)
print ("Number of training examples:" + str(train_val_list[0]))
print ("Number of test examples:" + str(test_list[0]))
print ("X_train shape:" + str(train_val_list[1]))
print ("y_train shape:" + str(train_val_list[2]))
print ("X_test shape:" + str(test_list[1]))
print ("y_test shape:" + str(test_list[2]))
plt.figure(figsize=(10, 10))
i=0
for image, label in (X_train[0:9],y_train[0:9]):
  ax = plt.subplot(3, 3, i + 1)
  plt.imshow(image[i].numpy().astype("uint8"))
  plt.title(label[i])
  plt.axis("off")
  i+=1<jupyter_output><empty_output><jupyter_text>**3. ResNet architecture**<jupyter_code>def convolutional_block(X, f, filters, stage, block, s = 2):
    """
    Implementation of the convolutional block 
  
    Returns:
    X -- output of the convolutional block, tensor of shape (n_H, n_W, n_C)
    """
    
    # defining name basis
    conv_name_base = 'res' + str(stage) + block + '_branch'
    bn_name_base = 'bn' + str(stage) + block + '_branch'
    
    # Retrieve Filters
    F1, F2, F3 = filters
    
    # Save the input value
    X_shortcut = X


    ##### MAIN PATH #####
    # First component of main path 
    X = Conv2D(F1, (1, 1), strides = (s,s), name = conv_name_base + '2a', kernel_initializer = glorot_uniform(seed=0),kernel_regularizer=regularizers.l2(0.01))(X)
    X = BatchNormalization(axis = 3, name = bn_name_base + '2a')(X)
    X = Activation('relu')(X)
    
    ### START CODE HERE ###

    # Second component of main path (≈3 lines)
    X = Conv2D(F2, (f, f), strides = (1,1),padding = "same", name = conv_name_base + '2b', kernel_initializer = glorot_uniform(seed=0),kernel_regularizer=regularizers.l2(0.01))(X)
    X = BatchNormalization(axis = 3, name = bn_name_base + '2b')(X)
    X = Activation('relu')(X)

    # Third component of main path (≈2 lines)
    X = Conv2D(F3, (1, 1), strides = (1,1), name = conv_name_base + '2c', kernel_initializer = glorot_uniform(seed=0),kernel_regularizer=regularizers.l2(0.01))(X)
    X = BatchNormalization(axis = 3, name = bn_name_base + '2c')(X)
    
    ##### SHORTCUT PATH #### (≈2 lines)
    X_shortcut = Conv2D(F3, (1, 1), strides = (s,s), name = conv_name_base + '1', kernel_initializer = glorot_uniform(seed=0),kernel_regularizer=regularizers.l2(0.01))(X_shortcut)
    X_shortcut = BatchNormalization(axis = 3, name = bn_name_base + '1')(X_shortcut)

    # Final step: Add shortcut value to main path, and pass it through a RELU activation (≈2 lines)
    X = Add()([X,X_shortcut])
    X = Activation('relu')(X)
    
    ### END CODE HERE ###
    
    return X
def identity_block(X, f, filters, stage, block):
    """
    Implementation of the identity block 
   
    Returns:
    X -- output of the identity block, tensor of shape (n_H, n_W, n_C)
    """
    
    # defining name basis
    conv_name_base = 'res' + str(stage) + block + '_branch'
    bn_name_base = 'bn' + str(stage) + block + '_branch'
    
    # Retrieve Filters
    F1, F2, F3 = filters
    
    # Save the input value. You'll need this later to add back to the main path. 
    X_shortcut = X
    
    # First component of main path
    X = Conv2D(filters = F1, kernel_size = (1, 1), strides = (1,1), padding = 'valid', name = conv_name_base + '2a', kernel_initializer = glorot_uniform(seed=0),kernel_regularizer=regularizers.l2(0.01))(X)
    X = BatchNormalization(axis = 3, name = bn_name_base + '2a')(X)
    X = Activation('relu')(X)
    
    ### START CODE HERE ###
    
    # Second component of main path (≈3 lines)
    X = Conv2D(filters = F2, kernel_size = (f, f), strides = (1,1), padding = 'same', name = conv_name_base + '2b', kernel_initializer = glorot_uniform(seed=0),kernel_regularizer=regularizers.l2(0.01))(X)
    X = BatchNormalization(axis = 3, name = bn_name_base + '2b')(X)
    X = Activation('relu')(X)
    

    # Third component of main path (≈2 lines)
    X = Conv2D(filters = F3, kernel_size = (1, 1), strides = (1,1), padding = 'valid', name = conv_name_base + '2c', kernel_initializer = glorot_uniform(seed=0),kernel_regularizer=regularizers.l2(0.01))(X)
    X = BatchNormalization(axis = 3, name = bn_name_base + '2c')(X)

    # Final step: Add shortcut value to main path, and pass it through a RELU activation (≈2 lines)
    X = Add()([X,X_shortcut])
    X = Activation("relu")(X)
    
    ### END CODE HERE ###
    
    return X

def ResNet(input_shape = (64, 64,1), classes = 2, rate = 0.3):
    """
    Implementation of the popular ResNet architecture

    Returns:
    model -- a Model() instance in Keras
    """
    
    # Define the input as a tensor with shape input_shape
    X_input = Input(input_shape)

    
    # Zero-Padding
    X = ZeroPadding2D((1, 1))(X_input)
    
    # Stage 1
    X = Conv2D(64, (3, 3), strides = (2, 2), name = 'conv1', kernel_initializer = glorot_uniform(seed=0))(X)
    X = BatchNormalization(axis = 3, name = 'bn_conv1')(X)
    X = Activation('relu')(X)
    X = MaxPooling2D((3, 3), strides=(2, 2))(X)

    # Stage 2
    X = convolutional_block(X, f = 3, filters = [64, 64, 128], stage = 2, block='a', s = 1)
    X = identity_block(X, 3, [64, 64, 128], stage=2, block='b')
    X = identity_block(X, 3, [64, 64, 128], stage=2, block='c')
    X = identity_block(X, 3, [64, 64, 128], stage=2, block='d')
    X = tf.keras.layers.Dropout(rate, seed=134)(X)

    # stage 3
    X = convolutional_block(X, f = 3, filters = [64, 64, 128], stage = 3, block='a', s = 1)
    X = identity_block(X, 3, [64, 64, 128], stage=3, block='b')
    X = identity_block(X, 3, [64, 64, 128], stage=3, block='c')
    X = identity_block(X, 3, [64, 64, 128], stage=3, block='d')
    X = tf.keras.layers.Dropout(rate, seed=234)(X)
  
    # Stage 4 
    X = convolutional_block(X, f = 3, filters = [128, 128, 256], stage = 4, block='a', s = 2)
    X = identity_block(X, 3, [128, 128, 256], stage=4, block='b')
    X = identity_block(X, 3, [128, 128, 256], stage=4, block='c')
    X = identity_block(X, 3, [128, 128, 256], stage=4, block='d')
    X = tf.keras.layers.Dropout(rate, seed=124)(X)

    # Stage 5
    X = convolutional_block(X, f = 3, filters = [128, 128, 256], stage = 5, block='a', s = 2)
    X = identity_block(X, 3, [128, 128, 256], stage=5, block='b')
    X = identity_block(X, 3, [128, 128, 256], stage=5, block='c')
    X = identity_block(X, 3, [128, 128, 256], stage=5, block='d')
    X = tf.keras.layers.Dropout(rate, seed=123)(X)

    # Stage 6 
    X = convolutional_block(X, f = 3, filters = [256, 256, 512], stage = 6, block='a', s = 2)
    X = identity_block(X, 3, [256, 256, 512], stage=6, block='b')
    X = identity_block(X, 3, [256, 256, 512], stage=6, block='c')
    X = identity_block(X, 3, [256, 256, 512], stage=6, block='d')
    X = tf.keras.layers.Dropout(rate, seed=14)(X)

    # Stage 7
    X = convolutional_block(X, f = 3, filters = [256, 256, 512], stage = 7, block='a', s = 2)
    X = identity_block(X, 3, [256, 256, 512], stage=7, block='b')
    X = identity_block(X, 3, [256, 256, 512], stage=7, block='c')
    X = identity_block(X, 3, [256, 256, 512], stage=7, block='d')
    X = tf.keras.layers.Dropout(rate, seed=34)(X)

    # output layer
    X = Flatten()(X)
    X = Dense(classes, activation='sigmoid', name='fc' + str(classes), kernel_initializer = glorot_uniform(seed=0))(X)
    
    # Create model
    model = Model(inputs = X_input, outputs = X, name='ResNet')

    return model<jupyter_output><empty_output><jupyter_text>**4. Model training**<jupyter_code>strategy = tf.distribute.TPUStrategy(resolver)
with strategy.scope():
  model = ResNet()
  model.compile(optimizer='adam',
                loss=BinaryCrossentropy(from_logits=True),
                metrics=Recall())
steps_per_epoch = 5160 // 120
validation_steps = 56 // 8

weight_for_normal = 5216 / (2 * 1341 )
weight_for_pneumonia = 5216 / (2 * 3875)

weights = {0:weight_for_normal, 1:weight_for_pneumonia}
from keras.callbacks import EarlyStopping, ModelCheckpoint, ReduceLROnPlateau
model_checkpoint = ModelCheckpoint('.mdl_wts.hdf5', save_best_only=True, monitor='val_recall_1', mode='max',verbose=1)
history = model.fit(train_ds,
          epochs=50,
          steps_per_epoch=steps_per_epoch,
          validation_data=val_ds, 
          validation_steps=validation_steps,
          callbacks=[model_checkpoint],
          class_weight=weights)
#model.load_weights('.mdl_wts.hdf5')
model.load_weights('/content/gdrive/MyDrive/Resnet_models/model_Resnet_4_weights_recall_988.h5')
model.save_weights('/content/gdrive/MyDrive/Resnet_models/model_Resnet_weights_recall_988.h5', overwrite=True)
preds = model.evaluate(test_ds,steps=39)
print ("Loss = " + str(preds[0]))
print ("Test Recall = " + str(preds[1]))
from sklearn.metrics import classification_report
predictions = model.predict(x_test,steps = 1)
#predictions = predictions.reshape(1,-1)[0]
predictions = np.argmax(predictions, axis=1)
#y_test = np.argmax(y_test, axis=1)
#report = classification_report(, predicted)

print(classification_report(np.argmax(y_test, axis=1), list(predictions), target_names = ['NORMAL (Class 0)','PNEUMONIA (Class 1)']))
from sklearn.metrics import confusion_matrix
confusion_matrix(np.argmax(y_test, axis=1),predictions)
255/(255+135)<jupyter_output><empty_output><jupyter_text><jupyter_code>epochs = 70
acc = history.history['recall']
val_acc = history.history['val_recall']

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 Recall')
plt.plot(epochs_range, val_acc, label='Validation Recall')
plt.legend(loc='lower right')
plt.title('Training and Validation Recall')

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()
model.summary()
len(model.layers)
plot_model(model, to_file='resnet_model.png')
SVG(model_to_dot(model).create(prog='dot', format='svg'))
imshow(X_test[0].astype("uint8"))
#plt.axis("off")<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13924128209842496, 'fraction_numerical': 0.03689403689403689, 'mean_word_length': 4.337912986588158, 'pattern_counts': {'":': 0, '<': 28, '<?xml version=': 0, '>': 28, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, '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': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '14973', 'license': 'MIT', 'n_tokens_mistral': 6385, 'n_tokens_neox': 5896, 'n_words': 1292, 'path': 'Pneumonia_classification.ipynb', 'repo_name': 'agam001/Pneumonia-Classification', 'size': '16317'}