Roboust_Route_Planner_FullDataset.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ROBOUST ROUTE PLANNER\n",
    "\n",
    "### Team: SAFJ\n",
    "\n",
    "### Members: Serif Soner Serbest - Jelena Banjac - Fatine Benhsain -  Asli Yorusun\n",
    "\n",
    "Using the inputs given, this notebook calculates possible routes from one station to another at a given time of a given day, taking into consideration the confidence of each connection / route."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook unites the information from Data Analysis and Distance Analysis notebooks. Therefore, please take a look on those if there are any sections considered uncommented.\n",
    "\n",
    "### TABLE OF CONTENTS\n",
    "\n",
    "#### 1. [Data](#1)\n",
    "\n",
    "In this section we import modify the dataset \n",
    "#### 2. [Confidence Calculation](#2)\n",
    "In this section we build necessary functions and dataframes to calculate confidence for a given route\n",
    "#### 3. [Connection Graph](#3)\n",
    "In this section we calculate the connection graph that shows the reachability of any station pairs.\n",
    "#### 4. [Timetable](#4)\n",
    "In this section we create timetable that provide information about departure and arrival stations and times for a given day\n",
    "#### 5. [Route](#5)\n",
    "In this section we calculate each possible routes by using our timetable and connection graph and provide confidence level of the route with confidence calculation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### System Parameters"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We start by defining the parameters used in our study : "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# change from one station to another in mins\n",
    "transfer_delay = 5 \n",
    "\n",
    "# average walking speed is assumed to be 4.5 km/h\n",
    "# ref: https://www.quora.com/What-is-the-average-walking-speed-of-a-human\n",
    "# in minutes\n",
    "max_walking_time = 5\n",
    "\n",
    "# m/min which corresponds to 4.5 km/h\n",
    "human_speed = 75 \n",
    "\n",
    "# as meters\n",
    "max_walking_distance = human_speed * max_walking_time "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Dependencies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pickle\n",
    "import socket\n",
    "import getpass\n",
    "import os\n",
    "\n",
    "import ast\n",
    "import numpy as np\n",
    "from scipy import stats\n",
    "import pandas as pd\n",
    "\n",
    "import networkx as nx\n",
    "from datetime import datetime, timedelta\n",
    "\n",
    "import math\n",
    "from math import sin, cos, sqrt, atan2, radians\n",
    "\n",
    "import random\n",
    "import json\n",
    "import matplotlib.patches as mpatches\n",
    "from py4j.protocol import Py4JJavaError\n",
    "import networkx as nx\n",
    "\n",
    "import matplotlib.pyplot as plt\n",
    "%matplotlib inline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We change the layout in order to be able to see spark dataframes:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style>.container { width:95% !important; }</style>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "plt.rcParams['figure.figsize'] = (10,6)\n",
    "plt.rcParams['font.size'] = 18\n",
    "plt.style.use('fivethirtyeight')\n",
    "\n",
    "def fix_layout(width:int=95):\n",
    "    from IPython.core.display import display, HTML\n",
    "    display(HTML('<style>.container { width:' + str(width) + '% !important; }</style>'))\n",
    "    \n",
    "fix_layout()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We set up Spark depending on whether we work locally or on the cluster :"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Spark Setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "username = getpass.getuser()\n",
    "\n",
    "SPARK_LOCAL = False\n",
    "\n",
    "# on the laptop\n",
    "if not 'iccluster' in socket.gethostname():\n",
    "    # set this to the base spark directory on your system\n",
    "    SPARK_LOCAL = True\n",
    "    \n",
    "    if username == \"fatine\":\n",
    "        spark_home = '/home/fatine/spark-2.4.1-bin-hadoop2.7'\n",
    "        \n",
    "        try:\n",
    "            import findspark\n",
    "            findspark.init(spark_home)\n",
    "        except ModuleNotFoundError as e:\n",
    "            print('Info: {}'.format(e))\n",
    "    elif username == \"soner\":\n",
    "        spark_home = '/home/soner/Desktop/DSLAB2019/spark-2.4.1-bin-hadoop2.7'\n",
    "        \n",
    "        try:\n",
    "            import findspark\n",
    "            findspark.init(spark_home)\n",
    "        except ModuleNotFoundError as e:\n",
    "            print('Info: {}'.format(e))\n",
    "            \n",
    "    elif username == \"jelena\":\n",
    "        pass\n",
    "        \n",
    "    \n",
    "        \n",
    "# cluster\n",
    "if username == \"jbanjac\":\n",
    "    ROOT_PATH = \"/home/jbanjac/robust-journey-planning\"\n",
    "    os.environ['PYSPARK_PYTHON'] = '/opt/anaconda3/bin/python'\n",
    "# local\n",
    "elif username == \"jelena\":\n",
    "    ROOT_PATH = os.getcwd()\n",
    "    os.environ['PYSPARK_PYTHON'] = '/home/jelena/anaconda3/bin/python'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql.functions import unix_timestamp, udf, desc\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql import Window\n",
    "from pyspark.sql.types import IntegerType\n",
    "\n",
    "from pyspark.ml.feature import VectorAssembler,Normalizer,  PCA,VectorIndexer\n",
    "from pyspark.ml.clustering import KMeans, BisectingKMeans, GaussianMixture"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "if username == \"jelena\"  or username == \"fatine\":\n",
    "    spark = (SparkSession \\\n",
    "                .builder \\\n",
    "                .appName('sbb-{0}'.format(getpass.getuser())) \\\n",
    "                .master('local[4]') \\\n",
    "                .config('spark.driver.memory', '10g') \\\n",
    "                .config('spark.executor.memory', '4g') \\\n",
    "                .config('spark.executor.instances', '5') \\\n",
    "                .config('spark.port.maxRetries', '100') \\\n",
    "                .getOrCreate())\n",
    "    CLUSTER_URL = \"hdfs://iccluster042.iccluster.epfl.ch:8020\"\n",
    "\n",
    "elif SPARK_LOCAL:\n",
    "    spark = SparkSession \\\n",
    "                .builder \\\n",
    "                .master(\"local\") \\\n",
    "                .appName(\"roboust-journey-planing\") \\\n",
    "                .config(\"spark.driver.host\", \"localhost\") \\\n",
    "                .getOrCreate()\n",
    "    CLUSTER_URL = \"\"\n",
    "else:\n",
    "    spark = SparkSession \\\n",
    "                .builder \\\n",
    "                .master(\"yarn\") \\\n",
    "                .appName('sbb-{0}'.format(getpass.getuser())) \\\n",
    "                .config('spark.executor.memory', '4g') \\\n",
    "                .config('spark.executor.instances', '5') \\\n",
    "                .config('spark.port.maxRetries', '100') \\\n",
    "                .getOrCreate()\n",
    "    CLUSTER_URL = \"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://nx-sb-srv-11-1.epfl.ch:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.4.1</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[4]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>sbb-fatine</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7f28973e41d0>"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc = spark.sparkContext\n",
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='1'></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## DATA"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In order to give the right routes and to build our confidence estimations, we need to import and use data extracted from the sbb websites.\n",
    "\n",
    "We start by reading the csv files containing our data :"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "if SPARK_LOCAL:\n",
    "    if username == \"fatine\":\n",
    "        df_zurich_full = spark.read.csv(f\"/home/fatine/Documents/Cours Semestre Printemps/Lab in DS/*/*.csv\", header=True, sep=';').cache()\n",
    "    elif username == \"soner\":\n",
    "        pass\n",
    "    elif username == \"jelena\":\n",
    "        df_zurich_full = spark.read.csv(f\"{CLUSTER_URL}/datasets/sbb/*/*/*.csv.bz2\", sep=';', header=True).cache()\n",
    "else:\n",
    "    df_zurich_full = spark.read.csv(f\"{CLUSTER_URL}/datasets/sbb/*/*/*.csv.bz2\", sep=';', header=True).cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can now inspect our data and clean it according to our needs, for example:\n",
    "    - we translate the columns from German to English\n",
    "    - we only keep the stations in the Zurich area\n",
    "    - we drop the connections that have null times because they are not useful for us without the adequate time\n",
    "    - we fix the station names that didn't get formatted correctly"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "def filtering_of_dataframe(df):\n",
    "    \"\"\" Filter dataframe\n",
    "    \n",
    "    This method cleans the unneeded rows from the dataframe that we concluded we don't need in Dataset Analysis notebook.\n",
    "    It will be both used for the whoe dataset as well as for the 1-day dataframes.\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    df: spark dataframe\n",
    "        Dataframe that contains trips information from SBB\n",
    "        \n",
    "    Returns\n",
    "    -------\n",
    "    df: spark dataframe\n",
    "        Dataframe that is filtered with data we need for the best route prediction\n",
    "    \"\"\"\n",
    "    df = df.withColumnRenamed(\"BETRIEBSTAG\", \"TRIP_DATE\")\\\n",
    "            .withColumnRenamed(\"FAHRT_BEZEICHNER\", \"TRIP_ID\")\\\n",
    "            .withColumnRenamed(\"BETREIBER_ID\", \"OPERATOR_ID\")\\\n",
    "            .withColumnRenamed(\"BETREIBER_ABK\", \"OPERATOR_ABK\")\\\n",
    "            .withColumnRenamed(\"BETREIBER_NAME\", \"OPERATOR_NAME\")\\\n",
    "            .withColumnRenamed(\"PRODUKT_ID\", \"TRANSPORT_TYPE\")\\\n",
    "            .withColumnRenamed(\"LINIEN_ID\", \"TRAIN_ID\")\\\n",
    "            .withColumnRenamed(\"LINIEN_TEXT\", \"TRAIN_NAME\")\\\n",
    "            .withColumnRenamed(\"UMLAUF_ID\", \"CIRCULATING_ID\")\\\n",
    "            .withColumnRenamed(\"VERKEHRSMITTEL_TEXT\", \"SERVICE_TYPE\")\\\n",
    "            .withColumnRenamed(\"ZUSATZFAHRT_TF\", \"ADDITIONAL_DRIVING\")\\\n",
    "            .withColumnRenamed(\"FAELLT_AUS_TF\", \"FAILED\")\\\n",
    "            .withColumnRenamed(\"BPUIC\", \"STATION_ID\")\\\n",
    "            .withColumnRenamed(\"HALTESTELLEN_NAME\", \"STATION_NAME\")\\\n",
    "            .withColumnRenamed(\"ANKUNFTSZEIT\", \"SCHEDULE_ARRIVE_TIME\")\\\n",
    "            .withColumnRenamed(\"AN_PROGNOSE\", \"ACTUAL_ARRIVE_TIME\")\\\n",
    "            .withColumnRenamed(\"AN_PROGNOSE_STATUS\", \"ACTUAL_ARRIVE_TIME_STATUS\")\\\n",
    "            .withColumnRenamed(\"ABFAHRTSZEIT\", \"SCHEDULE_DEPART_TIME\")\\\n",
    "            .withColumnRenamed(\"AB_PROGNOSE\", \"ACTUAL_DEPART_TIME\")\\\n",
    "            .withColumnRenamed(\"AB_PROGNOSE_STATUS\", \"ACTUAL_DEPART_TIME_STATUS\")\\\n",
    "            .withColumnRenamed(\"DURCHFAHRT_TF\", \"PASSES_BY\")\n",
    "    \n",
    "    \n",
    "    # load Zurich stations set\n",
    "    with open('distance/zurich_stations_set.pickle', 'rb') as handle:\n",
    "        zurich_stations_set = pickle.load(handle)\n",
    "\n",
    "    df = df.where(F.col(\"STATION_ID\").isin(zurich_stations_set)).cache()\n",
    "    \n",
    "    # For now we drop the connections with a null time\n",
    "    df = df.na.drop(subset=[\"SCHEDULE_ARRIVE_TIME\", \"SCHEDULE_DEPART_TIME\"])\n",
    "    \n",
    "    df = df.filter(\"TRANSPORT_TYPE is not null\")\n",
    "    df = df.filter(df.ADDITIONAL_DRIVING==False)\n",
    "    df = df.filter(df.FAILED==False)\n",
    "\n",
    "    @F.udf\n",
    "    def fix_station_name(station_name):\n",
    "        fixed_station_name = station_name.replace(\"�\", \"ü\")\n",
    "        return fixed_station_name\n",
    "\n",
    "    df = df.withColumn('STATION_NAME', fix_station_name(df.STATION_NAME))\n",
    "    df = df.filter(df.PASSES_BY == False)\n",
    "\n",
    "    df = df.filter(df.ACTUAL_ARRIVE_TIME_STATUS != \"UNBEKANNT\")\n",
    "    df = df.filter(df.ACTUAL_DEPART_TIME_STATUS != \"UNBEKANNT\")\n",
    "\n",
    "    df = df.filter(\"ACTUAL_ARRIVE_TIME is not null and SCHEDULE_ARRIVE_TIME is not null\")\n",
    "    df = df.filter(\"ACTUAL_ARRIVE_TIME is not null and SCHEDULE_ARRIVE_TIME is not null\")\n",
    "    df = df.filter(\"not(ACTUAL_DEPART_TIME is null and SCHEDULE_DEPART_TIME is null)\")\n",
    "    df = df.filter(\"ACTUAL_DEPART_TIME is not null and SCHEDULE_DEPART_TIME is not null\")\n",
    "    \n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- TRIP_DATE: string (nullable = true)\n",
      " |-- TRIP_ID: string (nullable = true)\n",
      " |-- OPERATOR_ID: string (nullable = true)\n",
      " |-- OPERATOR_ABK: string (nullable = true)\n",
      " |-- OPERATOR_NAME: string (nullable = true)\n",
      " |-- TRANSPORT_TYPE: string (nullable = true)\n",
      " |-- TRAIN_ID: string (nullable = true)\n",
      " |-- TRAIN_NAME: string (nullable = true)\n",
      " |-- CIRCULATING_ID: string (nullable = true)\n",
      " |-- SERVICE_TYPE: string (nullable = true)\n",
      " |-- ADDITIONAL_DRIVING: string (nullable = true)\n",
      " |-- FAILED: string (nullable = true)\n",
      " |-- STATION_ID: string (nullable = true)\n",
      " |-- STATION_NAME: string (nullable = true)\n",
      " |-- SCHEDULE_ARRIVE_TIME: string (nullable = true)\n",
      " |-- ACTUAL_ARRIVE_TIME: string (nullable = true)\n",
      " |-- ACTUAL_ARRIVE_TIME_STATUS: string (nullable = true)\n",
      " |-- SCHEDULE_DEPART_TIME: string (nullable = true)\n",
      " |-- ACTUAL_DEPART_TIME: string (nullable = true)\n",
      " |-- ACTUAL_DEPART_TIME_STATUS: string (nullable = true)\n",
      " |-- PASSES_BY: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_zurich_full = filtering_of_dataframe(df_zurich_full)\n",
    "df_zurich_full.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='2'></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## CONFIDENCE CALCULATION"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can now proceed to compute the confidence of connections.\n",
    "\n",
    "We start by creating some helper functions:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# creating helper UDF functions\n",
    "delay_arrive = F.unix_timestamp('ACTUAL_ARRIVE_TIME', format='dd.MM.yyyy HH:mm:ss') - F.unix_timestamp('SCHEDULE_ARRIVE_TIME', format='dd.MM.yyyy HH:mm')\n",
    "delay_depart = F.unix_timestamp('ACTUAL_DEPART_TIME', format='dd.MM.yyyy HH:mm:ss') - F.unix_timestamp('SCHEDULE_DEPART_TIME', format='dd.MM.yyyy HH:mm')\n",
    "\n",
    "@F.udf\n",
    "def convert_to_min(delay_sec):\n",
    "    minutes = math.ceil(delay_sec/60)\n",
    "    return minutes\n",
    "\n",
    "@F.udf\n",
    "def convert_to_weekday_1(date):\n",
    "    return str(datetime.strptime(date, '%d.%m.%Y').strftime('%w'))\n",
    "\n",
    "@F.udf\n",
    "def convert_to_weekday_2(date):\n",
    "    return str(datetime.strptime(date, '%d.%m.%Y %H:%M:%S').strftime('%w'))\n",
    "\n",
    "@F.udf\n",
    "def convert_to_hour(date):\n",
    "    return str(datetime.strptime(date, '%d.%m.%Y %H:%M:%S').hour)\n",
    "\n",
    "@F.udf\n",
    "def convert_to_month_1(date):\n",
    "    return str(datetime.strptime(date, '%d.%m.%Y').month)\n",
    "\n",
    "@F.udf\n",
    "def convert_to_month_2(date):\n",
    "    return str(datetime.strptime(date, '%d.%m.%Y %H:%M:%S').month)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using our defined methods, we can now proceed to add the delays to our dataframe through several columns for the minutes, weekday, month and hour."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# creating additional columns we need to store delay values\n",
    "df_zurich_full = df_zurich_full.withColumn(\"delay_arrive\", delay_arrive).cache()\n",
    "df_zurich_full = df_zurich_full.withColumn(\"delay_depart\", delay_depart)\n",
    "df_zurich_full = df_zurich_full.withColumn('delay_arrive', convert_to_min(df_zurich_full.delay_arrive))\n",
    "df_zurich_full = df_zurich_full.withColumn('delay_depart', convert_to_min(df_zurich_full.delay_depart))\n",
    "df_zurich_full = df_zurich_full.withColumn(\"TRIP_DATE_month\",convert_to_month_1(df_zurich_full['TRIP_DATE']))\\\n",
    "                               .withColumn('TRIP_DATE_week_day', convert_to_weekday_1(df_zurich_full['TRIP_DATE']))\n",
    "    \n",
    "df_zurich_full = df_zurich_full.withColumn(\"ACTUAL_ARRIVE_TIME_month\", convert_to_month_2(df_zurich_full['ACTUAL_ARRIVE_TIME']))\\\n",
    "                               .withColumn('ACTUAL_ARRIVE_TIME_week_day', convert_to_weekday_2(df_zurich_full['ACTUAL_ARRIVE_TIME']))\\\n",
    "                               .withColumn(\"ACTUAL_ARRIVE_TIME_hour\", convert_to_hour(df_zurich_full['ACTUAL_ARRIVE_TIME']))\n",
    "\n",
    "df_zurich_full = df_zurich_full.withColumn(\"ACTUAL_DEPART_TIME_month\", convert_to_month_2(df_zurich_full['ACTUAL_DEPART_TIME']))\\\n",
    "                               .withColumn('ACTUAL_DEPART_TIME_week_day', convert_to_weekday_2(df_zurich_full['ACTUAL_DEPART_TIME']))\\\n",
    "                               .withColumn(\"ACTUAL_DEPART_TIME_hour\", convert_to_hour(df_zurich_full['ACTUAL_DEPART_TIME']))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the Data Analysis notebook, we have previously defined some clusters for different attributes and saved them in json files for reuse, to avoid recomputing them each time and to use them here too. So now, we can read them from the respective files : Each one is a json containing the cluster numbers and the different entries for each cluster."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_cluster_num(s, clusters):\n",
    "    \"\"\" Get the cluster number of the element\n",
    "    \n",
    "    This method locates in which cluster the element belongs. E.g. service_type 85:11 is cluster 1, etc.\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    s: string\n",
    "        Element for which we want to get the cluster number\n",
    "    clusters: dict\n",
    "        Dictionary of cluster keys and element values. E.g. {'1': ['85:11', '85:30', ...], '2': [...], ...}\n",
    "        \n",
    "    Returns\n",
    "    -------\n",
    "    k: string\n",
    "        The number of the cluster\n",
    "    \"\"\"\n",
    "    for k, v in clusters.items():\n",
    "        if s in clusters[k]:\n",
    "            return k\n",
    "    return None\n",
    "\n",
    "def read_clusters(file_name):\n",
    "    \"\"\" Read clusters from the file (they were calculated in Dataset Analysis notebook)\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    file_name: string\n",
    "        File name where dictionary of clusters is stored\n",
    "        \n",
    "    Returns\n",
    "    -------\n",
    "    clusters: dict\n",
    "        Clusters dictionary\n",
    "    \"\"\"\n",
    "    with open(f'clusters/{file_name}', 'r') as f:\n",
    "        clusters = json.load(f)\n",
    "    return clusters\n",
    "\n",
    "cluster_SERVICE_TYPE = read_clusters('SERVICE_TYPE.json')\n",
    "cluster_OPERATOR_ID = read_clusters('OPERATOR_ID.json')\n",
    "cluster_STATION_ID = read_clusters('STATION_ID.json')\n",
    "\n",
    "cluster_ACTUAL_ARRIVE_TIME_month = read_clusters('ACTUAL_ARRIVE_TIME_month.json')\n",
    "cluster_ACTUAL_ARRIVE_TIME_hour = read_clusters('ACTUAL_ARRIVE_TIME_hour.json')\n",
    "cluster_ACTUAL_ARRIVE_TIME_week_day = read_clusters('ACTUAL_ARRIVE_TIME_week_day.json')\n",
    "\n",
    "cluster_ACTUAL_DEPART_TIME_month = read_clusters('ACTUAL_DEPART_TIME_month.json')\n",
    "cluster_ACTUAL_DEPART_TIME_hour = read_clusters('ACTUAL_DEPART_TIME_hour.json')\n",
    "cluster_ACTUAL_DEPART_TIME_week_day = read_clusters('ACTUAL_DEPART_TIME_week_day.json')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can now use these clusters to compute our probabilities.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "def arrive_distribution(arrive = df_zurich_full, date=None, time=None, trip_id=None, service_type=None, operator_id=None, transport_type=None, station_id=None):\n",
    "    \"\"\" Get arrival delay probability and arrival delay distribution coefficients\n",
    "    \n",
    "    This method calculates the arrival delay probability and it calculates the exponential distribution coefficients\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    arrive: spark dataframe\n",
    "        As default we get the full SBB dataset to calculate the probability\n",
    "    date: string\n",
    "        Date in format %d.%m.%Y\n",
    "    time: string\n",
    "        Time in format %H:%M\n",
    "    trip_id: string\n",
    "        Id of the trip\n",
    "    service_type: string\n",
    "        Type of service of the trip\n",
    "    operator_id: string\n",
    "        Operator id of the trip\n",
    "    transport_type: string\n",
    "        Transport type of the trip\n",
    "    station_id: string\n",
    "        Station ID\n",
    "        \n",
    "    Returns\n",
    "    -------\n",
    "    arrive_delay_distribution: tuple\n",
    "        Coefficients of the exponential distribution\n",
    "    arrive_delay_probability: float\n",
    "        Probability of this arrival setting to be delayed\n",
    "    \"\"\"\n",
    "    \n",
    "    # extract date information\n",
    "    if date:\n",
    "        week_day = str(datetime.strptime(date, '%d.%m.%Y').strftime('%w'))\n",
    "        month = str(datetime.strptime(date, '%d.%m.%Y').month)\n",
    "    if time:\n",
    "        hour = str(datetime.strptime(time, '%H:%M').hour)\n",
    "\n",
    "    if trip_id:\n",
    "        arrive = arrive.filter(arrive.TRIP_ID==trip_id).cache()\n",
    "    if transport_type:\n",
    "        arrive = arrive.filter(arrive.TRANSPORT_TYPE==transport_type).cache()\n",
    "    \n",
    "    oi_cn = get_cluster_num(operator_id, cluster_OPERATOR_ID)\n",
    "    if oi_cn: arrive = arrive.where(arrive.OPERATOR_ID.isin(cluster_OPERATOR_ID[oi_cn])).cache()\n",
    "\n",
    "    st_cn = get_cluster_num(service_type, cluster_SERVICE_TYPE)\n",
    "    if st_cn: arrive = arrive.where(arrive.SERVICE_TYPE.isin(cluster_SERVICE_TYPE[st_cn])).cache()\n",
    "\n",
    "        \n",
    "    si_cn = get_cluster_num(station_id, cluster_STATION_ID)\n",
    "    if si_cn: arrive = arrive.where(arrive.STATION_ID.isin(cluster_STATION_ID[si_cn])).cache()\n",
    "    \n",
    "    aat_m_cn = get_cluster_num(month, cluster_ACTUAL_ARRIVE_TIME_month)\n",
    "    if aat_m_cn: arrive = arrive.where(arrive.ACTUAL_ARRIVE_TIME_month.isin(cluster_ACTUAL_ARRIVE_TIME_month[aat_m_cn])).cache()\n",
    "        \n",
    "    aat_h_cn = get_cluster_num(hour, cluster_ACTUAL_ARRIVE_TIME_hour)\n",
    "    if aat_h_cn: arrive = arrive.where(arrive.ACTUAL_ARRIVE_TIME_hour.isin(cluster_ACTUAL_ARRIVE_TIME_hour[aat_h_cn])).cache()\n",
    "        \n",
    "    aat_wd_cn = get_cluster_num(week_day, cluster_ACTUAL_ARRIVE_TIME_week_day)\n",
    "    if aat_wd_cn: arrive = arrive.where(arrive.ACTUAL_ARRIVE_TIME_week_day.isin(cluster_ACTUAL_ARRIVE_TIME_week_day[aat_wd_cn])).cache()\n",
    "\n",
    "    # sum of all arrival trips\n",
    "    sum_arrive_trips = arrive.count()\n",
    "\n",
    "    # sum of all delayed trips\n",
    "    arrive = arrive.filter('delay_arrive > 0')\n",
    "    sum_delay_arrive_trips = arrive.count()\n",
    "\n",
    "    # get probabilities of arrival delay\n",
    "    arrive_delay_probability = sum_delay_arrive_trips/sum_arrive_trips\n",
    "    data = arrive.select('delay_arrive').toPandas()\n",
    "    arrive_delay_distribution = stats.expon.fit(np.array(list(map(int, data['delay_arrive'].values))), floc=0, scale=1)\n",
    "\n",
    "    return arrive_delay_distribution, arrive_delay_probability"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Probability of catching the next trip"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here we are calculating the probability of catching the next trip. We know the distribution of arrival delays, as well as the time difference between this and the next trip that make a connection. Also, we use the probability of arrival delay to be delayed. After the exploring following papers: \n",
    "\n",
    "- [Stochastic Modelling of Train Delays and Delay Propagation in Stations](https://repository.tudelft.nl/islandora/object/uuid:caa72522-26b1-4088-afc0-59c6e5c346f6/datastream/OBJ/download)\n",
    "- [Adi Botea, Stefano Braghin, \"Contingent versus Deterministic Plans in Multi-Modal Journey Planning\". ICAPS 2015: 268-272](https://dl.acm.org/citation.cfm?id=3038699)\n",
    "- [Mathematical modeling and methods for rescheduling\n",
    "trains under disrupted operations](https://tel.archives-ouvertes.fr/tel-00453640/document)\n",
    "- [Adi Botea, Stefano Braghin, \"Contingent versus Deterministic Plans in Multi-Modal Journey Planning\". ICAPS 2015: 268-272.](https://dl.acm.org/citation.cfm?id=3038699)\n",
    "- [Adi Botea, Evdokia Nikolova, Michele Berlingerio, \"Multi-Modal Journey Planning in the Presence of Uncertainty\". ICAPS 2013.](https://www.aaai.org/ocs/index.php/ICAPS/ICAPS13/paper/view/6023)\n",
    "\n",
    "we decided to use the next way of calculating the probability:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "def calculate_probability_to_catch(arrive_delay_distribution, timediff, arrive_delay_probability):\n",
    "    \"\"\" Calculating the probability of catching the next trip\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    arrive_delay_distribution: tuple\n",
    "        Coefficients of arrive delay exponential distribution\n",
    "    timediff: float\n",
    "        Time difference between this and next trip\n",
    "    arrive_delay_probability: float\n",
    "        Probability of arrival delay\n",
    "    \n",
    "    Returns\n",
    "    -------\n",
    "    p: float\n",
    "        Probability of catching the next trip\n",
    "    \"\"\"\n",
    "    timediff = int(timediff)\n",
    "    quantile = np.arange (0, timediff + 1, 1) \n",
    "    R = stats.expon.cdf(quantile, loc = 0, scale = arrive_delay_distribution[1])\n",
    "    \n",
    "    if arrive_delay_probability == 0:\n",
    "        p = 1\n",
    "    else:\n",
    "        p = (1 - arrive_delay_probability) + arrive_delay_probability * R[timediff]\n",
    "\n",
    "    return p"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "def calculate_confidence(connections_info):\n",
    "    \"\"\" Calculate the confidence of the whole route\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    connections_info: dict\n",
    "        Information of how one found route is connected\n",
    "        \n",
    "    Returns\n",
    "    -------\n",
    "    p: float\n",
    "        Confidence of this connection to succeed\n",
    "    \"\"\"\n",
    "    p = 1.0\n",
    "    \n",
    "    for idx in range(len(connections_info)-1):\n",
    "        \n",
    "        arrive_delay_distribution, arrive_delay_probability = None, None\n",
    "\n",
    "        if connections_info[idx]['transport_type'] != 'walk' and connections_info[idx+1]['transport_type'] != 'walk':\n",
    "\n",
    "            arrive_delay_distribution, arrive_delay_probability = arrive_distribution(date=connections_info[idx]['arrive_date'].strftime('%d.%m.%Y'),\n",
    "                                                          time=connections_info[idx]['arrive_date'].strftime('%H:%M'),\n",
    "                                                          trip_id=connections_info[idx]['trip_id'],\n",
    "                                                          service_type=connections_info[idx]['service_type'],\n",
    "                                                          operator_id=connections_info[idx]['operator_id'],\n",
    "                                                          transport_type=connections_info[idx]['transport_type'],\n",
    "                                                          station_id=connections_info[idx]['arrive_station_id'])\n",
    "\n",
    "            timediff = int((connections_info[idx+1]['depart_date']- connections_info[idx]['arrive_date']).seconds)/60\n",
    "\n",
    "            p *= calculate_probability_to_catch(arrive_delay_distribution, timediff, arrive_delay_probability)\n",
    "            \n",
    "        elif connections_info[idx]['transport_type'] == 'walk' and connections_info[idx+1]['transport_type'] != 'walk':\n",
    "            if idx == 0:\n",
    "                arrive_delay_probability = 0\n",
    "                arrive_delay_distribution = stats.expon.fit(np.zeros(1000), floc=0, scale=1)\n",
    "            \n",
    "            timediff = float((connections_info[idx+1]['depart_date'] - connections_info[idx]['arrive_date']).seconds)/60\n",
    "            \n",
    "            p *= calculate_probability_to_catch(arrive_delay_distribution, timediff, arrive_delay_probability)\n",
    "        \n",
    "        elif connections_info[idx]['transport_type'] != 'walk' and connections_info[idx+1]['transport_type'] == 'walk':\n",
    "            arrive_delay_distribution, arrive_delay_probability = arrive_distribution(date=connections_info[idx]['arrive_date'].strftime('%d.%m.%Y'),\n",
    "                                                          time=connections_info[idx]['arrive_date'].strftime('%H:%M'),\n",
    "                                                          trip_id=connections_info[idx]['trip_id'],\n",
    "                                                          service_type=connections_info[idx]['service_type'],\n",
    "                                                          operator_id=connections_info[idx]['operator_id'],\n",
    "                                                          transport_type=connections_info[idx]['transport_type'],\n",
    "                                                          station_id=connections_info[idx]['arrive_station_id'])\n",
    "            timediff = float((connections_info[idx+1]['depart_date']- connections_info[idx]['arrive_date']).seconds)/60\n",
    "\n",
    "            p *= calculate_probability_to_catch(arrive_delay_distribution, timediff, arrive_delay_probability)\n",
    "        \n",
    "    return p"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='3'></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# CONNECTION GRAPH"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Then we need an adjacency matrix that represents all the possible connections between stations without considering the time at first.\n",
    "\n",
    "For the step-by-step development of this section, please refer to the notebook Robust_Route_Planner_OneDay.ipynb, as well as the Distance_Analysis.ipynb"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import TimestampType, IntegerType, StructType, ArrayType, StructField"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "@F.udf('float')\n",
    "def calculateCrossDistance(lat1, lon1, lat2, lon2):\n",
    "    \"\"\" The distance between two stations, in meters\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    lat1: float\n",
    "        Latitude of 1\n",
    "    lon1: float\n",
    "        Longitude of 1\n",
    "    lat2: float\n",
    "        Latitude of 2\n",
    "    lon2: float\n",
    "        Longitude of 2\n",
    "        \n",
    "    Returns\n",
    "    -------\n",
    "    distance: float\n",
    "        Distance between these two points\n",
    "    \"\"\"\n",
    "    # approximate radius of earth in km\n",
    "    R = 6373.0\n",
    "\n",
    "    # latitude and longitude values of a given station in terms of radians \n",
    "    # for comparing with the Zurich HB's coordinates\n",
    "    lon_1 = radians(lon1)\n",
    "    lat_1 = radians(lat1)\n",
    "    lon_2 = radians(lon2)\n",
    "    lat_2 = radians(lat2)\n",
    "    \n",
    "    # calculates the distance by using Haversine formula\n",
    "    dlon = lon_2 - lon_1\n",
    "    dlat = lat_2 - lat_1\n",
    "    a = sin(dlat / 2)**2 + cos(lat_1) * cos(lat_2) * sin(dlon / 2)**2\n",
    "    c = 2 * atan2(sqrt(a), sqrt(1 - a))\n",
    "\n",
    "    distance = R * c * 1000 # as m\n",
    "    \n",
    "    return distance\n",
    "\n",
    "@F.udf\n",
    "def calculateDistance(latitude, longitude):\n",
    "    \"\"\" The distance between a coordinate and Zurich HB, in kilometers\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    latitude: float\n",
    "        Latitude of the station\n",
    "    longitude: float\n",
    "        Longitude of the station\n",
    "        \n",
    "    Returns\n",
    "    -------\n",
    "    distance: float\n",
    "        Distance from the station to Zurich HB\n",
    "    \"\"\" \n",
    "    \n",
    "    # approximate radius of earth in km\n",
    "    R = 6373.0\n",
    "    Zurich_HB_lat = 47.378178\n",
    "    Zurich_HB_lon = 8.540192\n",
    "    \n",
    "    # latitude and longitude values of Zurich HB in terms of radians\n",
    "    lat_Zurich_HB = radians(Zurich_HB_lat)\n",
    "    lon_Zurich_HB = radians(Zurich_HB_lon)\n",
    "\n",
    "    # latitude and longitude values of a given station in terms of radians \n",
    "    # for comparing with the Zurich HB's coordinates\n",
    "    lon = radians(longitude)\n",
    "    lat = radians(latitude)\n",
    "    \n",
    "    # calculates the distance by using Haversine formula\n",
    "    dlon = lon - lon_Zurich_HB\n",
    "    dlat = lat - lat_Zurich_HB\n",
    "    a = sin(dlat / 2)**2 + cos(lat_Zurich_HB) * cos(lat) * sin(dlon / 2)**2\n",
    "    c = 2 * atan2(sqrt(a), sqrt(1 - a))\n",
    "\n",
    "    distance = R * c # as km\n",
    "    \n",
    "    return distance\n",
    "\n",
    "def read_hrdf(file_name):\n",
    "    \"\"\" Read the HRDF file\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    file_name: string\n",
    "        File name of BFKOORD_GEO\n",
    "    \n",
    "    Returns\n",
    "    -------\n",
    "    data: list of lists\n",
    "        Data in the BFKOORD_GEO file\n",
    "    \"\"\"\n",
    "    with open(file_name, encoding='utf-8') as f:\n",
    "        lines = f.readlines()\n",
    "    \n",
    "    data = []\n",
    "    for line in lines:\n",
    "        data.append([line[:7].strip(), float(line[8:18].strip()), float(line[19:29].strip()), float(line[30:36].strip()), line[38:].strip()])\n",
    "    return data\n",
    "\n",
    "\n",
    "def get_distance_dictionary():\n",
    "    \"\"\" Get distance dictionary\n",
    "    \n",
    "    Returns\n",
    "    -------\n",
    "    distance_dictionary: dict\n",
    "        Get the dictionary with distances of stations\n",
    "    \"\"\"\n",
    "    df = pd.DataFrame(data=read_hrdf('BFKOORD_GEO'), columns=[\"stop_number\", \"longitude\", \"latitude\", \"elevation\", \"stop_name\"])\n",
    "    # convert to Spark DF\n",
    "    df_geo = spark.createDataFrame(df)\n",
    "\n",
    "    # Select the stations around 10 km of Zürich HB\n",
    "    df_zurich_geo = df_geo.withColumn('distance_to_Zurich', calculateDistance(df_geo.latitude,df_geo.longitude)).filter('distance_to_Zurich<=10.0')\n",
    "    \n",
    "    walk_from_df = df_zurich_geo.alias('walk_from_df').withColumnRenamed('longitude', 'longitude_from')\\\n",
    "                                                      .withColumnRenamed('latitude', 'latitude_from')\\\n",
    "                                                      .withColumnRenamed('stop_number', 'stop_number_from')\\\n",
    "                                                      .withColumnRenamed('stop_name', 'stop_name_from')\\\n",
    "                                                      .drop('elevation').drop('distance_to_Zurich')\n",
    "\n",
    "    walk_to_df = df_zurich_geo.alias(\"walk_to_df\").withColumnRenamed('longitude', 'longitude_to')\\\n",
    "                                                  .withColumnRenamed('latitude', 'latitude_to')\\\n",
    "                                                  .withColumnRenamed('stop_number', 'stop_number_to')\\\n",
    "                                                  .withColumnRenamed('stop_name', 'stop_name_to')\\\n",
    "                                                  .drop('elevation').drop('distance_to_Zurich')\n",
    "    # join these two dfs\n",
    "    df_joined = walk_from_df.crossJoin(walk_to_df)\n",
    "\n",
    "    # drop the row if arrival and departure stations are the same\n",
    "    mask = df_joined.stop_number_from == df_joined.stop_number_to\n",
    "    df_joined = df_joined[~mask]\n",
    "    df_distances = df_joined.withColumn('distance_as_meter', calculateCrossDistance(df_joined.latitude_from,df_joined.longitude_from,df_joined.latitude_to,df_joined.longitude_to))\\\n",
    "                                .drop('longitude_from').drop('longitude_to').drop('latitude_from').drop('latitude_to')\n",
    "\n",
    "    \n",
    "    distance_dictionary = {(row[0], row[1]):row[2] for row in df_distances.select('stop_number_from', 'stop_number_to', 'distance_as_meter').collect()}\n",
    "    \n",
    "    return distance_dictionary"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_transportation_matrix(df_trips, station_index_to_id, station_id_to_index):\n",
    "    \"\"\" Get transportation matrix\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    df_trips: spark dataframe\n",
    "        Dataframe containing trips info\n",
    "    station_index_to_id: dict\n",
    "        Dictionary to help convert station index to id\n",
    "    station_id_to_index: dict\n",
    "        Dictionary to help convert station id to index\n",
    "    \"\"\"\n",
    "\n",
    "    # window by trip id and sort by arrival time\n",
    "    trip_id_window = Window.partitionBy('TRIP_ID').orderBy(F.asc('SCHEDULE_ARRIVE_TIME'))\n",
    "    \n",
    "    # keep the order of stations in the trip\n",
    "    df_trips = df_trips.withColumn('TRIP_ORDER', F.rank().over(trip_id_window)).cache()\n",
    "                                   \n",
    "    # define udf and struct type to get all connected station pairs\n",
    "    schema = StructType([\n",
    "        StructField(\"DEPART\", ArrayType(IntegerType()), False),\n",
    "        StructField(\"ARRIVE\", ArrayType(IntegerType()), False)\n",
    "    ])\n",
    "\n",
    "    # iterate over each trip and connect every stations in the trip\n",
    "    def return_all_connected_station_pairs(station_indices):\n",
    "        depart = []\n",
    "        arrive = []\n",
    "        for i in range(len(station_indices)):\n",
    "            for j in range(i+1, len(station_indices)):\n",
    "                depart.append(station_indices[i])\n",
    "                arrive.append(station_indices[j])\n",
    "        return [depart, arrive]\n",
    "\n",
    "    return_all_connected_station_pairs_udf = F.udf(return_all_connected_station_pairs, schema)\n",
    "\n",
    "    # apply udf to df_trips\n",
    "    df_connections = df_trips.groupBy('TRIP_ID')\\\n",
    "                             .agg(F.collect_list('STATION_INDEX').alias('STATION_INDEX'))\\\n",
    "                             .withColumn('STATION_INDEX', return_all_connected_station_pairs_udf('STATION_INDEX'))\\\n",
    "                             .select(\"TRIP_ID\", \"STATION_INDEX.DEPART\", \"STATION_INDEX.ARRIVE\")\n",
    "                                   \n",
    "    transportation_matrix = np.zeros((len(station_index_to_id),len(station_index_to_id)))\n",
    "\n",
    "    # collect all pairs as (depart, arrive)\n",
    "    depart = df_connections.select('DEPART').rdd.flatMap(lambda x: x).collect()\n",
    "    arrive = df_connections.select('ARRIVE').rdd.flatMap(lambda x: x).collect()\n",
    "\n",
    "    # fill conection matrix with pairs\n",
    "    depart_list = [item for trip in depart for item in trip]\n",
    "    arrive_list = [item for trip in arrive for item in trip]\n",
    "\n",
    "    for i in range(len(depart_list)):\n",
    "        transportation_matrix[depart_list[i],arrive_list[i]] = 1\n",
    "                       \n",
    "    ### WALK MATRIX ###\n",
    "    distance_dictionary  = get_distance_dictionary()\n",
    "        \n",
    "    transfer_delay = 5 # change from one station to another in mins\n",
    "\n",
    "    # average walking speed is assumed to be 4.5 km/h\n",
    "    # ref: https://www.quora.com/What-is-the-average-walking-speed-of-a-human\n",
    "    max_walking_time = 5 # min\n",
    "    human_speed = 75 # m/min which corresponds to 4.5 km/h\n",
    "    max_walking_distance = human_speed * max_walking_time # as meters\n",
    "    \n",
    "    walk_matrix = np.zeros((len(station_index_to_id), len(station_index_to_id)))\n",
    "\n",
    "\n",
    "    for station_from, station_to in distance_dictionary:\n",
    "        # check if stations are exist in this day\n",
    "        if str(station_from) in  station_id_to_index and str(station_to) in station_id_to_index:\n",
    "            # check if the distance is acceptable\n",
    "            if distance_dictionary[(station_from, station_to)] <= max_walking_distance:\n",
    "                walk_matrix[station_id_to_index[str(station_from)], station_id_to_index[str(station_to)]] = 1\n",
    "    \n",
    "    return transportation_matrix, walk_matrix"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_connection_graph(df_trips, transportation_matrix, walk_matrix):\n",
    "    \"\"\" Get connections graph\n",
    "    \n",
    "    df_trips: spark dataframe\n",
    "        Dataframe of trips\n",
    "    transportation_matrix: dict\n",
    "        Transportation matrinx containing, same as in explained in other notebook\n",
    "    walk_matrix: dict\n",
    "        Walk matrix containg the info if there is a walk between the stations\n",
    "        \n",
    "    Returns\n",
    "    -------\n",
    "    connection_graph: nx.DiGraph\n",
    "        Connections graph\n",
    "    \"\"\"\n",
    "    # merged conenction_matrix and wal_matrix into a one single adjcency matrix of all possible paths\n",
    "    connection_matrix = np.logical_or(transportation_matrix, walk_matrix).astype(int)\n",
    "    connection_graph = nx.DiGraph(connection_matrix)\n",
    "    \n",
    "    return connection_graph"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='4'></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# TIMETABLE"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The matrix only contains 0 and 1 when a connection is possible at any given time. But we also need to know the times of the conneciton in addition to suggest to our user the best possible route. For that we form a timetable using the dataset already provided by doing a self join on the trip_id :"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_timetable_for_day(df_trips):     \n",
    "    \"\"\" Get the timetable of the day\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    df_trips: spark dataframe\n",
    "        Dataframe containing trips\n",
    "    \n",
    "    Returns\n",
    "    -------\n",
    "    timetable: spark dataframe\n",
    "        Contains timetable information\n",
    "    \"\"\"\n",
    "    # create timetable of every possible connection by merging df_depart and df_arrive\n",
    "    df_depart = df_trips.withColumnRenamed('STATION_ID', 'departure_station_id')\\\n",
    "                        .withColumnRenamed('STATION_NAME', 'departure_station_name')\\\n",
    "                        .withColumnRenamed('TRIP_ORDER', 'departure_trip_order')\\\n",
    "                        .withColumnRenamed('STATION_INDEX', 'departure_station_index')\\\n",
    "                        .drop('SCHEDULE_ARRIVE_TIME')\n",
    "\n",
    "    df_arrive = df_trips.withColumnRenamed('STATION_ID', 'arrival_station_id')\\\n",
    "                        .withColumnRenamed('STATION_NAME', 'arrival_station_name')\\\n",
    "                        .withColumnRenamed('TRIP_ORDER', 'arrival_trip_order')\\\n",
    "                        .withColumnRenamed('STATION_INDEX', 'arrival_station_index')\\\n",
    "                        .drop('SCHEDULE_DEPART_TIME').drop('type', 'OPERATOR_ID', 'SERVICE_TYPE', 'TRANSPORT_TYPE')\n",
    "\n",
    "\n",
    "    timetable = df_depart.join(df_arrive, on=['TRIP_ID'], how='left_outer').drop('departure_trip_order').drop('arrival_trip_order')\n",
    "\n",
    "    # drop columns with the same departure and arrival stations\n",
    "    mask = timetable.departure_station_name == timetable.arrival_station_name\n",
    "    timetable = timetable[~mask]\n",
    "\n",
    "    # time reverse connections\n",
    "    mask = timetable.SCHEDULE_DEPART_TIME > timetable.SCHEDULE_ARRIVE_TIME\n",
    "    timetable = timetable[~mask]\n",
    "    \n",
    "    return timetable"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='5'></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ROUTE"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can now use all the functions previously defined to give our user the best possible routes : considering on the one hand the duration and on the other hand the probability of catching the connection :"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "# load location dictionary\n",
    "with open('distance/location_dictionary.pickle', 'rb') as handle:\n",
    "    location_dictionary = pickle.load(handle)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We start by removing the paths that are walkable and keeping only the direct one (if many paths involve only walking):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "def filter_paths(connection_graph, departure_station_index, arrival_station_index, transportation_matrix, walk_matrix):\n",
    "    \"\"\" Filter paths we need\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    connection_graph: nx.DiGraph  \n",
    "        Connection graph\n",
    "    departure_station_index: int\n",
    "        Index of the departure station\n",
    "    arrival_station_index: int\n",
    "        Index of the arrival station\n",
    "    transportation_matrix: dict\n",
    "        Transportation matrix\n",
    "    walk_matrix: dict\n",
    "        Walk matrix\n",
    "    \n",
    "    Returns\n",
    "    -------\n",
    "    filtered_paths: list\n",
    "        Filtered paths\n",
    "    \"\"\"\n",
    "    filtered_paths = []\n",
    "    for path in nx.all_simple_paths(connection_graph, departure_station_index, arrival_station_index, 2):\n",
    "        if len(path) == 3:\n",
    "            # check if both connection is by walk or not\n",
    "            depart_index, arrive_index = path[0], path[1]\n",
    "            if walk_matrix[depart_index, arrive_index] == 1:\n",
    "                depart_index, arrive_index = path[1], path[2]\n",
    "                # if both connection is by walk, dont add to the paths\n",
    "                if walk_matrix[depart_index, arrive_index] == 1:\n",
    "                    continue\n",
    "        \n",
    "        filtered_paths.append(path)\n",
    "            \n",
    "    return filtered_paths, walk_matrix"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can now calculate the routes using networkx library and them check the times using the timetable. Finally we sort the routes found by probability and duration and show the shortest ones first :"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "def calculate_routes(departure_station, arrival_station, date, hour):  \n",
    "    \"\"\" Calculate possible routes based on user input\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    departure_station, \n",
    "        Departure station ID\n",
    "    arrival_station: string\n",
    "        Arrival station ID\n",
    "    date: string\n",
    "        Date in format %Y-%m-%d\n",
    "    hour: string\n",
    "        Hour in format %HH:%MM:%SS\n",
    "    \n",
    "    Returns\n",
    "    -------\n",
    "    all_possible_connections: list of dicts\n",
    "        List of all the possible connections\n",
    "    all_probabilities: list\n",
    "        List of probabilities of each of the connections\n",
    "    \"\"\"\n",
    "    \n",
    "    departure_day = datetime.strptime(date, '%Y-%m-%d')\n",
    "\n",
    "    if SPARK_LOCAL:\n",
    "        if username == \"fatine\":\n",
    "            df_zurich = spark.read.csv(f\"/home/fatine/Documents/Cours Semestre Printemps/Lab in DS/{departure_day.strftime('%Y-%m')}/{departure_day.strftime('%Y-%m-%d')}istdaten.csv\", header=True, sep=';').cache()\n",
    "\n",
    "        elif username == \"soner\":\n",
    "            df_zurich = spark.read.csv(f\"/home/soner/Desktop/DSLAB2019/Project/{departure_day.strftime('%Y-%m')}/{departure_day.strftime('%Y-%m-%d')}istdaten.csv\", header=True, sep=';').cache()\n",
    "        elif username == \"jelena\":\n",
    "            df_zurich = spark.read.csv(f\"{CLUSTER_URL}/datasets/sbb/{str(departure_day.year)}/{str(departure_day.month).zfill(2)}/{departure_day.strftime('%Y-%m-%d')}istdaten.csv.bz2\", sep=';', header=True).cache()\n",
    "    else:\n",
    "        df_zurich = spark.read.csv(f\"{CLUSTER_URL}/datasets/sbb/{str(departure_day.year)}/{str(departure_day.month).zfill(2)}/{departure_day.strftime('%Y-%m-%d')}istdaten.csv.bz2\", sep=';', header=True).cache()\n",
    "    \n",
    "    df_zurich = filtering_of_dataframe(df_zurich)\n",
    "                                   \n",
    "    # create mappings f of stations from id to index and index to id\n",
    "    station_index_to_id = list(df_zurich.select('STATION_ID').distinct().toPandas()['STATION_ID'])\n",
    "                                   \n",
    "    station_id_to_index = {}\n",
    "    for index, station_id in enumerate(station_index_to_id):\n",
    "        station_id_to_index[station_id] = index\n",
    "             \n",
    "    # helper methods for the spark dataframe\n",
    "    return_index = udf(lambda station_id: station_id_to_index[station_id], IntegerType())\n",
    "    return_datetime = udf(lambda date: datetime.strptime(date, \"%d.%m.%Y %H:%M\"), TimestampType())\n",
    "               \n",
    "    df_trips = df_zurich.filter(F.col('ADDITIONAL_DRIVING')=='false')\\\n",
    "                        .filter(F.col('PASSES_BY')=='false')\\\n",
    "                        .select(F.col('OPERATOR_ID'), F.col('SERVICE_TYPE'),F.col('TRIP_ID'),\n",
    "                                F.col('TRANSPORT_TYPE'),\n",
    "                                F.col('STATION_ID'),\n",
    "                                return_index('STATION_ID').astype('int').alias('STATION_INDEX'),\n",
    "                                F.col('STATION_NAME'),\n",
    "                                return_datetime(F.col('SCHEDULE_ARRIVE_TIME')).alias('SCHEDULE_ARRIVE_TIME'),\n",
    "                                return_datetime(F.col('SCHEDULE_DEPART_TIME')).alias('SCHEDULE_DEPART_TIME')).cache()                               \n",
    "                                   \n",
    "    # get timetable dataframe                               \n",
    "    timetable = get_timetable_for_day(df_trips)\n",
    "    # get transportation and walk matrices\n",
    "    transportation_matrix, walk_matrix = get_transportation_matrix(df_trips, station_index_to_id, station_id_to_index)\n",
    "    # get connection graph\n",
    "    connection_graph = get_connection_graph(df_trips, transportation_matrix, walk_matrix)\n",
    "    # get distance dictionary\n",
    "    distance_dictionary = get_distance_dictionary()                               \n",
    "    \n",
    "    # find indices of stations in the graph\n",
    "    departure_station_index = station_id_to_index[departure_station]\n",
    "    arrival_station_index = station_id_to_index[arrival_station]\n",
    "\n",
    "    all_possible_connections = []\n",
    "    all_probabilities = []\n",
    "                            \n",
    "    # find all simple paths in the graph\n",
    "    try:\n",
    "        paths, walk_matrix = filter_paths(connection_graph, departure_station_index, arrival_station_index, transportation_matrix, walk_matrix)\n",
    "    except nx.exception.NodeNotFound:\n",
    "        return None, None\n",
    "    \n",
    "    full_date = date + ' ' + hour                               \n",
    "                                   \n",
    "    i = 0\n",
    "    for path in paths:    \n",
    "        i = i + 1\n",
    "        arrival_date = datetime.strptime(full_date, '%Y-%m-%d %H:%M:%S')\n",
    "\n",
    "        connections_info = []\n",
    "\n",
    "        route_failed = False\n",
    "        for depart_index, arrive_index in zip(path[:-1], path[1:]):\n",
    "\n",
    "            # check if the path is by walk or by transportation\n",
    "            if walk_matrix[depart_index, arrive_index]:\n",
    "                mins = distance_dictionary[(str(station_index_to_id[depart_index]), str(station_index_to_id[arrive_index]))] / human_speed\n",
    "\n",
    "                departure_date = arrival_date \n",
    "                arrival_date = departure_date + timedelta(minutes=mins)\n",
    "\n",
    "                connections_info.append({'depart_station_id': station_index_to_id[depart_index] , \n",
    "                                         'arrive_station_id': station_index_to_id[arrive_index]  , \n",
    "                                         'depart_date': departure_date,\n",
    "                                         'arrive_date': arrival_date,\n",
    "                                         'transport_type': 'walk',\n",
    "                                         'trip_id': None,\n",
    "                                         'operator_id':None,\n",
    "                                         'service_type': None})\n",
    "\n",
    "            else:\n",
    "                # add walking time to change stations \n",
    "                arrival_date = arrival_date + timedelta(minutes=transfer_delay)\n",
    "                \n",
    "                # find earliest departure\n",
    "                try:\n",
    "                    connection = timetable.filter((F.col('departure_station_index') == depart_index) & (F.col('arrival_station_index') == arrive_index) & (F.col('SCHEDULE_DEPART_TIME') >= arrival_date)).first()\n",
    "                except Py4JJavaError:\n",
    "                    print('route failed')\n",
    "                    route_failed = True\n",
    "                    break\n",
    "                #print(\"connection\", connection)\n",
    "                if connection:\n",
    "                    arrival_date = connection.SCHEDULE_ARRIVE_TIME\n",
    "                    mins = arrival_date - connection.SCHEDULE_DEPART_TIME\n",
    "                \n",
    "                    connections_info.append({'depart_station_id': connection.departure_station_id, \n",
    "                                             'arrive_station_id': connection.arrival_station_id, \n",
    "                                             'depart_date': connection.SCHEDULE_DEPART_TIME, \n",
    "                                             'arrive_date': arrival_date, \n",
    "                                             'transport_type': connection.TRANSPORT_TYPE,\n",
    "                                             'trip_id': connection.TRIP_ID,\n",
    "                                             'operator_id': connection.OPERATOR_ID,\n",
    "                                             'service_type': connection.SERVICE_TYPE})\n",
    "                else:\n",
    "                    print('route failed')\n",
    "                    route_failed = True\n",
    "                    break\n",
    "\n",
    "        if not route_failed:            \n",
    "            p = calculate_confidence(connections_info)\n",
    "        else: \n",
    "            p = 0\n",
    "        \n",
    "        all_possible_connections.append(connections_info)\n",
    "        all_probabilities.append(p)\n",
    "                                           \n",
    "    # add Lon and Lat\n",
    "    for ci in all_possible_connections:\n",
    "        for trip in ci:\n",
    "            trip['depart_station_id_LON'] =  location_dictionary[trip['depart_station_id']][0]\n",
    "            trip['depart_station_id_LAT'] = location_dictionary[trip['depart_station_id']][1]\n",
    "            trip['arrive_station_id_LON'] = location_dictionary[trip['arrive_station_id']][0]\n",
    "            trip['arrive_station_id_LAT'] = location_dictionary[trip['arrive_station_id']][1]\n",
    "     \n",
    "                                   \n",
    "    # Let's sort the connections first by probability                               \n",
    "    max_confidence_indices = np.argsort(all_probabilities)[::-1]\n",
    "    \n",
    "    all_possible_connections = np.asarray(all_possible_connections)\n",
    "    all_possible_connections = all_possible_connections[max_confidence_indices]\n",
    "    \n",
    "    all_probabilities = np.asarray(all_probabilities)\n",
    "    all_probabilities = all_probabilities[max_confidence_indices]\n",
    "    # Let's sort the connections now by shortest time\n",
    "    durations = []\n",
    "    for i in range(len(all_possible_connections)):\n",
    "        d = 0\n",
    "        d = (all_possible_connections[i][-1]['arrive_date'] - all_possible_connections[i][0]['depart_date']).seconds/60\n",
    "        durations.append(d)\n",
    "\n",
    "    max_confidence_indices = np.argsort(durations)\n",
    "\n",
    "    all_possible_connections = np.asarray(all_possible_connections)\n",
    "    all_possible_connections = all_possible_connections[max_confidence_indices]\n",
    "\n",
    "    all_probabilities = np.asarray(all_probabilities)\n",
    "    all_probabilities = all_probabilities[max_confidence_indices]                               \n",
    "\n",
    "    durations= np.asarray(durations)\n",
    "    durations = durations[max_confidence_indices]                               \n",
    "\n",
    "                                        \n",
    "    i = 0 \n",
    "    for connections_info in all_possible_connections:    \n",
    "        print('route:', i)\n",
    "        p = all_probabilities[i]\n",
    "        d = durations[i]\n",
    "        i = i + 1\n",
    "        print('----------------')\n",
    "                                \n",
    "        for connection in connections_info:\n",
    "            print('departure: ',connection['depart_station_id'], 'arrival: ',connection['arrive_station_id'], 'departure_time: ', connection['depart_date'].strftime('%Y-%m-%d %H:%M:%S'), 'arrival_time: ', connection['arrive_date'].strftime('%Y-%m-%d %H:%M:%S'),'transport_type: ', connection['transport_type'])\n",
    "        print(f\"duration = {d} minutes\")\n",
    "        print(f\"confidence = {p}\")\n",
    "        print('----------------')\n",
    "                                       \n",
    "    \n",
    "    return list(all_possible_connections), list(all_probabilities), list(durations)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "route: 0\n",
      "----------------\n",
      "departure:  8503000 arrival:  8503016 departure_time:  2017-10-01 15:39:00 arrival_time:  2017-10-01 15:49:00 transport_type:  Zug\n",
      "duration = 10.0 minutes\n",
      "confidence = 1.0\n",
      "----------------\n",
      "route: 1\n",
      "----------------\n",
      "departure:  8503000 arrival:  8503011 departure_time:  2017-10-01 15:17:00 arrival_time:  2017-10-01 15:19:00 transport_type:  Zug\n",
      "departure:  8503011 arrival:  8503016 departure_time:  2017-10-01 15:34:00 arrival_time:  2017-10-01 15:56:00 transport_type:  Zug\n",
      "duration = 39.0 minutes\n",
      "confidence = 0.9999996940976795\n",
      "----------------\n",
      "route: 2\n",
      "----------------\n",
      "departure:  8503000 arrival:  8503010 departure_time:  2017-10-01 15:17:00 arrival_time:  2017-10-01 15:22:00 transport_type:  Zug\n",
      "departure:  8503010 arrival:  8503016 departure_time:  2017-10-01 15:33:00 arrival_time:  2017-10-01 15:56:00 transport_type:  Zug\n",
      "duration = 39.0 minutes\n",
      "confidence = 0.9999916491496048\n",
      "----------------\n",
      "route: 3\n",
      "----------------\n",
      "departure:  8503000 arrival:  8503006 departure_time:  2017-10-01 15:14:00 arrival_time:  2017-10-01 15:18:00 transport_type:  Zug\n",
      "departure:  8503006 arrival:  8503016 departure_time:  2017-10-01 15:52:00 arrival_time:  2017-10-01 15:56:00 transport_type:  Zug\n",
      "duration = 42.0 minutes\n",
      "confidence = 0.9999999999999996\n",
      "----------------\n",
      "route: 4\n",
      "----------------\n",
      "departure:  8503000 arrival:  8503015 departure_time:  2017-10-01 15:14:00 arrival_time:  2017-10-01 15:17:00 transport_type:  Zug\n",
      "departure:  8503015 arrival:  8503016 departure_time:  2017-10-01 15:47:00 arrival_time:  2017-10-01 15:56:00 transport_type:  Zug\n",
      "duration = 42.0 minutes\n",
      "confidence = 0.9999996940976795\n",
      "----------------\n",
      "route: 5\n",
      "----------------\n",
      "departure:  8503000 arrival:  8503307 departure_time:  2017-10-01 15:19:00 arrival_time:  2017-10-01 15:39:00 transport_type:  Zug\n",
      "departure:  8503307 arrival:  8503016 departure_time:  2017-10-01 15:56:00 arrival_time:  2017-10-01 16:01:00 transport_type:  Zug\n",
      "duration = 42.0 minutes\n",
      "confidence = 0.9999880327084439\n",
      "----------------\n",
      "route: 6\n",
      "----------------\n",
      "departure:  8503000 arrival:  8503009 departure_time:  2017-10-01 15:07:00 arrival_time:  2017-10-01 15:15:00 transport_type:  Zug\n",
      "departure:  8503009 arrival:  8503016 departure_time:  2017-10-01 15:28:00 arrival_time:  2017-10-01 15:56:00 transport_type:  Zug\n",
      "duration = 49.0 minutes\n",
      "confidence = 0.9998622141952314\n",
      "----------------\n",
      "route: 7\n",
      "----------------\n",
      "departure:  8503000 arrival:  8503200 departure_time:  2017-10-01 15:07:00 arrival_time:  2017-10-01 15:17:00 transport_type:  Zug\n",
      "departure:  8503200 arrival:  8503016 departure_time:  2017-10-01 15:26:00 arrival_time:  2017-10-01 15:56:00 transport_type:  Zug\n",
      "duration = 49.0 minutes\n",
      "confidence = 0.9980169982586669\n",
      "----------------\n",
      "route: 8\n",
      "----------------\n",
      "departure:  8503000 arrival:  8503202 departure_time:  2017-10-01 15:17:00 arrival_time:  2017-10-01 15:30:00 transport_type:  Zug\n",
      "departure:  8503202 arrival:  8503016 departure_time:  2017-10-01 15:52:00 arrival_time:  2017-10-01 16:26:00 transport_type:  Zug\n",
      "duration = 69.0 minutes\n",
      "confidence = 0.9999999998605267\n",
      "----------------\n",
      "route: 9\n",
      "----------------\n",
      "departure:  8503000 arrival:  8503201 departure_time:  2017-10-01 15:07:00 arrival_time:  2017-10-01 15:20:00 transport_type:  Zug\n",
      "departure:  8503201 arrival:  8503016 departure_time:  2017-10-01 15:53:00 arrival_time:  2017-10-01 16:26:00 transport_type:  Zug\n",
      "duration = 79.0 minutes\n",
      "confidence = 0.9999999997768426\n",
      "----------------\n"
     ]
    }
   ],
   "source": [
    "date = '2017-10-01'\n",
    "hour = '15:00:00'\n",
    "departure_station = 'Zürich HB'\n",
    "arrival_station = 'Zürich Flughafen'\n",
    "# min_confidence_level = 0.9\n",
    "\n",
    "departure_station_id = df_zurich_full.filter(\"STATION_NAME = '\" + departure_station + \"'\").first().STATION_ID\n",
    "arrival_station_id = df_zurich_full.filter(\"STATION_NAME = '\" + arrival_station + \"'\").first().STATION_ID\n",
    "all_possible_connections, all_probabilities, durations = calculate_routes(departure_station=departure_station_id, arrival_station=arrival_station_id, date=date, hour=hour)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "route failed\n",
      "route: 0\n",
      "----------------\n",
      "departure:  8503000 arrival:  8591067 departure_time:  2017-10-01 15:00:00 arrival_time:  2017-10-01 15:02:24 transport_type:  walk\n",
      "duration = 2.4 minutes\n",
      "confidence = 0.0\n",
      "----------------\n",
      "route: 1\n",
      "----------------\n",
      "departure:  8503000 arrival:  8587348 departure_time:  2017-10-01 15:00:00 arrival_time:  2017-10-01 15:01:37 transport_type:  walk\n",
      "departure:  8587348 arrival:  8591123 departure_time:  2017-10-01 15:30:00 arrival_time:  2017-10-01 15:36:00 transport_type:  Tram\n",
      "duration = 36.0 minutes\n",
      "confidence = 1.0\n",
      "----------------\n",
      "route: 2\n",
      "----------------\n",
      "departure:  8503000 arrival:  8591174 departure_time:  2017-10-01 15:00:00 arrival_time:  2017-10-01 15:04:51 transport_type:  walk\n",
      "departure:  8591174 arrival:  8591123 departure_time:  2017-10-01 15:50:00 arrival_time:  2017-10-01 15:52:00 transport_type:  Tram\n",
      "duration = 52.0 minutes\n",
      "confidence = 1.0\n",
      "----------------\n"
     ]
    }
   ],
   "source": [
    "all_possible_connections, all_probabilities, durations = calculate_routes(departure_station='8503000', arrival_station='8591123', date='2017-10-01', hour='15:00:00')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Visualization"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To run the visualization, please run the whole notebook and open the: http://0.0.0.0:5000/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "index_html = \"\"\"\n",
    "\n",
    "<!doctype html>\n",
    "<html lang=\"en\">\n",
    "<head>\n",
    "  <link rel=\"stylesheet\" href=\"https://cdn.rawgit.com/openlayers/openlayers.github.io/master/en/v5.3.0/css/ol.css\"\n",
    "    type=\"text/css\">\n",
    "  <script src=\"https://cdn.rawgit.com/openlayers/openlayers.github.io/master/en/v5.3.0/build/ol.js\"></script>\n",
    "</head>\n",
    "\n",
    "<body>\n",
    "  <div style=\"position: absolute; top: 10px; right: 10px; z-index: 100; background: white; padding: 8px;\">\n",
    "  <input type=\"text\" placeholder=\"Start Station ID...\" id=\"start_station_id\" />\n",
    "  <input type=\"text\" placeholder=\"End Station ID...\" id=\"end_station_id\" />\n",
    "  <input type=\"text\" placeholder=\"yyyy-mm-dd\" id=\"in_date\" />\n",
    "  <input type=\"text\" placeholder=\"HH:MM:SS\" id=\"in_time\" />\n",
    "  <button id=\"search\">Show</button>\n",
    "  <div id=\"message\"></div>\n",
    "</div>\n",
    "\n",
    "  <div id=\"map\" style=\"width: 100%; height: 100%\"></div>\n",
    "\n",
    "  <script type=\"text/javascript\">\n",
    "    var features = [];\n",
    "    var vectorLayer = new ol.layer.Vector({\n",
    "      style: function (feature, resolution) {\n",
    "        return feature.get('style');\n",
    "      },\n",
    "    });\n",
    "    var map = new ol.Map({\n",
    "      target: 'map',\n",
    "      layers: [\n",
    "        new ol.layer.Tile({\n",
    "          source: new ol.source.OSM()\n",
    "        }),\n",
    "      ],\n",
    "      view: new ol.View({\n",
    "        center: ol.proj.fromLonLat([8.540192, 47.378177]),\n",
    "        zoom: 14\n",
    "      })\n",
    "    });\n",
    "    map.addLayer(vectorLayer);\n",
    "\n",
    "    function addLine(start, final, color) {\n",
    "      var lineString = new ol.geom.LineString([start, final]);\n",
    "      lineString.transform('EPSG:4326', 'EPSG:3857');\n",
    "      var feature = new ol.Feature({\n",
    "        geometry: lineString,\n",
    "      });\n",
    "      feature.setStyle(new ol.style.Style({\n",
    "        stroke: new ol.style.Stroke({\n",
    "          color: color,\n",
    "          width: 5\n",
    "        })\n",
    "      }));\n",
    "      features.push(feature);\n",
    "      vectorLayer.setSource(\n",
    "        new ol.source.Vector({\n",
    "          features: features,\n",
    "        })\n",
    "      );\n",
    "    }\n",
    "\n",
    "    document.getElementById('search').addEventListener('click', function() {\n",
    "      document.getElementById(\"message\").innerHTML = '<br/> Loading... <br/>';\n",
    "      var start_station_id = document.getElementById('start_station_id').value;\n",
    "      var end_station_id = document.getElementById('end_station_id').value;\n",
    "      var in_date = document.getElementById('in_date').value;\n",
    "      var in_time = document.getElementById('in_time').value;\n",
    "\n",
    "      var xhttp = new XMLHttpRequest();\n",
    "\n",
    "      xhttp.onreadystatechange = function() {\n",
    "        if (this.readyState == 4 && this.status == 200) {\n",
    "          var data = JSON.parse(this.responseText);\n",
    "          features = [];\n",
    "          data.lines.forEach(function(line) {\n",
    "            addLine(line.start, line.end, line.color);\n",
    "          });\n",
    "          document.getElementById(\"message\").innerHTML = data.message;\n",
    "        }\n",
    "      };\n",
    "\n",
    "      xhttp.open('GET', '/search?start=' + start_station_id + '&end=' + end_station_id + '&date=' + in_date + '&time=' + in_time, true);\n",
    "      xhttp.send();\n",
    "    });\n",
    "\n",
    "  </script>\n",
    "</body>\n",
    "</html>\n",
    "\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " * Serving Flask app \"__main__\" (lazy loading)\n",
      " * Environment: production\n",
      "   WARNING: Do not use the development server in a production environment.\n",
      "   Use a production WSGI server instead.\n",
      " * Debug mode: off\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      " * Running on http://0.0.0.0:5000/ (Press CTRL+C to quit)\n",
      "127.0.0.1 - - [17/Jun/2019 18:59:58] \"\u001b[37mGET / HTTP/1.1\u001b[0m\" 200 -\n",
      "127.0.0.1 - - [17/Jun/2019 18:59:59] \"\u001b[37mGET / HTTP/1.1\u001b[0m\" 200 -\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "here...\n",
      "Ajax call executed! Start 8503000, end 8591123\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[2019-06-17 19:31:01,370] ERROR in app: Exception on /search [GET]\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/jelena/anaconda3/lib/python3.7/site-packages/flask/app.py\", line 2292, in wsgi_app\n",
      "    response = self.full_dispatch_request()\n",
      "  File \"/home/jelena/anaconda3/lib/python3.7/site-packages/flask/app.py\", line 1815, in full_dispatch_request\n",
      "    rv = self.handle_user_exception(e)\n",
      "  File \"/home/jelena/anaconda3/lib/python3.7/site-packages/flask/app.py\", line 1718, in handle_user_exception\n",
      "    reraise(exc_type, exc_value, tb)\n",
      "  File \"/home/jelena/anaconda3/lib/python3.7/site-packages/flask/_compat.py\", line 35, in reraise\n",
      "    raise value\n",
      "  File \"/home/jelena/anaconda3/lib/python3.7/site-packages/flask/app.py\", line 1813, in full_dispatch_request\n",
      "    rv = self.dispatch_request()\n",
      "  File \"/home/jelena/anaconda3/lib/python3.7/site-packages/flask/app.py\", line 1799, in dispatch_request\n",
      "    return self.view_functions[rule.endpoint](**req.view_args)\n",
      "  File \"<ipython-input-36-0b1000b4d6d9>\", line 45, in search\n",
      "    all_possible_connections, all_probabilities = calculate_routes(departure_station=start_arg, arrival_station=end_arg, date=date_arg, hour=time_arg)\n",
      "  File \"<ipython-input-24-28f1220b4515>\", line 61, in calculate_routes\n",
      "    transportation_matrix, walk_matrix = get_transportation_matrix(df_trips, station_index_to_id, station_id_to_index)\n",
      "  File \"<ipython-input-19-24c531dcc39d>\", line 58, in get_transportation_matrix\n",
      "    distance_dictionary  = get_distance_dictionary()\n",
      "  File \"<ipython-input-18-c7e46a5ff885>\", line 140, in get_distance_dictionary\n",
      "    distance_dictionary = {(row[0], row[1]):row[2] for row in df_distances.select('stop_number_from', 'stop_number_to', 'distance_as_meter').collect()}\n",
      "  File \"/home/jelena/anaconda3/lib/python3.7/site-packages/pyspark/sql/dataframe.py\", line 533, in collect\n",
      "    sock_info = self._jdf.collectToPython()\n",
      "  File \"/home/jelena/anaconda3/lib/python3.7/site-packages/py4j/java_gateway.py\", line 1257, in __call__\n",
      "    answer, self.gateway_client, self.target_id, self.name)\n",
      "  File \"/home/jelena/anaconda3/lib/python3.7/site-packages/pyspark/sql/utils.py\", line 63, in deco\n",
      "    return f(*a, **kw)\n",
      "  File \"/home/jelena/anaconda3/lib/python3.7/site-packages/py4j/protocol.py\", line 328, in get_return_value\n",
      "    format(target_id, \".\", name), value)\n",
      "py4j.protocol.Py4JJavaError: An error occurred while calling o18642.collectToPython.\n",
      ": org.apache.spark.SparkException: Job aborted due to stage failure: Task 15 in stage 528.0 failed 1 times, most recent failure: Lost task 15.0 in stage 528.0 (TID 12480, localhost, executor driver): TaskResultLost (result lost from block manager)\n",
      "Driver stacktrace:\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1889)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1877)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1876)\n",
      "\tat scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)\n",
      "\tat scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1876)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:926)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:926)\n",
      "\tat scala.Option.foreach(Option.scala:257)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:926)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2110)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2059)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2048)\n",
      "\tat org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:737)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2061)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2082)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2101)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2126)\n",
      "\tat org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:945)\n",
      "\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)\n",
      "\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)\n",
      "\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:363)\n",
      "\tat org.apache.spark.rdd.RDD.collect(RDD.scala:944)\n",
      "\tat org.apache.spark.sql.execution.SparkPlan.executeCollect(SparkPlan.scala:299)\n",
      "\tat org.apache.spark.sql.Dataset$$anonfun$collectToPython$1.apply(Dataset.scala:3257)\n",
      "\tat org.apache.spark.sql.Dataset$$anonfun$collectToPython$1.apply(Dataset.scala:3254)\n",
      "\tat org.apache.spark.sql.Dataset$$anonfun$53.apply(Dataset.scala:3364)\n",
      "\tat org.apache.spark.sql.execution.SQLExecution$$anonfun$withNewExecutionId$1.apply(SQLExecution.scala:78)\n",
      "\tat org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:125)\n",
      "\tat org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:73)\n",
      "\tat org.apache.spark.sql.Dataset.withAction(Dataset.scala:3363)\n",
      "\tat org.apache.spark.sql.Dataset.collectToPython(Dataset.scala:3254)\n",
      "\tat sun.reflect.GeneratedMethodAccessor141.invoke(Unknown Source)\n",
      "\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n",
      "\tat java.lang.reflect.Method.invoke(Method.java:498)\n",
      "\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\n",
      "\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\n",
      "\tat py4j.Gateway.invoke(Gateway.java:282)\n",
      "\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\n",
      "\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\n",
      "\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\n",
      "\tat java.lang.Thread.run(Thread.java:748)\n",
      "\n",
      "127.0.0.1 - - [17/Jun/2019 19:31:01] \"\u001b[1m\u001b[35mGET /search?start=8503000&end=8591123&date=2018-02-28&time=15:00:00 HTTP/1.1\u001b[0m\" 500 -\n"
     ]
    }
   ],
   "source": [
    "from flask import Flask, request, jsonify\n",
    "\n",
    "\n",
    "app = Flask(__name__)\n",
    "\n",
    "def beautify_trip(trip):\n",
    "    return f\"take a {trip['transport_type']} from {trip['depart_station_id']} at {trip['depart_date'].strftime('%d.%m.%Y %H:%M')} to {trip['arrive_station_id']} at {trip['arrive_date'].strftime('%d.%m.%Y %H:%M')}\"\n",
    "\n",
    "def beautify_print(all_possible_connections, all_probabilities):\n",
    "    colors = [\"green\", \"red\", \"blue\"]\n",
    "    ret_val = \"\"\n",
    "    for i, (p, ci) in enumerate(zip(all_probabilities, all_possible_connections)):\n",
    "        ret_val += f\"Connection {i+1} [{colors[i].upper()}] - Probability {p}:<br/>\"\n",
    "        for j, trip in enumerate(ci): \n",
    "            ret_val += f\"{j+1}) {beautify_trip(trip)} <br/>\"\n",
    "        ret_val += \"<br/><br/>\"\n",
    "    return ret_val\n",
    "\n",
    "def make_lines(all_possible_connections):\n",
    "    out_lines = []\n",
    "    colors = ['#00ff00', '#ff0000', '#0000ff']\n",
    "    for i, (p, ci) in enumerate(zip(all_probabilities, all_possible_connections)):\n",
    "        for j, trip in enumerate(ci): \n",
    "            out_lines.append({ 'start': [ trip['depart_station_id_LON'], trip['depart_station_id_LAT']], 'end': [trip['arrive_station_id_LON'], trip['arrive_station_id_LAT']], 'color': colors[i] })\n",
    "    return out_lines\n",
    "\n",
    "@app.route('/', methods=['GET'])\n",
    "def home():\n",
    "    return index_html\n",
    "\n",
    "\n",
    "@app.route('/search', methods=['GET'])\n",
    "def search():\n",
    "    print('here...')\n",
    "    # Grab arguments here\n",
    "    start_arg = request.args.get('start')\n",
    "    end_arg = request.args.get('end')\n",
    "    date_arg = request.args.get('date')\n",
    "    time_arg = request.args.get('time')\n",
    "    \n",
    "    print(f'Ajax call executed! Start {start_arg}, end {end_arg}')\n",
    "    \n",
    "    # Do some processing\n",
    "    #all_possible_connections, all_probabilities = find_route_with_probability(departure_station='8503000', arrival_station='8503016', date='2018-02-14', hour='00:00:00')\n",
    "    all_possible_connections, all_probabilities = calculate_routes(departure_station=start_arg, arrival_station=end_arg, date=date_arg, hour=time_arg)\n",
    "    all_possible_connections, all_probabilities = all_possible_connections[:3], all_probabilities[:3]\n",
    "    status_message = beautify_print(all_possible_connections, all_probabilities)\n",
    "    \n",
    "    out_lines = make_lines(all_possible_connections)\n",
    "    \n",
    "    # Return data in corresponding format\n",
    "    return jsonify({ 'message': status_message, 'lines': out_lines})\n",
    "\n",
    "\n",
    "app.run(host='0.0.0.0', port=5000)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}