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
|
#!/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)
|
