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

Completed

Push — master ( 8a6b55...7194ea )

unknown

created 2018-05-25 11:37 UTC

get_flanders_tidal_zone_geometry() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
c	0
b	0
f	0
dl	0
loc	3
rs	10

# -*- 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

from oe_geoutils.data.models import LocatieElement


def process_location_elements(crab_gateway, admin_grenzen_client, contour, municipalities_niscodes=None):
    """

    :param crab_gateway: gateway for accessing crab data
    :param admin_grenzen_client: client for accessing administrative borders
    :param contour: contour
    :param municipalities_niscodes: list of municipality niscodes not be considered
    :return: list of location elements
    """
    municipalities_niscodes = municipalities_niscodes if municipalities_niscodes else []
    admin_municipalities = admin_grenzen_client.get_admin_objecten(contour, 'gemeente')
    admin_municipalities_niscodes = set([int(admin_municipality['id']) for admin_municipality in admin_municipalities])

    location_elementens = []
    for municipality_niscode in admin_municipalities_niscodes.difference(municipalities_niscodes):
        crab_client = crab_gateway()
        crab_gem = crab_client.get_gemeente_by_niscode(municipality_niscode)
        gemeente_naam = crab_gem.naam
        gemeente_crab_id = crab_gem.id
        provincie_niscode = crab_gem.provincie.niscode
        provincie_naam = crab_gem.provincie.naam
        location_element = LocatieElement(
            type='https://id.erfgoed.net/vocab/ontology#LocatieElement',
            gemeente_niscode=municipality_niscode,
            gemeente_naam=gemeente_naam,
            gemeente_crab_id=gemeente_crab_id,
            provincie_niscode=provincie_niscode,
            provincie_naam=provincie_naam)
        location_elementens.append(location_element)

    return location_elementens


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, AttributeError):
        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, tolerance=10):
    """
    :param geojson_input: geojson geometrie
    :param tolerance: buffer tolerance distance of Flanders
    :return: True if the input geojson is within the Flanders region, otherwise False
    """
    if geometry.shape(geojson_input).within(get_flanders_geometry().buffer(tolerance)):
        return True
    return False


def check_in_flanders(geojson_input, check_getijdezone=0):
    """
    :param geojson_input: geojson geometrie
    :param check_getijdezone: bool, check in tidal zone is necessary
    :return: True if the input geojson intersects with Flanders region, otherwise False
    """
    in_flanders = False
    geojson_input_shape = geometry.shape(geojson_input)
    if geojson_input_shape.intersects(get_flanders_geometry()):
        in_flanders = True
    if not in_flanders and check_getijdezone:
        if geojson_input_shape.intersects(get_flanders_tidal_zone_geometry()):
            in_flanders = True
    return in_flanders


def _get_geometry(file_path):
    with open(file_path) as geof:
        features = json.load(geof)
    output_geojson = features['features'][0]['geometry']
    return geometry.shape(output_geojson)


def get_flanders_geometry():
    here = os.path.abspath(os.path.dirname(__file__))
    return _get_geometry(here + '/fixtures/vlaams_gewest.geojson')


