AdminGrenzenClient.check_erfgoedgemeenten() - Code Metrics - Inspection of "prepare release 0.4.0" - OnroerendErfgoed/oe_geoutils - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( b7e6d1...4d8241 )

unknown

created 2016-09-13 14:09 UTC

AdminGrenzenClient.check_erfgoedgemeenten() B

↳ Parent: AdminGrenzenClient

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	5
c	1
b	0
f	0
dl	0
loc	23
rs	8.2508

# -*- coding: utf8 -*-
from geoalchemy2 import WKTElement, shape
import json
import geojson
import re
from shapely import geometry
import requests
import os
from operator import attrgetter


def convert_wktelement_to_geojson(wktelement):
    """
    This function converts a wktelement (from GeoAlchemy) to a geojson
    :param wktelement: WKTElement
    :return: geojson
    """
    if wktelement is None:
        return None
    try:
        geom = shape.to_shape(wktelement)
        srid = wktelement.srid
        json_element = json.dumps(geometry.mapping(geom))
        geojson_element = geojson.loads(json_element)
        geojson_element['crs'] = {
            "type": "name",
            "properties": {
                "name": "urn:ogc:def:crs:EPSG::{0}".format(srid)
            }
        }

        return geojson_element

    except AssertionError:
        raise AssertionError("WKTelement is niet geldig: %s" % wktelement)


def convert_geojson_to_wktelement(value):
    """
    This function converts a geojson to a wktelement that can be used by GeoAlchemy.
    :param value: geojson
    :return: WKTElement
    """
    if value is None or not value:
        return None
    try:
        s = json.dumps(value)
        g1 = geojson.loads(s)
        g2 = geometry.shape(g1)
        srid = get_srid_from_geojson(g1)

        return WKTElement(data=g2.wkt, srid=srid)

    except ValueError:
        raise ValueError("GeoJson is niet geldig: %s" % value)


def convert_geojson_to_geometry(value):
    """
    This function converts a geojson to a Shapely.geometry.
    :param value: geojson
    :return: Shapely.geometry
    """
    geojson_element = geojson.loads(json.dumps(value))
    return geometry.shape(geojson_element)


def get_srid_from_geojson(value):
    """
    This function returns the srid of a geojson
    :param value: geojson
    :return: spatial reference identifier (srid)
    """
    if value is None or not value:
        return None
    try:
        if not value.get('crs'):
            return None
        srid = find_srid(value['crs']['properties']['name'])
        # todo make a better implementation for this?! Is there a fixed format in geojson to define CRS

        return srid

    except ValueError:  # pragma NO COVER
        # In deze situatie enkel key en attributeErrors mogelijk
        raise ValueError("Geen geldige Spatial Reference Identifier (SRID) gevonden" % value)


def find_srid(text):
    li = [int(x) for x in re.findall(r'\d+', text) if len(x) in [4, 5]]
    return max(li) if len(li) > 0 else None


def remove_dupl_values(list_coord):
    dupl_coord_indexes = [i for i in range(0, len(list_coord)) if i != 0 and
                          list_coord[i] == list_coord[i - 1]]
    [list_coord.pop(i) for i in sorted(dupl_coord_indexes, reverse=True)]


def remove_dupl_coords(coords, tree_types=(list, tuple)):
    """
    This function will remove duplicate consecutive coordinates
    :param coords: input coordinates
    """
    for o in coords:
        if isinstance(o[0][0], tree_types):
            remove_dupl_coords(o, tree_types)
        else:
            remove_dupl_values(o)


def check_within_flanders(geojson_input):
    """
    :param geojson_input: geojson geometrie
    :return: True if the input geojson is within the Flanders region, otherwise False
    """
    if geometry.shape(geojson_input).within(get_flanders_geometry()):
        return True
    return False


def check_in_flanders(geojson_input):
    """
    :param geojson_input: geojson geometrie
    :return: True if the input geojson intersects with Flanders region, otherwise False
    """
    if geometry.shape(geojson_input).intersects(get_flanders_geometry()):
        return True
    return False


def get_flanders_geometry():
    here = os.path.abspath(os.path.dirname(__file__))
    with open(here + '/fixtures/vlaams_gewest.geojson') as geof:
        flanders_features = json.load(geof)
    flanders_geojson = flanders_features['features'][0]['geometry']
    return geometry.shape(flanders_geojson)


