commute-map/gmap.py

import datetime, json
import threading
import time
from itertools import zip_longest
import requests
from staticmap import StaticMap, CircleMarker
from colour import Color
import argparse

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)

class Worker(threading.Thread):
    def __init__(self, coord_batch, map, destination, backwards = False):
        threading.Thread.__init__(self)

        self.work_items = coord_batch
        self.map = map
        self.destination = destination
        self.backwards = backwards

    def run(self):

        global app_id
        global app_code

        destinations = ""
        for idx, tup in enumerate(self.work_items):
            if tup is not None:
                destinations += "&destination" + str(idx) + "=" + str(tup[1]) + "," + str(tup[0])

        starts = ""
        for idx, tup in enumerate(self.work_items):
            if tup is not None:
                starts += "&start" + str(idx) + "=" + str(tup[1]) + "," + str(tup[0])

        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

        r = requests.get(urls)
        jsons = json.loads(r.content)

        for idx, tup in enumerate(self.work_items):
            try:
                timelen = jsons["response"]["matrixEntry"][idx]["summary"]["travelTime"]
            except:
                continue

            col_val = int((timelen / 60))
            if col_val >= 30:
                marker = CircleMarker(tup, Color("blue").get_hex_l() + "70", 12)
            else:
                marker = CircleMarker(tup, colors[col_val].get_hex_l() + "70", 12)
            self.map.add_marker(marker)


def entry():

    m = StaticMap(3000, 3000, url_template='http://a.tile.osm.org/{z}/{x}/{y}.png')

    destination_point = (-122.195211, 47.792218)
    #destination_point = (-122.295310, 47.441328)
    #destination_point = (-122.359465, 47.619038)

    home_points = []

    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))

    total = len(home_points)

    worker_pool = []

    for p in grouper(home_points, 100):

        w = Worker(p, m, destination_point)
        w.start()
        worker_pool.append(w)

    for w in worker_pool:
        total -= 100
        print("Total: {}".format(total))
        w.join()

    image = m.render(zoom=13)
    image.save('output-map.png')

import argparse
parser = argparse.ArgumentParser()
parser.add_argument("app_id", help="")
parser.add_argument("app_code", help="")
args = parser.parse_args()

app_id = args.app_id
app_code = args.app_code

entry()