353 lines
11 KiB
Python
353 lines
11 KiB
Python
"""
|
|
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, DEFAULT_SRID
|
|
|
|
|
|
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
|
|
|
|
@staticmethod
|
|
def is_valid_4326(geometry: MultiPolygon) -> bool:
|
|
""" Checks whether a given geometry's coordinates are in a valid range to be of EPSG:4326
|
|
|
|
Args:
|
|
geometry: The geometry
|
|
|
|
Returns:
|
|
ret_val (bool): Whether the geometry is valid EPSG:4326
|
|
"""
|
|
assert geometry.srid == DEFAULT_SRID
|
|
return GeometryProcessor.is_valid_coord_range(
|
|
geometry,
|
|
-90.0,
|
|
90.0,
|
|
-180.0,
|
|
180.0,
|
|
)
|
|
|
|
@staticmethod
|
|
def is_valid_25832(geometry: MultiPolygon) -> bool:
|
|
""" Checks whether a given geometry's coordinates are in a valid range to be of EPSG:25832
|
|
|
|
Args:
|
|
geometry: The geometry
|
|
|
|
Returns:
|
|
ret_val (bool): Whether the geometry is valid EPSG:25832
|
|
"""
|
|
assert geometry.srid == DEFAULT_SRID_RLP
|
|
return GeometryProcessor.is_valid_coord_range(
|
|
geometry,
|
|
166_000,
|
|
834_000,
|
|
0,
|
|
9_329_000,
|
|
)
|
|
|
|
@staticmethod
|
|
def is_valid_coord_range(geometry: MultiPolygon,
|
|
min_x:float|int,
|
|
max_x:float|int,
|
|
min_y:float|int,
|
|
max_y: float|int) -> bool:
|
|
""" Checks whether a given geometry's coordinates are in a valid range defined by input parameters
|
|
|
|
Args:
|
|
geometry: The geometry
|
|
min_x (float|int): The min x value
|
|
max_x (float|int): The max x value
|
|
min_y (float|int): The min y value
|
|
max_y (float|int): The max y value
|
|
|
|
Returns:
|
|
ret_val (bool): Whether the geometry's coordinates stays inside the defined range
|
|
"""
|
|
if not geometry.centroid.coords:
|
|
# No coordinates at all found, therefore technically proper 4326
|
|
return True
|
|
|
|
try:
|
|
env_min_x = geometry.envelope.min_x
|
|
env_max_x = geometry.envelope.max_x
|
|
env_min_y = geometry.envelope.min_y
|
|
env_max_y = geometry.envelope.max_x
|
|
|
|
valid = (
|
|
min_x <= env_min_x and
|
|
env_max_x <= max_x and
|
|
min_y <= env_min_y and
|
|
env_max_y <= max_y
|
|
)
|
|
return valid
|
|
except IndexError:
|
|
return False
|
|
|
|
@staticmethod
|
|
def cast_to_multipolygon(input_geom: MultiPolygon) -> MultiPolygon:
|
|
""" 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
|
|
|
|
@staticmethod
|
|
def cast_to_rlp_srid(input_geom: gdal.OGRGeometry) -> gdal.OGRGeometry:
|
|
""" If input_geom is not of RLP SRID (25832), cast to RLP SRID
|
|
|
|
Args:
|
|
input_geom ():
|
|
|
|
Returns:
|
|
output_geom
|
|
"""
|
|
output_geom = input_geom
|
|
if output_geom.srid != DEFAULT_SRID_RLP:
|
|
output_geom.transform(DEFAULT_SRID_RLP)
|
|
return output_geom
|
|
|
|
@staticmethod
|
|
def flatten_geom_to_2D(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
|
|
|
|
@staticmethod
|
|
def is_area_valid(geom: gdal.OGRGeometry) -> bool:
|
|
""" Checks whether the area is at least > 1m²
|
|
|
|
Returns:
|
|
|
|
"""
|
|
assert geom.srid == DEFAULT_SRID_RLP
|
|
is_area_valid = geom.area > 1 # > 1m² (SRID:25832)
|
|
return is_area_valid
|
|
|
|
|
|
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, srs: int) -> None:
|
|
assert input_geojson is not None
|
|
self._input_geojson = input_geojson
|
|
self._num_ignored_geometries = 0
|
|
self._errors = list()
|
|
self._srs = SpatialReference(srs)
|
|
|
|
@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", None)
|
|
|
|
is_input_geojson_empty = len(self._input_geojson) == 0
|
|
no_features_in_input_found = features is None
|
|
|
|
if not is_input_geojson_empty and no_features_in_input_found:
|
|
# check if _input_geojson is a feature itself
|
|
if self.__is_geojson_feature():
|
|
features = [
|
|
{
|
|
"geometry": self._input_geojson
|
|
}
|
|
]
|
|
else:
|
|
self.__add_error("Input does not seem to be geojson")
|
|
return
|
|
|
|
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=self._srs).unary_union
|
|
else:
|
|
# If no features have been processed, this indicates an empty geometry - so we store an empty geometry
|
|
multipolygon_geometry = MultiPolygon(srid=self._srs)
|
|
|
|
# Make sure to convert into a MultiPolygon. Relevant if a single Polygon is provided.
|
|
multipolygon_geometry = GeometryProcessor.cast_to_multipolygon(multipolygon_geometry)
|
|
self._validated_geometry = multipolygon_geometry
|
|
|
|
def __validate_single_features(self, features: list) -> list:
|
|
validated_features = []
|
|
|
|
# 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=self._srs)
|
|
if g.empty:
|
|
continue
|
|
g = GeometryProcessor.cast_to_rlp_srid(g)
|
|
if not GeometryProcessor.is_valid_25832(g):
|
|
raise AssertionError(_("This feature does not hold valid EPSG:25832 coordinates:\n {}".format(g.geojson)))
|
|
|
|
geometry_has_unwanted_dimensions = g.coord_dim > 2
|
|
if geometry_has_unwanted_dimensions:
|
|
g = GeometryProcessor.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 = GeometryProcessor.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 __is_geojson_feature(self):
|
|
""" Checks whether _input_geojson is a proper geojson feature
|
|
|
|
Returns:
|
|
|
|
"""
|
|
has_type = self._input_geojson.get("type", None) is not None
|
|
has_coordinates = self._input_geojson.get("coordinates", None) is not None
|
|
has_properties = self._input_geojson.get("properties", None) is not None
|
|
return has_type and has_coordinates and has_properties |