Hola! 😃
En este post, trataremos nuevamente el tema de clasificación de imágenes usando el método de Transfer Learning para diagnosticar, a partir de una imagen de rayos X de tórax, si un paciente tiene neumonía o no. A este tipo de clasificación se les denomina "binarias".
EfficientNet es una nueva (2019) arquitectura de Deep Learning para Redes Neurales Convolucionales. En el articulo "EfficientNet: Rethinking Model Scaling for Convolutional Neural Networks" https://arxiv.org/pdf/1905.11946.pdf se habla de una nueva forma de pensar las arquitecturas, limitando el ancho, largo y resolución de la Red Convolucional a relaciones fijas entre estos parámetros. El enlace al repositorio de Github de Efficientnet es el que sigue:
En en post anterior, vimos que el desbalance entre las clases es un problema para entrenar modelos de Machine Learning, ya que las clases con mayor cantidad de ejemplos tendrán preferencia en la predicción. Habíamos resuelto este problema haciendo downsampling a nuestros datos de entrenamiento, obteniendo buenos resultados.
En este post, usaremos el método de "clases pesadas" para resolver el problema de desbalance de datos. Este método consiste en calcular constantes llamadas pesos para cada clase, estos pesos se incluirán en la función de costo (Loss) como factores de corrección en los términos de la función.
La función de costo "Loss" es una métrica muy importante para medir el despeño de un modelo de clasificación, y depende de la diferencia entre la probabilidad predicha para una clase y el valor real de la clase. Por ejemplo:
Un modelo que predice 0.95 de probabilidad para diagnóstico de neumonía tendrá una diferencia con el valor real (1) de 0.05 mientras que otro modelo que predice una probabilidad de 0.70 para neumonía presenta una diferencia con el valor real de 0.3. Con un valor de corte de 0.5 (probabilidades mayores a 0.5 diagnostíca neumonía) ambos modelos predicen neumonía para el paciente, pero el segundo modelo tendrá una diferencia con el valor real mucho mayor que el primero.
Las diferencias entre el valor real y el predicho por el modelo nos indican cual modelo está separando mejor las clases, un modelo donde estas diferencias son pequeñas es un buen predictor. Es por esto que el objetivo siempre será minimizar el Loss.
La función de costo "usual" para los problemas de clasificación binaria de clases que están balanceadas es el binary-crossentropy, y está dada como sigue:
Donde "y" es el valor de la categoría (número entero, 0 ó 1) y "p" es la probabilidad calculada o predicha para esa categoría (número real, valores entre 0 y 1). El primer término de la ecuación corresponde al loss de la clase "1" y el segundo término corresponde al loss de la clase "0".
La función de costo está relacionada con la "entropía" del sistema, cuanto más ordenado esté el sistema menor será la entropía o función de costo; el objetivo al entrenar un modelo de clasificación es minimizar la entropía del sistema, pero cuando las clases están desbalanceadas el sumando que corresponde a la clase con más ejemplos contribuye más a la función de costo, por ende, nuestro modelo va a optimizar su entropía encontrando un mínimo que estará mejor adaptado a la clase predominante. Es por esto que cuando entrenamos un modelo con datos desbalanceados (sin hacer ningún ajuste) este tendrá preferencia de predicción sobre la clase dominante.
Para resolver este problema agregamos pesos a la fórmula, estos pesos son constantes que multiplican a los sumandos de la fórmula de loss, y que tranforman la escala de las contribuciones de cada clase, haciendo que todas contibuyan en proporciones iguales.
Cuando las clases "0" y "1" no están balanceadas la ecuación de costo se escribe como sigue:
Otro método que añadiremos es la estandarización de los datos, recordemos que para el código del post anterior dividimos los pixeles de las imágenes por 255, logrando que cada pixel tuviera un valor entre 0 y 1. Pero esto no es suficiente para hacer una comparación efectiva entre las imágenes. Es necesario aplicar un método de estadarización de los datos en cada imagen, haciendo que cada una de ellas tenga un promedio 0 y desviación estandar 1 en los valores de sus pixeles. Esto permitirá al modelo estudiar la forma de las distribuciones en cada imagen teniendo una base de medición estándar para todas ellas.
Nuestro plan de trabajo será el siguiente:
1) Carga de datos
2) Analisís de los datos y Visualización de imágenes
3) Normalización de datos y generación de datasets para entrenamiento, prueba y validación
4) Cálculo de los pesos para las clases positiva y negativa a usar en la función de Loss balanceado
5) Instalación de la arquitectura de Transfer Learning EfficientNet, y construcción de la red neural
6) Entrenamiento del modelo
7) Validación del modelo
Carga de Datos
Los datos han sido extraídos de la base de datos de Kaggle "Chest X-Ray Images (Pneumonia)" que contiene 5863 imágenes de rayos X de tórax divididos en 2 categorías: Normal y Pneumonia.
!pip install -U -q kaggle
!mkdir -p ~/.kaggle
from google.colab import files
files.upload()!cp kaggle.json ~/.kaggle/
!kaggle datasets download -d paultimothymooney/chest-xray-pneumoniafrom zipfile import ZipFile
import shutil
import glob
zip_file = ZipFile('/content/chest-xray-pneumonia.zip')
#opening the zip file in READ mode
with ZipFile('/content/chest-xray-pneumonia.zip', 'r') as zip:
# extracting all the files
print('Extracting all the files now...')
zip.extractall()
print('Done!')
!cd /contentimport os
#path to image data directories
base_dir = '/content/'
data_dir = os.path.join(base_dir, 'chest_xray')
train_dir = os.path.join(data_dir, 'train')
test_dir = os.path.join(data_dir, 'test')
val_dir = os.path.join(data_dir, 'val')
train_normal_dir = os.path.join(train_dir, 'NORMAL')
train_pneu_dir = os.path.join(train_dir, 'PNEUMONIA')
test_normal_dir = os.path.join(test_dir, 'NORMAL')
test_pneu_dir = os.path.join(test_dir, 'PNEUMONIA')
val_normal_dir = os.path.join(val_dir, 'NORMAL')
val_pneu_dir = os.path.join(val_dir, 'PNEUMONIA')
# list image file names
list_train_pneu = os.listdir(train_pneu_dir)
list_train_normal = os.listdir(train_normal_dir)
list_test_pneu = os.listdir(test_pneu_dir)
list_test_normal = os.listdir(test_normal_dir)
list_val_pneu = os.listdir(val_pneu_dir)
list_val_normal = os.listdir(val_normal_dir)Análisis de los datos
Como podemos ver en los gráficos de barras nuestros datos tienen desbalance de clases para los datasets de entrenamiento y pruebas. Mas adelante aplicaremos el método de pesos para solucionar este problema.
A continuación podemos observar los rayos X de tórax de los pacientes que presentaron neumonía en el cojunto de validación.
import numpy as np
%matplotlib inline
for i in range(8):
# define subplot
plt.subplot(330 + 1 + i)
# define filename
filename = val_pneu_dir + '/' + str(list_val_pneu[i])
# load image pixels
img = cv2.imread(filename)
# plot raw pixel data
img=cv2.resize(img,(200,200))
plt.imshow(img,cmap='gray')
plt.axis('off')
# show the figure
plt.show()
Y ahora veremos los rayos x de los pacientes que presentaron una condición "normal"
Se puede apreciar que los pacientes "normales" presentan imágenes de rayos X más nítidas en el área de los pulmones, mientras que los pacientes con diagnóstico de neumonía presentan opacidad, con concentración de zonas más blancas en los pulmones.
Haciendo histogramas de la escala de grises en las imágenes, donde el rango de grises va de 0 a 255, con 0 negro y 255 blanco, se pueden observar diferencias para los rayos X normales y con diagnóstico de neumonía.
Se observa claramente como este paciente sin neumonía presenta un pico grade para el color negro, mientras que hay una pequeña distribución de grises que se explica por la presencia de los huesos y otros tejidos del paciente.
En el caso de el paciente con neumonía vemos que la diferencia es dramática, observamos una disminución considerable del pico del negro en el histograma, lo que podría indicar presencia de fluidos en los pulmones, pero también podría deberse a opacidad debido a una toma defectuosa de rayos X. Nuestro modelo no solo tomará en cuenta la distribución de colores en los histogramas de las imágenes, sino también cómo los colores se distribuyen espacialmente en la imagen.
Veamos los histogramas de las imágenes para los casos normales en el conjunto de validación
En efecto, se observa que en todos, menos un histograma, se encuentra el pico en la región del negro que supera los 100.000 pixeles de frecuencia.
A continuación, los histogramas para las imágenes con diagnóstico de neumonía tienen este pico en el negro muy reducido, salvo uno de los histogramas. Como se mencionó antes, esto puede deberse a la presencia de nódulos de liquido concentrado en los pulmones, que no afectan la nitidez y contraste de la imagén, pero que sí representan un sintoma de neumonía.
Normalización de datos y generación de datasets para entrenamiento, prueba y validación
Para la normalización aplicamos los parámetros samplewise_center=True, samplewise_std_normalization= True al generador de datos ImageDataGenerator y luego usamos flow_from directory para traer los datos de los respectivos directorios. Estos parámetros normalizan los datos para que los valores en los pixeles de cada imagen tengan desviación estándar cero y valor medio 1. Además hemos incluido para el aumento de los datos de entrenamiento los siguientes parámetros :
rotation_range=5,
width_shift_range=0.1,
height_shift_range=0.1,
shear_range=0.5,
fill_mode='constant',
zoom_range=0.2,
brightness_range=[0.8, 1.2].
Debemos cuidar de no rotar demasiado las imágenes ni cambiar la orientación de éstas, ya que es importante preservar la posición normal de los órganos en el cuerpo.
Para crear el dataset de entrenamiento:
# create data generator for training
from keras.preprocessing.image import ImageDataGenerator
def get_train_generator(train_dir, batch_size=15,target_w = 299, target_h = 299):
"""
Return generator for training set, normalizing.
Args:
image_dir (str): directory where image files are held.
batch_size (int): images per batch to be fed into model during training.
seed (int): random seed.
Returns:
train_generator (DataFrameIterator): iterator over training set
"""
print("getting train generator...")
# normalize images and data augnmentation
image_generator = ImageDataGenerator(samplewise_center=True,
samplewise_std_normalization= True,
rotation_range=5,
width_shift_range=0.1,
height_shift_range=0.1,
shear_range=0.5,
fill_mode='constant',
zoom_range=0.2,
brightness_range=[0.8, 1.2])
# flow from directory with specified batch size
# and target image size
generator = image_generator.flow_from_directory(train_dir,
class_mode='binary,
batch_size=batch_size,
shuffle=True,
target_size=(target_w, target_h),
color_mode="rgb",
seed=42)
return generatorY para los datasets de validación y pruebas se ha hecho también la normalización de las imagenes sin hacer el aumento de datos, los datos para validación y pruebas no deben pasar por el proceso de aumento de datos porque presisamente queremos probar nuestro modelo sobre los datos que nos fueron dados en un principio, de lo contrario nuestro accuracy estaría viciado.
def get_test_and_valid_generator(train_dir,test_dir,valid_dir,
batch_size=15,target_w = 299, target_h = 299):
"""
Return generator for validation set and test test set using
normalization.
Args:
train_dir,test_dir,val_dir (str): directory where image files are held.
batch_size (int): images per batch to be fed into model during training.
seed (int): random seed.
target_w (int): final width of input images.
target_h (int): final height of input images.
Returns:
test_generator (DataFrameIterator) and valid_generator: iterators over test set and validation set respectively
"""
print("getting test and valid generators...")
# fit mean and std for test set generator
image_generator = ImageDataGenerator(samplewise_center=True,
samplewise_std_normalization= True)
# get test generator
valid_generator = image_generator.flow_from_directory(directory=val
id_dir, class_mode="binary",batch_size=1,shuffle=False,seed=42,
target_size=(target_w, target_h))
test_generator = image_generator.flow_from_directory(directory=test
_dir,class_mode="binary",batch_size=batch_size,shuffle=True,
seed=42,target_size=(target_w, target_h))
return valid_generator, test_generatorLlamamos las funciones creadas arriba para obtener nuestros dataset de trabajo.
train_it = get_train_generator(train_dir)
val_it,test_it=get_test_and_valid_generator(train_dir,test_dir,val_dir)
getting train generator...
Found 5216 images belonging to 2 classes.
getting test and valid generators...
Found 16 images belonging to 2 classes.
Found 624 images belonging to 2 classes.Cálculo de los pesos para las clases positiva y negativa a usar en la función de Loss balanceado
Una vez que tenemos nuestros datos listos para entrenamiento recordemos que debemos hallar los pesos de las clases para incluirlos como parámetros en la función ajuste del modelo ( model.fit() )
Para calcular los pesos simplemente calcularemos la frecuencia de cada clase, luego la frecuencia de los casos positivos será el peso de los casos negativos y viceversa.
# N is the number of patients
N = len(train_it.labels)
freq_pos= sum(train_it.labels)/N # Frecuency of positive cases
freq_neg = (N-sum(train_it.labels))/N # Frecuency of negative casespos_weight = freq_neg
neg_weight = freq_posAl multiplicar la cantidad de casos en cada clase por su correspondiente peso, vemos que las clases queda balanceadas!
plt.bar([1,2],np.array([len(list_train_pneu)*pos_weight,len(list_train_normal)*neg_weight]),color=['gray','purple'], alpha=0.5)
plt.title('Weighted Neumonia vs normal training')
plt.xticks([1,2],('Neumonia','normal'))
plt.ylabel('Count')
😍 ahora si! podemos entrenar. Pero antes, necesitamos una arquitectura de Deep Learning para entrenar no?
Instalación de la arquitectura de Transfer Learning EfficientNet, y construcción de la red neural
Vamos a pip instalar la arquitectura EficientNet desde el repositorio de Github
!pip install -U git+https://github.com/qubvel/efficientnetUna vez instalada, creamos la base para nuestra red neural con la arquitectura EfficientNet-B0, no usé otras arquitecturas mas avanzadas porque Google Colab colapsaba y no queremos eso 😅. Así que nos quedaremos con la más modesta versión B0.
import efficientnet.keras as efn
conv_base = efn.EfficientNetB0(weights='imagenet',include_top=False,input_shape=(299, 299, 3))
print(conv_base.summary())A esta arquitectura base le conectaremos una capa "flattern" para convertir los datos de salida en un vector de features, una capa Dense de 200 neuronas (fué la que me funcionó mejor) y la capa de salida de una neurona, ya que usaremos clasificación binaria con activación "sigmoide" y loss "binary crossentropy".
Las capas de dropout evitan que se tenga overfiting en el modelo y el optimizador que usaremos será "Adam" con learning rate lr=0.0001 (también porque fue el que mejor funcionó)
from keras.models import Sequential
from keras.layers import Dense
from keras.layers import Flatten
from keras.layers import Dropout
from keras import regularizers, losses, metrics
from keras.optimizers import Adam
model = Sequential()
model.add(conv_base)
model.add(Flatten())
model.add(Dropout(0.2))
model.add(Dense(200, activation='relu',kernel_regularizer=regularizers.l2(0.001)))
model.add(Dropout(0.2))
model.add(Dense(1, activation='sigmoid'))
model.compile(optimizer=Adam(lr=0.0001),
loss='binary_crossentropy',
metrics=['accuracy'])
print("model compiled")
print(model.summary())Al igual que la vez pasada, incluiremos un monitor de entrenamiento para que el learning rate sea ajustado cada vez que el loss no mejore para dos épocas (epoch) consecutivas.
from keras.callbacks import EarlyStopping, ModelCheckpoint, ReduceLROnPlateau
reduce_lr = ReduceLROnPlateau(
monitor='val_loss',
factor=0.5,
patience=2,
verbose=1,
mode='auto',
min_lr=0.000001)
early_stopping = EarlyStopping(
monitor='val_loss',
patience=10,
verbose=1,
mode='auto')
model_checkpoint = ModelCheckpoint(
filepath='weights.h5',
monitor='val_loss',
verbose=1,
save_best_only=True,
save_weights_only=True,
mode='auto')y para poder introducir los pesos en el entrenamiento debemos primero crear un diccionario de pesos
weights = {0:neg_weight,1:pos_weight}Entrenamiento del modelo
Veremos como solo 3 épocas logran un resultado muy bueno
# training
history = model.fit(train_it,
epochs=3,
class_weight=weights,
steps_per_epoch=train_it.n//train_it.batch_size,
validation_data = test_it,
validation_steps=test_it.n//test_it.batch_size,
callbacks=[reduce_lr, early_stopping, model_checkpoint])Hemos creado nuestro modelo de manera que guarda en un archivo .h5 el modelo con menor loss, así que debemos cargar el "mejor modelo" de nuevo en nuestro espacio de tabajo con la función model.load_weights para poder evaluarlo. Lo guardaremos nuevamente como un archivo .h5 con un nombre personalizado.
model.load_weights('weights.h5')
model.save('pneumonia-chest-x-ray-cnn.h5')Validación del modelo
Miremos como se comportaron el accuracy y el loss durante el entrenamiento
import sys
# plot diagnostic learning curves
def summarize_diagnostics(history):
# plot loss
plt.subplot(211)
plt.title('Cross Entropy Loss')
plt.plot(history.history['loss'], color='blue', label='train')
plt.plot(history.history['val_loss'], color='orange', label='test')
# plot accuracy
plt.subplot(212)
plt.title('Classification Accuracy')
plt.plot(history.history['accuracy'], color='blue', label='train')
plt.plot(history.history['val_accuracy'], color='orange', label='test')
plt.tight_layout()
plt.show()
plt.close()
summarize_diagnostics(history)En amarillo observamos los valores para test y en azul para training
Ahora evaluemos nuestro modelo sobre el conjunto de validación, este paso es importante porque nuestra función fue optimizada eligiendo el loss mínimo para el conjunto de test, mientras que el conjunto de validación no jugó ningún papel en el entrenamiento, por lo tanto es un conjunto de datos que nos dice como se comportará el modelo fuera de un ambiente controlado.
val_it.reset()
pred=model.predict_generator(val_it,
steps=val_it.n,
verbose=1)labels = (val_it.class_indices)
filenames=val_it.filenames
labels = dict((v,k) for k,v in labels.items())
predicted_class = pd.DataFrame(predicted_class_indices).replace(labels)
filenames=pd.DataFrame(filenames)
results=pd.concat([filenames,predicted_class],axis=1)
results.columns=['Filenames','Predictions']
results
Obtuvimos un resultado perfecto en el conjunto de validación!! Pero eso no es todo, nuestro loss y accuracy en el conjunto de testing mejoró con respecto a primer post, nuestro accuracy ahora es de 95% y nuestro loss es de 0.2.
Hasta pronto! 😛