|
1
|
|
|
# -*- coding: utf-8 -*- |
|
|
|
|
|
|
2
|
|
|
try: |
|
3
|
|
|
import logging |
|
4
|
|
|
import sys |
|
5
|
|
|
import sqlalchemy |
|
6
|
|
|
import geopandas as gpd |
|
|
|
|
|
|
7
|
|
|
import datetime |
|
8
|
|
|
from OSMPythonTools.overpass import Overpass |
|
9
|
|
|
from OSMPythonTools.nominatim import Nominatim |
|
10
|
|
|
from OSMPythonTools.overpass import overpassQueryBuilder |
|
11
|
|
|
except ImportError as err: |
|
12
|
|
|
logging.error('Error %s import module: %s', __name__, err) |
|
13
|
|
|
logging.exception('Exception occurred') |
|
14
|
|
|
|
|
15
|
|
|
sys.exit(128) |
|
16
|
|
|
|
|
17
|
|
|
|
|
18
|
|
|
def get_area_id(area): |
|
|
|
|
|
|
19
|
|
|
# Query Nominatom |
|
20
|
|
|
nominatim = Nominatim() |
|
21
|
|
|
return nominatim.query(area).areaId() |
|
22
|
|
|
|
|
23
|
|
|
|
|
24
|
|
|
def query_overpass(area_id, query_statement, element_type='node'): |
|
|
|
|
|
|
25
|
|
|
# Query Overpass based on area |
|
26
|
|
|
overpass = Overpass() |
|
27
|
|
|
query = overpassQueryBuilder(area=area_id, elementType=element_type, selector=query_statement) |
|
28
|
|
|
return overpass.query(query) |
|
29
|
|
|
|
|
30
|
|
|
|
|
31
|
|
|
def query_osm_postcode_gpd(session, lon, lat): |
|
|
|
|
|
|
32
|
|
|
if lat is None or lat == '' or lon == '' or lon is None: |
|
33
|
|
|
return None |
|
34
|
|
|
query = sqlalchemy.text(''' |
|
35
|
|
|
SELECT name |
|
36
|
|
|
FROM planet_osm_polygon, (SELECT ST_SetSRID(ST_MakePoint(:lon, :lat),4326) as geom) point |
|
37
|
|
|
WHERE boundary='postal_code' and ST_Contains(way, point.geom) ORDER BY name LIMIT 1;''') |
|
38
|
|
|
data = session.execute(query, {'lon': lon, 'lat': lat}).first() |
|
39
|
|
|
if data is None: |
|
40
|
|
|
return None |
|
41
|
|
|
row = dict(zip(data.keys(), data)) |
|
42
|
|
|
return int(row['name'].split(' ')[0]) if row['name'].split(' ')[0] is not None else None |
|
43
|
|
|
|
|
44
|
|
|
|
|
45
|
|
|
def query_postcode_osm_external(prefer_osm, session, lon, lat, postcode_ext): |
|
|
|
|
|
|
46
|
|
|
if prefer_osm is False and postcode_ext is not None: |
|
47
|
|
|
return postcode_ext |
|
48
|
|
|
query_postcode = query_osm_postcode_gpd(session, lon, lat) |
|
49
|
|
|
if prefer_osm is True and query_postcode is not None: |
|
|
|
|
|
|
50
|
|
|
return query_postcode |
|
51
|
|
|
elif prefer_osm is True and query_postcode is None: |
|
52
|
|
|
return postcode_ext |
|
53
|
|
|
|
|
54
|
|
|
|
|
55
|
|
|
def relationer(relation_text): |
|
|
|
|
|
|
56
|
|
|
if relation_text is None: |
|
57
|
|
|
return None |
|
58
|
|
|
data = [] |
|
59
|
|
|
for i in range(0, len(relation_text), 2): |
|
60
|
|
|
item = relation_text[i] |
|
61
|
|
|
if item[0] == 'n': |
|
62
|
|
|
tp = 'node' |
|
|
|
|
|
|
63
|
|
|
elif item[0] == 'w': |
|
64
|
|
|
tp = 'way' |
|
|
|
|
|
|
65
|
|
|
elif item[0] == 'r': |
|
66
|
|
|
tp = 'relation' |
|
|
|
|
|
|
67
|
|
|
rf = item[1:] |
|
|
|
|
|
|
68
|
|
|
rl = relation_text[i + 1] |
|
|
|
|
|
|
69
|
|
|
data.append({'type': tp, 'ref': rf, 'role': rl}) |
|
|
|
|
|
|
70
|
|
|
return data |
|
71
|
|
|
|
|
72
|
|
|
|
|
73
|
|
|
def timestamp_now(): |
|
|
|
|
|
|
74
|
|
|
return datetime.datetime.now() |
|
75
|
|
|
|
|
76
|
|
|
|
|
77
|
|
|
def osm_timestamp_now(): |
|
|
|
|
|
|
78
|
|
|
return '{:{dfmt}T{tfmt}Z}'.format(datetime.datetime.now(), dfmt='%Y-%m-%d', tfmt='%H:%M:%S') |
|
79
|
|
|
|
|
80
|
|
|
|
|
81
|
|
|
def query_osm_city_name_gpd(session, lon, lat): |
|
|
|
|
|
|
82
|
|
|
if lat is None or lat == '' or lon == '' or lon is None: |
|
83
|
|
|
return None |
|
84
|
|
|
query = sqlalchemy.text(''' |
|
85
|
|
|
SELECT name |
|
86
|
|
|
FROM planet_osm_polygon, (SELECT ST_SetSRID(ST_MakePoint(:lat,:lon),4326) as geom) point |
|
87
|
|
|
WHERE admin_level='8' and ST_Contains(way, point.geom) ORDER BY name LIMIT 1;''') |
|
88
|
|
|
data = session.execute(query, {'lon': lon, 'lat': lat}).first() |
|
89
|
|
|
if data is None: |
|
|
|
|
|
|
90
|
|
|
return None |
|
91
|
|
|
else: |
|
92
|
|
|
return data[0] |
|
93
|
|
|
|
|
94
|
|
|
|
|
95
|
|
|
def query_osm_city_name(session, name): |
|
|
|
|
|
|
96
|
|
|
query = sqlalchemy.text(''' |
|
97
|
|
|
SELECT name |
|
98
|
|
|
FROM planet_osm_polygon WHERE admin_level='8' and name=:name ORDER BY name LIMIT 1;''') |
|
99
|
|
|
data = session.execute(query, {'name': name}).first() |
|
100
|
|
|
if data is None: |
|
|
|
|
|
|
101
|
|
|
return None |
|
102
|
|
|
else: |
|
103
|
|
|
return data[0] |
|
104
|
|
|
|