{"cells":[{"metadata":{},"cell_type":"markdown","source":"#### This notebook implements and displays use of Random Cutout for image augmentation using new keras preprocessing layer. Adding the preprocessing layers helps in keeping the pipeline clean.\n\n> If this helps you, please do upvote :)"},{"metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"cell_type":"code","source":"import pandas as pd\nimport matplotlib\nimport matplotlib.pyplot as plt\nimport numpy as np\nimport seaborn as sns\nfrom sklearn.model_selection import train_test_split\nimport tensorflow as tf\nfrom tensorflow import keras\nimport tensorflow.keras.preprocessing as pp\nfrom tensorflow.keras import layers\nimport os \nimport random\n\n%matplotlib inline\nmatplotlib.style.use('ggplot')\n%config InlineBackend.figure_format = 'retina'\nsns.set(font_scale=1.5)","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":true},"cell_type":"code","source":"df = pd.read_csv('../input/cassava-leaf-disease-classification/train.csv')\nIMAGE_FOLDER='../input/cassava-leaf-disease-classification/train_images/'\n\nRANDOM_STATE=42\n\nfig_size=(15,8)\nsns.set(rc={'figure.figsize':fig_size})\ndef random_seed():\n    random.seed(RANDOM_STATE)\n    os.environ['PYTHONHASHSEED'] = str(RANDOM_STATE)\n    np.random.seed(RANDOM_STATE)\n    tf.random.set_seed(RANDOM_STATE)\n\nrandom_seed()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"AUTOTUNE = tf.data.experimental.AUTOTUNE\nSHUFFLE_BUFFER_SIZE = 2000","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"df['path'] = df['image_id'].apply(lambda x: IMAGE_FOLDER+x)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"X = df['path'].values\ny = df['label'].values","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def prepare_datasets(X, y, test_size, validation_size):\n    # create train, validation and test split\n    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=test_size,stratify=y)\n    X_train, X_validation, y_train, y_validation = train_test_split(X_train, y_train, test_size=validation_size, stratify=y_train)\n\n    return X_train, X_validation, X_test, y_train, y_validation, y_test","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Splitting the data in train, validation and test set"},{"metadata":{"trusted":true},"cell_type":"code","source":"X_train, X_validation, X_test, y_train, y_validation, y_test = prepare_datasets(X, y, 0.05, 0.15)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# https://cs230.stanford.edu/blog/datapipeline/\ndef load_image(image_path,label):\n    image = tf.io.read_file(image_path)\n    #Don't use tf.image.decode_image, or the output shape will be undefined\n    image = tf.image.decode_jpeg(image, channels=3)\n    return image, label","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def get_data_set(files, labels, batch_size, is_training=False):\n    dataset = tf.data.Dataset.from_tensor_slices((files, labels))\n    dataset = dataset.map(load_image, num_parallel_calls=AUTOTUNE)\n    if is_training:\n        dataset = dataset.shuffle(SHUFFLE_BUFFER_SIZE)\n    dataset = dataset.batch(batch_size)\n    dataset = dataset.prefetch(AUTOTUNE)\n    return dataset","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"sample_image_dataset = get_data_set(X_train, y_train, 32, is_training=True)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"sample_img_batch, sample_label_batch = next(iter(sample_image_dataset))\nprint(sample_img_batch.shape) \nprint(sample_label_batch.shape) ","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"import tensorflow_addons as tfa\nfrom tensorflow.python.keras.engine.input_spec import InputSpec\nfrom tensorflow.python.keras.engine.base_preprocessing_layer import PreprocessingLayer\nfrom tensorflow.python.keras.utils import tf_utils\n\nclass RandomCutout(PreprocessingLayer):\n    \"\"\"Creates random mask on the image.\n  Input shape:\n    4D tensor with shape:\n    `(samples, height, width, channels)`, data_format='channels_last'.\n  Output shape:\n    4D tensor with shape:\n    `(samples, height, width, channels)`, data_format='channels_last'.\n  Attributes:\n    mask: A tuple or a list with two values `mask-height` and `mask-width`.\n    seed: Integer. Used to create a random seed.\n    name: A string, the name of the layer.\n  Raise:\n    ValueError: if mask is not a list or tuple of two values.\n    InvalidArgumentError: if mask_size (mask_height x mask_width) can't be divisible by 2. \n  \"\"\"\n    def __init__(self, mask, seed=None, name=None, **kwargs):\n        self.mask = mask\n        if isinstance(mask, (tuple, list)) and len(mask) == 2:\n            self.lower = mask[0]\n            self.upper = mask[1]\n            \n        else:\n            raise ValueError('RandomCutout layer {name} received an invalid mask '\n                       'argument {arg}. only list or touple of size 2 should be passed'.format(name=name, arg=mask))\n\n        self.seed = seed\n        self.input_spec = InputSpec(ndim=4)\n        super(RandomCutout, self).__init__(name=name, **kwargs)\n\n    def call(self, inputs, training=True):\n        if training is None:\n            training = K.learning_phase()\n\n        def random_cutout_inputs():\n            return tfa.image.random_cutout(inputs, (self.lower, self.upper), constant_values = 0)\n\n        output = tf_utils.smart_cond(training, random_cutout_inputs,\n                                              lambda: inputs)\n        output.set_shape(inputs.shape)\n        return output\n\n    def compute_output_shape(self, input_shape):\n        return input_shape\n\n    def get_config(self):\n        config = {\n            'mask': self.mask,\n            'seed': self.seed,\n        }\n        \n        base_config = super(RandomCutout, self).get_config()\n        return dict(list(base_config.items()) + list(config.items()))","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"This class is for `TF 2.4.0`"},{"metadata":{"trusted":true},"cell_type":"code","source":"\n\nimport tensorflow_addons as tfa\nfrom tensorflow.python.keras.engine.input_spec import InputSpec\nfrom tensorflow.python.keras.engine.base_preprocessing_layer import PreprocessingLayer\nfrom tensorflow.python.keras.engine import base_preprocessing_layer\nfrom tensorflow.python.keras.utils import control_flow_util\n\nclass RandomCutout(PreprocessingLayer):\n    \"\"\"Creates random mask on the image.\n  Input shape:\n    4D tensor with shape:\n    `(samples, height, width, channels)`, data_format='channels_last'.\n  Output shape:\n    4D tensor with shape:\n    `(samples, height, width, channels)`, data_format='channels_last'.\n  Attributes:\n    mask: A tuple or a list with two values `mask-height` and `mask-width`.\n    seed: Integer. Used to create a random seed.\n    name: A string, the name of the layer.\n  Raise:\n    ValueError: if mask is not a list or tuple of two values.\n    InvalidArgumentError: if mask_size (mask_height x mask_width) can't be divisible by 2. \n  \"\"\"\n    def __init__(self, mask, seed=None, name=None, **kwargs):\n        self.mask = mask\n        if isinstance(mask, (tuple, list)) and len(mask) == 2:\n            self.lower = mask[0]\n            self.upper = mask[1]\n            \n        else:\n            raise ValueError('RandomCutout layer {name} received an invalid mask '\n                       'argument {arg}. only list or touple of size 2 should be passed'.format(name=name, arg=mask))\n\n        self.seed = seed\n        self.input_spec = InputSpec(ndim=4)\n        super(RandomCutout, self).__init__(name=name, **kwargs)\n        base_preprocessing_layer._kpl_gauge.get_cell('V2').set('RandomCutout')\n\n    def call(self, inputs, training=True):\n        if training is None:\n            training = K.learning_phase()\n\n        def random_cutout_inputs():\n            return tfa.image.random_cutout(inputs, (self.lower, self.upper), constant_values = 0)\n\n        output = control_flow_util.smart_cond(training, random_cutout_inputs,\n                                              lambda: inputs)\n        output.set_shape(inputs.shape)\n        return output\n\n    def compute_output_shape(self, input_shape):\n        return input_shape\n\n    def get_config(self):\n        config = {\n            'mask': self.mask,\n            'seed': self.seed,\n        }\n        \n        base_config = super(RandomCutout, self).get_config()\n        return dict(list(base_config.items()) + list(config.items()))","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"### Setting up the preprocessing and augmentation pipeline"},{"metadata":{"trusted":true},"cell_type":"code","source":"img_size = 512\nimg_preprocessing = tf.keras.Sequential([\n  layers.experimental.preprocessing.Resizing(img_size, img_size),\n  layers.experimental.preprocessing.Rescaling(1./255)\n],\nname= \"img_preprocessing\")\n\nimg_augmentation = tf.keras.Sequential(\n    [\n        RandomCutout(mask=(80,80))\n    ],\n    name=\"img_augmentation\",\n)\n\nimp_preprocessing_and_augmentation = tf.keras.Sequential([\n    img_preprocessing,\n    img_augmentation\n])\n","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Applying the image preprocessing and augmentation to each of the images in the batch"},{"metadata":{"trusted":true},"cell_type":"code","source":"for n in range(32):\n    plt.rcParams['figure.figsize'] = (25,15)\n    ax = plt.subplot(4, 8, n+1)\n    augmented_image = imp_preprocessing_and_augmentation(tf.expand_dims(sample_img_batch[n], 0))\n    plt.imshow(augmented_image[0].numpy())\n    plt.title(sample_label_batch[n].numpy())\n    plt.axis('off')","execution_count":null,"outputs":[]}],"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":4,"nbformat_minor":4}