{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# Image Segmentation\nAn image is a collection or set of different pixels. We group together the pixels that have similar attributes using image segmentation.. Thus, the task of image segmentation is to train a neural network to output a pixel-wise mask of the image. This helps in understanding the image at a much lower level, i.e., the pixel level.In image segmetation each pixel is given a label.","metadata":{}},{"cell_type":"markdown","source":"Before diving into code, first we will need to import all the required libraries","metadata":{}},{"cell_type":"markdown","source":"# Import libraries","metadata":{}},{"cell_type":"code","source":"import numpy as np \nimport pandas as pd \nimport tensorflow as tf\nfrom zipfile import ZipFile \nimport keras.backend as K\nfrom sklearn.model_selection import train_test_split\nimport matplotlib.pyplot as plt\nimport os","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Dataset Preparation","metadata":{}},{"cell_type":"markdown","source":"Let's unzip all the files","metadata":{}},{"cell_type":"code","source":"train_zip = \"/kaggle/input/carvana-image-masking-challenge/train.zip\"\nwith ZipFile(train_zip, 'r') as zip_: \n    zip_.extractall('/kaggle/working')","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_mask_zip = \"/kaggle/input/carvana-image-masking-challenge/train_masks.zip\"\nwith ZipFile(train_mask_zip, 'r') as zip_: \n    zip_.extractall('/kaggle/working')","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(\"Train set:  \", len(os.listdir(\"/kaggle/working/train\")))\nprint(\"Train masks:\", len(os.listdir(\"/kaggle/working/train_masks\")))","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"car_ids = []\npaths = []\nfor dirname, _, filenames in os.walk('/kaggle/working/train'):\n    for filename in filenames:\n        path = os.path.join(dirname, filename)    \n        paths.append(path)\n        \n        car_id = filename.split(\".\")[0]\n        car_ids.append(car_id)\n\nd = {\"id\": car_ids, \"car_path\": paths}\ndf = pd.DataFrame(data = d)\ndf = df.set_index('id')\ndf","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"car_ids = []\nmask_path = []\nfor dirname, _, filenames in os.walk('/kaggle/working/train_masks'):\n    for filename in filenames:\n        path = os.path.join(dirname, filename)\n        mask_path.append(path)\n        \n        car_id = filename.split(\".\")[0]\n        car_id = car_id.split(\"_mask\")[0]\n        car_ids.append(car_id)\n\n        \nd = {\"id\": car_ids,\"mask_path\": mask_path}\nmask_df = pd.DataFrame(data = d)\nmask_df = mask_df.set_index('id')\nmask_df","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"df[\"mask_path\"] = mask_df[\"mask_path\"]\ndf","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"Now we will perform a simple augmentation of flipping an image and then normalize the image pixel in between 0 and 1","metadata":{}},{"cell_type":"code","source":"img_size = [256,256]\n\ndef data_augmentation(car_img, mask_img):\n\n    if tf.random.uniform(()) > 0.5:\n        car_img = tf.image.flip_left_right(car_img)\n        mask_img = tf.image.flip_left_right(mask_img)\n\n    return car_img, mask_img\n\ndef preprocessing(car_path, mask_path):\n    car_img = tf.io.read_file(car_path) \n    car_img = tf.image.decode_jpeg(car_img, channels=3)\n    car_img = tf.image.resize(car_img, img_size)\n    car_img = tf.cast(car_img, tf.float32) / 255.0\n    \n    mask_img = tf.io.read_file(mask_path)\n    mask_img = tf.image.decode_jpeg(mask_img, channels=3)\n    mask_img = tf.image.resize(mask_img, img_size)\n    mask_img = mask_img[:,:,:1]    \n    mask_img = tf.math.sign(mask_img)\n    \n    \n    return car_img, mask_img\n\ndef create_dataset(df, train = False):\n    if not train:\n        ds = tf.data.Dataset.from_tensor_slices((df[\"car_path\"].values, df[\"mask_path\"].values))\n        ds = ds.map(preprocessing, tf.data.AUTOTUNE)\n    else:\n        ds = tf.data.Dataset.from_tensor_slices((df[\"car_path\"].values, df[\"mask_path\"].values))\n        ds = ds.map(preprocessing, tf.data.AUTOTUNE)\n        ds = ds.map(data_augmentation, tf.data.AUTOTUNE)\n\n    return ds","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"Now we will split the dataset into train and test","metadata":{}},{"cell_type":"code","source":"train_df, valid_df = train_test_split(df, random_state=42, test_size=.25)\ntrain = create_dataset(train_df, train = True)\nvalid = create_dataset(valid_df)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"TRAIN_LENGTH = len(train_df)\nBATCH_SIZE = 16\nBUFFER_SIZE = 1000","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_dataset = train.cache().shuffle(BUFFER_SIZE).batch(BATCH_SIZE).repeat()\ntrain_dataset = train_dataset.prefetch(buffer_size=tf.data.AUTOTUNE)\nvalid_dataset = valid.batch(BATCH_SIZE)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"Let's look the image and it's corresponding mask","metadata":{}},{"cell_type":"code","source":"def display(display_list):\n    plt.figure(figsize=(15, 15))\n\n    title = ['Input Image', 'True Mask', 'Predicted Mask']\n\n    for i in range(len(display_list)):\n        plt.subplot(1, len(display_list), i+1)\n        plt.title(title[i])\n        plt.imshow(tf.keras.preprocessing.image.array_to_img(display_list[i]))\n        plt.axis('off')\n    plt.show()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"for i in range(5):\n   for image, mask in train.take(i):\n        sample_image, sample_mask = image, mask\n        display([sample_image, sample_mask])","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Model","metadata":{}},{"cell_type":"markdown","source":"We are going to use U-Net model. A U-Net consists of an encoder (downsampler) and decoder (upsampler). In-order to learn robust features, and reduce the number of trainable parameters, a pretrained model can be used as the encoder.The encoder will be a pretrained MobileNetV2 model which is prepared and ready to use in tf.keras.applications. ","metadata":{}},{"cell_type":"code","source":"base_model = tf.keras.applications.MobileNetV2(input_shape=[256, 256, 3], include_top=False)\n\n# Use the activations of these layers\nlayer_names = [\n    'block_1_expand_relu',   # 64x64\n    'block_3_expand_relu',   # 32x32\n    'block_6_expand_relu',   # 16x16\n    'block_13_expand_relu',  # 8x8\n    'block_16_project',      # 4x4\n]\nbase_model_outputs = [base_model.get_layer(name).output for name in layer_names]\n\n# Create the feature extraction model\ndown_stack = tf.keras.Model(inputs=base_model.input, outputs=base_model_outputs)\ndown_stack.trainable = False","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def upsample(filters, size, norm_type='batchnorm', apply_dropout=False):\n    initializer = tf.random_normal_initializer(0., 0.02)\n    \n    result = tf.keras.Sequential()\n    result.add(\n      tf.keras.layers.Conv2DTranspose(filters, size, strides=2,\n                                      padding='same',\n                                      kernel_initializer=initializer,\n                                      use_bias=False))\n\n    if norm_type.lower() == 'batchnorm':\n        result.add(tf.keras.layers.BatchNormalization())\n    elif norm_type.lower() == 'instancenorm':\n        result.add(InstanceNormalization())\n\n    if apply_dropout:\n        result.add(tf.keras.layers.Dropout(0.5))\n\n        result.add(tf.keras.layers.ReLU())\n\n    return result\n\nup_stack = [\n    upsample(512, 3),  # 4x4 -> 8x8\n    upsample(256, 3),  # 8x8 -> 16x16\n    upsample(128, 3),  # 16x16 -> 32x32\n    upsample(64, 3),   # 32x32 -> 64x64\n]","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def unet_model(output_channels):\n    inputs = tf.keras.layers.Input(shape=[256, 256, 3])\n\n    # Downsampling through the model\n    skips = down_stack(inputs)\n    x = skips[-1]\n    skips = reversed(skips[:-1])\n\n  # Upsampling and establishing the skip connections\n    for up, skip in zip(up_stack, skips):\n        x = up(x)\n        concat = tf.keras.layers.Concatenate()\n        x = concat([x, skip])\n\n  # This is the last layer of the model\n    last = tf.keras.layers.Conv2DTranspose(\n      output_channels, 3, strides=2, activation='sigmoid',\n      padding='same')  #64x64 -> 128x128\n\n    x = last(x)\n\n    return tf.keras.Model(inputs=inputs, outputs=x)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Train the Model","metadata":{}},{"cell_type":"markdown","source":"Now let's compile the model and see the model architecture","metadata":{}},{"cell_type":"code","source":"def dice_coef(y_true, y_pred, smooth=1):\n    intersection = K.sum(y_true * y_pred, axis=[1,2,3])\n    union = K.sum(y_true, axis=[1,2,3]) + K.sum(y_pred, axis=[1,2,3])\n    return K.mean( (2. * intersection + smooth) / (union + smooth), axis=0)\n\ndef dice_loss(in_gt, in_pred):\n    return 1-dice_coef(in_gt, in_pred)\n\nmodel = unet_model(1)\n\nmodel.compile(optimizer='adam',\n              loss = dice_loss,\n              metrics=[dice_coef,'binary_accuracy'])\n\ntf.keras.utils.plot_model(model, show_shapes=True)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"Let's try out the model to see what it predicts before training.","metadata":{}},{"cell_type":"code","source":"for images, masks in train_dataset.take(1):\n    for img, mask in zip(images, masks):\n        sample_image = img\n        sample_mask = mask\n        break\ndef visualize(display_list):\n    plt.figure(figsize=(15, 15))\n    title = ['Input Image', 'True Mask', 'Predicted Mask']\n    for i in range(len(display_list)):\n        plt.subplot(1, len(display_list), i+1)\n        plt.title(title[i])\n        plt.imshow(tf.keras.preprocessing.image.array_to_img(display_list[i]))\n        plt.axis('off')\n    plt.show()\n\ndef show_predictions(sample_image, sample_mask):\n    pred_mask = model.predict(sample_image[tf.newaxis, ...])\n    pred_mask = pred_mask.reshape(img_size[0],img_size[1],1)\n    visualize([sample_image, sample_mask, pred_mask])\n    \nshow_predictions(sample_image, sample_mask)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"model.summary()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"Let's observe how the model improves while it is training. To accomplish this task, a callback function is defined below.","metadata":{}},{"cell_type":"code","source":"early_stop = tf.keras.callbacks.EarlyStopping(patience=4,restore_best_weights=True)\n\nclass DisplayCallback(tf.keras.callbacks.Callback):\n    def on_epoch_begin(self, epoch, logs=None):\n        if (epoch + 1) % 3 == 0:\n            show_predictions(sample_image, sample_mask)\nEPOCHS = 15\nSTEPS_PER_EPOCH = TRAIN_LENGTH // BATCH_SIZE\n\nmodel_history = model.fit(train_dataset, epochs=EPOCHS,\n                          steps_per_epoch=STEPS_PER_EPOCH,\n                          validation_data=valid_dataset,\n                          callbacks=[DisplayCallback(), early_stop])","metadata":{"trusted":true},"execution_count":null,"outputs":[]}]}