onestop_data_clean.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "data_clean.ipynb",
      "provenance": [],
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "accelerator": "GPU"
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/ppiont/tensor-flow-state/blob/master/onestop_data_clean.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "GQ7f4fQLqMXt",
        "colab_type": "code",
        "outputId": "a28f8181-db6f-418a-f45f-bb03272a36a2",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 35
        }
      },
      "source": [
        "from google.colab import drive\n",
        "drive.mount(\"/gdrive\", force_remount = True)"
      ],
      "execution_count": 0,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "Mounted at /gdrive\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "jnge-3rHqWUy",
        "colab_type": "code",
        "outputId": "85e99511-97a7-433e-84a3-73ce94de0734",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 35
        }
      },
      "source": [
        "%cd \"/gdrive/My Drive/tensor-flow-state/tensor-flow-state\""
      ],
      "execution_count": 0,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "/gdrive/My Drive/tensor-flow-state/tensor-flow-state\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "DtrvUXW94U4F",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import pandas as pd\n",
        "import numpy as np"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "2RnHSY3Xsbpy",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "data_dir = \"data/\"\n",
        "# Define sensors to process\n",
        "sensor_name_list = [\"RWS01_MONIBAS_0021hrl0414ra\", \"RWS01_MONIBAS_0021hrl0403ra\", \"RWS01_MONIBAS_0021hrl0409ra\", \"RWS01_MONIBAS_0021hrl0420ra\", \"RWS01_MONIBAS_0021hrl0426ra\"]"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "GhGB88g_-9Lu",
        "colab_type": "text"
      },
      "source": [
        "### ------------------------------------------------------------ START OF MESSING AROUND ------------------------------------------------------------"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "82mH2GauD06Y",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        ""
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "wcnAN544-21o",
        "colab_type": "text"
      },
      "source": [
        "### ------------------------------------------------------------ END OF MESSING AROUND ------------------------------------------------------------"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "a4GQG9DY3J_F",
        "colab_type": "text"
      },
      "source": [
        "### Clean sensor data"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "FmXOKIfAwgrw",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import datetime\n",
        "def dateparse (time_in_secs):\n",
        "    # Unix/epoch time to \"YYYY-MM-DD HH:MM:SS\"\n",
        "    return datetime.datetime.fromtimestamp(float(time_in_secs))\n",
        "\n",
        "def repair_datetime_index(df, freq = \"T\"):\n",
        "    df = df.loc[~df.index.duplicated(keep = \"first\")] # remove duplicate date time indexes\n",
        "    df = df.reindex(pd.date_range(start = df.index.min(), end = df.index.max(), freq = freq)) # add missing date time indexes\n",
        "    df.index = df.index.tz_localize(\"UTC\").tz_convert(\"Europe/Amsterdam\")\n",
        "\n",
        "    return df\n",
        "\n",
        "def fix_values(df):\n",
        "    # The order of these operations is currently important! Pay attention when making changes\n",
        "    df[\"speed_limit\"] = np.where((df.index.hour < 19) & (df.index.hour >= 6), 100, 130)\n",
        "    df.loc[df.flow < 0, \"flow\"] = np.nan # flow is either -2 (missing data) or 0 or positive. -2 to nan\n",
        "    df.loc[df.speed < -1, \"speed\"] = np.nan # -2 (missing data) as well as oddities (-1.33, an average over -2 and -1 lanes?) to nan \n",
        "    df.speed.mask(df.speed == -1, df.speed_limit, inplace = True) # -1 means no cars, setting it to speed limit\n",
        "    df.loc[(df.speed < 0) & (df.speed > -1), \"speed\"] = 0 # anything else below zero is between 0 and -1, occuring when some lanes have non-moving cars while others have have no cars.\n",
        "    df.speed.mask(df.speed > df.speed_limit, df.speed_limit, inplace = True) # cap speed at speed_limit, since higher speed dosn't add to representation\n",
        "    \n",
        "    return df\n",
        "\n",
        "import os\n",
        "def reduce_cols(sensors, path_in = \"data/ndw_raw/\", path_out = \"data/\"):\n",
        "    sensor_df_list = list()\n",
        "    for sensor in sensors:\n",
        "        df = pd.read_csv(os.path.join(path_in, sensor + \".csv\"), header = None, \\\n",
        "                         usecols = [0, 86, 87], names = [\"timestamp\", \"speed\", \"flow\"], \\\n",
        "                         index_col = \"timestamp\", parse_dates = True, date_parser = dateparse)\n",
        "        df.flow /= 60 # change flow unit to min^-1\n",
        "        df = repair_datetime_index(df)\n",
        "        df = fix_values(df)        \n",
        "        #df.to_csv(path_out + sensor)\n",
        "        sensor_df_list.append(df)\n",
        "\n",
        "    return sensor_df_list"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "iTUTI6Ivwgh2",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "sensor_df_list = reduce_cols(sensor_name_list)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "tNmTJSBP3ABY",
        "colab_type": "text"
      },
      "source": [
        "### Join Sensors"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "6Y0yV1GR3AE0",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "def join_sensors(sensor_df_list, sensor_name_list):\n",
        "    combined_df = pd.DataFrame({\"timestamp\": pd.date_range(start = \"2011-01-01\", end = \"2019-12-31\", freq = \"T\")})\n",
        "    combined_df.set_index(\"timestamp\", drop = True, inplace = True)\n",
        "    combined_df.index = combined_df.index.tz_localize(\"UTC\").tz_convert(\"Europe/Amsterdam\")\n",
        "    d = {}\n",
        "    for i, sensor in enumerate(sensor_df_list):\n",
        "        # only add speed limit on the final sensor\n",
        "        if i == len(sensor_df_list) - 1:\n",
        "            d[sensor_name_list[i]] = sensor_df_list[i]\n",
        "            combined_df = combined_df.join(d[sensor_name_list[i]], how = \"outer\", rsuffix = '_' + sensor_name_list[i])\n",
        "        else:\n",
        "            d[sensor_name_list[i]] = sensor_df_list[i].iloc[:, :2]\n",
        "            combined_df = combined_df.join(d[sensor_name_list[i]], how = \"outer\", rsuffix = \"_\" + sensor_name_list[i])\n",
        "    combined_df.dropna(how = \"all\", axis = 0, inplace = True) # this works in all cases because speed_limit is never NA on a sensor df\n",
        "    \n",
        "    return combined_df"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "UOF-7-DH3AIl",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# Join sensors to one table\n",
        "df = join_sensors(sensor_df_list, sensor_name_list)\n",
        "\n",
        "# Rename and reorder columns\n",
        "df.rename({\"speed_RWS01_MONIBAS_0021hrl0403ra\": \"speed_-2\", \"speed_RWS01_MONIBAS_0021hrl0409ra\": \"speed_-1\",\\\n",
        "           \"speed_RWS01_MONIBAS_0021hrl0420ra\": \"speed_+1\", \"speed_RWS01_MONIBAS_0021hrl0426ra\": \"speed_+2\",\\\n",
        "           \"flow_RWS01_MONIBAS_0021hrl0403ra\": \"flow_-2\", \"flow_RWS01_MONIBAS_0021hrl0409ra\": \"flow_-1\",\\\n",
        "           \"flow_RWS01_MONIBAS_0021hrl0420ra\": \"flow_+1\", \"flow_RWS01_MONIBAS_0021hrl0426ra\": \"flow_+2\"\\\n",
        "           }, axis = 1, inplace = True)\n",
        "col_order = [\"speed\", \"flow\", \"speed_-2\", \"speed_-1\",\"speed_+1\", \"speed_+2\", \"flow_-2\", \"flow_-1\", \"flow_+1\", \"flow_+2\", \"speed_limit\"]\n",
        "df = df[col_order]\n",
        "\n",
        "# Save table to csv\n",
        "#df.to_csv(data_dir + \"combined_df.csv\")"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "9dKPSgmOL31_",
        "colab_type": "code",
        "outputId": "e4e8b4df-0f58-4b4f-80a8-7732a1b33051",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 202
        }
      },
      "source": [
        "df.head()"
      ],
      "execution_count": 0,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/html": [
              "<div>\n",
              "<style scoped>\n",
              "    .dataframe tbody tr th:only-of-type {\n",
              "        vertical-align: middle;\n",
              "    }\n",
              "\n",
              "    .dataframe tbody tr th {\n",
              "        vertical-align: top;\n",
              "    }\n",
              "\n",
              "    .dataframe thead th {\n",
              "        text-align: right;\n",
              "    }\n",
              "</style>\n",
              "<table border=\"1\" class=\"dataframe\">\n",
              "  <thead>\n",
              "    <tr style=\"text-align: right;\">\n",
              "      <th></th>\n",
              "      <th>speed</th>\n",
              "      <th>flow</th>\n",
              "      <th>speed_-2</th>\n",
              "      <th>speed_-1</th>\n",
              "      <th>speed_+1</th>\n",
              "      <th>speed_+2</th>\n",
              "      <th>flow_-2</th>\n",
              "      <th>flow_-1</th>\n",
              "      <th>flow_+1</th>\n",
              "      <th>flow_+2</th>\n",
              "      <th>speed_limit</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>2011-01-03 01:00:00+01:00</th>\n",
              "      <td>122.000000</td>\n",
              "      <td>8.0</td>\n",
              "      <td>100.000000</td>\n",
              "      <td>130.000000</td>\n",
              "      <td>118.5</td>\n",
              "      <td>129.750000</td>\n",
              "      <td>8.0</td>\n",
              "      <td>6.0</td>\n",
              "      <td>4.0</td>\n",
              "      <td>6.0</td>\n",
              "      <td>130.0</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2011-01-03 01:01:00+01:00</th>\n",
              "      <td>130.000000</td>\n",
              "      <td>5.0</td>\n",
              "      <td>115.000000</td>\n",
              "      <td>126.333333</td>\n",
              "      <td>118.0</td>\n",
              "      <td>104.000000</td>\n",
              "      <td>18.0</td>\n",
              "      <td>6.0</td>\n",
              "      <td>5.0</td>\n",
              "      <td>2.0</td>\n",
              "      <td>130.0</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2011-01-03 01:02:00+01:00</th>\n",
              "      <td>111.000000</td>\n",
              "      <td>2.0</td>\n",
              "      <td>112.666667</td>\n",
              "      <td>104.000000</td>\n",
              "      <td>91.0</td>\n",
              "      <td>112.666667</td>\n",
              "      <td>14.0</td>\n",
              "      <td>2.0</td>\n",
              "      <td>4.0</td>\n",
              "      <td>8.0</td>\n",
              "      <td>130.0</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2011-01-03 01:03:00+01:00</th>\n",
              "      <td>105.333333</td>\n",
              "      <td>10.0</td>\n",
              "      <td>118.000000</td>\n",
              "      <td>108.666667</td>\n",
              "      <td>120.0</td>\n",
              "      <td>130.000000</td>\n",
              "      <td>14.0</td>\n",
              "      <td>6.0</td>\n",
              "      <td>9.0</td>\n",
              "      <td>10.0</td>\n",
              "      <td>130.0</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2011-01-03 01:04:00+01:00</th>\n",
              "      <td>130.000000</td>\n",
              "      <td>8.0</td>\n",
              "      <td>105.500000</td>\n",
              "      <td>121.000000</td>\n",
              "      <td>127.5</td>\n",
              "      <td>117.500000</td>\n",
              "      <td>21.0</td>\n",
              "      <td>13.0</td>\n",
              "      <td>7.0</td>\n",
              "      <td>5.0</td>\n",
              "      <td>130.0</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "                                speed  flow  ...  flow_+2  speed_limit\n",
              "2011-01-03 01:00:00+01:00  122.000000   8.0  ...      6.0        130.0\n",
              "2011-01-03 01:01:00+01:00  130.000000   5.0  ...      2.0        130.0\n",
              "2011-01-03 01:02:00+01:00  111.000000   2.0  ...      8.0        130.0\n",
              "2011-01-03 01:03:00+01:00  105.333333  10.0  ...     10.0        130.0\n",
              "2011-01-03 01:04:00+01:00  130.000000   8.0  ...      5.0        130.0\n",
              "\n",
              "[5 rows x 11 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 9
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "CrDjuh7i3AL0",
        "colab_type": "text"
      },
      "source": [
        "### Impute data"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Gu-qf0zY3APl",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "cols = col_order\n",
        "speed_cols = [\"speed\", \"speed_-2\", \"speed_-1\",\"speed_+1\", \"speed_+2\"]\n",
        "flow_cols = [\"flow\", \"flow_-2\", \"flow_-1\", \"flow_+1\", \"flow_+2\"]"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "cPqnzTiC3jv1",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# Where values are missing in one or more sensors, but are present in others, impute with mean of others\n",
        "def fill_na_row_mean(df):\n",
        "    row_avgs = df.mean(axis = 1).values.reshape(-1, 1)\n",
        "    df = df.fillna(0) + df.isna().values * row_avgs\n",
        "    return df"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "q57xMK6S3jy0",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "speed_df = fill_na_row_mean(df[speed_cols])\n",
        "flow_df = fill_na_row_mean(df[flow_cols])"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "xgWSgxrh3j2U",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "df = speed_df.join(flow_df, how = \"inner\").join(df[[\"speed_limit\"]], how = \"inner\")"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "GZC2hHl03j40",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# Interpolate null vals for the first week of data of speed and flow cols\n",
        "def interpolate_week(df, cols):\n",
        "    week = 7 * 24 * 60\n",
        "    for col in cols: \n",
        "        df.iloc[:week, df.columns.get_loc(col)] = df[col][:week].interpolate(method = \"time\")\n",
        "    return df\n",
        "\n",
        "# Replace remaining nulls with value from 1 week previous\n",
        "def shift_week(df, cols):\n",
        "    # Use RangeIndex for the this operation\n",
        "    df[\"timestamp\"] = df.index\n",
        "    df.reset_index(drop = True, inplace = True)\n",
        "    week = 7 * 24 * 60\n",
        "    for col in cols:\n",
        "        col_index = df.columns.get_loc(col)\n",
        "        for row in df.itertuples():\n",
        "            if np.isnan(row[col_index + 1]):\n",
        "                df.iat[row[0], col_index] = df.iat[(row[0] - week), col_index]\n",
        "    # Return to DateTimeIndex again\n",
        "    df.set_index(pd.to_datetime(df.timestamp.values), inplace = True) \n",
        "    df.drop(\"timestamp\", axis = 1, inplace = True)\n",
        "    return df"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "OLHJF-oi3ASk",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "df = interpolate_week(df, cols)\n",
        "df = shift_week(df, cols)\n",
        "#df.to_csv(\"data/df_imputed_week_shift.csv\")"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Mgz3j9ic4n5k",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import holidays\n",
        "df[\"density\"] = (df.flow * 60) / df.speed\n",
        "df[\"weekend\"] = np.where(df.index.weekday > 4, 1, 0).astype(np.int16)\n",
        "df[\"holiday\"] = np.array([int(x in holidays.NL()) for x in df.index]).astype(np.int16)\n",
        "df[\"speed_limit\"] =  np.where(df.speed_limit > 115, 1, 0)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "nM6GKSSZROoM",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "df.to_csv(\"data/df_imputed_week_shift_added_holiday_weekends_speed_limit_130.csv\")"
      ],
      "execution_count": 0,
      "outputs": []
    }
  ]
}