-
Notifications
You must be signed in to change notification settings - Fork 1
/
kml.py
79 lines (65 loc) · 3.35 KB
/
kml.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
from os import walk
from sys import argv
from time import sleep
from requests import post
from json import loads, dumps
API = 'https://api.open-elevation.com/api/v1/lookup'
def split_takeout_json():
"""
Split the monthly Google Takeout JSON files in json dir into a file per day.
Segments are filtered to only include related activities.
"""
for take in next(walk('takeout'), (None, None, []))[2]:
with open('takeout/' + take, 'r', encoding='utf-8') as t:
arr = [x['activitySegment'] for x in loads(t.read())['timelineObjects'] if
'activitySegment' in x and 'activityType' in x['activitySegment'] and
x['activitySegment']['activityType'] in ['SKIING', 'SNOWBOARDING', 'IN_GONDOLA_LIFT']]
days = set([x['duration']['endTimestamp'].split('T')[0] for x in arr])
for day in days:
with open('json/' + day + '.json', 'w') as x:
x.write('[')
for obj in arr:
with open('json/' + obj['duration']['endTimestamp'].split('T')[0] + '.json', 'a') as x:
x.write(dumps(obj) + ',')
for day in days:
with open('json/' + day + '.json', 'rb+') as x:
x.seek(-1, 2)
x.write(str.encode(']'))
if len(argv) > 1 and argv[1] == 'split':
split_takeout_json()
"""
Convert daily takeout JSON to GPX.
"""
for file in next(walk('json'), (None, None, []))[2]:
with open('json/' + file, 'r') as jsn:
with open('gpx/' + file.replace('json', 'gpx'), 'w') as gpx:
gpx.write('<?xml version="1.0" encoding="UTF-8" standalone="yes" ?>\n')
gpx.write('<gpx xmlns="http://www.topografix.com/GPX/1/1" '
'xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" '
'xsi:schemaLocation="http://www.topografix.com/GPX/1/1 http://www.topografix.com/GPX/1/1/gpx.xsd" '
'version="1.1" creator="gpx.py">\n')
gpx.write('\t<trk>\n')
gpx.write('\t\t<name>' + file.split('.')[0] + '</name>\n')
gpx.write('\t\t<type>Snowboarding</type>')
# Convert segments to trk
for segment in loads(jsn.read()):
if 'simplifiedRawPath' in segment:
gpx.write('\n\t\t<trkseg>\n')
# Gathers + fix coords
data = {'locations': []}
for p in segment['simplifiedRawPath']['points']:
data['locations'].append({"latitude": p["latE7"] / 1e7, "longitude": p["lngE7"] / 1e7})
# Get elevations
res = post(url=API, json=data)
while res.status_code == 429:
sleep(5)
res = post(url=API, json=data)
# Write track points
for c, p in zip(res.json()['results'], segment['simplifiedRawPath']['points']):
gpx.write(f'\t\t\t<trkpt lat="{c["latitude"]}" lon="{c["longitude"]}">\n')
gpx.write(f'\t\t\t\t<ele>{c["elevation"]}</ele>\n')
gpx.write(f'\t\t\t\t<time>{p["timestamp"]}</time>\n')
gpx.write(f'\t\t\t</trkpt>\n')
gpx.write('\t\t</trkseg>')
gpx.write('\n\t</trk>')
gpx.write('\n</gpx>')