forked from comp-journalism/2016-03-wapo-uber
-
Notifications
You must be signed in to change notification settings - Fork 3
/
gatherUberData.py
149 lines (123 loc) · 5.57 KB
/
gatherUberData.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
import ConfigParser
import json
import time
import grequests
import sys
import os
import csv
import urllib
import logging
import pytz
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler
# The API end points
price_url = "https://api.uber.com/v1/estimates/price"
product_url = "https://api.uber.com/v1/products"
time_url = "https://api.uber.com/v1/estimates/time"
# Parse in the configuration information to get uber server tokens
config = ConfigParser.ConfigParser()
config.read(os.path.dirname(os.path.abspath(__file__)) + "/config.conf")
uber_server_tokens = (config.get("MainConfig", "uber_server_tokens")).split(",")
# Parse in the locations and output file name
locations = json.loads(config.get("MainConfig", "locations"))
output_file_name = config.get("MainConfig", "output_file_name")
time_interval = int(config.get("MainConfig", "time_interval"))
# Create a file to store data
fileWriter = csv.writer(open(output_file_name, "w+"),delimiter=",")
fileWriter.writerow(["timestamp", "surge_multiplier", "expected_wait_time", "product_type", "low_estimate", "high_estimate", "start_location_id", "end_location_id"])
# These api param objects are used to send requests to the API, we create api_param objects for each location and endpoint we want to gather
api_params = []
for l in locations:
location_id = l["locations"]
price_parameters = {
'start_latitude': l["latitude"],
'end_latitude': l["latitude"],
'start_longitude': l["longitude"],
'end_longitude': l["longitude"]
}
time_parameters = {
'start_latitude': l["latitude"],
'start_longitude': l["longitude"],
}
api_params.append({"url": price_url, "location_id": location_id, "type": "price", "parameters": price_parameters})
api_params.append({"url": time_url, "location_id": location_id, "type": "time", "parameters": time_parameters})
# Create local time zone for DC
local_tz = pytz.timezone('US/Eastern')
def utc_to_local(utc_dt):
return utc_dt.replace(tzinfo=pytz.utc).astimezone(local_tz)
def aslocaltimestr(utc_dt):
return utc_to_local(utc_dt).strftime('%Y-%m-%d %H:%M:%S')
tokennum = 0
def gather_loop():
global tokennum
# A list to hold our things to do via async
async_action_items = []
common_data_dict = {}
for i, api_param in enumerate(api_params):
# Get the current time
api_param["datetime"] = aslocaltimestr(datetime.utcnow())
api_param["data"] = common_data_dict
api_param["parameters"]["server_token"] = uber_server_tokens[tokennum % len(uber_server_tokens)]
# From here: http://stackoverflow.com/questions/25115151/how-to-pass-parameters-to-hooks-in-python-grequests
action_item = grequests.get(api_param["url"]+"?"+urllib.urlencode(api_param["parameters"]), hooks={'response': [hook_factory(api_param)]})
async_action_items.append(action_item)
# increment the token num so that we use the next server key next time
tokennum = tokennum + 1
# Initiate both requests in parallel
grequests.map(async_action_items)
def hook_factory(*factory_args, **factory_kwargs):
def it_responded(res, **kwargs):
call_type = factory_args[0]["type"]
location_id = factory_args[0]["location_id"]
current_time = factory_args[0]["datetime"]
data_dict = factory_args[0]["data"]
try:
json_response = json.loads(res.content)
#print json_response
# Parse the data differently depending on the type of call it was
try:
if call_type == "time":
for t in json_response["times"]:
if t["display_name"] not in data_dict:
data_dict[t["display_name"]] = {}
data_dict[t["display_name"]]["expected_wait_time"] = t["estimate"]
elif call_type == "price":
for p in json_response["prices"]:
if p["display_name"] not in data_dict:
data_dict[p["display_name"]] = {}
data_dict[p["display_name"]]["surge_multiplier"] = p["surge_multiplier"]
data_dict[p["display_name"]]["product_type"] = p["display_name"]
data_dict[p["display_name"]]["low_estimate"] = p["low_estimate"]
if data_dict[p["display_name"]]["low_estimate"] != None:
data_dict[p["display_name"]]["low_estimate"] = int(data_dict[p["display_name"]]["low_estimate"])
data_dict[p["display_name"]]["high_estimate"] = p["high_estimate"]
if data_dict[p["display_name"]]["high_estimate"] != None:
data_dict[p["display_name"]]["high_estimate"] = int(data_dict[p["display_name"]]["high_estimate"])
#print data_dict
# For each product in the data dict write out a row
for p in data_dict:
data_dict[p]["timestamp"] = current_time
data_dict[p]["start_location_id"] = location_id
data_dict[p]["end_location_id"] = location_id
#print data_dict[p]
#if it has time and price then store it
if "expected_wait_time" in data_dict[p] and "surge_multiplier" in data_dict[p]:
#print data_dict[p]
fileWriter.writerow([data_dict[p]["timestamp"], data_dict[p]["surge_multiplier"], data_dict[p]["expected_wait_time"], data_dict[p]["product_type"], data_dict[p]["low_estimate"], data_dict[p]["high_estimate"], data_dict[p]["start_location_id"], data_dict[p]["end_location_id"]])
except TypeError as e:
print e
except Exception as e:
print "The response at " + str(aslocaltimestr(datetime.utcnow())) + " was: "
print json_response
print res.content
print e
return it_responded
# Create a scheduler to trigger every N seconds
# http://apscheduler.readthedocs.org/en/3.0/userguide.html#code-examples
logging.basicConfig()
scheduler = BackgroundScheduler()
scheduler.add_job(gather_loop, 'interval', seconds = time_interval)
scheduler.start()
while True:
time.sleep(1)
# nohup python -u gatherUberData.py > nohup.txt &