-
Notifications
You must be signed in to change notification settings - Fork 0
/
wmata.py
319 lines (268 loc) · 10.1 KB
/
wmata.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
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
# libraries
import requests
# project modules
from utils import config
API_BASE_URL = 'https://api.wmata.com/Bus.svc/json'
API_END_POINTS = {
'bus_position': f'{API_BASE_URL}/jBusPositions',
'path_details': f'{API_BASE_URL}/jRouteDetails',
'routes': f'{API_BASE_URL}/jRoutes',
'route_scheds ': f'{API_BASE_URL}/jRouteSchedule',
'stop_scheds ': f'{API_BASE_URL}/jStopSchedule',
'stops': f'{API_BASE_URL}/jStops',
'validate_key': 'https://api.wmata.com/Misc/Validate'
}
API_KEYS = config('wmata')
DATA_CHOICES = [
'positions', 'routes', 'route_scheds ',
'incident', 'stops', 'stop_scheds'
]
def validate_key(api_key: str) -> bool:
header = {
'api_key': api_key
}
r = requests.get(
url=API_END_POINTS['validate_key'],
headers=header
)
if r.status_code == requests.codes.ok:
return True
return False
def get_bus_position(route_id=None, lat=None, lon=None, radius=None):
"""
Returns bus positions for the given route, with an optional search radius.
If no parameters are specified, all bus positions are returned. Note that
the RouteID parameter accepts only base route names and no variations,
i.e.: use 10A instead of 10Av1 or 10Av2.
Bus positions are refreshed approximately every 7 to 10 seconds.
Args:
route_id (str): Optional; Base bus route, e.g.: 70, 10A.
lat (float): Optional; Center point Latitude, required if Longitude
and Radius are specified.
lon (float): Optional; Center point Longitude, required if Latitude
and Radius are specified.
radius (float): Optional; Radius (meters) to include in the search
area, required if Latitude and Longitude are specified.
Returns:
Request response object where json() method provides the following elements:
BusPositions - array containing bus position information.
"""
headers = {
'api_key': API_KEYS['bus_pos_key']
}
# configure api parameters
params = dict()
if route_id:
params['RouteID'] = route_id
if lat:
params['Lat'] = lat
if lon:
params['Lon'] = lon
if radius:
params['Radius'] = radius
r = requests.get(
url=API_END_POINTS['bus_position'],
headers=headers, params=params
)
return r
def get_path_details(route_id: str, date=None):
"""
For a given date, returns the set of ordered latitude/longitude points
along a route variant along with the list of stops served.
Args:
route_id (str): Bus route variant, e.g.: 70, 10A, 10Av1.
date (str): Optional; Date in YYYY-MM-DD format for which to retrieve
route and stop information. Defaults to today's date unless specified.
Returns:
Request response object where json() method provides the following elements:
Direction0 - structures describing path/stop information for the route.
Direction1 - structures describing path/stop information for the route.
Name - descriptive name for the route.
RouteID - bus ute variant (e.g.: 10A, 10Av1, ect.)
"""
headers = {
'api_key': API_KEYS['default']
}
# configure api parameters
params = dict()
if route_id:
params['RouteID'] = route_id
if date:
params['Date'] = date
r = requests.get(
url=API_END_POINTS['path_details'],
headers=headers, params=params
)
return r
def get_routes():
"""
Returns a list of all bus route variants (patterns).
For example, the 10A and 10Av1 are the same route, but may stop at slightly
different locations.
Returns:
Request response object where json() method provides the following elements:
Routes - array containing route variant information.
"""
headers = {
'api_key': API_KEYS['default']
}
r = requests.get(
url=API_END_POINTS['routes'],
headers=headers
)
return r
def get_schedule(route_id: str, date=None, including_variations=None):
"""Returns schedules for a given route variant for a given date.
Args:
route_id (str): Bus route variant, e.g.: 70, 10A, 10Av1.
date (str): Optional; Date in YYYY-MM-DD format for which to retrieve
schedule. Defaults to today's date unless specified.
including_variations (bool): Optional; Whether or not to include
variation
Returns:
Request response object where json() method provides the following elements:
Direction0 - structures describing path/stop information for the route.
Direction1 - structures describing path/stop information for the route.
Name - descriptive name for the route.
"""
headers = {
'api_key': API_KEYS['bus_route_sched_key']
}
# configure api parameters
params = dict()
if route_id:
params['RouteID'] = route_id
if date:
params['Date'] = date
if including_variations:
params['IncludingVariations'] = including_variations
r = requests.get(
url=API_END_POINTS['route_scheds '],
headers=headers, params=params
)
return r
def get_stop_schedule(stop_id: str, date=None):
"""Returns a set of buses scheduled at a stop for a given date.
Args:
stop_id (str): 7-digit regional stop ID.
date (str): Optional; Date in YYYY-MM-DD format for which to retrieve
schedule. Defaults to today's date unless specified.
Returns:
Request response object where json() method provides the following elements:
ScheduleArrivals - array containing scheduled arrival information.
Stop - structures describing stop information.
"""
headers = {
'api_key': API_KEYS['default']
}
# configure api parameters
params = dict()
if stop_id:
params['StopID'] = stop_id
if date:
params['Date'] = date
r = requests.get(
url=API_END_POINTS['stop_scheds '],
headers=headers, params=params
)
return r
def get_stops(lat=None, lon=None, radius=None):
"""
Returns a list of nearby bus stops based on latitude, longitude, and radius.
Omit all parameters to retrieve a list of all stops.
Args:
lat (float): Optional; Center point Latitude, required if Longitude
and Radius are specified.
lon (float): Optional; Center point Longitude, required if Latitude
and Radius are specified.
radius (float): Optional; Radius (meters) to include in the search
area, required if Latitude and Longitude are specified.
Returns:
Request response object where json() method provides the following elements:
Stops - array containing stop information.
"""
headers = {
'api_key': API_KEYS['bus_pos_key']
}
# configure api parameters
params = dict()
if lat:
params['Lat'] = lat
if lon:
params['Lon'] = lon
if radius:
params['Radius'] = radius
r = requests.get(
url=API_END_POINTS['stops'],
headers=headers, params=params
)
return r
def get_route_ids(routes_data: dict) -> list:
"""Return list of route ids from routes resp.json() data."""
route_ids = list()
for route in routes_data['Routes']:
route_ids.append(route['RouteID'])
return route_ids
def get_stop_ids(stops_data: dict) -> list:
"""Return list of stop ids from stops resp.json() data."""
stop_ids = list()
for stop in stops_data['Stops']:
stop_ids.append(stop['StopID'])
return stop_ids
def flatten_route_sched_data(resp_json: dict) -> list:
"""
Helper function to reformat bus schedule response data into flat format.
"""
# Remove and store name
name = resp_json['Name']
del resp_json['Name']
# Process direction, trip, and stop information
data = list()
for direction, trip_elem in resp_json.items(): # list of dicts
for trip in trip_elem: # trip_elem is list of dicts
for stop in trip['StopTimes']: # stops is list of dicts
row = {'Name': name}
row.update(trip) # add trip data for respective stop
del row['StopTimes'] # remove StopTimes dict object
row.update(stop) # store specific stop_time as row data
data.append(row)
return data
def flatten_stop_sched_data(resp_json: dict) -> list:
"""
Helper function to reformat stop schedule response data into flat format.
"""
# TODO: See flatten_route_sched_data function
raise NotImplementedError
def flatten_path_details_data(resp_json: dict) -> dict:
# Remove and store name and route id
name = resp_json['Name']
route_id = resp_json['RouteID']
del resp_json['Name']
del resp_json['RouteID']
# Process direction elements to get stops and shapes information
path_stops = list()
path_shapes = list()
data = {
'path_details_stops': path_stops,
'path_details_shapes': path_shapes
}
for direction, trip_elem in resp_json.items(): # direction is dict of dict
if not trip_elem: # skip if direction is empty dict
continue
# create path details stops rows
for stop_num, stop in enumerate(trip_elem['Stops'], 1): # list of dicts
row = {'RouteName': name, 'RouteID': route_id, 'StopNum': stop_num}
row.update(trip_elem) # add trip data for respective stop
del row['Shape'] # remove Shape dict object
del row['Stops'] # remove StopTimes dict object
row.update(stop) # store specific stop_time as row data
path_stops.append(row)
# create path details shapes rows
for shape in trip_elem['Shape']: # list of dicts
row = {'RouteName': name, 'RouteID': route_id}
row.update(trip_elem) # add trip data for respective stop
del row['Shape'] # remove Shape dict object
del row['Stops'] # remove StopTimes dict object
row.update(shape) # store specific shapes as row data
path_shapes.append(row)
return data