Skip to content

Instantly share code, notes, and snippets.

@jalessio
Forked from irees/README.md
Last active Sep 13, 2016
Embed
What would you like to do?
Forked: Transit dimensions: Transitland Schedule API

Minor updates by jalessio 2016-08-04

  • Reduce PER_PAGE from 1000 to 500 to get around API gateway timeout errors
  • Remove call to meta['total'] which is no longer part of the API response JSON
  • Update date to August 2016

Transitland Frequency Visualization

Accompanies blog post: Transit dimensions: Transitland Schedule API

The frequency.py script:

  • Fetches all trips on a given date, between a start time and end time, inside of a bounding box
  • Calculates the number of connections between every stop
  • Uses a colormap and line width to show more frequent service
  • Outputs a GeoJSON map as output.geojson

An example GeoJSON output

The script defines all the query parameters as constants; feel free to use it as a jumping off point. The example interface is an excerpt from a more fully featured client library in the works.

Please check out the Transitland Datastore Github Repository for the full schedule API documentation.

Note that your results may vary slightly from the image used in the blog post.

"""Transitland Schedule API: create GeoJSON map of transit frequency."""
import json
import urllib
import urllib2
import datetime
import math
import os
##########################################################
##### Transitland Datastore Interface #####
##########################################################
class Datastore(object):
"""A simple interface to the Transitland Datastore."""
def __init__(self, host):
self.host = host
def _request(self, uri):
print uri
req = urllib2.Request(uri)
req.add_header('Content-Type', 'application/json')
response = urllib2.urlopen(req)
return json.loads(response.read())
def request(self, endpoint, **data):
"""Request with JSON response."""
return self._request(
'%s%s?%s'%(self.host, endpoint, urllib.urlencode(data or {}))
)
def paginated(self, endpoint, key, **data):
"""Request with paginated JSON response. Returns generator."""
response = self.request(endpoint, **data)
while response:
meta = response['meta']
print '%s: %s -> %s'%(
key,
meta['offset'],
meta['offset']+meta['per_page']
)
for entity in response[key]:
yield entity
if meta.get('next'):
response = self._request(meta.get('next'))
else:
response = None
def schedule_stop_pairs(self, **data):
"""Request Schedule Stop Pairs."""
return self.paginated(
'/api/v1/schedule_stop_pairs',
'schedule_stop_pairs',
**data
)
def stops(self, **data):
"""Request Stops"""
return self.paginated(
'/api/v1/stops',
'stops',
**data
)
def stop(self, onestop_id):
"""Request a Stop by Onestop ID."""
return self.request('/api/v1/stops/%s'%onestop_id)
def duration(t1, t2):
"""Return the time between two HH:MM:SS times, in seconds."""
fmt = '%H:%M:%S'
t1 = datetime.datetime.strptime(t1, fmt)
t2 = datetime.datetime.strptime(t2, fmt)
return (t2 - t1).seconds
##########################################################
##### Count trips between stops, output GeoJSON #####
##########################################################
HOST = 'http://transit.land'
PER_PAGE = 500
BBOX = [
-122.554,
37.668,
-122.085,
37.912
]
DATE = '2016-08-01'
BETWEEN = [
'07:00:00',
'09:00:00'
]
HOURS = duration(BETWEEN[0], BETWEEN[1]) / 3600.0
# Minimum vehicles per hour
# http://colorbrewer2.org/
COLORMAP = {
0: '#fef0d9',
3: '#fdcc8a',
6: '#fc8d59',
10: '#d7301f'
}
OUTFILE = 'output.geojson'
if os.path.exists(OUTFILE):
raise Exception("File exists: %s"%OUTFILE)
# Transitland Datastore API
ds = Datastore(HOST)
# Group SSPs by (origin, destination) and count
edges = {}
ssps = ds.schedule_stop_pairs(
bbox=','.join(map(str, BBOX)),
origin_departure_between=','.join(BETWEEN),
date=DATE,
per_page=PER_PAGE
)
for ssp in ssps:
key = ssp['origin_onestop_id'], ssp['destination_onestop_id']
if key not in edges:
edges[key] = 0
edges[key] += 1
# Get Stop geometries
stops = {}
for stop in ds.stops(per_page=PER_PAGE, bbox=','.join(map(str, BBOX))):
stops[stop['onestop_id']] = stop
# Create GeoJSON Features
colorkeys = sorted(COLORMAP.keys())
features = []
edges_sorted = sorted(edges.items(), key=lambda x:x[1])
for (origin_onestop_id,destination_onestop_id),trips in edges_sorted:
# Origin and destination geometries
origin = stops.get(origin_onestop_id)
destination = stops.get(destination_onestop_id)
if not (origin and destination):
# Outside bounding box
continue
# Frequency is in trips per hour
frequency = trips / HOURS
frequency_class = [i for i in colorkeys if frequency >= i][-1]
print "Origin: %s Destination: %s Trips: %s Frequency: %s Freq. class: %s"%(
origin_onestop_id,
destination_onestop_id,
trips,
frequency,
frequency_class
)
# Create the GeoJSON Feature
features.append({
"type": "Feature",
"name": "%s -> %s"%(origin['name'], destination['name']),
"properties": {
"origin_onestop_id": origin_onestop_id,
"destination_onestop_id": destination_onestop_id,
"trips": trips,
"frequency": frequency,
"frequency_class": frequency_class,
"stroke": COLORMAP[frequency_class],
"stroke-width": frequency_class+1,
"stroke-opacity": 1.0
},
"geometry": {
"type": "LineString",
"coordinates": [
origin['geometry']['coordinates'],
destination['geometry']['coordinates']
]
}
})
# Create the GeoJSON Feature Collection
fc = {
"type": "FeatureCollection",
"features": features
}
with open(OUTFILE, 'wb') as f:
json.dump(fc, f)
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment