Completed
Push — master ( 3a7275...213df2 )
by
unknown
01:12
created

AdminGrenzenClient.get_admin_objecten()   B

Complexity

Conditions 2

Size

Total Lines 26

Duplication

Lines 0
Ratio 0 %
Metric Value
cc 2
dl 0
loc 26
rs 8.8571
1
# -*- coding: utf8 -*-
2
from geoalchemy2 import WKTElement, shape
3
import json
4
import geojson
5
import re
6
from shapely import geometry
7
import requests
8
import os
9
from operator import attrgetter
10
11
12
def convert_wktelement_to_geojson(wktelement):
13
    """
14
    This function converts a wktelement (from GeoAlchemy) to a geojson
15
    :param wktelement: WKTElement
16
    :return: geojson
17
    """
18
    if wktelement is None:
19
        return None
20
    try:
21
        geom = shape.to_shape(wktelement)
22
        srid = wktelement.srid
23
        json_element = json.dumps(geometry.mapping(geom))
24
        geojson_element = geojson.loads(json_element)
25
        geojson_element['crs'] = {
26
            "type": "name",
27
            "properties": {
28
                "name": "urn:ogc:def:crs:EPSG::{0}".format(srid)
29
            }
30
        }
31
32
        return geojson_element
33
34
    except AssertionError:
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable AssertionError does not seem to be defined.
Loading history...
35
        raise AssertionError("WKTelement is niet geldig: %s" % wktelement)
36
37
38
def convert_geojson_to_wktelement(value):
39
    """
40
    This function converts a geojson to a wktelement that can be used by GeoAlchemy.
41
    :param value: geojson
42
    :return: WKTElement
43
    """
44
    if value is None or not value:
45
        return None
46
    try:
47
        s = json.dumps(value)
48
        g1 = geojson.loads(s)
49
        g2 = geometry.shape(g1)
50
        srid = get_srid_from_geojson(g1)
51
52
        return WKTElement(data=g2.wkt, srid=srid)
53
54
    except ValueError:
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable ValueError does not seem to be defined.
Loading history...
55
        raise ValueError("GeoJson is niet geldig: %s" % value)
56
57
58
def convert_geojson_to_geometry(value):
59
    """
60
    This function converts a geojson to a Shapely.geometry.
61
    :param value: geojson
62
    :return: Shapely.geometry
63
    """
64
    geojson_element = geojson.loads(json.dumps(value))
65
    return geometry.shape(geojson_element)
66
67
68
def get_srid_from_geojson(value):
69
    """
70
    This function returns the srid of a geojson
71
    :param value: geojson
72
    :return: spatial reference identifier (srid)
73
    """
74
    if value is None or not value:
75
        return None
76
    try:
77
        if not value.get('crs'):
78
            return None
79
        srid = find_srid(value['crs']['properties']['name'])
80
        # todo make a better implementation for this?! Is there a fixed format in geojson to define CRS
81
82
        return srid
83
84
    except ValueError:  # pragma NO COVER
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable ValueError does not seem to be defined.
Loading history...
85
        # In deze situatie enkel key en attributeErrors mogelijk
86
        raise ValueError("Geen geldige Spatial Reference Identifier (SRID) gevonden" % value)
87
88
89
def find_srid(text):
90
    li = [int(x) for x in re.findall(r'\d+', text) if len(x) in [4, 5]]
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable x does not seem to be defined.
Loading history...
91
    return max(li) if len(li) > 0 else None
92
93
94
def check_within_flanders(geojson_input):
95
    """
96
    :param geojson_input: geojson geometrie
97
    :return: True if the input geojson is within the Flanders region, otherwise False
98
    """
99
    if geometry.shape(geojson_input).within(get_flanders_geometry()):
100
        return True
101
    return False
102
103
104
def check_in_flanders(geojson_input):
105
    """
106
    :param geojson_input: geojson geometrie
107
    :return: True if the input geojson intersects with Flanders region, otherwise False
108
    """
109
    if geometry.shape(geojson_input).intersects(get_flanders_geometry()):
110
        return True
111
    return False
112
113
114
def get_flanders_geometry():
115
    here = os.path.abspath(os.path.dirname(__file__))
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable __file__ does not seem to be defined.
Loading history...
116
    with open(here + '/fixtures/vlaams_gewest.geojson') as geof:
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable here does not seem to be defined.
Loading history...
Comprehensibility Best Practice introduced by
The variable geof does not seem to be defined.
Loading history...
117
        flanders_features = json.load(geof)
