From a067aafde14a4699c801eba6b60c48f1402593d1 Mon Sep 17 00:00:00 2001 From: mpeltriaux Date: Sat, 12 Sep 2026 16:34:34 +0200 Subject: [PATCH] # Refactoring geometry processing * adds GeometryProcessor * adds GeoJsonValidator * refactors SimpleGeomForm is_valid() logic using these new classes --- konova/forms/geometry_form.py | 137 ++---------- konova/utils/geometry/__init__.py | 5 + konova/utils/geometry/geometry_validator.py | 228 ++++++++++++++++++++ 3 files changed, 250 insertions(+), 120 deletions(-) create mode 100644 konova/utils/geometry/__init__.py create mode 100644 konova/utils/geometry/geometry_validator.py diff --git a/konova/forms/geometry_form.py b/konova/forms/geometry_form.py index cc4fcaf5..1e067bb4 100644 --- a/konova/forms/geometry_form.py +++ b/konova/forms/geometry_form.py @@ -7,9 +7,7 @@ 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.contrib.gis.geos import MultiPolygon from django.utils.translation import gettext_lazy as _ from django.forms import JSONField @@ -18,6 +16,7 @@ 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 konova.utils.geometry.geometry_validator import GeoJsonValidator, GeometryProcessor from user.models import UserActionLogEntry @@ -60,6 +59,11 @@ class SimpleGeomForm(BaseForm): self.initialize_form_field("output", geom) def is_valid(self): + """ Custom is_valid method for SimpleGeomForm + + Returns: + + """ super().is_valid() is_valid = True @@ -75,118 +79,21 @@ class SimpleGeomForm(BaseForm): 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 + geojson_validator = GeoJsonValidator(geom) + geojson_validator.validate() - # 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) + if geojson_validator.errors: + self.add_error("output", geojson_validator.errors) + return False - 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 cleaned_data is not proper empty dict, initialize one if self.cleaned_data is None: self.cleaned_data = {} - self.cleaned_data["output"] = form_geom.ewkt + self.cleaned_data["output"] = geojson_validator.validated_geometry.ewkt + self._num_geometries_ignored = geojson_validator.num_ignored_geometries 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 @@ -213,9 +120,9 @@ class SimpleGeomForm(BaseForm): created=action, ) - is_vertices_num_valid = self.__is_vertices_num_valid() + is_vertices_num_valid = GeometryProcessor.is_vertices_num_valid(geometry.geom) if not is_vertices_num_valid: - geometry.geom = self.__simplify_geometry(geometry.geom, max_vert=GEOM_MAX_VERTICES) + geometry.geom = GeometryProcessor.simplify_geometry(geometry.geom, max_vert=GEOM_MAX_VERTICES) geometry.save() self._geometry_simplified = True @@ -240,16 +147,6 @@ class SimpleGeomForm(BaseForm): """ 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 diff --git a/konova/utils/geometry/__init__.py b/konova/utils/geometry/__init__.py new file mode 100644 index 00000000..afc28f66 --- /dev/null +++ b/konova/utils/geometry/__init__.py @@ -0,0 +1,5 @@ +""" +Author: Michel Peltriaux +Created on: 12.09.26 + +""" diff --git a/konova/utils/geometry/geometry_validator.py b/konova/utils/geometry/geometry_validator.py new file mode 100644 index 00000000..8d0c970e --- /dev/null +++ b/konova/utils/geometry/geometry_validator.py @@ -0,0 +1,228 @@ +""" +Author: Michel Peltriaux +Created on: 12.09.26 + +""" +import json + +from django.contrib.gis import gdal +from django.contrib.gis.gdal.srs import SpatialReference +from django.contrib.gis.geos import Polygon, MultiPolygon +from django.contrib.gis.geos.prototypes.io import WKTWriter +from django.utils.translation import gettext_lazy as _ + +from konova.settings import GEOM_MAX_VERTICES +from konova.sub_settings.lanis_settings import DEFAULT_SRID_RLP + + +class GeometryProcessor: + """ GeometryProcessor wraps methods for checking or processing geometry objects + + """ + @staticmethod + def simplify_geometry(geom: MultiPolygon, 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 + + @staticmethod + def is_vertices_num_valid(geometry: MultiPolygon): + """ Checks whether the number of vertices in the geometry is not too high + + Returns: + + """ + return geometry.num_coords <= GEOM_MAX_VERTICES + +class GeoJsonValidator: + """ GeoJson Validator validates geojson (e.g. from API or form input) + + """ + _errors: list + _input_geojson: dict + _validated_geometry: MultiPolygon + _num_ignored_geometries: int + + _accepted_ogr_types = [ + "Polygon", + "Polygon25D", + "MultiPolygon", + "MultiPolygon25D", + ] + + def __init__(self, input_geojson: dict) -> None: + assert input_geojson is not None + self._input_geojson = input_geojson + self._num_ignored_geometries = 0 + self._errors = list() + + @property + def num_ignored_geometries(self): + """ Getter + + Returns: + + """ + return self._num_ignored_geometries + + @property + def input_geojson(self): + """ Getter + + Returns: + + """ + return self._input_geojson + + @property + def validated_geometry(self): + """ Getter + + Returns: + + """ + return self._validated_geometry + + @property + def errors(self): + """ Getter + + Returns: + + """ + return self._errors + + def __add_error(self, text: str): + """ Wraps pushing new error into error list + + Args: + text (str): The error text + + Returns: + + """ + self._errors.append(text) + + def validate(self): + """ Validates input geojson + + Returns: + + """ + features = self._input_geojson.get("features", []) + try: + validated_features = self.__validate_single_features(features) + except AssertionError as e: + self.__add_error(str(e)) + return + + # Unionize all polygon features into one new MultiPolygon + if validated_features: + multipolygon_geometry = MultiPolygon(*validated_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 + multipolygon_geometry = MultiPolygon(srid=DEFAULT_SRID_RLP) + + # Make sure to convert into a MultiPolygon. Relevant if a single Polygon is provided. + multipolygon_geometry = self.cast_to_multipolygon(multipolygon_geometry) + self._validated_geometry = multipolygon_geometry + + def __validate_single_features(self, features: list) -> list: + validated_features = [] + + srs = SpatialReference(DEFAULT_SRID_RLP) + # Check validity for each feature of the geometry + for feature in features: + 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=srs) + + 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 self._accepted_ogr_types + if geometry_type_is_accepted: + raise AssertionError(_("Only surfaces allowed. Points or lines must be buffered.")) + + 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_ignored_geometries += 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) + if not g.valid: + raise AssertionError(g.valid_reason) + + # If the resulting polygon is just a single polygon, we add it to the list of properly created features + if isinstance(g, Polygon): + validated_features.append(g) + elif isinstance(g, MultiPolygon): + # The resulting polygon could be of type MultiPolygon (due to multiple surfaces) + # If so, we extract each single polygons from the MultiPolygon and add them to the created features list + validated_features.extend(list(g)) + + return validated_features + + + def __flatten_geom_to_2D(self, geom: gdal.OGRGeometry) -> gdal.OGRGeometry: + """ + 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 __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 + + @staticmethod + def cast_to_multipolygon(input_geom): + """ If input_geom is not a MultiPolygon, cast to MultiPolygon + + Args: + input_geom (): + + Returns: + output_geom + """ + output_geom = input_geom + if not isinstance(input_geom, MultiPolygon): + try: + output_geom = MultiPolygon(input_geom, srid=DEFAULT_SRID_RLP) + except TypeError as e: + raise AssertionError(f"Only (Multi)Polygon allowed! Could not convert {input_geom.geom_type} to MultiPolygon") + return output_geom +