""" Author: Michel Peltriaux Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany Contact: ksp-servicestelle@sgdnord.rlp.de Created on: 15.08.22 """ import json from django.contrib.gis import gdal from django.contrib.gis.geos import MultiPolygon, Polygon from django.contrib.gis.geos.prototypes.io import WKTWriter from django.utils.translation import gettext_lazy as _ from django.forms import JSONField from konova.forms.base_form import BaseForm from konova.models import Geometry from konova.settings import GEOM_MAX_VERTICES from konova.tasks import celery_update_parcels, celery_check_for_geometry_conflicts from konova.sub_settings.lanis_settings import DEFAULT_SRID_RLP from user.models import UserActionLogEntry class SimpleGeomForm(BaseForm): """ A geometry form for rendering geometry read-only using a widget """ read_only: bool = True _geometry_simplified: bool = False output = JSONField( label=_("Geometry"), help_text=_(""), label_suffix="", required=False, disabled=False, ) _num_geometries_ignored: int = 0 empty = False def __init__(self, *args, **kwargs): self.read_only = kwargs.pop("read_only", True) super().__init__(*args, **kwargs) # Initialize geometry try: geom = self.instance.geometry.geom self.empty = geom.empty if self.empty: raise AttributeError geojson = self.instance.geometry.as_feature_collection(srid=DEFAULT_SRID_RLP) geojson = self._set_geojson_properties(geojson, title=self.instance.identifier or None) geom = json.dumps(geojson) except AttributeError: # If no geometry exists for this form, we simply set the value to None and zoom to the maximum level geom = json.dumps({}) self.empty = True self.initialize_form_field("output", geom) def is_valid(self): super().is_valid() is_valid = True # Make sure invalid geometry is properly rendered again to the user # Therefore: write submitted data back into form field # (does not matter whether we know if it is valid or invalid) submitted_data = self.data["output"] submitted_data = json.loads(submitted_data) submitted_data = self._set_geojson_properties(submitted_data) self.initialize_form_field("output", json.dumps(submitted_data)) # Get geojson from form for validity checking geom = self.data.get("output", json.dumps({})) geom = json.loads(geom) # Initialize features list with empty MultiPolygon, so that an empty input will result in a # proper empty MultiPolygon object features = [] features_json = geom.get("features", []) accepted_ogr_types = [ "Polygon", "Polygon25D", "MultiPolygon", "MultiPolygon25D", ] # Check validity for each feature of the geometry for feature in features_json: feature_geom = feature.get("geometry", feature) if feature_geom is None: # Fallback for rare cases where a feature does not contain any geometry continue # Try to create a geometry object from the single feature feature_geom = json.dumps(feature_geom) g = gdal.OGRGeometry(feature_geom, srs=DEFAULT_SRID_RLP) geometry_has_unwanted_dimensions = g.coord_dim > 2 if geometry_has_unwanted_dimensions: g = self.__flatten_geom_to_2D(g) geometry_type_is_accepted = g.geom_type not in accepted_ogr_types if geometry_type_is_accepted: self.add_error("output", _("Only surfaces allowed. Points or lines must be buffered.")) is_valid &= False return is_valid is_area_valid = self.__is_area_valid(g) if not is_area_valid: # Geometries with an invalid size will not be saved to the db # We assume these are malicious snippets which are not supposed to be in the geometry in the first place self._num_geometries_ignored += 1 continue # Whatever this geometry object is -> try to create a Polygon from it # The resulting polygon object automatically detects whether a valid polygon has been created or not g = Polygon.from_ewkt(g.ewkt) is_valid &= g.valid if not g.valid: self.add_error("output", g.valid_reason) return is_valid # If the resulting polygon is just a single polygon, we add it to the list of properly casted features if isinstance(g, Polygon): features.append(g) elif isinstance(g, MultiPolygon): # The resulting polygon could be of type MultiPolygon (due to multiple surfaces) # If so, we extract all polygons from the MultiPolygon and extend the casted features list features.extend(list(g)) # Unionize all polygon features into one new MultiPolygon if features: form_geom = MultiPolygon(*features, srid=DEFAULT_SRID_RLP).unary_union else: # If no features have been processed, this indicates an empty geometry - so we store an empty geometry form_geom = MultiPolygon(srid=DEFAULT_SRID_RLP) # Make sure to convert into a MultiPolygon. Relevant if a single Polygon is provided. form_geom = Geometry.cast_to_multipolygon(form_geom) # Write unionized Multipolygon back into cleaned data if self.cleaned_data is None: self.cleaned_data = {} self.cleaned_data["output"] = form_geom.ewkt return is_valid def __is_vertices_num_valid(self): """ Checks whether the number of vertices in the geometry is not too high Returns: """ geom = self.cleaned_data.get("output") g = gdal.OGRGeometry(geom, srs=DEFAULT_SRID_RLP) num_vertices = g.num_coords return num_vertices <= GEOM_MAX_VERTICES def __is_area_valid(self, geom: gdal.OGRGeometry): """ Checks whether the area is at least > 1m² Returns: """ is_area_valid = geom.area > 1 # > 1m² (SRID:25832) return is_area_valid def __simplify_geometry(self, geom, max_vert: int): """ Simplifies a geometry Geometry will be simplified until a threshold of max vertices has been reached. Args: geom (MultiPolygon): The geometry max_vert (int): Threshold of maximum vertices in geometry Returns: geom (MultiPolygon): The simplified geometry """ tolerance = 0.1 n = geom.num_coords while(n > max_vert): geom = geom.simplify(tolerance) n = geom.num_coords tolerance += 0.1 return geom def save(self, action: UserActionLogEntry): """ Saves the form's geometry Creates a new geometry entry if none is set, yet Args: action (): Returns: """ try: if self.instance is None or self.instance.geometry is None: raise LookupError geometry = self.instance.geometry geometry.geom = self.cleaned_data.get("output", MultiPolygon(srid=DEFAULT_SRID_RLP)) geometry.modified = action geometry.save() except LookupError: # No geometry or linked instance holding a geometry exist --> create a new one! geometry = Geometry.objects.create( geom=self.cleaned_data.get("output", MultiPolygon(srid=DEFAULT_SRID_RLP)), created=action, ) is_vertices_num_valid = self.__is_vertices_num_valid() if not is_vertices_num_valid: geometry.geom = self.__simplify_geometry(geometry.geom, max_vert=GEOM_MAX_VERTICES) geometry.save() self._geometry_simplified = True # Start parcel update and geometry conflict checking procedure in a background process celery_update_parcels.delay(geometry.id) celery_check_for_geometry_conflicts.delay(geometry.id) return geometry def get_num_geometries_ignored(self): """ Returns the number of geometries which had to be ignored for various reasons Returns: """ return self._num_geometries_ignored def has_geometry_simplified(self): """ Returns whether the geometry has been simplified or not. Returns: """ return self._geometry_simplified def __flatten_geom_to_2D(self, geom): """ Enforces a given OGRGeometry from higher dimensions into 2D """ wkt_w = WKTWriter(dim=2) g_wkt = wkt_w.write(geom.geos).decode("utf-8") geom = gdal.OGRGeometry(g_wkt) return geom def _set_geojson_properties(self, geojson: dict, title: str = None): """ Toggles the editable property of the geojson for proper handling in map client Args: geojson (dict): The GeoJson title (str): An alternative title for the geometry Returns: geojson (dict): The altered GeoJson """ features = geojson.get("features", []) for feature in features: if not feature.get("properties", None): feature["properties"] = {} feature["properties"]["editable"] = not self.read_only if title: feature["properties"]["title"] = title return geojson