Editing Public Transport Route System

In this example we show how to add a new public transport route, beginning by adding new transit stops all the way to adding trips for an entire day.

Imports

from math import ceil

from shapely.ops import linemerge

from polaris import Polaris
from polaris.runs.static_assignment.static_graph import StaticGraph
from polaris.utils.testing.temp_model import TempModel

Open the network and open the network database sphinx_gallery_thumbnail_path = ‘../../examples/editing_models/bus_map.png’

project_dir = TempModel("Bloomington")
pol = Polaris.from_dir(project_dir)

network = pol.network
transit = network.transit

Let’s assume we want to add a new long-distance bus route between two nodes in the model And that we want this route to not go through Freeways, Expressways or Ramps, if possible So the idea is to build the alignment of the route first, using the path over the network between these two nodes

# But feel free to bring your own route shapes and stop locations/sequence !!!!

The two nodes we want to connect with a bus route are the following:

origin = 14
destination = 2064

Let’s get the road network graph to compute the best path between the two nodes

graph_maker = StaticGraph(network.path_to_file)
graph = graph_maker.graph

:   0%|          | 0/185 [00:00<?, ?it/s]

Now that we have the graph, we need to penalize freeways, expressways and ramps heavily (10x)

# Let's read the links from the network
links = network.tables.get("Link")

# Identify the freeways
freeways = links[links["type"].isin(['FREEWAY', 'RAMP', 'EXPRESSWAY'])].link.to_numpy()

# We penalize the field in the graph, we don't mess with the geometries or the link table directly
# JUST A GRAPH PENALIZATION
# Let's penalize both distance and time in both directions of each link, even though we will not use distance in this
# example
# This is done using AequilibraE, so documentation for this code would be found in its website
graph.network.loc[graph.network.link_id.isin(freeways), ['time_ab', 'time_ba', 'distance']] *= 10
graph.prepare_graph(graph.centroids)
graph.set_blocked_centroid_flows(False)

# We minimize time, could also do distance
# graph.set_graph("distance")
graph.set_graph("time")

Now we compute the path between our origin and our destination The node_offset is necessary because of the way the graph is built internally, as POLARIS models have an overlap between zone IDs and node IDs The resulting path object contains all the information about the links traversed including the direction of traversal for each link

# 'path' is an AequilibraE path object
path = graph.compute_path(origin + graph_maker.node_offset, destination + graph_maker.node_offset)

We loop through all the links in our path to get their geometries and build the full route geometry

all_geometry = []
for link, link_dir in zip(path.path, path.path_link_directions):
    # We get the link geometry directly from our link GeoDataFrame, so geometries are true to our model
    geo = links.loc[links.link == link, 'geo'].values[0]
    # If we traversed the link in the BA direction, we need to reverse the geometry
    geo = geo.reverse() if link_dir < 0 else geo
    all_geometry.append(geo)

# We merge all geometries into a single linestring
route_geo = linemerge(all_geometry)

Let’s find the locations of the stops we want to add along the route We will add a stop every 750 meters

spacing = 750
num_stops = ceil(route_geo.length / spacing)
stop_geos = [route_geo.interpolate(x / num_stops, normalized=True) for x in range(num_stops + 1)]

route_id = transit.edit.add_route(agency_id=2, mode_id=3, shape=route_geo, route_name="my long distance route")

# We will add 3 new stops

stop_sequence =[]
for i, geo in enumerate(stop_geos):
    # Note that latitude is Y and longitude is X
    s_id = transit.edit.add_stop(agency_id=2, route_type=3, stop_code=f"stp_{i}", latitude=geo.y, longitude=geo.x)
    stop_sequence.append(s_id)

Our pattern will include the three stops we added, plus one pre-existing stop

pattern_id = transit.edit.add_route_pattern(route_id=route_id, stop_sequence=stop_sequence)

# We can also create the reverse pattern in one go
pattern_return = transit.edit.add_route_pattern(route_id=route_id, stop_sequence=stop_sequence[::-1])

