konova/konova/utils/wfs/spatial.py

"""
Author: Michel Peltriaux
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
Contact: michel.peltriaux@sgdnord.rlp.de
Created on: 17.12.21

"""
from abc import abstractmethod
from time import sleep

import requests
import xmltodict
from django.contrib.gis.db.models.functions import AsGML, Transform
from requests.auth import HTTPDigestAuth

from konova.settings import DEFAULT_SRID_RLP, PARCEL_WFS_USER, PARCEL_WFS_PW, PROXIES


class AbstractWFSFetcher:
    """ Base class for fetching WFS data

    """
    # base_url represents not the capabilities url but the parameter-free base url
    base_url = None
    version = None
    auth_user = None
    auth_pw = None
    auth_digest_obj = None

    class Meta:
        abstract = True

    def __init__(self, base_url: str, version: str = "1.1.0", auth_user: str = None, auth_pw: str = None, *args, **kwargs):
        self.base_url = base_url
        self.version = version
        self.auth_pw = auth_pw
        self.auth_user = auth_user

        self._create_auth_obj()

    def _create_auth_obj(self):
        if self.auth_pw is not None and self.auth_user is not None:
            self.auth_digest_obj = HTTPDigestAuth(
                self.auth_user,
                self.auth_pw
            )

    @abstractmethod
    def get_features(self, feature_identifier: str, filter_str: str):
        raise NotImplementedError


class ParcelWFSFetcher(AbstractWFSFetcher):
    """ Fetches features from a special parcel WFS

    """
    geometry_id = None
    geometry_property_name = None
    count = 100

    def __init__(self, geometry_id: str, geometry_property_name: str = "msGeometry", *args, **kwargs):
        super().__init__(
            version="2.0.0",
            base_url="https://www.geoportal.rlp.de/registry/wfs/519",
            auth_user=PARCEL_WFS_USER,
            auth_pw=PARCEL_WFS_PW,
            *args,
            **kwargs
        )
        self.geometry_id = geometry_id
        self.geometry_property_name = geometry_property_name

    def _create_spatial_filter(self,
                               geometry_operation: str,
                               filter_srid: str = None):
        """ Creates a xml spatial filter according to the WFS filter specification

        Args:
            geometry_operation (str): One of the WFS supported spatial filter operations (according to capabilities)
            filter_srid (str): Used to transform the geometry into the spatial reference system identified by this srid

        Returns:
            spatial_filter (str): The spatial filter element
        """
        from konova.models import Geometry
        if filter_srid is None:
            filter_srid = DEFAULT_SRID_RLP
        geom_gml = Geometry.objects.filter(
            id=self.geometry_id
        ).annotate(
            transformed=Transform(srid=filter_srid, expression="geom")
        ).annotate(
            gml=AsGML('transformed')
        ).first().gml
        spatial_filter = f"<Filter><{geometry_operation}><PropertyName>{self.geometry_property_name}</PropertyName>{geom_gml}</{geometry_operation}></Filter>"
        return spatial_filter

    def _create_post_data(self,
                          geometry_operation: str,
                          filter_srid: str = None,
                          typenames: str = None,
                          start_index: int = 0,
                          ):
        """ Creates a POST body content for fetching features

        Args:
            geometry_operation (str): One of the WFS supported spatial filter operations (according to capabilities)
            filter_srid (str): Used to transform the geometry into the spatial reference system identified by this srid

        Returns:
            _filter (str): A proper xml WFS filter
        """
        start_index = str(start_index)
        spatial_filter = self._create_spatial_filter(
            geometry_operation,
            filter_srid
        )
        _filter = f'<wfs:GetFeature service="WFS" version="{self.version}" xmlns:wfs="http://www.opengis.net/wfs/2.0" xmlns:fes="http://www.opengis.net/fes/2.0" xmlns:myns="http://www.someserver.com/myns" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.opengis.net/wfs/2.0 http://schemas.opengis.net/wfs/2.0.0/wfs.xsd" count="{self.count}" startindex="{start_index}"><wfs:Query typeNames="{typenames}">{spatial_filter}</wfs:Query></wfs:GetFeature>'
        return _filter

    def get_features(self,
                     typenames: str,
                     spatial_operator: str = "Intersects",
                     filter_srid: str = None,
                     start_index: int = 0,
                     rerun_on_exception: bool = True
                     ):
        """ Fetches features from the WFS using POST

        POST is required since GET has a character limit around 4000. Having a larger filter would result in errors,
        which do not occur in case of POST.

        Args:
            typenames (str): References to parameter 'typenames' in a WFS GetFeature request
            spatial_operator (str): Defines the spatial operation for filtering
            filter_srid (str): Defines the spatial reference system, the geometry shall be transformed into for filtering
            start_index (str): References to parameter 'startindex' in a

        Returns:
            features (list): A list of returned features
        """
        features = []
        while start_index is not None:
            post_body = self._create_post_data(
                spatial_operator,
                filter_srid,
                typenames,
                start_index
            )
            response = requests.post(
                url=self.base_url,
                data=post_body,
                auth=self.auth_digest_obj,
                proxies=PROXIES,
            )

            content = response.content.decode("utf-8")
            content = xmltodict.parse(content)
            collection = content.get(
                "wfs:FeatureCollection",
                {},
            )

            # Check if collection is an exception and does not contain the requested data
            if len(collection) == 0:
                exception = content.get(
                    "ows:ExceptionReport",
                    {}
                )
                if len(exception) > 0 and rerun_on_exception:
                    # Wait a second before another try
                    sleep(1)
                    self.get_features(
                        typenames,
                        spatial_operator,
                        filter_srid,
                        start_index,
                        rerun_on_exception=False
                    )

            members = collection.get(
                "wfs:member",
                [],
            )
            if len(members) > 1:
                features += members
            else:
                features += [members]

            if collection.get("@next", None) is not None:
                start_index += self.count
            else:
                start_index = None

        return features