{"cells":[{"metadata":{},"cell_type":"markdown","source":"In this notebook, I'd like to share how to label encode all wifi files in parallel using dask. On my workstation with 20 cpu cores, it took only 10 mins to encode all the data. Due to the limitation of kaggle kernel, I have to set number of workers to 2. "},{"metadata":{"trusted":true},"cell_type":"code","source":"! ls ../input/indoor-location-navigation","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"from glob import glob\nimport os\nimport sys\nfrom sklearn.preprocessing import LabelEncoder\n\nimport numpy as np\nimport pandas as pd\nfrom tqdm import tqdm","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"import pandas as pd\nimport numpy as np\nfrom tqdm import tqdm\nfrom sklearn.preprocessing import LabelEncoder\nfrom glob import glob\nfrom dask.distributed import wait\n\nSENSORS = ['acce','acce_uncali','gyro',\n           'gyro_uncali','magn','magn_uncali','ahrs']\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}\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\ndef to_frame(data, col):\n    cols = FIELDS[col]\n    if data.shape[0]>0:\n        df = pd.DataFrame(data, columns=cols)\n    else:\n        df = create_dummy_df(cols)\n    for col in df.columns:\n        if 'timestamp' in col:\n            df[col] = df[col].astype('int64')\n    return df\n\ndef create_dummy_df(cols):\n    df = pd.DataFrame()\n    for col in cols:\n        df[col] = [0]\n        if col in ['ssid','bssid']:\n            df[col] = df[col].map(str)\n    return df\n\ndef 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])-1\n    elif floor.endswith('f'):\n        return int(floor[0])-1\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","execution_count":null,"outputs":[]},{"metadata":{"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)\n","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":{"trusted":true},"cell_type":"code","source":"# set n_workers to number of cores\nclient = Client(n_workers=2, \n                threads_per_worker=1)\nclient","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"### Label encode wifi ID features"},{"metadata":{"trusted":true},"cell_type":"code","source":"PATH = '../input/indoor-location-navigation'\nOUT = './wifi_lbl_encode'\nos.mkdir(OUT)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def read_wifi_ssid_bssid(name):\n    data = read_data_file(name)\n    dw = to_frame(data.wifi,'wifi')\n    ss = dw['ssid'].unique()\n    bs = dw['bssid'].unique()\n    return ss,bs\n\ndef encode_wifi_ssid_bssid(name, lbl_ssid, lbl_bssid):\n    data = read_data_file(name)\n    dw = to_frame(data.wifi,'wifi')\n    dw['ssid'] = lbl_ssid.transform(dw['ssid'])\n    dw['bssid'] = lbl_bssid.transform(dw['bssid'])\n    name = name.replace(PATH,OUT)\n    dw.to_csv(name,index=False)\n    return dw","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"### Run"},{"metadata":{"trusted":true},"cell_type":"code","source":"train_files = glob(f'{PATH}/train/*/*/*.txt')\ntest_files = glob(f'{PATH}/test/*.txt')\nlen(test_files),len(train_files)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"name = train_files[0]\nss,bs = read_wifi_ssid_bssid(name)\ntype(ss),ss.shape,bs.shape,name","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"os.mkdir(f'{OUT}/train')\nos.mkdir(f'{OUT}/test')\nfor site in os.listdir(f'{PATH}/train'):\n    os.mkdir(f'{OUT}/train/{site}')\n    for floor in os.listdir(f'{PATH}/train/{site}'):\n        os.mkdir(f'{OUT}/train/{site}/{floor}')","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)","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(test_files+used):\n    f = client.submit(read_wifi_ssid_bssid,fname) \n    futures.append(f) \n_ = wait(futures)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\nss = []\nbs = []\nfor f in tqdm(futures):\n    s,b = f.result()\n    ss.append(s)\n    bs.append(b)\nss = np.concatenate(ss)\nbs = np.concatenate(bs)\nss.shape, bs.shape","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n\nlbl_ssid = LabelEncoder()\nlbl_ssid.fit(ss)\nprint(lbl_ssid.classes_.shape)\nlbl_ssid.classes_","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n\nlbl_bssid = LabelEncoder()\nlbl_bssid.fit(bs)\nprint(lbl_bssid.classes_.shape)\nlbl_bssid.classes_","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Let's look at one encode result."},{"metadata":{"trusted":false},"cell_type":"code","source":"encode_wifi_ssid_bssid(name, lbl_ssid, lbl_bssid)\nname = name.replace(PATH,OUT)\ndf = pd.read_csv(name)\ndf.head()","execution_count":null,"outputs":[]},{"metadata":{"trusted":false},"cell_type":"code","source":"%%time\n\nlbl_ssid_f = client.scatter(lbl_ssid)\nlbl_bssid_f = client.scatter(lbl_bssid)\n_ = wait(lbl_ssid_f)\n_ = wait(lbl_bssid_f)","execution_count":null,"outputs":[]},{"metadata":{"trusted":false},"cell_type":"code","source":"%%time\nfutures = [] # save the future since dask is lazy, otherwise nothing is executed.\n\nfor fname in tqdm(test_files+used):\n    f = client.submit(encode_wifi_ssid_bssid, fname, lbl_ssid_f, lbl_bssid_f)\n    futures.append(f) \n_ = wait(futures)","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}