provincie_niscode() - Code Metrics - Inspection of "Merge pull request #13 from OnroerendErfgoed/DEV_0..." - OnroerendErfgoed/oe_geoutils - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 180df1...ad4d03 )

by Bart

created 2016-12-08 15:53 UTC

provincie_niscode() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
dl	0
loc	12
rs	9.4285
c	0
b	0
f	0

# -*- 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 provincie_niscode(gemeente_niscode):
    """
    geef de niscode van de provincie op basis van de niscode van de gemeente

    :param gemeente_niscode: niscode van de gemeente
    :return: niscode van de provincie
    """
    pcode = str(gemeente_niscode)[0]
    if pcode == '2':
        return int('20001')
    else:
        return int('{0}0000'.format(pcode))


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 provincie_niscode(gemeente_niscode):
141			"""
142			geef de niscode van de provincie op basis van de niscode van de gemeente
143
144			:param gemeente_niscode: niscode van de gemeente
145			:return: niscode van de provincie
146			"""
147			pcode = str(gemeente_niscode)[0]
148			if pcode == '2':
149			return int('20001')
150			else:
151			return int('{0}0000'.format(pcode))
152
153
154			def get_centroid_xy(value):
155			"""
156			Returns the centroid x,y of a geojson polygon.
157			If the geojson is a multipolygon. It return the centroid of the biggest polygon
158			:param value: geojson
159			:return: x,y
160			"""
161			geometry_element = convert_geojson_to_geometry(value)
162			if geometry_element.type == 'MultiPolygon':
163			geometry_element_polygons = list(geometry_element.geoms)
164			largest_polygon = max(geometry_element_polygons, key=attrgetter('area'))
165			centroid = largest_polygon.centroid.wkt
166			else:
167			centroid = geometry_element.centroid.wkt
168			return ','.join([x for x in re.findall(r'\d+.\d+', centroid)])
169
170
171			def add_crab_ids(crabpy_gateway, address):
172			gemeente = address.get("gemeente")
173			if gemeente:
174			gewest_ids = [2, 1, 3]
175			gemeente_val = None
176			for gewest_id in gewest_ids:
177			gemeenten = crabpy_gateway.list_gemeenten(gewest_id)
178			gemeente_val = next((g for g in gemeenten if g.naam.lower() == gemeente.lower()), None)
179			if gemeente_val:
180			address['gemeente'] = "" + gemeente_val.naam
181			address['gemeente_id'] = gemeente_val.id
182			break
183			straat_naam = address.get("straat")
184			if gemeente_val and straat_naam:
185			straat_val = next((s for s in gemeente_val.straten if s.label.lower() == straat_naam.lower()), None)
186			if straat_val:
187			address["straat"] = "" + straat_val.label
188			address["straat_id"] = straat_val.id
189			huisnummer_naam = address.get("huisnummer")
190			if huisnummer_naam:
191			huisnummer_val = next((n for n in straat_val.huisnummers
192			if n.huisnummer.lower() == huisnummer_naam.lower()), None)
193			if huisnummer_val:
194			address["huisnummer"] = "" + huisnummer_val.huisnummer
195			address["huisnummer_id"] = huisnummer_val.id
196			return address
197
198
199			def nearest_location(geojson_element, crabpy_gateway=None):
200			"""
201			Returns the nearest location. If a crabpy_gateway is given, the crab id's will be added to the address.
202			Uses the agiv service https://loc.geopunt.be/geolocation/Location?xy=<x,y>
203			where <x,y> are the coordinates of the centroid of the geometry (polygon).
204			The agiv service returns crab, urbis en poi location types only in Flanders and Brussels.
205			:param value: geojson
206			:return: - crab address
207			- None if the nearest address is nog crab type
208			"""
209			xy = get_centroid_xy(geojson_element)
210
211			r = requests.get('https://loc.geopunt.be/geolocation/Location?xy={0}'.format(xy))
212			result = r.json()
213			if 'LocationResult' not in result \
214			or len(result['LocationResult']) == 0 \
215			or 'crab' not in result['LocationResult'][0]['LocationType']:
216			return None
217			locres = result['LocationResult'][0]['FormattedAddress']
218			straat_huisnummer, postcode_gemeente = locres.rsplit(', ', 1)
219			straat, huisnummer = straat_huisnummer.rsplit(' ', 1)
220			postcode, gemeente = postcode_gemeente.split(' ', 1)
221			address = {
222			'straat': straat,
223			'huisnummer': huisnummer,
224			'omschrijving_straat': straat + ', ' + huisnummer,
225			'postcode': postcode,
226			'gemeente': gemeente,
227			'land': 'BE'
228			}
229			if crabpy_gateway:
230			return add_crab_ids(crabpy_gateway, address)
231			else:
232			return address
233
234
235			class AdminGrenzenClient:
236			def __init__(self, base_url):
237			self.base_url = base_url
238
239			def get_admin_objecten(self, geojson_input, admin_grens_type, return_geometry=0):
240			"""
241			This function returns the administrative areas objects which intersects with the input geojson
242			:param geojson_input: geojson
243			:param admin_grens_type: type of returned administrative objects
244			:param return_geometry: boolean indicating whether the geometry of the returned objects should also be included
245			:
246			:return: list of administrative areas objects
247			"""
248			if check_in_flanders(geojson_input):
249			data = {
250			'geef_geometrie': return_geometry,
251			'type': admin_grens_type,
252			'geometrie': geojson_input
253			}
254			res = requests.post(
255			url=self.base_url,
256			json=data,
257			headers={
258			'Accept': 'application/json',
259			'Content-Type': 'application/json'
260			}
261			)
262			res.raise_for_status()
263			return json.loads(res.text)
264			else:
265			return []
266
267			def get_gemeenten(self, geojson_input):
268			"""
269			This function returns the names of the municipalities which intersects with the input geojson
270			:param geojson_input: geojson
271			:return: list of municipalities
272			"""
273			results = self.get_admin_objecten(geojson_input, 'gemeente')
274			return [gemeente['naam'] for gemeente in results]
275
276			def get_provincies(self, geojson_input):
277			"""
278			This function returns the names of the provinces which intersects with the input geojson
279			:param geojson_input: geojson
280			:return: list of municipalities
281			"""
282			results = self.get_admin_objecten(geojson_input, 'provincie')
283			return [provincie['naam'] for provincie in results]
284
285			@staticmethod
286			def _get_lagest_admin_obj(results, geojson_input):
287			if len(results) == 0:
288			return None
289			elif len(results) == 1:
290			return {'niscode': results[0]['id'], 'naam': results[0]['naam']}
291			else:
292			input_polygon = convert_geojson_to_geometry(geojson_input)
293			for result in results:
294			admin_geom = convert_geojson_to_geometry(result['geometrie'])
295			result['intersection_area'] = admin_geom.intersection(input_polygon).area
296			largest_m = max(results, key=lambda x: x['intersection_area'])
297			return {'niscode': largest_m['id'], 'naam': largest_m['naam']}
298
299			def get_gemeente(self, geojson_input):
300			"""
301			This function returns the name of the municipality which intersects with the input geojson.
302			If more municipalities intersect. The municipality is returned with the largest intersection
303			:param value: administratievegrenzen_url
304			:param value: geojson
305			:return: municipality
306			"""
307			results = self.get_admin_objecten(geojson_input, 'gemeente', 1)
308			return self._get_lagest_admin_obj(results, geojson_input)
309
310			def get_provincie(self, geojson_input):
311			"""
312			This function returns the name of the province which intersects with the input geojson.
313			If more provinces intersect. The province is returned with the largest intersection.
314			:param value: administratievegrenzen_url
315			:param value: geojson
316			:return: province
317			"""
318			results = self.get_admin_objecten(geojson_input, 'provincie', 1)
319			return self._get_lagest_admin_obj(results, geojson_input)
320
321			def check_erfgoedgemeenten(self, geojson_input, erfgoedgemeenten_list):
322			"""
323			:param geojson_input: geojson geometry
324			:param erfgoedgemeenten_list: list of niscodes of municipalities
325			:return: dict including key status and optional key message
326			status "ok" when the geojson does not intersect with one of given municipalities
327			status "warn" when the geojson intersects partially with one of given municipalities
328			status "error" when the geojson is within one of given municipalities
329			"""
330			gemeenten = self.get_admin_objecten(geojson_input, 'gemeente')
331			erfgoedgemeenten = [gemeente for gemeente in gemeenten if int(gemeente['id']) in json.loads(erfgoedgemeenten_list)]
332			if len(erfgoedgemeenten) > 0:
333			if len(gemeenten) == len(erfgoedgemeenten):
334			return {
335			"status": "error",
336			"message": "Let op, de zone van deze melding is gelegen in een onroerenderfgoedgemeente en "
337			"kan niet bewaard worden. Gelieve de melding in te dienen bij deze gemeente."}
338			else:
339			return {
340			"status": "warn",
341			"message": "Let op, deze melding ligt deels in een onroerenderfgoedgemeente . "
342			"Gelieve de melding vooronderzoek eveneens in te dienen bij deze gemeente."}
343			return {"status": "ok"}
344

OnroerendErfgoed / oe_geoutils

Push — master ( 180df1...ad4d03 )

provincie_niscode() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like