Final Demo

app.py

@@ -1,16 +1,8 @@
 import gradio as gr
 
-# ------------- Global Variables
-DAYS = [
-    "Su, 26.11.2023 ",
-    "Mo, 27.11.2023",
-    "Tu, 28.11.2023",
-    "We, 29.11.2023",
-    "Th, 30.11.2023",
-    "Fr, 01.12.2023",
-    "Sa, 02.12.2023",   
-]
+from utils import get_best_path
 
+# ------------- Constants
 TITLE = "Travel Planner" # TODO: Change title as we need
 
 # ------------- Main App
@@ -34,7 +26,7 @@ with gr.Blocks(
 
         with gr.Row() as metainfo:
             limit = gr.Slider(minimum=1, maximum=5, step=1, label="Limit", interactive=True, value=3)
-            days = gr.Dropdown(DAYS, label="Day", value=DAYS[-1])
+            days = gr.Textbox(label="Day", placeholder="YYYY-MM-DD")
             time = gr.Textbox(label="Time", placeholder="hh:mm")
 
         submit = gr.Button(value="Search", variant="primary")
@@ -53,6 +45,8 @@ with gr.Blocks(
             ---
             """
 
+            md = get_best_path(A, B, day, time, limit)
+
             return {
                 output: gr.Column(visible=True),
                 route: md

graph.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:d389ba765d0f6688beb576d4cc55ced125252491e5a20f49e124373128386331
+size 11175230

utils.py

@@ -0,0 +1,518 @@
+"""
+Some utility functions.
+"""
+
+import re
+import argparse
+import requests
+import pandas as pd
+import requests
+from geopy.geocoders import Nominatim
+from geopy.distance import distance
+import heapq
+import pickle
+
+PR_STATIONS = "./data/pr_stations"
+
+# Average speed in m/s
+AVG_SPEED = {
+    "foot": 1.4,
+    "bike": 4.17,
+    "car": 13.89,
+}
+
+# Max travel time in seconds
+MAX_TRAVEL_TIME = {
+    "foot": 60 * 60 // 4,  # 15min
+    "bike": 60 * 60 // 2,  # 30min
+    "car": 60 * 60,  # 1h
+}
+
+PENALITIES = {
+    "foot": 1,
+    "bike": 1,
+    "car": 1,
+    "train": 1,
+}
+
+
+def get_penalties(sustainability: bool):
+    if sustainability:
+        PENALITIES["bike"] *= 1.1
+        PENALITIES["train"] *= 1.2
+        PENALITIES["car"] *= 5
+    else:
+        PENALITIES["car"] *= 2
+
+    return PENALITIES
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+
+    # Default date and time args
+    # current_date = date.today().strftime("%Y-%m-%d")
+    # current_time = datetime.now().strftime("%H:%M")
+    current_date = "2023-12-01"
+    current_time = "12:00"
+
+    # Help messages
+    loc_types = ["address", "station name", "station abbreviation", "coordinate"]
+    transportation_types = ["train", "tram", "ship", "bus", "cableway"]
+    start_help = f"Start location (Specify either {', '.join(loc_types)})"
+    # via_help = f"Locations to pass through (Specify either {', '.join(loc_types)})"
+    stop_help = f"Stop location (Specify either {', '.join(loc_types)})"
+    date_help = "Date of departure (Format: YYYY-MM-DD). Default: Today"
+    time_help = "Time of departure (Format: YYYY-MM-DD). Default: Now"
+    transportation_help = f"Modes of transportation (Specify from {', '.join(transportation_types)}). Default: All"
+    outage_simulation = f"Simulate outage of a station"
+
+    # Specify line arguments
+    parser.add_argument("--start", type=str, required=True, help=start_help)
+    # parser.add_argument("--via", type=list[str], help=via_help)
+    parser.add_argument("--end", type=str, required=True, help=stop_help)
+    parser.add_argument("--date", type=str, default=current_date, help=date_help)
+    parser.add_argument("--time", type=str, default=current_time, help=time_help)
+    parser.add_argument(
+        "--limit", type=int, default=3, help="Number of journeys to return"
+    )
+    parser.add_argument(
+        "--transportations",
+        type,
+        choices=transportation_types,
+        default=["train"],
+        help=transportation_help,
+    )
+    parser.add_argument("--exact-travel-time", action="store_true", help=time_help)
+    parser.add_argument(
+        "--change-penalty", type=int, default=300, help="Change penalty"
+    )
+    parser.add_argument(
+        "--sustainability",
+        action="store_true",
+        help="Sustainability of journey",
+    )
+    parser.add_argument("--outage", action="store_true", help=outage_simulation)
+
+    return parser.parse_args()
+
+
+def get_location(G, address: str) -> str:
+    """
+    Converts an address to coordinates (latitude, longitude)
+
+    Address can be:
+        - Full name
+        - Coordinates  (Format: "latitude, longitude")
+
+    Args:
+        address (str): Address to convert to coordinates.
+
+    Returns:
+        str: Coordinates of the address (Format: "latitude, longitude").
+    """
+    pattern = re.compile(r"^\s*-?\d+\.\d+\s*,\s*-?\d+\.\d+\s*$")
+
+    # Check if it is already a coordinate
+    if bool(pattern.match(address)):
+        return address
+
+    # Check if already stationo
+    if address in G.nodes():
+        return G.nodes[address]["pos"]
+
+    # Use geopy to convert coordinates to address
+    geolocator = Nominatim(user_agent="sbb_project")
+    location = geolocator.geocode(
+        address, country_codes="ch", language="de", exactly_one=True
+    )
+
+    return "{}, {}".format(location.latitude, location.longitude)
+
+
+def get_distance(start, end):
+    """
+    Calculate distance between two coordinates in meters.
+    Coordinates must be in the format "latitude, longitude".
+
+    Args:
+        start (str): Start coordinate.
+        end (str): End coordinate.
+
+    Returns:
+        int: Distance between the two coordinates in meters.
+    """
+    start_converted = tuple(map(float, start.split(", ")))
+    end_converted = tuple(map(float, end.split(", ")))
+    return distance(start_converted, end_converted).meters
+
+
+def get_exact_travel_time(start, end, method="car"):
+    """
+    Calculate travel time between two coordinates in seconds.
+    """
+    endpoint = "http://www.mapquestapi.com/directions/v2/route"
+    if method not in ["foot", "bike", "car"]:
+        raise ValueError("Method must be either foot, bike or car")
+
+    names = {
+        "foot": "pedestrian",
+        "bike": "bicycle",
+        "car": "fastest",
+    }
+    # Search params
+    params = {
+        "key": "HnDX3JAuALRTge28jbZVWO1L538fJbZE",
+        "from": start,
+        "to": end,
+        "unit": "k",  # Use km instead of miles
+        "narrativeType": "none",  # Just some other parameters to omit information we don't care about
+        "sideOfStreetDisplay": False,
+        "routeType": names[method],
+    }
+
+    # Do GET request and read JSON
+    response = requests.get(endpoint, params=params)
+    data = response.json()
+    seconds = data["route"]["time"]
+
+    return seconds
+
+
+def get_approx_travel_time(dist, method="car"):
+    """
+    Calculate the approximate travel time between two coordinates in seconds.
+
+    Args:
+        dist (int): Distance between the two coordinates in meters.
+        method (str, optional): Method of transportation. Defaults to "car".
+    """
+    return dist / AVG_SPEED[method]
+
+
+def dijkstra(G, start, end, start_time, change_penalty=300, mode_penalties=PENALITIES):
+    # Initialize distances and time dictionary with all distances set to infinity
+    assert start in G.nodes(), f"Start node {start} not in graph"
+    assert end in G.nodes(), f"End node {end} not in graph"
+
+    distances = {
+        node: {
+            "distance": float("infinity"),
+            "time": start_time,
+        }
+        for node in list(G.nodes())
+    }
+
+    # Initialise dictionary to keep track of the edges that lead to the node with the smallest distance
+    edges_to = {node: None for node in list(G.nodes())}
+
+    # Set the distance from the start node to itself to 0
+    distances[start]["distance"] = 0
+    edges_to[start] = ("", "Start", {"type": "foot", "journey_id": None, "duration": 0})
+
+    # Priority queue to keep track of nodes with their current distances
+    priority_queue = [(0, start)]
+
+    while priority_queue:
+        # Get the node with the smallest distance from the priority queue
+        current_distance, current_node = heapq.heappop(priority_queue)
+
+        if current_node == end:
+            return distances[end], edges_to
+
+        # Check if the current distance is smaller than the stored distance
+        if current_distance > distances[current_node]["distance"]:
+            continue
+
+        for _, neighbor, attributes in G.out_edges(current_node, data=True):
+            no_duration_avail = not (type(attributes["duration"]) in [int, float])
+            if no_duration_avail:
+                continue
+
+            # Initialise distance to neighbor
+            distance = attributes["duration"]
+
+            # Add penalty
+            distance *= mode_penalties[attributes["type"]]
+
+            # Compute wait time for train (if next edge is a train)
+            wait = 0  # number of seconds you have to wait for the train
+            next_is_train = attributes["type"] == "train"
+            prev_is_train = edges_to[current_node][-1]["type"] == "train"
+            changed_trip = False
+            if next_is_train:
+                wait = (
+                    attributes["departure"] - distances[current_node]["time"]
+                ).total_seconds()
+                train_departed = (
+                    distances[current_node]["time"] > attributes["departure"]
+                )
+
+                if train_departed:
+                    continue
+
+                if prev_is_train:
+                    changed_trip = (
+                        edges_to[current_node][-1]["journey_id"]
+                        != attributes["journey_id"]
+                    )
+
+            # Penalise changing and waiting times
+            changed_mode = edges_to[current_node][-1]["type"] != attributes["type"]
+
+            # Add max of waiting time or changing penalty to current dist
+            if changed_mode or changed_trip:
+                distance += max(wait, change_penalty)
+
+            # Overall dist (prev dist + dist to neighbor)
+            distance = current_distance + distance
+
+            # If the new distance is smaller, update the distance and add to the priority queue
+            if distance < distances[neighbor]["distance"]:
+                # Update distance and time of arrival for neighbor
+                time_of_arrival = distances[current_node]["time"] + pd.Timedelta(
+                    seconds=distance
+                )
+                distances[neighbor]["distance"] = distance
+                distances[neighbor]["time"] = time_of_arrival
+
+                # Add edge that leads to neighbor
+                edges_to[neighbor] = (current_node, neighbor, attributes)
+
+                # Update priority queue
+                heapq.heappush(priority_queue, (distance, neighbor))
+
+    print(f"Probably there is no path between {start} and {end}")
+
+    return distances, edges_to
+
+
+def reconstruct_edges(edges_to, start: str, end: str):
+    """
+    Given shortest path from start to end, reconstruct edges.
+    """
+    path = [end]
+    edges = [edges_to[end]]
+    while path[-1] != start:
+        path.append(edges_to[path[-1]][0])
+        edges.append(edges_to[path[-1]])
+
+    return edges
+
+
+def postprocess_path(edges):
+    """Merge two edges if they have same transport type.
+
+        Edge Structure: (
+           start,
+           end,
+           {
+               type,
+               duration,
+               departure  (only for train),
+               arrival    (only for train),
+               journey_id (only for train),
+               trip_name  (only for train),
+        )
+
+    Args:
+        edges (list[tuple]): list of travel edges.
+
+    Returns:
+        list[tuple]: post-processed list of travel edges.
+    """
+    traversed = []
+    prev = None
+
+    for edge in edges[::-1]:
+        if prev is None:
+            prev = edge
+            traversed.append(edge)
+            continue
+
+        # Need to check the transport type and, if it is train,
+        # the journey id
+        if edge[2]["type"] == prev[2]["type"] and (
+            edge[2]["type"] != "train" or edge[2]["journey_id"] == prev[2]["journey_id"]
+        ):
+            prev = traversed.pop()
+
+            # Merge the two edges
+            new_edge = (
+                prev[0],
+                edge[1],
+                {
+                    "type": prev[2]["type"],
+                    "duration": prev[2]["duration"] + edge[2]["duration"],
+                },
+            )
+
+            if edge[2]["type"] == "train":
+                new_edge[2]["departure"] = prev[2]["departure"]
+                new_edge[2]["arrival"] = edge[2]["arrival"]
+                new_edge[2]["journey_id"] = edge[2]["journey_id"]
+                new_edge[2]["trip_name"] = edge[2]["trip_name"]
+        else:
+            new_edge = edge
+
+        prev = new_edge
+        traversed.append(new_edge)
+
+    return traversed
+
+
+def pretty_time_delta(seconds):
+    seconds = int(seconds)
+    days, seconds = divmod(seconds, 86400)
+    hours, seconds = divmod(seconds, 3600)
+    minutes, seconds = divmod(seconds, 60)
+    if days > 0:
+        return "%dd%dh%dm%ds" % (days, hours, minutes, seconds)
+    elif hours > 0:
+        return "%dh%dm%ds" % (hours, minutes, seconds)
+    elif minutes > 0:
+        return "%dm%ds" % (minutes, seconds)
+    else:
+        return "%ds" % (seconds,)
+
+
+def pretty_print(edges, args):
+    """Prints the edges in a nice format.
+
+    Args:
+        edges (list[tuple]): list of travel edges.
+    """
+    print(f"# Your Journey from {args.start} to {args.end}")
+
+    print(f"\nDate/ Time: {args.date} at {args.time}")
+    print(f"Sustainability: {args.sustainability}\n")
+
+    print("### Travel Information\n---\n")
+
+    for i, (src, dst, attr) in enumerate(edges):
+        if src == "Start":
+            src = args.start
+        if dst == "End":
+            dst = args.end
+
+        duration = pretty_time_delta(attr["duration"])
+
+        print(f"{i+1}. Go by {attr['type']} from {src} to {dst} for {duration}")
+
+"""
+    Remove all the trains from one station to another to simulate an outage.
+"""
+
+
+def remove_all_trains(G, from_station, to_station):
+    edges_to_remove = []
+
+    for edge in G.out_edges(from_station, data=True):
+        if edge[2]["type"] == "train" and edge[1] == to_station:
+            edges_to_remove.append(edge[:2])
+
+    print(
+        f"Removing all the edges from {from_station} to {to_station} : {len(edges_to_remove)}"
+    )
+    for edge in edges_to_remove:
+        G.remove_edge(*edge)
+
+
+def get_final_path_md(edges, start, end, date, time, sustainability):
+    
+    md = f"## Your Journey from {start} to {end}\n\n"
+    md += f"📅 Date/ Time: {date} at {time}\n"
+    md += "### Travel Information\n"
+
+    for i, (src, dst, attr) in enumerate(edges):
+        if src == "Start":
+            src = start
+        if dst == "End":
+            dst = end
+
+        duration = pretty_time_delta(attr["duration"])
+
+        travel_type = attr['type']
+        emoji = "🚉" if travel_type == "train" else "🚗" if travel_type == "car" else "🚶" if travel_type == "foot" else "🚀"
+
+        md += f"{i+1}. {emoji} Go by {travel_type} from {src} to {dst} for {duration}\n\n"
+    
+    return md
+
+def get_best_path(start, end, date, time, limit, exact_travel_time=False, outage=False, sustainability=False, change_penalty=300):
+    
+    # Load graph
+    with open("graph.pkl", "rb") as f:
+        G = pickle.load(f)
+
+    if outage:
+        # Remove all the edges from 2 stations to simulate an outage
+        remove_all_trains(G, from_station="St-Maurice", to_station="Martigny")
+
+    # Convert start and destination to location (lon, lat)
+    start_loc = get_location(G, start)
+    end_loc = get_location(G, end)
+
+    # Add both locations to graph
+    G.add_node("Start", pos=start_loc)
+    G.add_node("End", pos=end_loc)
+
+    # Find k-closest stations from start and end
+    dists_from_start = []
+    dists_to_end = []
+    for station, attr in G.nodes(data=True):
+        try:
+            station_pos = attr["pos"]
+            dists_from_start.append(
+                (station, get_distance(start_loc, station_pos))
+            )
+            dists_to_end.append((station, get_distance(end_loc, station_pos)))
+        except Exception as e:
+            continue
+
+    # Sort the distances in place
+    start_k_closest = sorted(dists_from_start, key=lambda x: x[1])[: limit]
+    end_k_closest = sorted(dists_to_end, key=lambda x: x[1])[: limit]
+
+    # Compute travel time from start to k closest stations
+    for mode in ["foot", "bike", "car"]:
+        for station, dist in start_k_closest:
+            if exact_travel_time:
+                travel_time = get_exact_travel_time(
+                    start_loc, station, method=mode
+                )
+                G.add_edge("Start", station, duration=travel_time, type=mode)
+            else:
+                travel_time = get_approx_travel_time(dist, method=mode)
+                G.add_edge("Start", station, duration=travel_time, type=mode)
+
+        for station, dist in end_k_closest:
+            if exact_travel_time:
+                travel_time = get_exact_travel_time(dist, method=mode)
+                G.add_edge(station, "End", duration=travel_time, type=mode)
+            else:
+                travel_time = get_approx_travel_time(dist, method=mode)
+                G.add_edge(station, "End", duration=travel_time, type=mode)
+
+    # Run Dijkstra on graph
+    start_time = pd.to_datetime(f"{date} {time}")
+    mode_penalties = get_penalties(sustainability)
+    dists, edges_to = dijkstra(
+        G,
+        "Start",
+        "End",
+        start_time=start_time,
+        change_penalty=change_penalty,
+        mode_penalties=mode_penalties,
+    )
+
+    # Reconstruct path
+    edges = reconstruct_edges(edges_to, "Start", "End")
+
+    # Postprocess path
+    path = postprocess_path(edges[:-1])
+
+    # Print journey
+    md = get_final_path_md(path, start, end, date, time, sustainability)
+
+    return md