118
    flanders_geojson = flanders_features['features'][0]['geometry']
119
    return geometry.shape(flanders_geojson)
120
121
122
def get_centroid_xy(value):
123
    """
124
    Returns the centroid x,y of a geojson polygon.
125
    If the geojson is a multipolygon. It return the centroid of the biggest polygon
126
    :param value: geojson
127
    :return: x,y
128
    """
129
    geometry_element = convert_geojson_to_geometry(value)
130
    if geometry_element.type == 'MultiPolygon':
131
        geometry_element_polygons = list(geometry_element.geoms)
132
        largest_polygon = max(geometry_element_polygons, key=attrgetter('area'))
133
        centroid = largest_polygon.centroid.wkt
134
    else:
135
        centroid = geometry_element.centroid.wkt
136
    return ','.join([x for x in re.findall(r'\d+.\d+', centroid)])
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable x does not seem to be defined.
Loading history...
137
138
139
def add_crab_ids(crabpy_gateway, address):
140
    gemeente = address.get("gemeente")
141
    if gemeente:
142
        gewest_ids = [2, 1, 3]
143
        gemeente_val = None
144
        for gewest_id in gewest_ids:
145
            gemeenten = crabpy_gateway.list_gemeenten(gewest_id)
146
            gemeente_val = next((g for g in gemeenten if g.naam.lower() == gemeente.lower()), None)
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable g does not seem to be defined.
Loading history...
147
            if gemeente_val:
148
                address['gemeente'] = "" + gemeente_val.naam
149
                address['gemeente_id'] = gemeente_val.id
150
                break
151
        straat_naam = address.get("straat")
152
        if gemeente_val and straat_naam:
153
            straat_val = next((s for s in gemeente_val.straten if s.label.lower() == straat_naam.lower()), None)
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable s does not seem to be defined.
Loading history...
154
            if straat_val:
155
                address["straat"] = "" + straat_val.label
156
                address["straat_id"] = straat_val.id
157
                huisnummer_naam = address.get("huisnummer")
158
                if huisnummer_naam:
159
                    huisnummer_val = next((n for n in straat_val.huisnummers
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable n does not seem to be defined.
Loading history...
160
                                           if n.huisnummer.lower() == huisnummer_naam.lower()), None)
161
                    if huisnummer_val:
162
                        address["huisnummer"] = "" + huisnummer_val.huisnummer
163
                        address["huisnummer_id"] = huisnummer_val.id
164
        return address
165
166
167
def nearest_location(geojson_element, crabpy_gateway=None):
168
    """
169
    Returns the nearest location. If a crabpy_gateway is given, the crab id's will be added to the address.
170
    Uses the agiv service https://loc.geopunt.be/geolocation/Location?xy=<x,y>
171
    where <x,y> are the coordinates of the centroid of the geometry (polygon).
172
    The agiv service returns crab, urbis en poi location types only in Flanders and Brussels.
173
    :param value: geojson
174
    :return: - crab address
175
             - None if the nearest address is nog crab type
176
    """
177
    xy = get_centroid_xy(geojson_element)
178
179
    r = requests.get('https://loc.geopunt.be/geolocation/Location?xy={0}'.format(xy))
180
    result = r.json()
181
    if 'LocationResult' not in result \
182
            or len(result['LocationResult']) == 0 \
183
            or 'crab' not in result['LocationResult'][0]['LocationType']:
184
        return None
185
    locres = result['LocationResult'][0]['FormattedAddress']
186
    straat_huisnummer, postcode_gemeente = locres.rsplit(', ', 1)
187
    straat, huisnummer = straat_huisnummer.rsplit(' ', 1)
188
    postcode, gemeente = postcode_gemeente.split(' ', 1)
189
    address = {
190
        'straat': straat,
191
        'huisnummer': huisnummer,
192
        'omschrijving_straat': straat + ', ' + huisnummer,
193
        'postcode': postcode,
194
        'gemeente': gemeente,
195
        'land': 'BE'
196
    }
197
    if crabpy_gateway:
198
        return add_crab_ids(crabpy_gateway, address)
199
    else:
200
        return address
201
202
203
class AdminGrenzenClient:
204
    def __init__(self, base_url):
205
        self.base_url = base_url
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable base_url does not seem to be defined.
Loading history...
206
207
    def get_admin_objecten(self, geojson_input, admin_grens_type, return_geometry=0):