# We could have also done that as below
# pattern_return = transit.edit.flip_pattern(pattern_id, keep_original_pattern=True)

We Compute the departure time for each trip in our period of interest

# Let's say we want to add trips for the first three hours of the day, each hour with a different headway
# We have a neat little API for coalescing the departure times for all of the intervals
intervals = [
    {"start": 3600, "end": 7200, "headway": 1200},
    {"start": 7200, "end": 10800, "headway": 1000},
]

trip_starts = transit.edit.suggest_departures(intervals)
print(trip_starts)

[4200.0, 5400.0, 6600.0, 7700.0, 8700.0, 9700.0, 10700.0]

We add a new trip for each trip departure time in our period of interest

added_trips = []
for instant in trip_starts:
    # This is the most vanilla alternative, where we add a new trip with a default speed
    # that will only be used if we don't find a speed value from the database (from route or gtfs mode, in that order)
    # Speeds are in m/s
    trip_id = transit.edit.add_pt_trip(pattern_id=pattern_id, departure_time=instant, default_speed=11)
    trip_id_return = transit.edit.add_pt_trip(pattern_id=pattern_return, departure_time=instant, default_speed=11)

    # We can also request that only trips that begin within a certain interval can be used for deriving the new
    # trip's speed
    # trip_id = transit.edit.add_pt_trip(pattern_id=pattern_id, departure_time=instant, default_speed=15,
    #                                            horizon_begin=28800, horizon_end=36000)

    # We can also enforce the use of the default speed we are providing
    # trip_id = transit.edit.add_pt_trip(pattern_id=pattern_id, departure_time=instant, default_speed=15,
    #                                            force_default_speed=True)

    # We can also tell polaris to apply a speedup factor to the speed found in the database, in case these trips are
    # expected to be faster (factor>1) or slower (factor<1) than what is found on average
    # This factor does NOT apply to the *default_speed*
    # trip_id = transit.edit.add_pt_trip(pattern_id=pattern_id, departure_time=instant, default_speed=15,
    #                                            speed_factor=1.15)
    added_trips.extend([trip_id, trip_id_return])

/home/gitlab-runner/builds/polaris/code/polarislib/polaris/network/transit/edit_supply.py:461: UserWarning: Caching data
  warnings.warn("Caching data")

# Let's say we want to ensure a certain frequency for a pattern within an interval
# In that case, we can use a dedicated API call that adds all trips necessary to keep that headway
# and deletes the trips that previously existed there

# First we clear the data cache so we can be sure to incorporate all recently added trips to our computations
transit.edit.clear_data_cache()

trip_starts = transit.edit.suggest_departures([{"start": 0, "end": 86400, "headway": 30000}])

added, deleted = transit.edit.ensure_pattern_departures(
    pattern_id=pattern_id, departure_times=trip_starts, default_speed=30
)

print("Added trips: ", added)
print("Deleted trips: ", deleted)

/home/gitlab-runner/builds/polaris/code/polarislib/polaris/network/transit/edit_supply.py:461: UserWarning: Caching data
  warnings.warn("Caching data")
Added trips:  [2001600010008, 2001600010009, 2001600010010]
Deleted trips:  [2001600010001, 2001600010002, 2001600010003, 2001600010004, 2001600010005, 2001600010006, 2001600010007]

Visualizing the results

patterns = network.tables.get("Transit_Patterns")
patterns = patterns[patterns.pattern_id == pattern_id]
stops_df = network.tables.get("Transit_Stops")
stops_df = stops_df[stops_df.stop_id.isin(stop_sequence)]


m = patterns.explore()
stops_df.explore(color='red', m=m)

Make this Notebook Trusted to load map: File -> Trust Notebook

In practice, ou would also have to rebuild the network graphs to incorporate the new transit routes

# from polaris.network.traffic.road_connectors import RoadConnectors
#
# active = network.active
# active.build()
#
# # And the ferry connectors (if they exist)
# rc = RoadConnectors(supply_path=network.path_to_file)
# # And we add connectors that are no longer than 50 meters
# rc.connect_ferries(max_distance=50)