{"metadata": {"kernelspec": {"display_name": "Python 3", "name": "python3", "language": "python"}, "language_info": {"codemirror_mode": {"name": "ipython", "version": 3}, "version": "3.6.1", "pygments_lexer": "ipython3", "nbconvert_exporter": "python", "name": "python", "file_extension": ".py", "mimetype": "text/x-python"}}, "nbformat_minor": 1, "cells": [{"cell_type": "markdown", "metadata": {"_cell_guid": "50c3936f-4053-4ea7-a337-80a318081d3e", "_uuid": "e43e9a511480d5ef6ea2b11e830804ad64cedb6a"}, "source": ["Credits to *Peter Giannakopoulos* , I've taken his starter code and added some modifications so that the all the image processing for training and testing to utilize all available cores on a multicore machine. If you have a machine with a bunch of cores and lots of RAM you'll definitely might find this useful. Even if you don't have a lot of RAM you can modify the code to suit your needs. Let's check the vm of kaggle kernels.\n"]}, {"execution_count": null, "cell_type": "code", "metadata": {}, "source": ["!cat /proc/cpuinfo"], "outputs": []}, {"execution_count": null, "cell_type": "code", "metadata": {}, "source": ["!free -h"], "outputs": []}, {"cell_type": "markdown", "metadata": {}, "source": ["It seem that there's an 8-core machine with 36 GB of free RAM."]}, {"execution_count": null, "cell_type": "code", "metadata": {}, "source": ["from subprocess import check_output\n", "print(check_output([\"ls\", \"../input\"]).decode(\"utf8\"))\n", "\n", "# Any results you write to the current directory are saved as output."], "outputs": []}, {"execution_count": null, "cell_type": "code", "metadata": {}, "source": ["import numpy as np\n", "import pandas as pd\n", "import matplotlib.pyplot as plt\n", "from keras.models import Sequential, Model, model_from_json\n", "from keras.layers import Dense, Conv2D, Input, MaxPool2D, UpSampling2D, Concatenate, Conv2DTranspose\n", "from keras.layers import BatchNormalization, Dropout, AveragePooling2D\n", "from keras.callbacks import ModelCheckpoint\n", "import tensorflow as tf\n", "from keras.optimizers import Adam\n", "# from tqdm import tqdm\n", "import multiprocessing as mp\n", "from multiprocessing import cpu_count\n", "import os, cv2, time\n", "from itertools import repeat\n", "from functools import partial\n", "from keras.preprocessing.image import load_img, ImageDataGenerator\n", "%matplotlib inline"], "outputs": []}, {"execution_count": null, "cell_type": "code", "metadata": {"collapsed": true}, "source": ["# set the necessary directories\n", "data_dir = \"../input/train/\"\n", "mask_dir = \"../input/train_masks/\"\n", "test_dir = \"../input/test/\"\n", "all_images = os.listdir(data_dir)\n", "test_images = os.listdir(test_dir)\n", "# %ls data"], "outputs": []}, {"execution_count": null, "cell_type": "code", "metadata": {"collapsed": true}, "source": ["def preprocess(data_dir, img_name, dims, rles):\n", "    img = load_img(data_dir+img_name)\n", "    img = np.array(img, dtype='float32')/255.\n", "    if rles:\n", "        img = cv2.resize(img, (1918, 1280))\n", "        mask = img > 0.5\n", "        img = rle(mask)\n", "    else:\n", "        img = cv2.resize(img, dims)\n", "    return img"], "outputs": []}, {"execution_count": null, "cell_type": "code", "metadata": {"collapsed": true}, "source": ["# generator that we will use to read the data from the directory\n", "def process_data(data_dir, mask_dir, batch_size, dims, images):\n", "    \"\"\"\n", "    data_dir: where the actual images are kept\n", "    mask_dir: where the actual masks are kept\n", "    images: the filenames of the images we want to generate batches from\n", "    batch_size: self explanatory\n", "    dims: the dimensions in which we want to rescale our images\n", "    \"\"\"\n", "    imgs = []\n", "    labels = []\n", "    # images\n", "    img = preprocess(data_dir, images, dims, False)\n", "    imgs.append(img)\n", "\n", "    # masks\n", "    mask = preprocess(mask_dir, images.split(\".\")[0] + '_mask.gif', dims, False)\n", "    labels.append(mask[:, :, 0])\n", "    return imgs, labels"], "outputs": []}, {"cell_type": "markdown", "metadata": {}, "source": ["In my machine I have lots of RAM so loading all the trainig data and the mask is not an issue. It requires approximately 17GB. If you want you can change the default setting \"batch_size=len(all_images)\" to a moderate stting like e.g. 64."]}, {"execution_count": null, "cell_type": "code", "metadata": {"collapsed": true}, "source": ["def multicore_generator(images, batch_size=len(all_images)):\n", "    ix = np.random.choice(np.arange(len(images)), batch_size) # from len(train_images) choose batch_size=64\n", "    tic = time.time()\n", "    pool = mp.Pool(processes=cpu_count())\n", "    train_gen = partial(process_data, data_dir, mask_dir, batch_size, (256, 256))\n", "    gen = pool.map_async(train_gen, list(np.array(images)[ix]), chunksize=8)\n", "    gen.wait()\n", "    results = gen.get()\n", "    pool.close()\n", "    pool.join()\n", "    pool.terminate()\n", "    x, y = zip(*results)\n", "    x = np.array(x, dtype='float32').reshape(-1, 256, 256, 3)\n", "    y = np.array(y, dtype='int32').reshape(-1, 256, 256, 1)\n", "    print((time.time() - tic)/60.)\n", "    return x, y"], "outputs": []}, {"execution_count": null, "cell_type": "code", "metadata": {"collapsed": true}, "source": ["# Now let's use Tensorflow to write dice_coeficcient metric\n", "def dice_coef(y_true, y_pred):\n", "    smooth = 1e-5\n", "    \n", "    y_true = tf.round(tf.reshape(y_true, [-1]))\n", "    y_pred = tf.round(tf.reshape(y_pred, [-1]))\n", "    \n", "    isct = tf.reduce_sum(y_true * y_pred)\n", "    \n", "    return 2 * isct / (tf.reduce_sum(y_true) + tf.reduce_sum(y_pred))"], "outputs": []}, {"execution_count": null, "cell_type": "code", "metadata": {"collapsed": true}, "source": ["# First let's define the two different types of layers that we will be using.\n", "def down(input_layer, filters, pool=True):\n", "    conv1 = Conv2D(filters, (3, 3), padding='same', activation='elu')(input_layer)\n", "    conv2 = Conv2D(filters, (3, 3), padding='same', activation='elu')(conv1)\n", "    residual = BatchNormalization(axis=3)(conv2)\n", "    if pool:\n", "        max_pool = MaxPool2D()(residual)\n", "#         max_pool = AveragePooling2D()(residual)\n", "        return max_pool, residual\n", "    else:\n", "        return residual\n", "\n", "def up(input_layer, residual, filters):\n", "    filters=int(filters)\n", "    upsample = UpSampling2D()(input_layer)\n", "    upconv = Conv2D(filters, (2, 2), padding=\"same\")(upsample)\n", "    concat = Concatenate(axis=3)([residual, upconv])\n", "    drop = Dropout(0.25)(concat)\n", "    conv1 = Conv2D(filters, (3, 3), padding='same', activation='elu')(drop)\n", "    conv2 = Conv2D(filters, (3, 3), padding='same', activation='elu')(conv1)\n", "    return conv2"], "outputs": []}, {"execution_count": null, "cell_type": "code", "metadata": {}, "source": ["# Make a custom U-nets implementation.\n", "filters = 64\n", "input_layer = Input(shape = [256, 256, 3])\n", "layers = [input_layer]\n", "residuals = []\n", "\n", "# Down 1, 128\n", "d1, res1 = down(input_layer, filters)\n", "residuals.append(res1)\n", "\n", "filters *= 2\n", "\n", "# Down 2, 64\n", "d2, res2 = down(d1, filters)\n", "residuals.append(res2)\n", "\n", "filters *= 2\n", "\n", "# Down 3, 32\n", "d3, res3 = down(d2, filters)\n", "residuals.append(res3)\n", "\n", "filters *= 2\n", "\n", "# Down 4, 16\n", "d4, res4 = down(d3, filters)\n", "residuals.append(res4)\n", "\n", "filters *= 2\n", "\n", "# Down 5, 8\n", "d5 = down(d4, filters, pool=False)\n", "\n", "# Up 1, 16\n", "up1 = up(d5, residual=residuals[-1], filters=filters/2)\n", "\n", "filters /= 2\n", "\n", "# Up 2,  32\n", "up2 = up(up1, residual=residuals[-2], filters=filters/2)\n", "\n", "filters /= 2\n", "\n", "# Up 3, 64\n", "up3 = up(up2, residual=residuals[-3], filters=filters/2)\n", "\n", "filters /= 2\n", "\n", "# Up 4, 128\n", "up4 = up(up3, residual=residuals[-4], filters=filters/2)\n", "\n", "out = Conv2D(filters=1, kernel_size=(1, 1), activation=\"sigmoid\")(up4)\n", "\n", "model = Model(input_layer, out)\n", "model.compile(optimizer=Adam(1e-4), loss='binary_crossentropy', metrics=[dice_coef])\n", "model.summary()"], "outputs": []}, {"execution_count": null, "cell_type": "code", "metadata": {"collapsed": true}, "source": ["def rle(img):\n", "    '''\n", "    img: numpy array, 1 - mask, 0 - background\n", "    Returns run length as string formated\n", "    '''\n", "    bytes = np.where(img.flatten() == 1)[0]\n", "    runs = []\n", "    prev = -2\n", "    for b in bytes:\n", "        if (b > prev + 1): runs.extend((b + 1, 0))\n", "        runs[-1] += 1\n", "        prev = b\n", "\n", "    return ' '.join([str(i) for i in runs])"], "outputs": []}, {"execution_count": null, "cell_type": "code", "metadata": {"collapsed": true}, "source": ["def predict_masks(test_images, dims=(256, 256), batch_size=32):\n", "    valid_imgs = []\n", "    rles = []\n", "    tic = time.time()\n", "    pool = mp.Pool(processes=cpu_count())\n", "    for batch in xrange(0, len(test_images), batch_size):\n", "        resized = pool.map_async(preprocess, zip(repeat(test_dir), test_images[batch:batch+batch_size], \n", "                                                 repeat(dims), repeat(False)))\n", "        resized.wait()\n", "        predictions = model.predict_on_batch(np.array(resized.get()))\n", "        valid_imgs.append(np.squeeze(predictions))\n", "        masks = pool.map_async(preprocess, zip(repeat(test_dir), test_images[batch:batch+batch_size],\n", "                                               repeat(dims), repeat(True)))\n", "        masks.wait()\n", "        rles.append(masks.get())\n", "        \n", "        \n", "        print(\"{}:{}, {}, {}\".format(batch,\n", "                                     batch+batch_size, \n", "                                     len(test_images[batch:batch+batch_size]),\n", "                                     np.array(resized.get()).shape, \n", "                                     len(masks.get())\n", "                                )\n", "         )\n", "        \n", "    pool.close()\n", "    pool.join()\n", "    pool.terminate()\n", "    print(\"{} min.\".format((time.time() - tic)/60.))"], "outputs": []}, {"execution_count": null, "cell_type": "code", "metadata": {}, "source": ["if __name__==\"__main__\":\n", "    x, y = multicore_generator(all_images)\n", "    print(x.shape, y.shape)\n", "    model.fit(x, y, batch_size=12, epochs=10, validation_split=0.2)\n", "    predictions = predict_masks(test_images)"], "outputs": []}, {"cell_type": "markdown", "metadata": {}, "source": ["Notice: keras has an imageDataGenerator object but it is not multicore and hence very slow. On my machine (24core, 64GB RAM) using it with a batch_size=32 and 100 iterations i.e. 3200 images it took 60.73 min. while with the above code in approximately 60 min. I had finished on all 100K images. Enjoy!"]}], "nbformat": 4}