208
        """
209
        This function returns the administrative areas objects which intersects with the input geojson
210
        :param geojson_input: geojson
211
        :param admin_grens_type: type of returned administrative objects
212
        :param return_geometry: boolean indicating whether the geometry of the returned objects should also be included
213
        :
214
        :return: list of administrative areas objects
215
        """
216
        if check_in_flanders(geojson_input):
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable geojson_input does not seem to be defined.
Loading history...
217
            params = {
218
                'geef_geometrie': return_geometry,
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable return_geometry does not seem to be defined.
Loading history...
219
                'type': admin_grens_type,
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable admin_grens_type does not seem to be defined.
Loading history...
220
                'geometrie': json.dumps(geojson_input)
221
            }
222
            res = requests.get(
223
                url=self.base_url,
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable self does not seem to be defined.
Loading history...
224
                params=params,
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable params does not seem to be defined.
Loading history...
225
                headers={
226
                    'Accept': 'application/json'
227
                }
228
            )
229
            res.raise_for_status()
230
            return json.loads(res.text)
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable res does not seem to be defined.
Loading history...
231
        else:
232
            return []
233
234
    def get_gemeenten(self, geojson_input):
235
        """
236
        This function returns the names of the municipalities which intersects with the input geojson
237
        :param geojson_input: geojson
238
        :return: list of municipalities
239
        """
240
        results = self.get_admin_objecten(geojson_input, 'gemeente')
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable geojson_input does not seem to be defined.
Loading history...
241
        return [gemeente['naam'] for gemeente in results]
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable gemeente does not seem to be defined.
Loading history...
Comprehensibility Best Practice introduced by
The variable results does not seem to be defined.
Loading history...
242
243
    def get_provincies(self, geojson_input):
244
        """
245
        This function returns the names of the provinces which intersects with the input geojson
246
        :param geojson_input: geojson
247
        :return: list of municipalities
248
        """
249
        results = self.get_admin_objecten(geojson_input, 'provincie')
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable geojson_input does not seem to be defined.
Loading history...
250
        return [provincie['naam'] for provincie in results]
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable results does not seem to be defined.
Loading history...
Comprehensibility Best Practice introduced by
The variable provincie does not seem to be defined.
Loading history...
251
252
    @staticmethod
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable staticmethod does not seem to be defined.
Loading history...
253
    def _get_lagest_admin_obj(results, geojson_input):
254
        if len(results) == 0:
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable results does not seem to be defined.
Loading history...
255
            return None
256
        elif len(results) == 1:
257
            return {'niscode': results[0]['id'], 'naam': results[0]['naam']}
258
        else:
259
            input_polygon = convert_geojson_to_geometry(geojson_input)
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable geojson_input does not seem to be defined.
Loading history...
260
            for result in results:
261
                admin_geom = convert_geojson_to_geometry(result['geometrie'])
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable result does not seem to be defined.
Loading history...
262
                result['intersection_area'] = admin_geom.intersection(input_polygon).area
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable input_polygon does not seem to be defined.
Loading history...
263
            largest_m = max(results, key=lambda x: x['intersection_area'])
264
            return {'niscode': largest_m['id'], 'naam': largest_m['naam']}
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable largest_m does not seem to be defined.
Loading history...
265
266
    def get_gemeente(self, geojson_input):
267
        """
268
        This function returns the name of the municipality which intersects with the input geojson.
269
        If more municipalities intersect. The municipality is returned with the largest intersection
270
        :param value: administratievegrenzen_url
271
        :param value: geojson
272
        :return: municipality
273
        """
274
        results = self.get_admin_objecten(geojson_input, 'gemeente', 1)
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable geojson_input does not seem to be defined.
Loading history...
275
        return self._get_lagest_admin_obj(results, geojson_input)
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable results does not seem to be defined.
Loading history...
276
277
    def get_provincie(self, geojson_input):
278
        """
279
        This function returns the name of the province which intersects with the input geojson.
280
        If more provinces intersect. The province is returned with the largest intersection.
281
        :param value: administratievegrenzen_url
282
        :param value: geojson
283
        :return: province
284
        """
285
        results = self.get_admin_objecten(geojson_input, 'provincie', 1)
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable geojson_input does not seem to be defined.
Loading history...
286
        return self._get_lagest_admin_obj(results, geojson_input)
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable results does not seem to be defined.
Loading history...
287