def get_centroid_xy(value):
    """
    Returns the centroid x,y of a geojson polygon.
    If the geojson is a multipolygon. It return the centroid of the biggest polygon
    :param value: geojson
    :return: x,y
    """
    geometry_element = convert_geojson_to_geometry(value)
    if geometry_element.type == 'MultiPolygon':
        geometry_element_polygons = list(geometry_element.geoms)
        largest_polygon = max(geometry_element_polygons, key=attrgetter('area'))
        centroid = largest_polygon.centroid.wkt
    else:
        centroid = geometry_element.centroid.wkt
    return ','.join([x for x in re.findall(r'\d+.\d+', centroid)])


def add_crab_ids(crabpy_gateway, address):
    gemeente = address.get("gemeente")
    if gemeente:
        gewest_ids = [2, 1, 3]
        gemeente_val = None
        for gewest_id in gewest_ids:
            gemeenten = crabpy_gateway.list_gemeenten(gewest_id)
            gemeente_val = next((g for g in gemeenten if g.naam.lower() == gemeente.lower()), None)
            if gemeente_val:
                address['gemeente'] = "" + gemeente_val.naam
                address['gemeente_id'] = gemeente_val.id
                break
        straat_naam = address.get("straat")
        if gemeente_val and straat_naam:
            straat_val = next((s for s in gemeente_val.straten if s.label.lower() == straat_naam.lower()), None)
            if straat_val:
                address["straat"] = "" + straat_val.label
                address["straat_id"] = straat_val.id
                huisnummer_naam = address.get("huisnummer")
                if huisnummer_naam:
                    huisnummer_val = next((n for n in straat_val.huisnummers
                                           if n.huisnummer.lower() == huisnummer_naam.lower()), None)
                    if huisnummer_val:
                        address["huisnummer"] = "" + huisnummer_val.huisnummer
                        address["huisnummer_id"] = huisnummer_val.id
        return address


def nearest_location(geojson_element, crabpy_gateway=None):
    """
    Returns the nearest location. If a crabpy_gateway is given, the crab id's will be added to the address.
    Uses the agiv service https://loc.geopunt.be/geolocation/Location?xy=<x,y>
    where <x,y> are the coordinates of the centroid of the geometry (polygon).
    The agiv service returns crab, urbis en poi location types only in Flanders and Brussels.
    :param value: geojson
    :return: - crab address
             - None if the nearest address is nog crab type
    """
    xy = get_centroid_xy(geojson_element)

    r = requests.get('https://loc.geopunt.be/geolocation/Location?xy={0}'.format(xy))
    result = r.json()
    if 'LocationResult' not in result \
            or len(result['LocationResult']) == 0 \
            or 'crab' not in result['LocationResult'][0]['LocationType']:
        return None
    locres = result['LocationResult'][0]['FormattedAddress']
    straat_huisnummer, postcode_gemeente = locres.rsplit(', ', 1)
    straat, huisnummer = straat_huisnummer.rsplit(' ', 1)
    postcode, gemeente = postcode_gemeente.split(' ', 1)
    address = {
        'straat': straat,
        'huisnummer': huisnummer,
        'omschrijving_straat': straat + ', ' + huisnummer,
        'postcode': postcode,
        'gemeente': gemeente,
        'land': 'BE'
    }
    if crabpy_gateway:
        return add_crab_ids(crabpy_gateway, address)
    else:
        return address