def get_flanders_tidal_zone_geometry():
    here = os.path.abspath(os.path.dirname(__file__))
    return _get_geometry(here + '/fixtures/vlaams_gewest_getijdenzone.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)
    gemeente = gemeente.replace(' ({})'.format(postcode), '')
    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			from oe_geoutils.data.models import LocatieElement
12
13
14			def process_location_elements(crab_gateway, admin_grenzen_client, contour, municipalities_niscodes=None):
15			"""
16
17			:param crab_gateway: gateway for accessing crab data
18			:param admin_grenzen_client: client for accessing administrative borders
19			:param contour: contour
20			:param municipalities_niscodes: list of municipality niscodes not be considered
21			:return: list of location elements
22			"""
23			municipalities_niscodes = municipalities_niscodes if municipalities_niscodes else []
24			admin_municipalities = admin_grenzen_client.get_admin_objecten(contour, 'gemeente')
25			admin_municipalities_niscodes = set([int(admin_municipality['id']) for admin_municipality in admin_municipalities])
26
27			location_elementens = []
28			for municipality_niscode in admin_municipalities_niscodes.difference(municipalities_niscodes):
29			crab_client = crab_gateway()
30			crab_gem = crab_client.get_gemeente_by_niscode(municipality_niscode)
31			gemeente_naam = crab_gem.naam
32			gemeente_crab_id = crab_gem.id
33			provincie_niscode = crab_gem.provincie.niscode
34			provincie_naam = crab_gem.provincie.naam
35			location_element = LocatieElement(
36			type='https://id.erfgoed.net/vocab/ontology#LocatieElement',
37			gemeente_niscode=municipality_niscode,
38			gemeente_naam=gemeente_naam,
39			gemeente_crab_id=gemeente_crab_id,
40			provincie_niscode=provincie_niscode,
41			provincie_naam=provincie_naam)
42			location_elementens.append(location_element)
43
44			return location_elementens
45
46
47			def convert_wktelement_to_geojson(wktelement):
48			"""
49			This function converts a wktelement (from GeoAlchemy) to a geojson
50			:param wktelement: WKTElement
51			:return: geojson
52			"""
53			if wktelement is None:
54			return None
55			try:
56			geom = shape.to_shape(wktelement)
57			srid = wktelement.srid
58			json_element = json.dumps(geometry.mapping(geom))
59			geojson_element = geojson.loads(json_element)
60			geojson_element['crs'] = {
61			"type": "name",
62			"properties": {
63			"name": "urn:ogc:def:crs:EPSG::{0}".format(srid)
64			}
65			}
66
67			return geojson_element
68
69			except (AssertionError, AttributeError):
70			raise AssertionError("WKTelement is niet geldig: %s" % wktelement)
71
72
73			def convert_geojson_to_wktelement(value):
74			"""
75			This function converts a geojson to a wktelement that can be used by GeoAlchemy.
76			:param value: geojson
77			:return: WKTElement
78			"""
79			if value is None or not value:
80			return None
81			try:
82			s = json.dumps(value)
83			g1 = geojson.loads(s)
84			g2 = geometry.shape(g1)
85			srid = get_srid_from_geojson(g1)
86
87			return WKTElement(data=g2.wkt, srid=srid)
88
89			except ValueError:
90			raise ValueError("GeoJson is niet geldig: %s" % value)
91
92
93			def convert_geojson_to_geometry(value):
94			"""
95			This function converts a geojson to a Shapely.geometry.
96			:param value: geojson
97			:return: Shapely.geometry
98			"""
99			geojson_element = geojson.loads(json.dumps(value))
100			return geometry.shape(geojson_element)
101
102
103			def get_srid_from_geojson(value):
104			"""
105			This function returns the srid of a geojson
106			:param value: geojson
107			:return: spatial reference identifier (srid)
108			"""
109			if value is None or not value:
110			return None
111			try:
112			if not value.get('crs'):
113			return None
114			srid = find_srid(value['crs']['properties']['name'])
115			# todo make a better implementation for this?! Is there a fixed format in geojson to define CRS
116
117			return srid
118
119			except ValueError: # pragma NO COVER
120			# In deze situatie enkel key en attributeErrors mogelijk
121			raise ValueError("Geen geldige Spatial Reference Identifier (SRID) gevonden" % value)
122
123
124			def find_srid(text):
125			li = [int(x) for x in re.findall(r'\d+', text) if len(x) in [4, 5]]
126			return max(li) if len(li) > 0 else None
127
128
129			def remove_dupl_values(list_coord):
130			dupl_coord_indexes = [i for i in range(0, len(list_coord)) if i != 0 and
131			list_coord[i] == list_coord[i - 1]]
132			[list_coord.pop(i) for i in sorted(dupl_coord_indexes, reverse=True)]
133
134
135			def remove_dupl_coords(coords, tree_types=(list, tuple)):
136			"""
137			This function will remove duplicate consecutive coordinates
138			:param coords: input coordinates
139			"""
140			for o in coords:
141			if isinstance(o[0][0], tree_types):
142			remove_dupl_coords(o, tree_types)
143			else:
144			remove_dupl_values(o)
145
146
147			def check_within_flanders(geojson_input, tolerance=10):
148			"""
149			:param geojson_input: geojson geometrie
150			:param tolerance: buffer tolerance distance of Flanders
151			:return: True if the input geojson is within the Flanders region, otherwise False
152			"""
153			if geometry.shape(geojson_input).within(get_flanders_geometry().buffer(tolerance)):
154			return True
155			return False
156
157
158			def check_in_flanders(geojson_input, check_getijdezone=0):
159			"""
160			:param geojson_input: geojson geometrie
161			:param check_getijdezone: bool, check in tidal zone is necessary
162			:return: True if the input geojson intersects with Flanders region, otherwise False
163			"""
164			in_flanders = False
165			geojson_input_shape = geometry.shape(geojson_input)
166			if geojson_input_shape.intersects(get_flanders_geometry()):
167			in_flanders = True
168			if not in_flanders and check_getijdezone:
169			if geojson_input_shape.intersects(get_flanders_tidal_zone_geometry()):
170			in_flanders = True
171			return in_flanders
172
173
174			def _get_geometry(file_path):
175			with open(file_path) as geof:
176			features = json.load(geof)
177			output_geojson = features['features'][0]['geometry']
178			return geometry.shape(output_geojson)
179
180
181			def get_flanders_geometry():
182			here = os.path.abspath(os.path.dirname(__file__))
183			return _get_geometry(here + '/fixtures/vlaams_gewest.geojson')
184
185
186			def get_flanders_tidal_zone_geometry():
187			here = os.path.abspath(os.path.dirname(__file__))
188			return _get_geometry(here + '/fixtures/vlaams_gewest_getijdenzone.geojson')
189
190
191			def provincie_niscode(gemeente_niscode):
192			"""
193			geef de niscode van de provincie op basis van de niscode van de gemeente
194
195			:param gemeente_niscode: niscode van de gemeente
196			:return: niscode van de provincie
197			"""
198			pcode = str(gemeente_niscode)[0]
199			if pcode == '2':
200			return int('20001')
201			else:
202			return int('{0}0000'.format(pcode))
203
204
205			def get_centroid_xy(value):
206			"""
207			Returns the centroid x,y of a geojson polygon.
208			If the geojson is a multipolygon. It return the centroid of the biggest polygon
209			:param value: geojson
210			:return: x,y
211			"""
212			geometry_element = convert_geojson_to_geometry(value)
213			if geometry_element.type == 'MultiPolygon':
214			geometry_element_polygons = list(geometry_element.geoms)
215			largest_polygon = max(geometry_element_polygons, key=attrgetter('area'))
216			centroid = largest_polygon.centroid.wkt
217			else:
218			centroid = geometry_element.centroid.wkt
219			return ','.join([x for x in re.findall(r'\d+.\d+', centroid)])
220
221
222			def add_crab_ids(crabpy_gateway, address):
223			gemeente = address.get("gemeente")
224			if gemeente:
225			gewest_ids = [2, 1, 3]
226			gemeente_val = None
227			for gewest_id in gewest_ids:
228			gemeenten = crabpy_gateway.list_gemeenten(gewest_id)
229			gemeente_val = next((g for g in gemeenten if g.naam.lower() == gemeente.lower()), None)
230			if gemeente_val:
231			address['gemeente'] = "" + gemeente_val.naam
232			address['gemeente_id'] = gemeente_val.id
233			break
234			straat_naam = address.get("straat")
235			if gemeente_val and straat_naam:
236			straat_val = next((s for s in gemeente_val.straten if s.label.lower() == straat_naam.lower()), None)
237			if straat_val:
238			address["straat"] = "" + straat_val.label
239			address["straat_id"] = straat_val.id
240			huisnummer_naam = address.get("huisnummer")
241			if huisnummer_naam:
242			huisnummer_val = next((n for n in straat_val.huisnummers
243			if n.huisnummer.lower() == huisnummer_naam.lower()), None)
244			if huisnummer_val:
245			address["huisnummer"] = "" + huisnummer_val.huisnummer
246			address["huisnummer_id"] = huisnummer_val.id
247			return address
248
249
250			def nearest_location(geojson_element, crabpy_gateway=None):
251			"""
252			Returns the nearest location. If a crabpy_gateway is given, the crab id's will be added to the address.
253			Uses the agiv service https://loc.geopunt.be/geolocation/Location?xy=<x,y>
254			where <x,y> are the coordinates of the centroid of the geometry (polygon).
255			The agiv service returns crab, urbis en poi location types only in Flanders and Brussels.
256			:param value: geojson
257			:return: - crab address
258			- None if the nearest address is nog crab type
259			"""
260			xy = get_centroid_xy(geojson_element)
261
262			r = requests.get('https://loc.geopunt.be/geolocation/Location?xy={0}'.format(xy))
263			result = r.json()
264			if 'LocationResult' not in result \
265			or len(result['LocationResult']) == 0 \
266			or 'crab' not in result['LocationResult'][0]['LocationType']:
267			return None
268			locres = result['LocationResult'][0]['FormattedAddress']
269			straat_huisnummer, postcode_gemeente = locres.rsplit(', ', 1)
270			straat, huisnummer = straat_huisnummer.rsplit(' ', 1)
271			postcode, gemeente = postcode_gemeente.split(' ', 1)
272			gemeente = gemeente.replace(' ({})'.format(postcode), '')
273			address = {
274			'straat': straat,
275			'huisnummer': huisnummer,
276			'omschrijving_straat': straat + ', ' + huisnummer,
277			'postcode': postcode,
278			'gemeente': gemeente,
279			'land': 'BE'
280			}
281			if crabpy_gateway:
282			return add_crab_ids(crabpy_gateway, address)
283			else:
284			return address
285
286
287			class AdminGrenzenClient:
288			def __init__(self, base_url):
289			self.base_url = base_url
290
291			def get_admin_objecten(self, geojson_input, admin_grens_type, return_geometry=0):
292			"""
293			This function returns the administrative areas objects which intersects with the input geojson
294			:param geojson_input: geojson
295			:param admin_grens_type: type of returned administrative objects
296			:param return_geometry: boolean indicating whether the geometry of the returned objects should also be included
297			:
298			:return: list of administrative areas objects
299			"""
300			if check_in_flanders(geojson_input):
301			data = {
302			'geef_geometrie': return_geometry,
303			'type': admin_grens_type,
304			'geometrie': geojson_input
305			}
306			res = requests.post(
307			url=self.base_url,
308			json=data,
309			headers={
310			'Accept': 'application/json',
311			'Content-Type': 'application/json'
312			}
313			)
314			res.raise_for_status()
315			return json.loads(res.text)
316			else:
317			return []
318
319			def get_gemeenten(self, geojson_input):
320			"""
321			This function returns the names of the municipalities which intersects with the input geojson
322			:param geojson_input: geojson
323			:return: list of municipalities
324			"""
325			results = self.get_admin_objecten(geojson_input, 'gemeente')
326			return [gemeente['naam'] for gemeente in results]
327
328			def get_provincies(self, geojson_input):
329			"""
330			This function returns the names of the provinces which intersects with the input geojson
331			:param geojson_input: geojson
332			:return: list of municipalities
333			"""
334			results = self.get_admin_objecten(geojson_input, 'provincie')
335			return [provincie['naam'] for provincie in results]
336
337			@staticmethod
338			def _get_lagest_admin_obj(results, geojson_input):
339			if len(results) == 0:
340			return None
341			elif len(results) == 1:
342			return {'niscode': results[0]['id'], 'naam': results[0]['naam']}
343			else:
344			input_polygon = convert_geojson_to_geometry(geojson_input)
345			for result in results:
346			admin_geom = convert_geojson_to_geometry(result['geometrie'])
347			result['intersection_area'] = admin_geom.intersection(input_polygon).area
348			largest_m = max(results, key=lambda x: x['intersection_area'])
349			return {'niscode': largest_m['id'], 'naam': largest_m['naam']}
350
351			def get_gemeente(self, geojson_input):
352			"""
353			This function returns the name of the municipality which intersects with the input geojson.
354			If more municipalities intersect. The municipality is returned with the largest intersection
355			:param value: administratievegrenzen_url
356			:param value: geojson
357			:return: municipality
358			"""
359			results = self.get_admin_objecten(geojson_input, 'gemeente', 1)
360			return self._get_lagest_admin_obj(results, geojson_input)
361
362			def get_provincie(self, geojson_input):
363			"""
364			This function returns the name of the province which intersects with the input geojson.
365			If more provinces intersect. The province is returned with the largest intersection.
366			:param value: administratievegrenzen_url
367			:param value: geojson
368			:return: province
369			"""
370			results = self.get_admin_objecten(geojson_input, 'provincie', 1)
371			return self._get_lagest_admin_obj(results, geojson_input)
372
373			def check_erfgoedgemeenten(self, geojson_input, erfgoedgemeenten_list):
374			"""
375			:param geojson_input: geojson geometry
376			:param erfgoedgemeenten_list: list of niscodes of municipalities
377			:return: dict including key status and optional key message
378			status "ok" when the geojson does not intersect with one of given municipalities
379			status "warn" when the geojson intersects partially with one of given municipalities
380			status "error" when the geojson is within one of given municipalities
381			"""
382			gemeenten = self.get_admin_objecten(geojson_input, 'gemeente')
383			erfgoedgemeenten = [gemeente for gemeente in gemeenten if
384			int(gemeente['id']) in json.loads(erfgoedgemeenten_list)]
385			if len(erfgoedgemeenten) > 0:
386			if len(gemeenten) == len(erfgoedgemeenten):
387			return {
388			"status": "error",
389			"message": "Let op, de zone van deze melding is gelegen in een onroerenderfgoedgemeente en "
390			"kan niet bewaard worden. Gelieve de melding in te dienen bij deze gemeente."}
391			else:
392			return {
393			"status": "warn",
394			"message": "Let op, deze melding ligt deels in een onroerenderfgoedgemeente . "
395			"Gelieve de melding vooronderzoek eveneens in te dienen bij deze gemeente."}
396			return {"status": "ok"}
397

OnroerendErfgoed / oe_geoutils

Push — master ( 8a6b55...7194ea )

get_flanders_tidal_zone_geometry() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like