-
Notifications
You must be signed in to change notification settings - Fork 0
/
import_osm.py
104 lines (76 loc) · 2.67 KB
/
import_osm.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
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
import requests
from osm2geojson import json2geojson
import json
import logging
import os
from time import sleep
from typing import Any, Dict, Optional
QUERY_DIR = 'queries'
RESULT_DIR = 'osm_data'
CFG_LAYERS = [
{
'query_filename': os.path.join(
QUERY_DIR,
'indoormark.overpassql'
),
'result_filename': os.path.join(RESULT_DIR, 'indoormark.geojson'),
},
{
'query_filename': os.path.join(
QUERY_DIR,
'traffic_signals.overpassql'
),
'result_filename': os.path.join(RESULT_DIR, 'traffic_signals.geojson'),
},
]
OVERPASS_API_URL = 'https://lz4.overpass-api.de/api/interpreter'
OVERPASS_RETRIES = 5
OVERPASS_TIMEOUT = 30 # seconds
def download_data(overpass_query_file: str) -> Optional[Dict[Any, Any]]:
with open(overpass_query_file, 'r') as f:
query = f.read().strip()
logging.info(f'Read overpass query from file: {overpass_query_file}')
logging.info(f'Downloading overpass data...')
for _ in range(OVERPASS_RETRIES):
try:
response = requests.get(OVERPASS_API_URL, params={'data': query})
if response.status_code != 200:
logging.warning(
f'Incorrect status code: {response.status_code}'
)
continue
return response.json()
except Exception as e:
logging.error(f'Error with downloading/parsing data: {e}')
sleep(OVERPASS_TIMEOUT)
def filter_data(overpass_data: Dict[str, Any]) -> Dict[str, Any]:
for element in overpass_data['elements']:
if 'tags' not in element:
continue
element['tags'].pop('phone', None)
element['tags'].pop('payment', None)
tags = element['tags']
for key in tags.keys():
if key.startswith('payment:'):
element['tags'].pop(key)
return overpass_data
def main():
logging.basicConfig(
format='%(asctime)s,%(msecs)d %(name)s %(levelname)s %(message)s',
datefmt='%Y-%m-%d,%H:%M:%S',
level=logging.INFO
)
for layer in CFG_LAYERS:
overpass_data = download_data(layer['query_filename'])
if overpass_data is None:
logging.info('Empty overpass data. Exiting!')
exit(1)
logging.info(f'Filtering data...')
overpass_data = filter_data(overpass_data)
logging.info(f'Parsing overpass data to geojson...')
geojson = json2geojson(overpass_data)
with open(layer['result_filename'], 'w') as f:
json.dump(geojson, f)
logging.info(f'Saved {layer["result_filename"]}.')
if __name__ == '__main__':
main()