mitchellhansen
/
commute-map
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Compare commits

base: mitchellhansen:master
Branches Tags
mitchellhansen:master
mitchellhansen:branch
..
compare: mitchellhansen:branch
Branches Tags
mitchellhansen:master
mitchellhansen:branch

1 Commits
master ... branch

Author SHA1 Message Date
mitchellhansen c2ded6a9c6 old screenshot
4 years ago
1 changed files with 60 additions and 151 deletions
Show Stats

211
gmap.py
Unescape View File

@ -6,198 +6,107 @@ import requests
from staticmap import StaticMap, CircleMarker
from colour import Color
import argparse
import oauthlib
import requests
from requests_oauthlib import OAuth1Session, OAuth1
token_url = 'https://account.api.here.com/oauth2/token'
matrix_url = "https://matrix.router.hereapi.com/v8/matrix"
matrix_status_url = "https://matrix.router.hereapi.com/v8/matrix/{0}/status"
matrix_retrieve_url = "https://matrix.router.hereapi.com/v8/matrix/{0}"
granularity = 0.0050
commute_range = 75
work_group_size = 100
color_scale_width = 60 # this is equal to minutes until full red saturation, 60 == 60 minutes
alpha = "70"
blot_size = 13
map_scale = 11
reverse = True
def get_bearer_token():
data = {"grant_type": "client_credentials"}
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
oauth = OAuth1(client_key, client_secret=client_secret)
r = requests.post(url=token_url, auth=oauth, headers=headers, data=data)
access_token = json.loads(r.content)['access_token']
return access_token
# Groups home_points into n groups, fills the non-fits with fillvalue
def group_elements(home_points, n, fillvalue=None):
args = [iter(home_points)] * n # this is some 9000IQ syntax
return zip_longest(*args, fillvalue=fillvalue)
green = Color("green")
colors = list(green.range_to(Color("red"),30))
app_id = 0
app_code = 0
def grouper(iterable, n, fillvalue=None):
args = [iter(iterable)] * n
return zip_longest(*args, fillvalue=fillvalue)
# This is called with the
class Worker(threading.Thread):
def __init__(self, destination, coord_batch, graphical_map, access_token, backwards=False):
def __init__(self, coord_batch, map, destination, backwards = False):
threading.Thread.__init__(self)
self.coord_batch = coord_batch
self.access_token = access_token
self.graphical_map = graphical_map
self.work_items = coord_batch
self.map = map
self.destination = destination
self.backwards = backwards
def run(self):
global app_id
global app_code
start_points = []
for idx, tup in enumerate(self.coord_batch):
destinations = ""
for idx, tup in enumerate(self.work_items):
if tup is not None:
start_points.append({'lat': tup[0], 'lng': tup[1]})
if self.backwards is True:
origins = [self.destination]
destinations = start_points
else:
origins = start_points
destinations = [self.destination]
request_payload = {
"origins": origins,
"destinations": destinations,
"regionDefinition": {
"type": "world",
# "type": "circle",
# "center": self.destination,
# "radius": 10000
}
}
headers = {
'Authorization': 'Bearer {}'.format(self.access_token),
'Content-Type': 'application/json'
}
response = requests.request("POST", matrix_url, headers=headers, data=json.dumps(request_payload))
post_response = json.loads(response.text)
matrix_id = post_response['matrixId']
while True:
response = requests.request("GET", matrix_status_url.format(matrix_id), headers=headers, allow_redirects=False)
jsons = json.loads(response.text)
if jsons['status'] == "inProgress" or jsons['status'] == "accepted":
time.sleep(10)
continue
elif jsons['status'] == "completed":
break
else:
print("big problems " + response.text)
break
destinations += "&destination" + str(idx) + "=" + str(tup[1]) + "," + str(tup[0])
headers = {
'Authorization': 'Bearer {}'.format(self.access_token),
# 'Content-Type': 'application/json',
'Accept - Encoding': 'gzip'
}
starts = ""
for idx, tup in enumerate(self.work_items):
if tup is not None:
starts += "&start" + str(idx) + "=" + str(tup[1]) + "," + str(tup[0])
response = requests.request("GET", matrix_retrieve_url.format(matrix_id), headers=headers)
jsons = json.loads(response.text)
urls = "https://matrix.route.api.here.com/routing/7.2/calculatematrix.json?app_id={0}&app_code={1}&departure=2018-12-21T01:00:00&mode=fastest;car;traffic:enabled&summaryAttributes=traveltime&start=".format(app_id, app_code)
urls += str(self.destination[1]) + "," + str(self.destination[0])
urls += destinations
green = Color("green")
colors = list(green.range_to(Color("red"), color_scale_width))
r = requests.get(urls)
jsons = json.loads(r.content)
for idx, tup in enumerate(self.coord_batch):
for idx, tup in enumerate(self.work_items):
try:
timelen = jsons["matrix"]["travelTimes"][idx]
timelen = jsons["response"]["matrixEntry"][idx]["summary"]["travelTime"]
except:
continue
try:
if jsons["matrix"]["errorCodes"][idx] != 0:
error = True
col_val = int((timelen / 60))
if col_val >= 30:
marker = CircleMarker(tup, Color("blue").get_hex_l() + "70", 12)
else:
error = False
except:
error = False
pass
tup = (tup[1], tup[0])
col_val = int((timelen / 60)) # set the color bucket to the number of minutes
# so this means that full saturation will == the number of minutes when setting color_scale_width
if error is True:
marker = CircleMarker(tup, "blue", 6)
elif col_val >= color_scale_width:
# the + 70 here is just setting the aplha value
marker = CircleMarker(tup, Color("red").get_hex_l() + alpha, blot_size)
else:
if col_val == color_scale_width / 2:
marker = CircleMarker(tup, colors[col_val].get_hex_l() + "FF", blot_size)
else:
marker = CircleMarker(tup, colors[col_val].get_hex_l() + alpha, blot_size)
self.graphical_map.add_marker(marker)
print("Done with this one")
marker = CircleMarker(tup, colors[col_val].get_hex_l() + "70", 12)
self.map.add_marker(marker)
def entry():
access_token = get_bearer_token()
m = StaticMap(3000, 3000, url_template='http://a.tile.osm.org/{z}/{x}/{y}.png')
url = 'http://a.tile.osm.org/{z}/{x}/{y}.png'
url = 'https://cartodb-basemaps-b.global.ssl.fastly.net/light_nolabels/{z}/{x}/{y}.png '
graphical_map = StaticMap(3000, 3000, url_template=url)
destination_point = (-122.195211, 47.792218)
#destination_point = (-122.295310, 47.441328)
#destination_point = (-122.359465, 47.619038)
# Since we are testing commute-to, we set the destination of "work"
destination_point = {'lat': 47.5121104, "lng": -121.8852897}
home_points = []
# Populate this list with an complete xy grid of points centered on destination_point
coord_grid = []
range_val = 20
for x in range(-range_val, range_val):
for y in range(-range_val, range_val):
#if (x > 148 or x < -148) or (y > 148 or y < -148):
x_point = destination_point[0] + 0.0012 * x
y_point = destination_point[1] + 0.0012 * y
home_points.append((x_point, y_point))
for x in range(-commute_range, commute_range):
for y in range(-commute_range, commute_range):
x_point = destination_point['lat'] + granularity * x
y_point = destination_point['lng'] + granularity * y
coord_grid.append((x_point, y_point))
total = len(home_points)
coords_left_to_query = len(coord_grid)
# dispatch the workers to the pool
worker_pool = []
for coord_batch in group_elements(coord_grid, work_group_size):
w = Worker(destination_point, coord_batch, graphical_map, access_token, backwards=reverse)
for p in grouper(home_points, 100):
w = Worker(p, m, destination_point)
w.start()
worker_pool.append(w)
# Print out our progress as wel join up the workers
for w in worker_pool:
total -= 100
print("Total: {}".format(total))
w.join()
coords_left_to_query -= work_group_size
print("Coords left to query: {}".format(coords_left_to_query))
marker = CircleMarker((destination_point['lng'], destination_point['lat']), "red", 12)
graphical_map.add_marker(marker)
# finish up our work with the map
image = graphical_map.render(zoom=map_scale)
image.save('output-map.png')
# Program start
image = m.render(zoom=13)
image.save('maps.png')
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--key", help="")
parser.add_argument("--secret", help="")
parser.add_argument("app_id", help="")
parser.add_argument("app_code", help="")
args = parser.parse_args()
client_key = args.key
client_secret = args.secret
app_id = args.app_id
app_code = args.app_code
entry()

Write Preview
Loading…
Cancel
Save