class AdminGrenzenClient:
    def __init__(self, base_url):
        self.base_url = base_url

    def get_admin_objecten(self, geojson_input, admin_grens_type, return_geometry=0):
        """
        This function returns the administrative areas objects which intersects with the input geojson
        :param geojson_input: geojson
        :param admin_grens_type: type of returned administrative objects
        :param return_geometry: boolean indicating whether the geometry of the returned objects should also be included
        :
        :return: list of administrative areas objects
        """
        if check_in_flanders(geojson_input):
            data = {
                'geef_geometrie': return_geometry,
                'type': admin_grens_type,
                'geometrie': geojson_input
            }
            res = requests.post(
                url=self.base_url,
                json=data,
                headers={
                    'Accept': 'application/json',
                    'Content-Type': 'application/json'
                }
            )
            res.raise_for_status()
            return json.loads(res.text)
        else:
            return []

    def get_gemeenten(self, geojson_input):
        """
        This function returns the names of the municipalities which intersects with the input geojson
        :param geojson_input: geojson
        :return: list of municipalities
        """
        results = self.get_admin_objecten(geojson_input, 'gemeente')
        return [gemeente['naam'] for gemeente in results]

    def get_provincies(self, geojson_input):
        """
        This function returns the names of the provinces which intersects with the input geojson
        :param geojson_input: geojson
        :return: list of municipalities
        """
        results = self.get_admin_objecten(geojson_input, 'provincie')
        return [provincie['naam'] for provincie in results]

    @staticmethod
    def _get_lagest_admin_obj(results, geojson_input):
        if len(results) == 0:
            return None
        elif len(results) == 1:
            return {'niscode': results[0]['id'], 'naam': results[0]['naam']}
        else:
            input_polygon = convert_geojson_to_geometry(geojson_input)
            for result in results:
                admin_geom = convert_geojson_to_geometry(result['geometrie'])
                result['intersection_area'] = admin_geom.intersection(input_polygon).area
            largest_m = max(results, key=lambda x: x['intersection_area'])
            return {'niscode': largest_m['id'], 'naam': largest_m['naam']}

    def get_gemeente(self, geojson_input):
        """
        This function returns the name of the municipality which intersects with the input geojson.
        If more municipalities intersect. The municipality is returned with the largest intersection
        :param value: administratievegrenzen_url
        :param value: geojson
        :return: municipality
        """
        results = self.get_admin_objecten(geojson_input, 'gemeente', 1)
        return self._get_lagest_admin_obj(results, geojson_input)

    def get_provincie(self, geojson_input):
        """
        This function returns the name of the province which intersects with the input geojson.
        If more provinces intersect. The province is returned with the largest intersection.
        :param value: administratievegrenzen_url
        :param value: geojson
        :return: province
        """
        results = self.get_admin_objecten(geojson_input, 'provincie', 1)
        return self._get_lagest_admin_obj(results, geojson_input)

    def check_erfgoedgemeenten(self, geojson_input, erfgoedgemeenten_list):
        """
        :param geojson_input: geojson geometry
        :param erfgoedgemeenten_list: list of niscodes of municipalities
        :return: dict including key status and optional key message
                 status "ok" when the geojson does not intersect with one of given municipalities
                 status "warn" when the geojson intersects partially with one of given municipalities
                 status "error" when the geojson is within one of given municipalities
        """
        gemeenten = self.get_admin_objecten(geojson_input, 'gemeente')
        erfgoedgemeenten = [gemeente for gemeente in gemeenten if int(gemeente['id']) in json.loads(erfgoedgemeenten_list)]
        if len(erfgoedgemeenten) > 0:
            if len(gemeenten) == len(erfgoedgemeenten):
                return {
                    "status": "error",
                    "message": "Let op, de zone van deze melding is gelegen in een onroerenderfgoedgemeente en "
                               "kan niet bewaard worden. Gelieve de melding in te dienen bij deze gemeente."}
            else:
                return {
                    "status": "warn",
                    "message": "Let op, deze melding ligt deels in een onroerenderfgoedgemeente . "
                               "Gelieve de melding vooronderzoek eveneens in te dienen bij deze gemeente."}
        return {"status": "ok"}


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:
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:
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
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]]
91			return max(li) if len(li) > 0 else None
92
93
94			def remove_dupl_values(list_coord):
95			dupl_coord_indexes = [i for i in range(0, len(list_coord)) if i != 0 and
96			list_coord[i] == list_coord[i - 1]]
97			[list_coord.pop(i) for i in sorted(dupl_coord_indexes, reverse=True)]
98
99
100			def remove_dupl_coords(coords, tree_types=(list, tuple)):
101			"""
102			This function will remove duplicate consecutive coordinates
103			:param coords: input coordinates
104			"""
105			for o in coords:
106			if isinstance(o[0][0], tree_types):
107			remove_dupl_coords(o, tree_types)
108			else:
109			remove_dupl_values(o)
110
111
112			def check_within_flanders(geojson_input):
113			"""
114			:param geojson_input: geojson geometrie
115			:return: True if the input geojson is within the Flanders region, otherwise False
116			"""
117			if geometry.shape(geojson_input).within(get_flanders_geometry()):
118			return True
119			return False
120
121
122			def check_in_flanders(geojson_input):
123			"""
124			:param geojson_input: geojson geometrie
125			:return: True if the input geojson intersects with Flanders region, otherwise False
126			"""
127			if geometry.shape(geojson_input).intersects(get_flanders_geometry()):
128			return True
129			return False
130
131
132			def get_flanders_geometry():
133			here = os.path.abspath(os.path.dirname(__file__))
134			with open(here + '/fixtures/vlaams_gewest.geojson') as geof:
135			flanders_features = json.load(geof)
136			flanders_geojson = flanders_features['features'][0]['geometry']
137			return geometry.shape(flanders_geojson)
138
139
140			def get_centroid_xy(value):
141			"""
142			Returns the centroid x,y of a geojson polygon.
143			If the geojson is a multipolygon. It return the centroid of the biggest polygon
144			:param value: geojson
145			:return: x,y
146			"""
147			geometry_element = convert_geojson_to_geometry(value)
148			if geometry_element.type == 'MultiPolygon':
149			geometry_element_polygons = list(geometry_element.geoms)
150			largest_polygon = max(geometry_element_polygons, key=attrgetter('area'))
151			centroid = largest_polygon.centroid.wkt
152			else:
153			centroid = geometry_element.centroid.wkt
154			return ','.join([x for x in re.findall(r'\d+.\d+', centroid)])
155
156
157			def add_crab_ids(crabpy_gateway, address):
158			gemeente = address.get("gemeente")
159			if gemeente:
160			gewest_ids = [2, 1, 3]
161			gemeente_val = None
162			for gewest_id in gewest_ids:
163			gemeenten = crabpy_gateway.list_gemeenten(gewest_id)
164			gemeente_val = next((g for g in gemeenten if g.naam.lower() == gemeente.lower()), None)
165			if gemeente_val:
166			address['gemeente'] = "" + gemeente_val.naam
167			address['gemeente_id'] = gemeente_val.id
168			break
169			straat_naam = address.get("straat")
170			if gemeente_val and straat_naam:
171			straat_val = next((s for s in gemeente_val.straten if s.label.lower() == straat_naam.lower()), None)
172			if straat_val:
173			address["straat"] = "" + straat_val.label
174			address["straat_id"] = straat_val.id
175			huisnummer_naam = address.get("huisnummer")
176			if huisnummer_naam:
177			huisnummer_val = next((n for n in straat_val.huisnummers
178			if n.huisnummer.lower() == huisnummer_naam.lower()), None)
179			if huisnummer_val:
180			address["huisnummer"] = "" + huisnummer_val.huisnummer
181			address["huisnummer_id"] = huisnummer_val.id
182			return address
183
184
185			def nearest_location(geojson_element, crabpy_gateway=None):
186			"""
187			Returns the nearest location. If a crabpy_gateway is given, the crab id's will be added to the address.
188			Uses the agiv service https://loc.geopunt.be/geolocation/Location?xy=<x,y>
189			where <x,y> are the coordinates of the centroid of the geometry (polygon).
190			The agiv service returns crab, urbis en poi location types only in Flanders and Brussels.
191			:param value: geojson
192			:return: - crab address
193			- None if the nearest address is nog crab type
194			"""
195			xy = get_centroid_xy(geojson_element)
196
197			r = requests.get('https://loc.geopunt.be/geolocation/Location?xy={0}'.format(xy))
198			result = r.json()
199			if 'LocationResult' not in result \
200			or len(result['LocationResult']) == 0 \
201			or 'crab' not in result['LocationResult'][0]['LocationType']:
202			return None
203			locres = result['LocationResult'][0]['FormattedAddress']
204			straat_huisnummer, postcode_gemeente = locres.rsplit(', ', 1)
205			straat, huisnummer = straat_huisnummer.rsplit(' ', 1)
206			postcode, gemeente = postcode_gemeente.split(' ', 1)
207			address = {
208			'straat': straat,
209			'huisnummer': huisnummer,
210			'omschrijving_straat': straat + ', ' + huisnummer,
211			'postcode': postcode,
212			'gemeente': gemeente,
213			'land': 'BE'
214			}
215			if crabpy_gateway:
216			return add_crab_ids(crabpy_gateway, address)
217			else:
218			return address
219
220
221			class AdminGrenzenClient:
222			def __init__(self, base_url):
223			self.base_url = base_url
224
225			def get_admin_objecten(self, geojson_input, admin_grens_type, return_geometry=0):
226			"""
227			This function returns the administrative areas objects which intersects with the input geojson
228			:param geojson_input: geojson
229			:param admin_grens_type: type of returned administrative objects
230			:param return_geometry: boolean indicating whether the geometry of the returned objects should also be included
231			:
232			:return: list of administrative areas objects
233			"""
234			if check_in_flanders(geojson_input):
235			data = {
236			'geef_geometrie': return_geometry,
237			'type': admin_grens_type,
238			'geometrie': geojson_input
239			}
240			res = requests.post(
241			url=self.base_url,
242			json=data,
243			headers={
244			'Accept': 'application/json',
245			'Content-Type': 'application/json'
246			}
247			)
248			res.raise_for_status()
249			return json.loads(res.text)
250			else:
251			return []
252
253			def get_gemeenten(self, geojson_input):
254			"""
255			This function returns the names of the municipalities which intersects with the input geojson
256			:param geojson_input: geojson
257			:return: list of municipalities
258			"""
259			results = self.get_admin_objecten(geojson_input, 'gemeente')
260			return [gemeente['naam'] for gemeente in results]
261
262			def get_provincies(self, geojson_input):
263			"""
264			This function returns the names of the provinces which intersects with the input geojson
265			:param geojson_input: geojson
266			:return: list of municipalities
267			"""
268			results = self.get_admin_objecten(geojson_input, 'provincie')
269			return [provincie['naam'] for provincie in results]
270
271			@staticmethod
272			def _get_lagest_admin_obj(results, geojson_input):
273			if len(results) == 0:
274			return None
275			elif len(results) == 1:
276			return {'niscode': results[0]['id'], 'naam': results[0]['naam']}
277			else:
278			input_polygon = convert_geojson_to_geometry(geojson_input)
279			for result in results:
280			admin_geom = convert_geojson_to_geometry(result['geometrie'])
281			result['intersection_area'] = admin_geom.intersection(input_polygon).area
282			largest_m = max(results, key=lambda x: x['intersection_area'])
283			return {'niscode': largest_m['id'], 'naam': largest_m['naam']}
284
285			def get_gemeente(self, geojson_input):
286			"""
287			This function returns the name of the municipality which intersects with the input geojson.
288			If more municipalities intersect. The municipality is returned with the largest intersection
289			:param value: administratievegrenzen_url
290			:param value: geojson
291			:return: municipality
292			"""
293			results = self.get_admin_objecten(geojson_input, 'gemeente', 1)
294			return self._get_lagest_admin_obj(results, geojson_input)
295
296			def get_provincie(self, geojson_input):
297			"""
298			This function returns the name of the province which intersects with the input geojson.
299			If more provinces intersect. The province is returned with the largest intersection.
300			:param value: administratievegrenzen_url
301			:param value: geojson
302			:return: province
303			"""
304			results = self.get_admin_objecten(geojson_input, 'provincie', 1)
305			return self._get_lagest_admin_obj(results, geojson_input)
306
307			def check_erfgoedgemeenten(self, geojson_input, erfgoedgemeenten_list):
308			"""
309			:param geojson_input: geojson geometry
310			:param erfgoedgemeenten_list: list of niscodes of municipalities
311			:return: dict including key status and optional key message
312			status "ok" when the geojson does not intersect with one of given municipalities
313			status "warn" when the geojson intersects partially with one of given municipalities
314			status "error" when the geojson is within one of given municipalities
315			"""
316			gemeenten = self.get_admin_objecten(geojson_input, 'gemeente')
317			erfgoedgemeenten = [gemeente for gemeente in gemeenten if int(gemeente['id']) in json.loads(erfgoedgemeenten_list)]
318			if len(erfgoedgemeenten) > 0:
319			if len(gemeenten) == len(erfgoedgemeenten):
320			return {
321			"status": "error",
322			"message": "Let op, de zone van deze melding is gelegen in een onroerenderfgoedgemeente en "
323			"kan niet bewaard worden. Gelieve de melding in te dienen bij deze gemeente."}
324			else:
325			return {
326			"status": "warn",
327			"message": "Let op, deze melding ligt deels in een onroerenderfgoedgemeente . "
328			"Gelieve de melding vooronderzoek eveneens in te dienen bij deze gemeente."}
329			return {"status": "ok"}
330

OnroerendErfgoed / oe_geoutils

Push — master ( b7e6d1...4d8241 )

AdminGrenzenClient.check_erfgoedgemeenten() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like