#!/usr/bin/env python import requests from datetime import datetime, timedelta from collections import defaultdict import json import os, sys import pytz import argparse URL= "https://maps.googleapis.com/maps/api/distancematrix/json" parser = argparse.ArgumentParser(description='Plot traffic information for various origin-destinations') parser.add_argument('-c', '--config', type=str, default='config.json', help='Config file to load (default config.jsonà') parser.add_argument('-n', '--dry-run', action='store_true', help='Run in dry-run mode (don’t make any api call)') parser.add_argument('-d', '--date', type=str, default='now', help='date (%%Y-%%m-%%d) for which to generate the plots') parser.add_argument('-s', '--stats', action='store_true', help='only print number of weekly calls for the config') args = parser.parse_args() WEEKDAYS = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"] SETTINGS = json.load(open(args.config, "r")) if args.date == "now": date = datetime.now(tz=pytz.timezone("Europe/Paris")) else: date = datetime.strptime(args.date, "%Y-%m-%d").replace(tzinfo=pytz.timezone("Europe/Paris")) def parse_plot(info): def condition(d): return "days" not in info or WEEKDAYS[d.weekday()] in info["days"] return { "name": info["name"], "origin": SETTINGS.get("locations", {}).get(info["origin"], info["origin"]), "destination": SETTINGS.get("locations", {}).get(info["destination"], info["destination"]), "condition": condition, "times": [t.split(":") for t in info["times"]], } plots = {} for name,info in SETTINGS["plots"].items(): plots[name] = parse_plot(info) def to_time(date, hour, minute): return date.replace(date.year, date.month, date.day, int(hour), int(minute), 0, 0) def get(time, origin, destination): keys = { "origins": origin, "destinations": destination, "departure_time": int(time.timestamp()), "key": SETTINGS["api_key"], } return requests.get(URL, params=keys) def stats(): total_calls_per_week = 0 for k, infos in plots.items(): calls_per_week = 0 for i in range(0, 7): if infos["condition"](date + timedelta(i)): calls_per_week += 1 total_calls_per_week += calls_per_week * len(infos["times"]) print("{} calls per week to api".format(total_calls_per_week)) if args.stats: stats() sys.exit(0) gnuplot = """ reset set terminal png size 1024,768 enhanced font "Helvetica" set output 'outputs/{fname}.png' set datafile separator "\\t" set xlabel "Time" set ylabel "Duration" set xdata time set timefmt "%H:%M" set format x "%H:%M" set title "{date} {name}" plot '{fname}.dat' using 1:2 with linespoints pointtype 5 title "" """ def get_times(name, infos, date, dry_run=False): result = {} if "condition" in infos and not infos["condition"](date): return result for time in infos["times"]: t = to_time(date, time[0], time[1]) try: if dry_run: v = 42 else: v = get(t, infos["origin"], infos["destination"]).json() v = int(v["rows"][0]["elements"][0]["duration_in_traffic"]["value"] / 60) except: continue result[t.strftime("%H:%M")] = v with open(date.strftime("%Y-%m-%d_{}.json".format(name)), "w") as f: f.write(json.dumps(result, indent=True)) fname = date.strftime("%Y-%m-%d_{}".format(name)) with open("{}.dat".format(fname), "w") as f: for time, duration in result.items(): f.write('{}\t{}\n'.format(time, duration)) with open("{}.gnuplot".format(fname), "w") as f: f.write(gnuplot.format(**{ "fname": fname, "name": infos["name"], "date": date.strftime("%Y-%m-%d") })) os.system("gnuplot -c {}.gnuplot".format(fname)) return result for name, infos in plots.items(): get_times(name, infos, date, dry_run=args.dry_run)