Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Retrieve all cemeteries in a given region (here states of Austria) with their municipality
import itertools
import logging
import json
import time
import overpy
api = overpy.Overpass()
RELATIONS_QUERY = """
[out:json][timeout:180];
area[wikidata="{wikidata}"]->.boundaryarea;
(
relation[{query}](area.boundaryarea);
);
foreach(
out center;
way(r:"outer") -> .outer;
node(w.outer)->.n;
.n is_in->.a;
area.a[name][boundary=administrative][admin_level="8"] -> .a;
convert way ::=::,
::id = id(),
admin_level_8=a.set(t[name]),
wikidata_item=a.set(t[wikidata]);
out body;
);
"""
WAYS_QUERY = """
[out:json][timeout:180];
area[wikidata="{wikidata}"]->.boundaryarea;
(
way[{query}](area.boundaryarea);
);
foreach(
out center;
node(w)->.n;
.n is_in->.a;
area.a[name][boundary=administrative][admin_level="8"] -> .a;
convert relation ::=::,
::id = id(),
admin_level_8=a.set(t[name]),
wikidata_item=a.set(t[wikidata]);
out body;
);
"""
NODES_QUERY = """
[out:json][timeout:180];
area[wikidata="{wikidata}"]->.boundaryarea;
(
node[{query}](area.boundaryarea);
);
out body;
is_in->.a;
area.a[name][boundary=administrative][admin_level="8"] -> .a;
convert relation ::=::,
::id = id(),
admin_level_8=a.set(t[name]),
wikidata_item=a.set(t[wikidata]);
out body;
"""
def extract_data(element, fake_element):
poi = {}
poi['name'] = element.tags.get('name', '')
poi['lat'] = element.center_lat.to_eng_string()
poi['lon'] = element.center_lon.to_eng_string()
poi['municipality'] = fake_element.tags.get('admin_level_8', '')
poi['municipality_wikidata'] = fake_element.tags.get('wikidata_item', '')
return poi
def query_relations(region, query_element):
logging.info("Getting all relations %s in region %s", query_element, region)
pois = []
overpass_query = RELATIONS_QUERY.format(wikidata=region, query=query_element)
result = api.query(overpass_query)
for element, fake_element in itertools.izip(result.relations, result.ways):
pois.append(extract_data(element, fake_element))
return pois
def query_ways(region, query_element):
logging.info("Getting all ways %s in region %s", query_element, region)
pois = []
overpass_query = WAYS_QUERY.format(wikidata=region, query=query_element)
result = api.query(overpass_query)
for element, fake_element in itertools.izip(result.ways, result.relations):
pois.append(extract_data(element, fake_element))
return pois
def get_elements(region, query_element):
logging.info("Getting all elements %s in region %s", query_element, region)
elements = []
elements += query_ways(region, query_element)
cool_off(60)
elements += query_relations(region, query_element)
return elements
def cool_off(seconds):
logging.info("Sleeping for %s seconds" % seconds)
time.sleep(seconds)
def main():
regions = [
"Q1741", # Vienna
"Q37985", # Carinthia
"Q38981", # Vorarlberg
"Q41358", # Styria
"Q41967", # Upper Austria
"Q42497", # Lower Austria
"Q42880", # Tyrol
"Q43210", # Burgenland
"Q43325", # Salzbourg
]
for region in regions:
cemeteries = get_elements(region, '"landuse"="cemetery"')
with open("Cemeteries-%s.json" % region, 'w') as fp:
json.dump(cemeteries, fp, indent=4, separators=(',', ': '))
cool_off(300)
graveyards = get_elements(region, '"amenity"="grave_yard"')
with open("Graveyards-%s.json" % region, 'w') as fp:
json.dump(graveyards, fp, indent=4, separators=(',', ': '))
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO)
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.