text_plain_output_1.png
from IPython.display import clear_output

from keras.layers import (Activation, Input, MaxPooling2D, BatchNormalization,

from keras.models import Model

from keras.preprocessing import image

from pathlib import Path

from tensorflow.keras.utils import plot_model

import keras

import matplotlib.pyplot as plt

import numpy as np # linear algebra

import tensorflow as tf

img_root_path = Path('../input/chest-xray-masks-and-labels/Lung Segmentation/CXR_png/')
mask_root_path = Path('../input/chest-xray-masks-and-labels/Lung Segmentation/masks')
test_root_path = Path('../input/chest-xray-masks-and-labels/Lung Segmentation/test')
img_paths = sorted(img_root_path.iterdir())
mask_paths = sorted(mask_root_path.iterdir())
test_paths = sorted(test_root_path.iterdir())

images = []
masks = []
test_images = []
image_size = (256, 256)
for img, mask, test_img in zip(img_paths, mask_paths, test_paths):
    images.append(image.load_img(img, target_size=image_size))
    masks.append(image.load_img(mask, target_size=image_size))
    test_images.append(image.load_img(test_img, target_size=image_size))

X = []
for img in images:
    x = image.img_to_array(img)
    X.append(x)
X = np.array(X)

y = []
for mask in masks:
    y.append(image.img_to_array(mask))
y = np.array(y)

import tensorflow as tf
y = np.where(y[..., 0] > 0, 1, 0)[..., tf.newaxis]

(X_train.shape, y_train.shape, X_test.shape, y_test.shape)

sample_image = X_test[0]
sample_mask = y_test[0]

import keras
from IPython.display import clear_output

def show_predictions(model):
    pred_mask = model.predict(sample_image[None])
    print(pred_mask.shape)

    fig, ax = plt.subplots(1, 3, figsize=(15, 8))

    ax[0].imshow(sample_image)
    ax[1].imshow(sample_mask[..., 0])
    ax[2].imshow(np.squeeze(pred_mask, axis=0))
    plt.show()

class DisplayCallback(keras.callbacks.Callback):
    def __init__(self, patience=1):
        super().__init__()
        self.patience = patience
    
    def on_train_begin(self, logs=None):
        self.wait = 0

    def on_epoch_end(self, epoch, logs=None):
        self.wait += 1
        if self.wait >= self.patience:
            clear_output(wait=True)
            show_predictions(self.model)
            print(f'\nSample Prediction after epoch {epoch+1}')
            self.wait = 0

from keras.models import Model
from keras.layers import Activation, Input, MaxPooling2D, BatchNormalization, Conv2D, Conv2DTranspose, concatenate, Dropout
from tensorflow.keras.utils import plot_model
def Unet(num_classes=1, input_shape=(256, 256, 3), start_num_masks=64, deep=4, mask_size=(3, 3), conv_padding='same', input_layers=None, skip_connection=None, only_layers=False):
    if not input_layers:
        skip_connection = []
        img = Input(shape=input_shape)
        current_num_masks = start_num_masks
        x = img
        for i in range(deep):
            x = Conv2D(current_num_masks, mask_size, padding='same', name=f'bl_{i + 1}_conv1')(x)
            x = BatchNormalization()(x)
            x = Activation('relu')(x)
            x = Conv2D(current_num_masks, mask_size, padding='same', name=f'bl_{i + 1}_conv2')(x)
            x = BatchNormalization()(x)
            x = Activation('relu')(x)
            skip_connection.append(x)
            x = MaxPooling2D()(x)
            current_num_masks = current_num_masks * 2
        current_num_masks = current_num_masks / 2
    else:
        current_num_masks = sum([start_num_masks for _ in range(4)])
        for layer in input_layers.layers[:-1]:
            layer.trainable = False
        x = input_layers.layers[-1].output
        img = input_layers.inputs
    for i in range(deep):
        x = Conv2DTranspose(current_num_masks, (2, 2), strides=(2, 2), padding='same')(x)
        x = BatchNormalization()(x)
        x = Activation('relu')(x)
        x = concatenate([x, skip_connection[-(i + 1)]])
        x = Conv2D(current_num_masks, mask_size, padding='same')(x)
        x = BatchNormalization()(x)
        x = Activation('relu')(x)
        x = Conv2D(current_num_masks, mask_size, padding='same')(x)
        x = BatchNormalization()(x)
        x = Activation('relu')(x)
        current_num_masks = current_num_masks / 2
    if only_layers:
        return (img, x)
    x = Conv2D(num_classes, (3, 3), activation='sigmoid', padding='same')(x)
    model = Model(img, x)
    return model

model = Unet(start_num_masks=128, deep=5)

plot_model(model, show_shapes=True)