{"nbformat": 4, "cells": [{"source": ["### Tested: \n", "######## the resulting file is 85GB , run time on 8 cores: 53 minutes\n", "<img src=\"https://sites.google.com/site/sgdysregulation/img/imag.png\" >"], "metadata": {"_cell_guid": "eead97aa-8cc3-43ce-a98d-6680bec6de55", "_uuid": "b5053f3df75e90409cf19050de146489beb2840e"}, "cell_type": "markdown"}, {"source": ["### Multiprocessing imporvement inspired by \n", "[StackOverflow:Understanding Multiprocessing: Shared Memory Management, Locks and Queues in Python](https://stackoverflow.com/questions/20742637/understanding-multiprocessing-shared-memory-management-locks-and-queues-in-pyt)"], "metadata": {"_cell_guid": "7eda1a8b-0647-46e7-a529-7bc4b6cd80ff", "_uuid": "7770adc16c22afd27460d8ef7e7669a8effeaedb"}, "cell_type": "markdown"}, {"source": ["import pandas as pd\n", "import numpy as np\n", "import bson\n", "import h5py\n", "\n", "import os\n", "\n", "\n", "import multiprocessing as mp\n", "\n", "\n", "import cv2 #opencv helpful for storing image as array\n", "import itertools #helps in parallel processing\n", "\n", "import scipy as scp\n", "\n", "import matplotlib.pyplot as plt\n", "import matplotlib as mpl\n", "import time\n", "import base64\n"], "outputs": [], "metadata": {"_cell_guid": "9160d4a0-50d0-4b87-8953-a6197ee6ccc1", "collapsed": true, "_uuid": "9b26281c57bd266764c6439bbb0be7589b22ee4c"}, "execution_count": null, "cell_type": "code"}, {"source": ["REC_SIZE = 7069896\n", "path = '../input/'\n", "kaggle =True\n", "# path='./'\n", "# kaggle = False\n", "!ls \"$path\""], "outputs": [], "metadata": {"_cell_guid": "dd8db1d5-9033-4857-b547-12b2dc2fa446", "_uuid": "35b9f8927499714bc7ea6638af41c1e92c02727e"}, "execution_count": null, "cell_type": "code"}, {"source": ["#### seprerate lock and queue used for reading/writing"], "metadata": {"_cell_guid": "bd3561ab-bfa8-4434-9b92-4ce3210e935c", "_uuid": "2c7721df25bf1ecb6df7f78361cb3e120285e4cc"}, "cell_type": "markdown"}, {"source": ["def process_batch(args):\n", "    \"\"\"\n", "    INPUT: args where args[0] is a list of tuples of size 4\n", "    (itr,CHUNK_SIZE ,in_file,out_file)\n", "    reads a batch of data from the BSON file \"in_file\" \n", "    puts the part of the input itr into a read queue.\n", "    converts the images to base64\n", "    write the images to dataset \"imgs\" in h5 file out_file\n", "    writes the columns [category_id, _id, img_num] \n", "    corresponding to the imgs into dataset \"meta\" in h5 file out_file\n", "    finally puts the part of the input itr into a write queue.\n", "    \"\"\"\n", "    t0 = time.time()\n", "    \n", "    itr,CHUNK_SIZE ,in_file,out_file= args[0]\n", "    if not os.path.exists(in_file):\n", "        return\n", "    print('Processing Batch {} , Batch size: {}'.format(itr,CHUNK_SIZE))\n", "    \n", "    lock_r = args[1]\n", "    queue_r = args[2]\n", "    lock_w = args[3] \n", "    queue_w = args[4] \n", "    \n", "    lock_r.acquire()\n", "    with open(in_file,'rb') as b:\n", "        iterator = bson.decode_file_iter(b)\n", "        df = pd.DataFrame(list(itertools.islice(iterator,\n", "                                itr*CHUNK_SIZE,\n", "                                (itr+1)*CHUNK_SIZE))).set_index(['category_id','_id'])\n", "    lock_r.release()\n", "    queue_r.put(itr)\n", "    \n", "    df = df['imgs'].apply(pd.Series).stack().apply(\n", "        lambda x: base64.b64encode(x['picture']))\n", "    data = df.index.to_frame().reset_index(drop=True),np.vstack(df.values)\n", "    df = None\n", "    lock_w.acquire()\n", "    try:\n", "        with h5py.File(out_file) as hdf:\n", "            if 'imgs' not in hdf.keys():\n", "                dt = h5py.special_dtype(vlen=bytes)\n", "                dset = hdf.create_dataset('imgs', shape= (data[1].shape),\n", "                                          maxshape=(None, data[1].shape[1]), chunks=True,\n", "                                          compression=\"lzf\",dtype=dt)\n", "                iset = hdf.create_dataset(name = 'meta', shape= data[0].shape,\n", "                                          maxshape=(None,data[0].shape[1]),\n", "                                          chunks=True,\n", "                                          dtype=np.int64,compression=\"lzf\")\n", "            else:\n", "                dset = hdf['imgs']\n", "                dset.resize((dset.shape[0]+data[1].shape[0],1)) \n", "                iset = hdf['meta']\n", "                iset.resize((iset.shape[0]+data[0].shape[0],iset.shape[1])) \n", "            dset[-data[1].shape[0]:,...] = data [1]\n", "            iset[-data[0].shape[0]:,...] = data [0]\n", "            hdf.close()\n", "    except Exception as e:\n", "        print('write failed',e)\n", "    data = None\n", "    lock_w.release()\n", "    t1= time.time()-t0\n", "    print('Batch {} processing Done! Time: {:} mins, {:.2f}secs'.format(itr, t1//60,t1%60))\n", "    queue_w.put(itr)"], "outputs": [], "metadata": {"_cell_guid": "70ac9985-f71f-4404-9132-d130cd44ab03", "collapsed": true, "_uuid": "1134ee949dd81e9f4153f5830c92ec6d4ca19bef"}, "execution_count": null, "cell_type": "code"}, {"source": ["\n", "def read_queue(queue):\n", "    \"\"\"Turns a qeue into a normal python list.\"\"\"\n", "    results = []\n", "    while not queue.empty():\n", "        result = queue.get()\n", "        results.append(result)\n", "    return results"], "outputs": [], "metadata": {"_cell_guid": "cb457d6c-b422-43f5-921b-df61a0df9b2b", "collapsed": true, "_uuid": "4bc455f76bfdde9c2d3951d330dd47496ed5b53b"}, "execution_count": null, "cell_type": "code"}, {"source": ["def make_iterator(args, lock_r, queue_r,lock_w, queue_w):\n", "    \"\"\"Makes an iterator over args and passes the lock an queue to each element.\"\"\"\n", "    return ((arg, lock_r, queue_r,lock_w, queue_w) for arg in args)"], "outputs": [], "metadata": {"_cell_guid": "9396b10b-2b44-4b9a-b54e-3561e38ed5f6", "collapsed": true, "_uuid": "475a2ab2e84329d6f3489200e708c552e55b18f3"}, "execution_count": null, "cell_type": "code"}, {"source": ["def start_processing(in_file,\n", "                     out_file, \n", "                     CHUNK_SIZE,\n", "                     EP,\n", "                     SP,\n", "                     ncores=4):\n", "    \"\"\"Starts the manager\n", "    \n", "    :param in_file the BSON file with byte images\n", "    :param out_file the HDF file with base64 images \n", "    in \"image\" dataset and meta data in \"meta\" dataset\n", "    :param CHUNK_SIZE the batch size per process\n", "    :param  EP the end postion (last batch)\n", "    :param  SP the start postion (first batch)\n", "    :param  ncores the number of CPU cores to use\n", "    \"\"\"\n", "    \n", "    args = list(zip(range(EP-1,SP-1,-1),\n", "                    [CHUNK_SIZE]*(EP-SP),\n", "                    [in_file]*(EP-SP),\n", "                    [out_file]*(EP-SP)))\n", "\n", "    result =  manager(process_batch, args, ncores)\n", "    return result"], "outputs": [], "metadata": {"_cell_guid": "7824e025-452e-43bb-9fd1-00310b4834c5", "collapsed": true, "_uuid": "c2710ae4f05a0a50fbb30dc9066a5392c19faaf5"}, "execution_count": null, "cell_type": "code"}, {"source": ["def manager(jobfunc, args, ncores):\n", "    \"\"\"Runs a pool of processes WITH a Manager for the lock and queue.\n", "\n", "    \"\"\"\n", "    mypool = mp.Pool(ncores)\n", "    lock_r = mp.Manager().Lock()\n", "    queue_r = mp.Manager().Queue()\n", "    lock_w = mp.Manager().Lock()\n", "    queue_w = mp.Manager().Queue()\n", "    iterator = make_iterator(args, lock_r, queue_r,lock_w, queue_w)\n", "    mypool.map(jobfunc, iterator)\n", "    mypool.close()\n", "    mypool.join()\n", "\n", "    return read_queue(queue_r),read_queue(queue_w)"], "outputs": [], "metadata": {"_cell_guid": "d9463d50-dce1-4085-9b63-267eda0bb0b9", "collapsed": true, "_uuid": "f8ec7d17403f2b85ffc00a1cdf9388291553ac4b"}, "execution_count": null, "cell_type": "code"}, {"source": [], "metadata": {"_cell_guid": "49f8d189-af97-4139-b379-68b9df1e8dac", "_uuid": "bb398d5c1ed783d4f06aea9e46593bb897614baa"}, "cell_type": "markdown"}, {"source": ["\"\"\"Run \"\"\"\n", "\n", "in_file= '{}train.bson'.format(path)\n", "out_file= 'train.h5'\n", "CHUNK_SIZE = 2**17\n", "#default values\n", "EP = 1+(REC_SIZE//CHUNK_SIZE)\n", "SP = 0\n", "if kaggle:\n", "    !rm \"$out_file\"\n", "    CHUNK_SIZE = 2**8\n", "    EP = 1+(REC_SIZE//CHUNK_SIZE)\n", "    SP = EP-16\n", "print(SP,':',EP)\n", "\n", "t0 = time.time()\n", "try:\n", "    res = start_processing(in_file = in_file,\n", "                     out_file = out_file,\n", "                     CHUNK_SIZE = CHUNK_SIZE,\n", "                     EP = EP,\n", "                    SP = SP,ncores = min(mp.cpu_count(),8))\n", "except Exception as e:\n", "    print('Failed:',e)\n", "t1= time.time()-t0\n", "print('-'*40)\n", "print('Done! Total Processing Time: {:} mins, {:.2f}secs'.format(t1//60,t1%60))"], "outputs": [], "metadata": {"_cell_guid": "cbfe8fb9-0825-4bff-b232-2414d9598cb9", "_uuid": "3e26a1b0dbb7257b6e9dae9d6ad5af6952fb38c6", "scrolled": true}, "execution_count": null, "cell_type": "code"}, {"source": [], "outputs": [], "metadata": {"_cell_guid": "6c6d475f-d2b9-417b-a0a4-46b509c24c7a", "collapsed": true, "_uuid": "9792178a60fe03cc3bd1a6b5e14960a4ff296a1d"}, "execution_count": null, "cell_type": "code"}, {"source": ["### Retrieve files"], "metadata": {"_cell_guid": "f29a4668-c92c-4448-ba1d-1d64b7bff60c", "_uuid": "f5658f926066a82c868ca52fada11d5e9293829f"}, "cell_type": "markdown"}, {"source": ["def read_h5(file,num_of_rec,frRec=0):\n", "    \"\"\"\n", "    retrieves certain number of records `num_od_rec` from the file `file`\n", "    records start at `frRec` \n", "    \"\"\"\n", "    cols =['category_id','_id','img_num']\n", "    with h5py.File(file) as hdf:\n", "\n", "        data = hdf['imgs'][frRec:num_of_rec]\n", "        index = hdf['meta'][frRec:num_of_rec]\n", "        hdf.close()\n", "    df = pd.DataFrame(index,columns= cols)\n", "    df['imgs'] = data\n", "    df['imgs'] = df['imgs'].apply(\n", "        lambda bStr: cv2.imdecode(\n", "                            np.fromstring(\n", "                                base64.b64decode(bStr),\n", "                                dtype='uint8'),\n", "                            cv2.IMREAD_COLOR))\n", "    return df.set_index(cols)\n", "df = read_h5(out_file,10)"], "outputs": [], "metadata": {"_cell_guid": "4a95597a-5945-4809-a75c-4efe4a08268c", "collapsed": true, "_uuid": "ea7ea9872dc732f7631c30cd36004ebb5ee0d1c9"}, "execution_count": null, "cell_type": "code"}, {"source": ["df.head()"], "outputs": [], "metadata": {"_cell_guid": "db8a1664-edd1-400f-b7de-7de90bec1ab4", "_uuid": "531d84715d819db89928f987e937e0784652c205"}, "execution_count": null, "cell_type": "code"}, {"source": ["df.info()"], "outputs": [], "metadata": {"_cell_guid": "14f2a27b-acf0-4c4d-8286-f06eaa9c71e2", "_uuid": "05a07f112bc87c26bb529e3c0ca447222a7cb9f3"}, "execution_count": null, "cell_type": "code"}, {"source": ["n=np.random.randint(0,len(df))\n", "img= df.iloc[n,-1]\n", "imgVSH = cv2.cvtColor(img,cv2.COLOR_RGB2HSV)[...,::-1] #change back to RGB\n", "\n", "fig,axs = plt.subplots(2,2,figsize=(16,8))\n", "axs = axs.flatten()\n", "\n", "titles = ['Intensity(Value)','Saturation','Hue']\n", "cmaps = ['gray',mpl.cm.GnBu,mpl.cm.GnBu]\n", "for i,ax in enumerate(axs[1:]):\n", "    ax.imshow(imgVSH[...,i],cmap=cmaps[i])\n", "    ax.set_title(titles[i])\n", "    ax.axis('off')\n", "axs[0].imshow(img[...,::-1])\n", "axs[0].set_title('Original')\n", "axs[0].axis('off');\n", "\n", "plt.tight_layout();\n", "plt.show();"], "outputs": [], "metadata": {"_cell_guid": "ed578494-ba55-4385-a06e-2f11d3ca7630", "_uuid": "9563acc8496e3c9a056762ddcad41c45f74ba9dd"}, "execution_count": null, "cell_type": "code"}, {"source": ["# TODO show how to process data on disk"], "outputs": [], "metadata": {"_cell_guid": "965fb120-7ed3-4f05-b6ff-2ab741d6e3b5", "collapsed": true, "_uuid": "f5c2b651c30ab121ca9a13b5e6ba7136b53683fa"}, "execution_count": null, "cell_type": "code"}, {"source": [], "outputs": [], "metadata": {"_cell_guid": "ef23a154-37e2-4ee7-8840-990844506c39", "collapsed": true, "_uuid": "2e1d62a514a8d3bf3ae0dd9c696eddcb62ad82d8"}, "execution_count": null, "cell_type": "code"}], "nbformat_minor": 1, "metadata": {"kernelspec": {"display_name": "Python 3", "language": "python", "name": "python3"}, "language_info": {"pygments_lexer": "ipython3", "codemirror_mode": {"version": 3, "name": "ipython"}, "file_extension": ".py", "version": "3.6.1", "name": "python", "mimetype": "text/x-python", "nbconvert_exporter": "python"}}}