{"cells":[{"metadata":{},"cell_type":"markdown","source":"# SE-ResNetimage classification with 1 channel image with auxiliary freqency input\n- handle the long audio (> 2min) by spliting into clip of 1 minutes.\n- Select multiple local maximum on melspectrogram to propose 200 pixel by 200 pixel cropping windows from the spectrom by Short-time Fourier transform (STFT).\n- store the images in 128 pixel by 128 pixel\n- data augmentation\n- use iterator to feed data in order to reduce the memory load.\n- 3 x 1 filter to extract the feature at each stft bin\n- Sequence-Excitation-resnet for Channel Attention and Layer Attention ","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"# packages\n## system\nimport os, warnings, sys, pathlib\nfrom pathlib import Path\nwarnings.filterwarnings('ignore')\n\n## data structure\nimport pandas as pd\nimport numpy as np\nimport tensorflow as tf\nfrom tensorflow import keras\nimport tensorflow.keras.backend as K\nfrom tensorflow.python.keras.utils.data_utils import Sequence\n\n## utils\nimport math\nimport scipy\nimport scipy.ndimage as ndimage\nimport scipy.ndimage.filters as filters\nimport random\nimport datetime\nimport time\nfrom tqdm import tqdm\nfrom sklearn.pipeline import Pipeline\nfrom sklearn.utils import class_weight\n\n## image \nimport cv2\nfrom PIL import Image\nfrom skimage.transform import rotate\nimport imgaug.augmenters as iaa\n\n## audio\nimport soundfile as sf\nimport librosa\nfrom pydub import AudioSegment\n\n## data cleaning\nfrom sklearn.preprocessing import StandardScaler, normalize, LabelEncoder, OneHotEncoder, scale\nfrom sklearn.decomposition import PCA\nfrom sklearn.metrics import confusion_matrix, f1_score, ConfusionMatrixDisplay\n\n## graphing\nimport librosa.display\nimport seaborn as sns\nimport matplotlib.pyplot as plt\nfrom matplotlib.collections import PatchCollection\nfrom matplotlib.patches import Rectangle\n%matplotlib inline","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# print(tf.test.is_gpu_available(), tf.config.list_physical_devices('GPU'))","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# keras dataloader: 1-channel with aux on the fly","execution_count":null},{"metadata":{},"cell_type":"markdown","source":"## training data path","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"# get image path\n# define audio dataset path\n\nDATASET_AUDIO = \"train_audio/\"\nDATASET_SPEC = \"/kaggle/input/\" + \"birdsong-recognition/\" + DATASET_AUDIO\n\nprint(DATASET_SPEC)\n","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"## lable encoder ","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"# get index of label and freq \nlabel_names = sorted(set(os.listdir(DATASET_SPEC)))\n\n# onehot encode label and freq\nlabel_to_onehot = OneHotEncoder(sparse = False)\nlabel_to_onehot.fit(np.array(label_names).reshape(-1, 1))\nlabel_dict = dict(zip(label_names, label_to_onehot.fit_transform(np.array(label_names).reshape(-1, 1))))\n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"pd.DataFrame(label_dict)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"## freq encoder ","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"freq_bin = np.arange(0,5000)\ndef split(word): \n    return [char for char in word] \nfreq_bin_encode = [split(str(freq).zfill(4)) for freq in freq_bin]\nfreq_bin_encode = [[int(digit) for digit in freq]  for freq in freq_bin_encode]\nfreq_dict = dict(zip(list(freq_bin), freq_bin_encode))\n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"pd.DataFrame(freq_dict)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"## training image paths","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"# define audio dataset path\n\n# DATASET_AUDIO = \"train_audio_small_sort\"\n# DATASET_AUDIO = \"train_audio_medium_sort\"\n# DATASET_AUDIO = \"train_audio_large_sort\"\nDATASET_AUDIO = \"train_audio_sort\"\n\n# DATASET_SPEC = \"../input\" + \"/birdsong-recognition\" + \"/img_\" + DATASET_AUDIO # local\nDATASET_SPEC = \"../input\" + \"/img-train-audio-sort\" + \"/img_\" + DATASET_AUDIO # kaggle  \n\nprint(DATASET_SPEC)\ndata_dir = DATASET_SPEC\ndata_root = pathlib.Path(data_dir)\nall_image_path = data_root.rglob('*.jpg')\nall_image_path = [str(pathlib.Path(path)) for path in all_image_path]\n\nall_label = [path.split('/')[-2] for path in all_image_path]","execution_count":null,"outputs":[]},{"metadata":{"scrolled":true,"trusted":true},"cell_type":"code","source":"class DataGenerator(Sequence):\n    \"\"\"\n    https://bbs.cvmart.net/topics/1545\n    \"\"\"\n    def __init__(self, filepath, label_dict, freq_dict, batch_size=8, imgshape=(256, 472),\n                 n_channels=3, n_classes=13, shuffle=True, ):\n        # initiation method\n        self.filepath=filepath\n        self.batch_size = batch_size\n        self.imgshape = imgshape\n        self.n_channels = n_channels\n        self.shuffle = shuffle\n        \n        self.pathlist= [str(pathlib.Path(path)) for path in pathlib.Path(self.filepath).rglob('*.jpg')]\n        self.on_epoch_end()\n        self.label_dict = label_dict\n        self.freq_dict = freq_dict\n\n    def __getitem__(self, index):\n        # generate batch index\n        indexes = self.indexes[index * self.batch_size:(index + 1) * self.batch_size]\n        # generate list of batch index\n        batch_pathlist = [self.pathlist[k] for k in indexes]\n        # generate data\n        images = self._generate_images(batch_pathlist)\n        freqs = self._generate_freqs(batch_pathlist)\n        labels = self._generate_labels(batch_pathlist)\n        return (images, freqs), labels\n\n    def __len__(self):\n        # return the number of batch\n        return int(np.floor(len(self.pathlist) / self.batch_size))\n    \n    def _load_image(self, image_path):\n        def gasuss_noise(image, mean=0, var=0.01):\n            noise = np.random.normal(mean, var ** 0.5, image.shape)\n            out = image + noise\n            if out.min() < 0:\n                low_clip = -1.\n            else:\n                low_clip = 0.\n            out = np.clip(out, low_clip, 1.0)\n            return out\n        img = cv2.imread(image_path, cv2.IMREAD_GRAYSCALE)/255 # grey 1 channel\n        img = cv2.flip(img, flipCode= random.randint(-1,1)) # flip up or down\n#         img = rotate(img, angle=random.randint(-5,5), mode='reflect') # rotate\n        img = cv2.warpAffine(img,\n                             M = np.float32([[1, 0, random.randint(-28,28)],\n                                             [0, 1, random.randint(-28,28)]]),\n                             dsize = img.shape)\n        img = gasuss_noise(img, var = random.randint(1,10)/1000)\n        if self.imgshape != img.shape:\n            img = cv2.resize(img, self.imgshape)\n            \n        return np.expand_dims(img, -1)\n    \n    def _generate_images(self, batch_pathlist):\n        # generate images for a batach\n        images = np.empty((self.batch_size, *self.imgshape, self.n_channels))\n        for i, path in enumerate(batch_pathlist):\n            images[i,] = self._load_image(path)\n        return images\n\n    def _generate_labels(self, batch_pathlist):\n        # generate labels for a batch\n        labels = np.empty((self.batch_size, len(self.label_dict) ), dtype=int)\n        for i, path in enumerate(batch_pathlist):\n            # Store sample\n            labels[i,] = label_dict.get(path.split('/')[-2])\n        return labels\n    \n    def _generate_freqs(self, batch_pathlist):\n        # generatre freqs for a batch\n        freqs = np.empty((self.batch_size, len(self.freq_dict.get(0))), dtype=int)\n        # Generate data\n        for i, path in enumerate(batch_pathlist):\n            # Store sample\n            freqs[i,]= freq_dict.get(int(path.split('/')[-1].split('_')[1]))\n        return freqs\n\n    def on_epoch_end(self):\n        # update index at the end of each epoch\n        self.indexes = np.arange(len(self.pathlist))\n        if self.shuffle == True:\n            np.random.shuffle(self.indexes)\n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Parameters\nbatch_size = 32\nimgshape = (128,128)\nn_channels = 1\nparams = {'batch_size': batch_size,\n        'n_channels': n_channels,\n        'shuffle': True,\n        'label_dict': label_dict, \n        'freq_dict': freq_dict,\n        'imgshape': imgshape}\ntrain_filepath = DATASET_SPEC + \"/train\" # for local\nvalid_filepath = DATASET_SPEC + \"/val\" # for local\nall_filepath = DATASET_SPEC\n# Generators\ntrain_generator = DataGenerator(train_filepath, **params)\nvalid_generator = DataGenerator(valid_filepath, **params)\nall_generator = DataGenerator(all_filepath, **params)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# getting weight for the unbalance data\ncw = class_weight.compute_class_weight('balanced',\n                                                 np.unique(all_label),\n                                                 all_label)\ncw = dict(enumerate(cw))\n","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"## preview image","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"def getkeybyval(my_dict, val):\n    key_list = list(my_dict.keys()) \n    val_list = list(my_dict.values())\n    return key_list[val_list.index(list(val))]\n\ndef show_batch(image_batch, freq_batch, label_batch, label_to_onehot, freq_dict):\n    \"\"\"\n    https://stackoverflow.com/questions/60129658/def-show-batch-not-showing-my-train-images\n    \"\"\"\n    plt.figure(figsize=(20,20))\n    for n in range(25):\n        ax = plt.subplot(5,5,n+1)\n        plt.imshow(image_batch[n,:,:,0])\n        plt.title(\"class: {} ({})\".format(\n            label_to_onehot.inverse_transform([label_batch[n]]), \n            getkeybyval(freq_dict, freq_batch[n])))\n        plt.axis('off')","execution_count":null,"outputs":[]},{"metadata":{"scrolled":false,"trusted":true},"cell_type":"code","source":"i = 0\nfor image_freq, label in train_generator:    \n    print(\"[batch {}/{}] shape: [image={}, freq = {}], label = {}\".\n          format(i+1,len(train_generator), image_freq[0].shape, image_freq[1].shape, label.shape))\n    show_batch(image_freq[0], image_freq[1], label, label_to_onehot, freq_dict)\n    \n    i += 1\n    if i > 0:\n        break \n        ","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Model Outline","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"initializer = tf.keras.initializers.VarianceScaling()\nimages_shape = [128,128,1]\nfreqs_shape = len(freq_dict.get(0))\noutput_size = len(label_dict)\n\nprint(\"initializer: \", initializer)\nprint(\"images_shape: \", images_shape)\nprint(\"freqs_shape: \", freqs_shape)\nprint(\"output_size: \", output_size)","execution_count":null,"outputs":[]},{"metadata":{"scrolled":false,"trusted":true},"cell_type":"code","source":"# batchnormalization and activation\ndef BatchActivate(x):\n    x = keras.layers.BatchNormalization()(x)\n    x = keras.layers.Activation('relu')(x)\n    return x\n\ndef convolution2D_block(x, filters, kernel_size = (3,3), strides=(1,1),\n                        padding='same', BatchAct=True, initializer = tf.keras.initializers.VarianceScaling()):\n    x = keras.layers.Conv2D(filters = filters, strides=strides, padding=padding,\n                            kernel_size = kernel_size, kernel_initializer = initializer)(x)\n    if BatchAct == True:\n        x = BatchActivate(x)\n    return x\n# conv block\ndef convolution1D_block(x, filters, size, strides=(1,1), padding='same', BatchAct=True, initializer = tf.keras.initializers.VarianceScaling()):\n    x = keras.layers.Conv1D(filters, size, strides=strides, padding=padding, kernel_initializer = initializer)(x)\n    if BatchAct == True:\n        x = BatchActivate(x)\n    return x\n\ndef residual1D_block(x, filters, conv_num=3, activation=\"relu\", padding=\"same\"):\n    # Shortcut\n    s = keras.layers.Conv1D(filters, kernel_size = 1, padding=padding)(x)\n    for i in range(conv_num - 1):\n        x = keras.layers.Conv1D(filters, kernel_size= 3, padding=padding)(x)\n        x = keras.layers.Activation(activation)(x)\n    x = keras.layers.Conv1D(filters, 3, padding=padding)(x)\n    x = keras.layers.Add()([x, s])\n    x = keras.layers.Activation(activation)(x)\n    return keras.layers.MaxPool1D(pool_size=2, strides=2)(x)\n\n\ndef squeeze_excite_block(x, ratio=8):\n    '''\n    https://github.com/titu1994/keras-squeeze-excite-network\n    '''\n    init = x\n    filters = init.shape[3]\n    se_shape = (1, 1, filters)\n\n    se = keras.layers.GlobalAveragePooling2D()(x)\n    se = keras.layers.Dense(filters // ratio, activation='relu', kernel_initializer='he_normal', use_bias=False)(se)\n    se = keras.layers.Dense(filters, activation='sigmoid', kernel_initializer='he_normal', use_bias=False)(se)\n    se = keras.layers.Reshape(se_shape)(se)\n    x = keras.layers.multiply([x, se])\n    return x\n\ndef squeeze_excite_block1d(x, ratio=8):\n    '''\n    https://github.com/titu1994/keras-squeeze-excite-network\n    '''\n    init = x\n    filters = init.shape[1]\n    se = keras.layers.Dense(filters // ratio, activation='relu', kernel_initializer='he_normal', use_bias=False)(x)\n    se = keras.layers.Dense(filters, activation='sigmoid', kernel_initializer='he_normal', use_bias=False)(se)\n    x = keras.layers.multiply([x, se])\n    return x\n\n\ndef residual2D_block(x, filters = 128, conv_num=2, downscale_filters = 32, kernel_size = (3,1), batch_activate = True, initializer = tf.keras.initializers.VarianceScaling()):\n    # dimension reduction\n    if filters != x.shape[3]:\n        x = keras.layers.Conv2D(filters, kernel_size = (1,1), padding=\"same\", kernel_initializer = initializer)(x) \n        c = convolution2D_block(x, downscale_filters, kernel_size = kernel_size, initializer = initializer)\n    else:\n        c = convolution2D_block(x, downscale_filters, kernel_size = kernel_size, initializer = initializer)\n    # convolution2D block series\n    for i in range(conv_num - 2): \n        c = convolution2D_block(c, downscale_filters, kernel_size = kernel_size, initializer = initializer)\n    # restore dimension\n    c = keras.layers.Conv2D(filters, (1,1), padding=\"same\", kernel_initializer = initializer)(c) #\n    c = squeeze_excite_block(c)\n    x = keras.layers.Add()([c, x])\n    return x\n\n\ndef resNet_model(image_inputs, initializer = initializer, DropoutRatio = 0.5):\n    \n    # expanding receptive field\n    x = convolution2D_block(image_inputs, filters = 64, kernel_size = (3,3))\n    x = convolution2D_block(x, filters = 256, kernel_size = (3,3))\n    \n    # residual_block_sequence with short\n    x = keras.layers.MaxPooling2D((2, 2))(x)\n    s1 = x\n    x = residual2D_block(x, filters = 256, downscale_filters = 128, conv_num=3, kernel_size = (3,1), batch_activate = True, initializer = initializer)    \n#     x = keras.layers.MaxPooling2D((2, 2))(x)\n    s1 = keras.layers.Conv2D(256, (1,1), padding=\"same\", kernel_initializer = initializer)(s1)\n#     s1 = keras.layers.MaxPooling2D((2, 2))(s1)\n    x = residual2D_block(x, filters = 256, downscale_filters = 128, conv_num=3, kernel_size = (3,1), batch_activate = True, initializer = initializer)\n    x = keras.layers.Add()([s1, x])\n    \n    x = keras.layers.MaxPooling2D((2, 2))(x)\n    s2 = x\n    x = residual2D_block(x, filters = 256, downscale_filters = 128, conv_num=4, kernel_size = (3,3), batch_activate = True, initializer = initializer)\n#     x = keras.layers.MaxPooling2D((2, 2))(x)\n    s2 = keras.layers.Conv2D(256, (1,1), padding=\"same\", kernel_initializer = initializer)(s2)\n#     s2 = keras.layers.MaxPooling2D((2, 2))(s2)\n    x = residual2D_block(x, filters = 256, downscale_filters = 128, conv_num=4, kernel_size = (3,3), batch_activate = True, initializer = initializer)\n    x = keras.layers.Add()([s2, x])\n    \n    x = keras.layers.MaxPooling2D((2, 2))(x)\n    s3 = x\n    x = residual2D_block(x, filters = 256, downscale_filters = 128, conv_num=3, kernel_size = (3,1), batch_activate = True, initializer = initializer)\n#     x = keras.layers.MaxPooling2D((2, 2))(x)\n    s3 = keras.layers.Conv2D(256, (1,1), padding=\"same\", kernel_initializer = initializer)(s3)\n#     s3 = keras.layers.MaxPooling2D((2, 2))(s3)\n    x = residual2D_block(x, filters = 256, downscale_filters = 128, conv_num=3, kernel_size = (3,1), batch_activate = True, initializer = initializer)\n    x = keras.layers.Add()([s3, x])\n\n    x = keras.layers.MaxPooling2D((2, 2))(x)\n    s4 = x\n    x = residual2D_block(x, filters = 256, downscale_filters = 128, conv_num=3, kernel_size = (3,1), batch_activate = True, initializer = initializer)\n#     x = keras.layers.MaxPooling2D((2, 2))(x)\n    s4 = keras.layers.Conv2D(256, (1,1), padding=\"same\", kernel_initializer = initializer)(s4)\n#     s4 = keras.layers.MaxPooling2D((2, 2))(s4)\n    x = residual2D_block(x, filters = 256, downscale_filters = 128, conv_num=3, kernel_size = (3,1), batch_activate = True, initializer = initializer)\n    x = keras.layers.Add()([s4, x])\n\n    # output\n    x = keras.layers.AveragePooling2D(pool_size=(2, 2))(x)\n    x = keras.layers.Flatten()(x)\n    x = keras.layers.Dropout(DropoutRatio)(x)\n    model=tf.keras.models.Model(inputs=image_inputs,outputs=x)\n    return model\n\ndef build_model(input_shape, num_classes, DropoutRatio, initializer = tf.keras.initializers.VarianceScaling()):\n    # image decoding\n    image_inputs = keras.layers.Input(shape = input_shape[0], name = \"image_input_layer\")\n    resNet = resNet_model(image_inputs = image_inputs, initializer = initializer, DropoutRatio = DropoutRatio)\n    x = resNet.output\n    x = keras.layers.Dense(1024, activation='relu')(x)\n    x = squeeze_excite_block1d(x)\n#     x = keras.layers.Dropout(DropoutRatio)(x)\n    # merge freq\n    freq_inputs = keras.layers.Input(shape = input_shape[1], name = \"freq_input_layer\")\n    x = keras.layers.Concatenate(axis= 1)([x, freq_inputs])\n    # multilayer perceptron\n    x = keras.layers.Dropout(DropoutRatio)(x)\n    # output\n    outputs = tf.keras.layers.Dense(num_classes, activation=\"softmax\",\n                                    kernel_initializer = initializer, name=\"output\")(x) # or sigmoid\n    return tf.keras.Model(inputs=[image_inputs, freq_inputs], outputs=outputs)\n\n\nmodel = build_model(input_shape = (images_shape, freqs_shape), num_classes = output_size, initializer = initializer, DropoutRatio = 0)\nmodel.summary()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Model compilation","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"# Adaptive Moment Estimation (Adam) \nadam_learning_rate = 0.0001  # initial learning rate\nadam = keras.optimizers.Adam(learning_rate=adam_learning_rate)\n\n# RMSprop\nrmsprop_learning_rate = 0.01\nrmsprop = tf.keras.optimizers.RMSprop(learning_rate=rmsprop_learning_rate)\n\n# Adagrad\nadagrad_learning_rate = 0.01\nadagrad = tf.keras.optimizers.Adagrad(learning_rate=adagrad_learning_rate)\n\n# Stochastic gradient descent (sgd)\nsgd_learning_rate = 0.01 # initial learning rate\nsgd_decay_rate = 0.1\nsgd_momentum = 0.8\nsgd = keras.optimizers.SGD(learning_rate = sgd_learning_rate )\n\n# Adadelta\nadadelta_learning_rate=0.002\nadadelta_rho=0.99\nadadelta_epsilon=1e-07\nadadelta = tf.keras.optimizers.Adadelta(learning_rate = adadelta_learning_rate,\n                                        rho = adadelta_rho, \n                                        epsilon = adadelta_epsilon)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# compilation configuration\noptimizer = adadelta\n\n# loss = \"sparse_categorical_crossentropy\" # for integer encoding output\nloss = \"categorical_crossentropy\" # for onehot encoding output\n\nmetrics = [\"acc\"]","execution_count":null,"outputs":[]},{"metadata":{"scrolled":true,"trusted":true},"cell_type":"code","source":"# compile\nmodel.compile(optimizer = optimizer, \n              loss = loss, \n              metrics = metrics)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Trainning","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"# ModelCheckpoint\nmodel_path = \"./model/\"\nif not os.path.exists(model_path):\n    os.mkdir(model_path)\nmdlcheckpoint_cb = keras.callbacks.ModelCheckpoint(model_save_filename = \"model.h5\", monitor = \"val_accuracy\", \n                                                   filepath = model_path, save_best_only = True)\n\n# learning schedule\nlogdir = './logs'\nif not os.path.exists(logdir):\n    os.mkdir(logdir)\nlogdir = logdir + \"/\" + datetime.datetime.now().strftime(\"%Y%m%d_%H%M%S\")\nif not os.path.exists(logdir):\n    os.mkdir(logdir)\nfile_writer_path = logdir + \"/metrics\"\nif not os.path.exists(file_writer_path):\n    os.mkdir(file_writer_path)\nfile_writer = tf.summary.create_file_writer(file_writer_path)\nfile_writer.set_as_default()\ndef exp_decay(epoch, lr):\n    decay_rate = 0.07\n    decay_step = 40\n    if epoch % decay_step == 0 and epoch:\n        return lr * decay_rate\n    #tf.summary.scalar('learning rate', data=lr, step=epoch)\n    return lr\nlearningrate_cb = keras.callbacks.LearningRateScheduler(exp_decay)\n\n# TensorBoard\ntensorboard_cb = keras.callbacks.TensorBoard(logdir)\n\n# EarlyStopping\nearlystopping_cb = keras.callbacks.EarlyStopping(patience=15, restore_best_weights=True)\n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# traning configuration\nNUM_EPOCHS = 15\ncallbacks_list = [#earlystopping_cb,\n                  mdlcheckpoint_cb, \n                  tensorboard_cb,\n#                   learningrate_cb\n                 ]\n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"model.load_weights(\"../input/model-v2-e45/model_v2_e45.h5\", by_name=False)","execution_count":null,"outputs":[]},{"metadata":{"scrolled":false,"trusted":true},"cell_type":"code","source":"# Training\n!rm -rf ./logs/ # Clear any logs from previous runs\nhistory = model.fit(\n    x = train_generator,\n    epochs = NUM_EPOCHS,\n    validation_data = valid_generator,\n    callbacks = callbacks_list,\n    verbose = 1,\n    class_weight = cw\n)\n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"model.save(model_path + \"model_v2_e60.h5\")","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Evaluation","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"# load dataset and preprocess images（the preprocess function in map  can preprocess all previous pictures)\ndef load_and_preprocess_image(img_path):\n    # read pictures\n    img_raw = tf.io.read_file(img_path)\n    # decode pictures\n    img_tensor = tf.image.decode_jpeg(img_raw, channels=1) # change channels to 3 result in 3-channel image\n    # resize\n#     img_tensor = tf.image.resize(img_tensor, [128, 128])\n    #tf.cast() function is a type conversion function that converts the data format of x into dtype\n    img_tensor = tf.cast(img_tensor, tf.float32)\n    # normalization\n    img_tensor = img_tensor / 255.0\n    # flip left or right\n#     img_tensor = tf.image.random_flip_left_right(img_tensor)\n    return img_tensor\n\n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# define valid path\nvalid_root = pathlib.Path(valid_filepath)\nvalid_image_path = valid_root.rglob('*.jpg')\nvalid_image_path = [str(pathlib.Path(path)) for path in valid_image_path]\nprint(\"total valid: \", len(valid_image_path))\n\n# get index of label and freq \nvalid_freq = [int(path.split('/')[-1].split('_')[1]) for path in valid_image_path]\nvalid_label = [path.split('/')[-2] for path in valid_image_path]\noh_valid_freq = [freq_dict.get(freq) for freq in valid_freq]\noh_valid_label = label_to_onehot.transform(np.array(valid_label).reshape(-1, 1))\n\n# convert to tensor and zip\nvalid_image_ds = tf.data.Dataset.from_tensor_slices(valid_image_path).map(load_and_preprocess_image)\nvalid_freq_ds = tf.data.Dataset.from_tensor_slices(oh_valid_freq)\nvalid_label_ds = tf.data.Dataset.from_tensor_slices(oh_valid_label)\nvalid_ds = tf.data.Dataset.zip(((valid_image_ds, valid_freq_ds), valid_label_ds))\nvalid_ds = valid_ds.batch(len(valid_image_path)).repeat()\niteration_valid = iter(valid_ds)","execution_count":null,"outputs":[]},{"metadata":{"scrolled":false,"trusted":true},"cell_type":"code","source":"pred_prob = model.predict(iteration_valid.get_next())\npred_class_prob = np.amax(pred_prob, 1)\npred_class_integer = np.argmax(pred_prob, 1)\npred_class = [list(label_dict)[i] for i in pred_class_integer]\ncm = confusion_matrix(valid_label, pred_class)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"plt.figure(figsize=(15, 5))\nConfusionMatrixDisplay(cm, display_labels = list(label_dict.keys())).plot(cmap = \"BuGn\", \n                                                                          include_values= False)\nplt.xticks(rotation=45)","execution_count":null,"outputs":[]},{"metadata":{"scrolled":true,"trusted":true},"cell_type":"code","source":"df = pd.DataFrame(data=pred_prob, columns=list(label_dict.keys()))\ndf[\"actual_class\"] = valid_label\ndf = df.reset_index()\ndf = df.melt(id_vars = [\"index\", \"actual_class\"], var_name = \"pred_class\", value_name = \"probability\")\ndf = df.sort_values(by = ['index', 'actual_class']).reset_index(drop = True)\ngrouped_max = df[['index', 'probability']].groupby('index').max().reset_index().rename(columns={\"probability\": \"max_probability\"})\ndf = pd.merge(df, grouped_max, on='index')\ndf['max_probability'] = np.where(df['probability'] == df['max_probability'], True, False)\ndf = df[['index', 'actual_class', 'pred_class', 'max_probability', 'probability']]\n\ndf['correctly_pred'] = np.where((df['pred_class'] == df['actual_class']) & (df['max_probability'] == True), True, False)\ndf = df[(df['correctly_pred']==True) | (df['max_probability']==True) ]","execution_count":null,"outputs":[]},{"metadata":{"scrolled":false,"trusted":true},"cell_type":"code","source":"plt.figure(figsize=(15, 5))\nsns.boxplot(x=\"actual_class\", y=\"probability\", \n            hue = \"correctly_pred\", data = df)\nplt.legend(loc='upper left')\nplt.xticks(rotation=45)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"plt.figure(figsize=(15, 5))\nsns.countplot(x=\"actual_class\", hue = \"correctly_pred\", data = df)\nplt.legend(loc='upper left')\nplt.xticks(rotation=45)","execution_count":null,"outputs":[]},{"metadata":{"scrolled":true,"trusted":true},"cell_type":"code","source":"f1_score(valid_label, pred_class, average= \"weighted\")","execution_count":null,"outputs":[]},{"metadata":{"scrolled":true,"trusted":true},"cell_type":"code","source":"# %reload_ext tensorboard\n# %tensorboard --logdir {logdir}","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Grad-CAM and Guided Backpropagation","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"# test_loss, test_accuracy = model.evaluate(test_ds)\n# print('Test loss: {0:.2f}. Test accuracy: {1:.2f}%'.format(test_loss, test_accuracy*100.))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"from tensorflow.keras.models import Model\nfrom tensorflow.python.framework import ops\n\nclass GradCAM:\n    # Adapted with some modification from https://www.pyimagesearch.com/2020/03/09/grad-cam-visualize-class-activation-maps-with-keras-tensorflow-and-deep-learning/\n    def __init__(self, model, layerName=None):\n        \"\"\"\n        model: pre-softmax layer (logit layer)\n        \"\"\"\n        self.model = model\n        self.layerName = layerName\n\n    def compute_heatmap(self, input_ds, classIdx, upsample_size, eps=1e-5):\n        gradModel = Model(inputs=[self.model.inputs],\n                          outputs=[self.model.get_layer(self.layerName).output, self.model.output])\n        # record operations for automatic differentiation\n        with tf.GradientTape() as tape:\n            (convOuts, preds) = gradModel(input_ds)  # preds after softmax\n            loss = preds[:, classIdx]\n        # compute gradients with automatic differentiation\n        grads = tape.gradient(loss, convOuts)\n        \n        # discard batch\n        convOuts = convOuts[0]\n        grads = grads[0]\n        \n        # normalize grads\n        norm_grads = tf.divide(grads, tf.reduce_mean(tf.square(grads)) + tf.constant(eps))\n        \n        # compute weights\n        weights = tf.reduce_mean(norm_grads, axis=(0, 1))\n        cam = tf.reduce_sum(tf.multiply(weights, convOuts), axis=-1)\n        \n        # Apply reLU\n        cam = np.maximum(cam, 0)\n        cam = cam / np.max(cam)\n        cam = cv2.resize(cam, upsample_size, cv2.INTER_LINEAR)\n\n#         print(\"[convOuts] shape: {}, max: {}, min: {}\".format(convOuts.numpy().shape, convOuts.numpy().max(), convOuts.numpy().min()))\n#         print(\"[loss] shape: {}, max: {}, min: {}\".format(loss.numpy().shape, loss.numpy().max(), loss.numpy().min()))\n#         print(\"[grads] shape: {}, max: {}, min: {}\".format(grads.numpy().shape, grads.numpy().max(), grads.numpy().min()))\n        \n        return cam\ndef overlay_gradCAM(img, cam):\n    def remap(x, out_min, out_max):\n        return (x - x.min()) * (out_max - out_min) / (x.max() - x.min()) + out_min\n    cam = remap(cam, 0, 255)\n    cam = np.uint8(cam)\n    new_img = 0.3 * cam + 0.5 * img\n    new_img = np.array(new_img)\n    new_img = remap(new_img, 0, 255)\n    \n    return (new_img * 255.0 / new_img.max()).astype(\"uint8\")\n\ndef guidedRelu(x):\n    def grad(dy):\n        return tf.cast(dy>0,\"float32\") * tf.cast(x>0, \"float32\") * dy\n    return tf.nn.relu(x), grad\n\n# Reference: https://github.com/eclique/keras-gradcam with adaption to tensorflow 2.0  \nclass GuidedBackprop:\n    def __init__(self,model, layerName=None):\n        self.model = model\n        self.layerName = layerName\n        self.gbModel = self.build_guided_model()\n\n    def build_guided_model(self):\n        gbModel = Model(\n            inputs = [self.model.inputs],\n            outputs = [self.model.get_layer(self.layerName).output]\n        )\n        layer_dict = [layer for layer in gbModel.layers[1:] if hasattr(layer,\"activation\")]\n        for layer in layer_dict:\n            if layer.activation == tf.keras.activations.relu:\n                layer.activation = guidedRelu        \n        return gbModel\n    \n    def guided_backprop(self, input_ds, upsample_size):\n        \"\"\"Guided Backpropagation method for visualizing input saliency.\"\"\"\n        with tf.GradientTape() as tape:\n            tape.watch(input_ds)\n            outputs = self.gbModel(input_ds)\n        grads = tape.gradient(outputs, input_ds)[0][0,:,:,:]\n        saliency = cv2.resize(src = np.float32(grads), dsize = upsample_size, interpolation = cv2.INTER_AREA)\n#         saliency = grads\n        return saliency\n\n\ndef deprocess_image(x):\n    \"\"\"Same normalization as in:\n    https://github.com/fchollet/keras/blob/master/examples/conv_filter_visualization.py\n    \"\"\"\n    # normalize tensor: center on 0., ensure std is 0.25\n    x = x.copy()\n    x -= x.mean()\n    x /= (x.std() + K.epsilon())\n    x *= 0.25\n\n    # clip to [0, 1]\n    x += 0.5\n    x = np.clip(x, 0, 1)\n\n    # convert to RGB array\n    x *= 255\n    if K.image_data_format() == 'channels_first':\n        x = x.transpose((1, 2, 0))\n    x = np.clip(x, 0, 255).astype('uint8')\n#     def remap(x, out_min, out_max):\n#         return (x - x.min()) * (out_max - out_min) / (x.max() - x.min()) + out_min\n#     x = remap(x, 0, 255).astype('uint8')\n    \n    return x","execution_count":null,"outputs":[]},{"metadata":{"scrolled":false,"trusted":true},"cell_type":"code","source":"def show_gradCAMs(model, gradCAM, GuidedBP, image_inputs, aux_inputs, classIdx , upsample_size,n=3, decode={}):\n    \"\"\"\n    this work for tensorflow 2.x for input image with shape (none, x, x, 1)\n    modified from\n    https://github.com/nguyenhoa93/GradCAM_and_GuidedGradCAM_tf2/blob/master/src/guidedBackprop.py\n    \"\"\"\n    plt.subplots(figsize=(25, 10*n))\n    k=1\n    for j in range(n):\n        # define instance\n        i = random.randint(0,n)\n        image_input = load_and_preprocess_image(image_inputs[i])\n        aux_input = aux_inputs[i]\n        input_ds = [tf.cast(np.expand_dims(image_input, axis=0), tf.float32), \n                    tf.cast(np.expand_dims(aux_input, axis=0), tf.float32)]\n#         input_ds = tf.concat([x, y], axis = 1)\n        image_input = image_input[:,:,0]\n        # record the image size\n        upsample_size = (image_input.shape[1], image_input.shape[0])\n        \n        # Show original image\n        plt.subplot(n,3,k)\n        plt.imshow(image_input)\n#         plt.title(\"class: {}\".format(classIdx), fontsize=20)\n        plt.axis(\"off\")\n        \n        # Show overlayed grad\n        plt.subplot(n,3,k+1)\n        preds = model.predict(input_ds)\n        idx = preds.argmax()\n        \n        # decode result in form of [class, prob]\n        if len(decode)==0:\n            res = tf.keras.applications.imagenet_utils.decode_predictions(preds)#[0][0][1:]\n        else:\n            res = [list(decode)[idx], preds.max()]    \n            \n        # compute cam and gb\n        cam = gradCAM.compute_heatmap(input_ds = input_ds, classIdx=idx, upsample_size=upsample_size)\n        gb = GuidedBP.guided_backprop(input_ds = input_ds, upsample_size = upsample_size)\n        \n#         print(\"[image_input] shape: {}, max: {}, min: {}\".format(image_input.shape, image_input.max(), image_input.min()))\n#         print(\"[cam] shape: {}, max: {}, min: {}\".format(cam.shape, cam.max(), cam.min()))\n#         print(\"[gb] shape: {}, max: {}, min: {}\".format(gb.shape, gb.max(), gb.min()))\n        \n        # Show Gradient CAM \n        gradCAM_img = overlay_gradCAM(image_input, cam)\n#         new_img = cv2.cvtColor(new_img, cv2.COLOR_BGR2RGB)\n        plt.imshow(gradCAM_img)\n        plt.title(\"GradCAM - Pred: {}. Prob: {}\".format(res[0],res[1]), fontsize=20)\n        plt.axis(\"off\")\n        \n        # Show guided GradCAM\n        plt.subplot(n,3,k+2)\n        guided_gradcam = deprocess_image(gb* cam)\n        plt.imshow(guided_gradcam)\n#         plt.imshow(gb, cmap=plt.cm.BuGn)\n        plt.title(\"Guided GradCAM\", fontsize=20)\n        plt.axis(\"off\")\n        \n        k += 3\n    plt.show()\n","execution_count":null,"outputs":[]},{"metadata":{"scrolled":true,"trusted":true},"cell_type":"code","source":"# layer and class\nlayerName = \"activation_19\"\nactual_class = \"aldfly\"\n\n# index\nclass_indices = label_dict\nclassIdx = label_dict.get(actual_class)\n\n# input image path\nclass_root = pathlib.Path(valid_filepath + \"/\" +actual_class)\nclass_image_path = class_root.rglob('*.jpg')\nclass_image_path = [str(pathlib.Path(path)) for path in class_image_path]\n\n# get index of label and freq \nclass_freq = [int(path.split('/')[-1].split('_')[1]) for path in class_image_path]\nclass_label = [path.split('/')[-2] for path in class_image_path]\nencode_class_freq = [freq_dict.get(freq) for freq in class_freq]\noh_class_label = label_to_onehot.transform(np.array(class_label).reshape(-1, 1))\n\nimage_inputs = class_image_path\naux_inputs = list(encode_class_freq)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# launch\ngradCAM = GradCAM(model = model, layerName = layerName)\nguidedBP = GuidedBackprop(model = model, layerName = layerName)\nprint(\"actual class: \",actual_class)\nshow_gradCAMs(model, gradCAM, guidedBP, \n              image_inputs = image_inputs, aux_inputs = aux_inputs, \n              decode= class_indices, classIdx = classIdx, \n              upsample_size = (128, 128), n=20)\n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"","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}