{"cells":[{"metadata":{},"cell_type":"markdown","source":"### This dataset contains a lot of files. [Dask](https://dask.org/) is a great library to accelerate such workloads in parallel. In this notebook, we show how to use dask to engineer features in parallel and train a xgboost model.\n\n### On a machine with 16 cores, the feature engineering time is reduced from 1 hour to 5 minutes."},{"metadata":{"trusted":true},"cell_type":"code","source":"from glob import glob\nfrom collections import Counter\nimport os\nimport sys\n\nimport numpy as np\nimport pandas as pd\nfrom tqdm import tqdm\nfrom sklearn.preprocessing import OneHotEncoder\nfrom sklearn.model_selection import KFold\nimport xgboost as xgb","execution_count":null,"outputs":[]},{"metadata":{"_kg_hide-input":true,"trusted":true},"cell_type":"code","source":"from dataclasses import dataclass\n\nimport numpy as np\n\n\n@dataclass\nclass ReadData:\n    acce: np.ndarray\n    acce_uncali: np.ndarray\n    gyro: np.ndarray\n    gyro_uncali: np.ndarray\n    magn: np.ndarray\n    magn_uncali: np.ndarray\n    ahrs: np.ndarray\n    wifi: np.ndarray\n    ibeacon: np.ndarray\n    waypoint: np.ndarray\n\n\ndef read_data_file(data_filename):\n    acce = []\n    acce_uncali = []\n    gyro = []\n    gyro_uncali = []\n    magn = []\n    magn_uncali = []\n    ahrs = []\n    wifi = []\n    ibeacon = []\n    waypoint = []\n\n    with open(data_filename, 'r', encoding='utf-8') as file:\n        lines = file.readlines()\n\n    for line_data in lines:\n        line_data = line_data.strip()\n        if not line_data or line_data[0] == '#':\n            continue\n\n        line_data = line_data.split('\\t')\n\n        if line_data[1] == 'TYPE_ACCELEROMETER':\n            acce.append([int(line_data[0]), float(line_data[2]), float(line_data[3]), float(line_data[4])])\n            continue\n\n        if line_data[1] == 'TYPE_ACCELEROMETER_UNCALIBRATED':\n            acce_uncali.append([int(line_data[0]), float(line_data[2]), float(line_data[3]), float(line_data[4])])\n            continue\n\n        if line_data[1] == 'TYPE_GYROSCOPE':\n            gyro.append([int(line_data[0]), float(line_data[2]), float(line_data[3]), float(line_data[4])])\n            continue\n\n        if line_data[1] == 'TYPE_GYROSCOPE_UNCALIBRATED':\n            gyro_uncali.append([int(line_data[0]), float(line_data[2]), float(line_data[3]), float(line_data[4])])\n            continue\n\n        if line_data[1] == 'TYPE_MAGNETIC_FIELD':\n            magn.append([int(line_data[0]), float(line_data[2]), float(line_data[3]), float(line_data[4])])\n            continue\n\n        if line_data[1] == 'TYPE_MAGNETIC_FIELD_UNCALIBRATED':\n            magn_uncali.append([int(line_data[0]), float(line_data[2]), float(line_data[3]), float(line_data[4])])\n            continue\n\n        if line_data[1] == 'TYPE_ROTATION_VECTOR':\n            if len(line_data)>=5:\n                ahrs.append([int(line_data[0]), float(line_data[2]), float(line_data[3]), float(line_data[4])])\n            continue\n\n        if line_data[1] == 'TYPE_WIFI':\n            sys_ts = line_data[0]\n            ssid = line_data[2]\n            bssid = line_data[3]\n            rssi = line_data[4]\n            lastseen_ts = line_data[6]\n            wifi_data = [sys_ts, ssid, bssid, rssi, lastseen_ts]\n            wifi.append(wifi_data)\n            continue\n\n        if line_data[1] == 'TYPE_BEACON':\n            ts = line_data[0]\n            uuid = line_data[2]\n            major = line_data[3]\n            minor = line_data[4]\n            rssi = line_data[6]\n            ibeacon_data = [ts, '_'.join([uuid, major, minor]), rssi]\n            ibeacon.append(ibeacon_data)\n            continue\n\n        if line_data[1] == 'TYPE_WAYPOINT':\n            waypoint.append([int(line_data[0]), float(line_data[2]), float(line_data[3])])\n\n    acce = np.array(acce)\n    acce_uncali = np.array(acce_uncali)\n    gyro = np.array(gyro)\n    gyro_uncali = np.array(gyro_uncali)\n    magn = np.array(magn)\n    magn_uncali = np.array(magn_uncali)\n    ahrs = np.array(ahrs)\n    wifi = np.array(wifi)\n    ibeacon = np.array(ibeacon)\n    waypoint = np.array(waypoint)\n\n    return ReadData(acce, acce_uncali, gyro, gyro_uncali, magn, magn_uncali, ahrs, wifi, ibeacon, waypoint)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"import dask\nfrom dask.distributed import Client, wait, LocalCluster","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"We use only 2 workers on kaggle kernel. On your local machine, you could increase the number of workers/threads."},{"metadata":{"trusted":true},"cell_type":"code","source":"client = Client(n_workers=2, \n                threads_per_worker=1)\nclient","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"The dashboard is a great feature to monitor the progress of dask. Since dask is asynchronous, only the progress bar in the dashboard reflects the real progress."},{"metadata":{},"cell_type":"markdown","source":"### Functions"},{"metadata":{"trusted":true},"cell_type":"code","source":"def mpe(yp, y):\n    e1 = (yp[:,0] - y[:,0])**2 + (yp[:,1] - y[:,1])**2\n    e2 = 15*np.abs(yp[:,2] - y[:,2])\n    return np.mean(e1**0.5 + e2)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def get_building_floor(fname):\n    xx = fname.split('/')\n    return xx[-3],xx[-2]\n\ndef get_test_building(name):\n    with open(name) as f:\n        for c,line in enumerate(f):\n            if c==1:\n                x = line.split()[1].split(':')[1]\n                return x  \n\ndef get_floor_target(floor):\n    floor = floor.lower()\n    if floor in ['bf','bm']:\n        return None\n    elif floor == 'b':\n        return -1\n    if floor.startswith('f'):\n        return int(floor[1])\n    elif floor.endswith('f'):\n        return int(floor[0])\n    elif floor.startswith('b'):\n        return -int(floor[1])\n    elif floor.endswith('b'):\n        return -int(floor[0])\n    else:\n        return None\n        \nACOLS = ['timestamp','x','y','z']\n        \nFIELDS = {\n    'acce': ACOLS,\n    'acce_uncali': ACOLS,\n    'gyro': ACOLS,\n    'gyro_uncali': ACOLS,\n    'magn': ACOLS,\n    'magn_uncali': ACOLS,\n    'ahrs': ACOLS,\n    'wifi': ['timestamp','ssid','bssid','rssi','last_timestamp'],\n    'ibeacon': ['timestamp','code','rssi'],\n    'waypoint': ['timestamp','x','y']\n}\n\nNFEAS = {\n    'acce': 3,\n    'acce_uncali': 3,\n    'gyro': 3,\n    'gyro_uncali': 3,\n    'magn': 3,\n    'magn_uncali': 3,\n    'ahrs': 3,\n    'wifi': 1,\n    'ibeacon': 1,\n    'waypoint': 3\n}","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def build_fea_one_file(data):\n    feas = []\n    target = None\n    for k,v in vars(data).items():\n        if k == 'waypoint':\n            if len(v.shape)==2 and v.shape[1] == 3:\n                target = v[:,1:]\n            else:\n                target = None\n            continue\n        if k in ['wifi','ibeacon']:\n            continue\n        if v.shape[0] == 0:\n            feas.extend([None]*NFEAS[k]*2)\n            continue\n        df = pd.DataFrame(v, columns=FIELDS[k])\n        for col in df.columns[1:]:\n            if df[col].dtype!='O' and 'time' not in col:\n                feas.extend([df[col].mean(),df[col].std()])\n    return np.array(feas),target\n\ndef fe(name):\n    data = read_data_file(name)\n    x,y = build_fea_one_file(data)\n    assert len(x) == 42\n    return x,y","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"### Build Features"},{"metadata":{"trusted":true},"cell_type":"code","source":"!ls ../input/indoor-location-navigation","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"PATH = '../input/indoor-location-navigation'\ntrain_files = glob(f'{PATH}/train/*/*/*.txt')\nlen(train_files)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\nbuildings = []\nfloors = []\nused = []\nfor fname in tqdm(train_files):\n    b,f = get_building_floor(fname)\n    f = get_floor_target(f)\n    if f is None:\n        continue\n    used.append(fname)\n    buildings.append(b)\n    floors.append(f)\ny = np.array(floors)\nb = np.array(buildings)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\nenc = OneHotEncoder()\nbs = enc.fit_transform(np.expand_dims(b,1))\nbs.shape","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\nfutures = [] # save the future since dask is lazy, otherwise nothing is executed.\nfor fname in tqdm(used):\n    f = client.submit(fe,fname) \n    futures.append(f) ","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\nX = [i.result() for i in futures]\nys = np.vstack([np.mean(i[1],axis=0) for i in X])\nX = np.vstack([i[0] for i in X])\nX.shape,ys.shape","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"df = pd.DataFrame(ys,columns=['w_x','w_y'])\ndf['building'] = b\ndf['floors'] = y\ndf.head()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"X = np.hstack([X,bs.toarray()])\nprint(X.shape)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"test_files = glob(f'{PATH}/test/*.txt')\nlen(test_files)","execution_count":null,"outputs":[]},{"metadata":{"trusted":false},"cell_type":"code","source":"%%time\ntest_b = []\nfor name in tqdm(test_files):\n    test_b.append(get_test_building(name))\ntest_b = np.array(test_b)\nbs = enc.transform(np.expand_dims(test_b,1))","execution_count":null,"outputs":[]},{"metadata":{"trusted":false},"cell_type":"code","source":"%%time\nfutures = [] # save the future since dask is lazy, otherwise nothing is executed.\nfor fname in tqdm(test_files):\n    f = client.submit(fe,fname) \n    futures.append(f) ","execution_count":null,"outputs":[]},{"metadata":{"trusted":false},"cell_type":"code","source":"%%time\nXt = [i.result()[0] for i in futures]\nXt = np.vstack(Xt)\nXt = np.hstack([Xt,bs.toarray()])\nXt.shape","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"### Train XGB"},{"metadata":{"trusted":false},"cell_type":"code","source":"params = {\n        'objective': 'reg:linear',\n        'eval_metric': 'rmse',\n        'eta':0.1,\n        'depth':7,\n        'nthread':2,\n        'verbosity': 0,\n    }","execution_count":null,"outputs":[]},{"metadata":{"collapsed":true,"jupyter":{"outputs_hidden":true},"trusted":false},"cell_type":"code","source":"N = 5\ndtest = xgb.DMatrix(data=Xt)\nysub = np.zeros([Xt.shape[0],3])\n\nkf = KFold(n_splits=N,shuffle=True,random_state=42)\n\nmsgs = []\nfor i,(train_index, test_index) in enumerate(kf.split(X)):\n    X_train, X_test = X[train_index], X[test_index]\n    yps = np.zeros([X_test.shape[0],3])\n    yrs = yps.copy()\n    for c,col in enumerate(['w_x','w_y','floors']):\n        y = df[col].values\n        y_train, y_test = y[train_index], y[test_index]\n        \n        dtrain = xgb.DMatrix(data=X_train, label=y_train)\n        dvalid = xgb.DMatrix(data=X_test, label=y_test)\n        watchlist = [(dtrain, 'train'), (dvalid, 'eval')] \n\n        clf = xgb.train(params, dtrain=dtrain,\n                    num_boost_round=70,evals=watchlist,\n                    early_stopping_rounds=10,\n                    verbose_eval=100)\n        yp = clf.predict(dvalid)\n        yps[:,c] = yp\n        yrs[:,c] = y_test\n        ysub[:,c] += clf.predict(dtest)\n    msg = f'Fold {i}: MPE {mpe(yps, yrs):.4f}'\n    print(msg)\n    msgs.append(msg)\nysub = ysub/N","execution_count":null,"outputs":[]},{"metadata":{"trusted":false},"cell_type":"code","source":"msgs","execution_count":null,"outputs":[]},{"metadata":{"trusted":false},"cell_type":"code","source":"sub = pd.read_csv(f'{PATH}/sample_submission.csv')\nsub.head()","execution_count":null,"outputs":[]},{"metadata":{"trusted":false},"cell_type":"code","source":"sub.shape","execution_count":null,"outputs":[]},{"metadata":{"trusted":false},"cell_type":"code","source":"sub['site'] = sub['site_path_timestamp'].apply(lambda x: x.split('_')[0])\nsub.head()","execution_count":null,"outputs":[]},{"metadata":{"trusted":false},"cell_type":"code","source":"test_map = {i:j for i,j in zip(test_b, test_files)}\nsub['filename'] = sub['site'].apply(lambda x: test_map[x])\nsub.head()","execution_count":null,"outputs":[]},{"metadata":{"trusted":false},"cell_type":"code","source":"ds = pd.DataFrame(ysub,columns=['x','y','floor'])\nds.head()","execution_count":null,"outputs":[]},{"metadata":{"trusted":false},"cell_type":"code","source":"ds['filename'] = test_files\nds.head()","execution_count":null,"outputs":[]},{"metadata":{"trusted":false},"cell_type":"code","source":"sub = sub.drop(['x','y','floor'],axis=1).merge(ds,on='filename',how='left')\nprint(sub.shape)\nsub.head()","execution_count":null,"outputs":[]},{"metadata":{"trusted":false},"cell_type":"code","source":"for i in sub.columns:\n    print(i,sub[i].isnull().sum())","execution_count":null,"outputs":[]},{"metadata":{"trusted":false},"cell_type":"code","source":"sub['floor'] = sub['floor'].astype('int')\nsub.head()","execution_count":null,"outputs":[]},{"metadata":{"trusted":false},"cell_type":"code","source":"sub.drop(['site','filename'],axis=1).to_csv('submission.csv',index=False)","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}