11282283562

Committed 10 Oct 2024 09:29PM UTC coverage: 60.113% (-0.02%) from 60.128%

Build # 11282283562

Build Type

Pull #949

github

Committed by

web-flow

Commit Message

Merge 1058b0b85 into 852fe6d4b

Pull Request Pull Request #949: Remove dependency on the Pytz library

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

82 existing lines in 4 files now uncovered.

8533 of 14195 relevant lines covered (60.11%)

0.6 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

61.98

/owslib/feature/wfs200.py

# =============================================================================
# OWSLib. Copyright (C) 2005 Sean C. Gillies
#
# Contact email: sgillies@frii.com
#
# $Id: wfs.py 503 2006-02-01 17:09:12Z dokai $
# =============================================================================

from io import BytesIO
import logging
from urllib.parse import urlencode

# owslib imports:
from owslib import util
from owslib.fgdc import Metadata
from owslib.iso import MD_Metadata
from owslib.iso3 import MD_Metadata as MD_Metadata3  # ISO 19115 Part 3 XML
from owslib.ows import Constraint, ServiceIdentification, ServiceProvider, OperationsMetadata
from owslib.etree import etree
from owslib.util import nspath, testXMLValue, openURL, Authentication
from owslib.crs import Crs
from owslib.feature import WebFeatureService_
from owslib.feature.common import (
    WFSCapabilitiesReader,
    AbstractContentMetadata
)
from owslib.namespaces import Namespaces

LOGGER = logging.getLogger(__name__)

n = Namespaces()
WFS_NAMESPACE = n.get_namespace("wfs20")
OWS_NAMESPACE = n.get_namespace("ows110")
OGC_NAMESPACE = n.get_namespace("ogc")
GML_NAMESPACE = n.get_namespace("gml")
FES_NAMESPACE = n.get_namespace("fes")


class ServiceException(Exception):
    pass


class WebFeatureService_2_0_0(WebFeatureService_):
    """Abstraction for OGC Web Feature Service (WFS).

    Implements IWebFeatureService.
    """

    def __new__(
        self,
        url,
        version,
        xml,
        parse_remote_metadata=False,
        timeout=30,
        headers=None,
        username=None,
        password=None,
        auth=None,
    ):
        """ overridden __new__ method

        @type url: string
        @param url: url of WFS capabilities document
        @type xml: string
        @param xml: elementtree object
        @type parse_remote_metadata: boolean
        @param parse_remote_metadata: whether to fully process MetadataURL elements
        @param headers: HTTP headers to send with requests
        @param timeout: time (in seconds) after which requests should timeout
        @param username: service authentication username
        @param password: service authentication password
        @param auth: instance of owslib.util.Authentication
        @return: initialized WebFeatureService_2_0_0 object
        """
        obj = object.__new__(self)
        obj.__init__(
            url,
            version,
            xml,
            parse_remote_metadata,
            timeout,
            headers=headers,
            username=username,
            password=password,
            auth=auth,
        )
        return obj

    def __getitem__(self, name):
        """ check contents dictionary to allow dict like access to service layers"""
        if name in list(self.__getattribute__("contents").keys()):
            return self.__getattribute__("contents")[name]
        else:
            raise KeyError("No content named %s" % name)

    def __init__(
        self,
        url,
        version,
        xml=None,
        parse_remote_metadata=False,
        timeout=30,
        headers=None,
        username=None,
        password=None,
        auth=None,
    ):
        """Initialize."""
        if auth:
            if username:
                auth.username = username
            if password:
                auth.password = password
        else:
            auth = Authentication()
        super(WebFeatureService_2_0_0, self).__init__(auth)
        LOGGER.debug("building WFS %s" % url)
        self.url = url
        self.version = version
        self.timeout = timeout
        self.headers = headers
        self._capabilities = None
        reader = WFSCapabilitiesReader(self.version, headers=self.headers, auth=self.auth)
        if xml:
            self._capabilities = reader.readString(xml)
        else:
            self._capabilities = reader.read(self.url, self.timeout)
        self._buildMetadata(parse_remote_metadata)

    def _buildMetadata(self, parse_remote_metadata=False):
        """set up capabilities metadata objects: """

        self.updateSequence = self._capabilities.attrib.get("updateSequence")

        # serviceIdentification metadata
        serviceidentelem = self._capabilities.find(nspath("ServiceIdentification"))
        if serviceidentelem is not None:
            self.identification = ServiceIdentification(serviceidentelem)
        # need to add to keywords list from featuretypelist information:
        featuretypelistelem = self._capabilities.find(
            nspath("FeatureTypeList", ns=WFS_NAMESPACE)
        )
        if featuretypelistelem is not None:
            featuretypeelems = featuretypelistelem.findall(
                nspath("FeatureType", ns=WFS_NAMESPACE)
            )
            if serviceidentelem is not None:
                for f in featuretypeelems:
                    kwds = f.findall(nspath("Keywords/Keyword", ns=OWS_NAMESPACE))
                    if kwds is not None:
                        for kwd in kwds[:]:
                            if kwd.text not in self.identification.keywords:
                                self.identification.keywords.append(kwd.text)

        # TODO: update serviceProvider metadata, miss it out for now
        serviceproviderelem = self._capabilities.find(nspath("ServiceProvider"))
        if serviceproviderelem is not None:
            self.provider = ServiceProvider(serviceproviderelem)

        # serviceOperations metadata
        self.operations = []

        for elem in self._capabilities.find(nspath("OperationsMetadata"))[:]:
            if elem.tag != nspath("ExtendedCapabilities"):
                self.operations.append(OperationsMetadata(elem))
        self.constraints = {}
        for elem in self._capabilities.findall(
            nspath("OperationsMetadata/Constraint", ns=WFS_NAMESPACE)
        ):
            self.constraints[elem.attrib["name"]] = Constraint(
                elem, self.owscommon.namespace
            )
        self.parameters = {}
        for elem in self._capabilities.findall(
            nspath("OperationsMetadata/Parameter", ns=WFS_NAMESPACE)
        ):
            self.parameters[elem.attrib["name"]] = Parameter(
                elem, self.owscommon.namespace
            )

        # serviceContents metadata: our assumption is that services use a top-level
        # layer as a metadata organizer, nothing more.

        self.contents = {}
        featuretypelist = self._capabilities.find(
            nspath("FeatureTypeList", ns=WFS_NAMESPACE)
        )
        features = self._capabilities.findall(
            nspath("FeatureTypeList/FeatureType", ns=WFS_NAMESPACE)
        )
        for feature in features:
            cm = ContentMetadata(
                feature, featuretypelist, parse_remote_metadata, auth=self.auth
            )
            self.contents[cm.id] = cm

        # exceptions
        self.exceptions = [
            f.text for f in self._capabilities.findall("Capability/Exception/Format")
        ]

    def getcapabilities(self):
        """Request and return capabilities document from the WFS as a
        file-like object.
        NOTE: this is effectively redundant now"""
        reader = WFSCapabilitiesReader(self.version, auth=self.auth)
        return openURL(
            reader.capabilities_url(self.url), timeout=self.timeout,
            headers=self.headers, auth=self.auth
        )

    def items(self):
        """supports dict-like items() access"""
        items = []
        for item in self.contents:
            items.append((item, self.contents[item]))
        return items

    def getfeature(
        self,
        typename=None,
        filter=None,
        bbox=None,
        featureid=None,
        featureversion=None,
        propertyname=None,
        maxfeatures=None,
        srsname=None,
        storedQueryID=None,
        storedQueryParams=None,
        method="Get",
        outputFormat=None,
        startindex=None,
        sortby=None,
    ):
        """Request and return feature data as a file-like object.

        #TODO: NOTE: have changed property name from ['*'] to None - check the use of this in WFS 2.0

        Parameters
        ----------
        typename : list
            List of typenames (string)
        filter : string
            XML-encoded OGC filter expression.
        bbox : tuple
            (left, bottom, right, top) in the feature type's coordinates == (minx, miny, maxx, maxy)
        featureid : list
            List of unique feature ids (string)
        featureversion : string
            Default is most recent feature version.
        propertyname : list
            List of feature property names. For Get request, '*' matches all.
            For Post request, leave blank (None) to get all properties.
        maxfeatures : int
            Maximum number of features to be returned.
        srsname: string
            EPSG code to request the data in
        storedQueryID : string
            A name identifying a prepared set available in WFS-service
        storedQueryParams : dict
            Variable amount of extra information sent to server related to
            storedQueryID to further define the requested data
            {'parameter_name': parameter_value}
        method : string
            Qualified name of the HTTP DCP method to use.
        outputFormat: string (optional)
            Requested response format of the request.
        startindex: int (optional)
            Start position to return feature set (paging in combination with maxfeatures)
        sortby: list (optional)
            List of property names whose values should be used to order
            (upon presentation) the set of feature instances that
            satify the query.

        There are 5 different modes of use

        1) typename and bbox (simple spatial query)
        2) typename and filter (==query) (more expressive)
        3) featureid (direct access to known features)
        4) storedQueryID and optional storedQueryParams
        5) filter only via Post method

        Raises:
            ServiceException: If there is an error during the request

        Returns:
            BytesIO -- Data returned from the service as a file-like object
        """
        storedQueryParams = storedQueryParams or {}
        url = data = None
        if typename and type(typename) == type(""):  # noqa: E721
            typename = [typename]
        if method.upper() == "GET":
            (url) = self.getGETGetFeatureRequest(
                typename,
                filter,
                bbox,
                featureid,
                featureversion,
                propertyname,
                maxfeatures,
                srsname,
                storedQueryID,
                storedQueryParams,
                outputFormat,
                "Get",
                startindex,
                sortby,
            )
            LOGGER.debug("GetFeature WFS GET url %s" % url)
        else:
            url, data = self.getPOSTGetFeatureRequest(
                typename,
                filter,
                bbox,
                featureid,
                featureversion,
                propertyname,
                maxfeatures,
                storedQueryID,
                storedQueryParams,
                outputFormat,
                "Post",
                startindex,
                sortby)

        u = openURL(url, data, method, timeout=self.timeout, headers=self.headers, auth=self.auth)

        # check for service exceptions, rewrap, and return
        # We're going to assume that anything with a content-length > 32k
        # is data. We'll check anything smaller.
        if "Content-Length" in u.info():
            length = int(u.info()["Content-Length"])
            have_read = False
        else:
            data = u.read()
            have_read = True
            length = len(data)

        if length < 32000:
            if not have_read:
                data = u.read()

            try:
                tree = etree.fromstring(data)
            except BaseException:
                # Not XML
                return BytesIO(data)
            else:
                if tree.tag == "{%s}ServiceExceptionReport" % OGC_NAMESPACE:
                    se = tree.find(nspath("ServiceException", OGC_NAMESPACE))
                    raise ServiceException(str(se.text).strip())
                else:
                    return BytesIO(data)
        else:
            if have_read:
                return BytesIO(data)
            return u

    def getpropertyvalue(
        self,
        query=None,
        storedquery_id=None,
        valuereference=None,
        typename=None,
        method=nspath("Get"),
        **kwargs
    ):
        """ the WFS GetPropertyValue method"""
        try:
            base_url = next(
                (
                    m.get("url")
                    for m in self.getOperationByName("GetPropertyValue").methods
                    if m.get("type").lower() == method.lower()
                )
            )
        except StopIteration:
            base_url = self.url
        request = {
            "service": "WFS",
            "version": self.version,
            "request": "GetPropertyValue",
        }
        if query:
            request["query"] = str(query)
        if valuereference:
            request["valueReference"] = str(valuereference)
        if storedquery_id:
            request["storedQuery_id"] = str(storedquery_id)
        if typename:
            request["typename"] = str(typename)
        if kwargs:
            for kw in list(kwargs.keys()):
                request[kw] = str(kwargs[kw])
        encoded_request = urlencode(request)
        u = openURL(base_url + encoded_request, timeout=self.timeout, headers=self.headers, auth=self.auth)
        return u.read()

    def _getStoredQueries(self):
        """ gets descriptions of the stored queries available on the server """
        sqs = []
        # This method makes two calls to the WFS - one ListStoredQueries, and one DescribeStoredQueries.
        # The information is then aggregated in 'StoredQuery' objects
        method = nspath("Get")

        # first make the ListStoredQueries response and save the results in a dictionary
        # if form {storedqueryid:(title, returnfeaturetype)}
        try:
            base_url = next(
                (
                    m.get("url")
                    for m in self.getOperationByName("ListStoredQueries").methods
                    if m.get("type").lower() == method.lower()
                )
            )
        except StopIteration:
            base_url = self.url

        request = {
            "service": "WFS",
            "version": self.version,
            "request": "ListStoredQueries",
        }
        encoded_request = urlencode(request)
        u = openURL(
            base_url, data=encoded_request, timeout=self.timeout, headers=self.headers, auth=self.auth
        )
        tree = etree.fromstring(u.read())
        tempdict = {}
        for sqelem in tree[:]:
            title = rft = id = None
            id = sqelem.get("id")
            for elem in sqelem[:]:
                if elem.tag == nspath("Title", WFS_NAMESPACE):
                    title = elem.text
                elif elem.tag == nspath("ReturnFeatureType", WFS_NAMESPACE):
                    rft = elem.text
            tempdict[id] = (title, rft)  # store in temporary dictionary

        # then make the DescribeStoredQueries request and get the rest of the information about the stored queries
        try:
            base_url = next(
                (
                    m.get("url")
                    for m in self.getOperationByName("DescribeStoredQueries").methods
                    if m.get("type").lower() == method.lower()
                )
            )
        except StopIteration:
            base_url = self.url
        request = {
            "service": "WFS",
            "version": self.version,
            "request": "DescribeStoredQueries",
        }
        encoded_request = urlencode(request)
        u = openURL(
            base_url, data=encoded_request, timeout=self.timeout, headers=self.headers, auth=self.auth
        )
        tree = etree.fromstring(u.read())
        tempdict2 = {}
        for sqelem in tree[:]:
            params = []  # list to store parameters for the stored query description
            id = sqelem.get("id")
            for elem in sqelem[:]:
                abstract = ""
                if elem.tag == nspath("Abstract", WFS_NAMESPACE):
                    abstract = elem.text
                elif elem.tag == nspath("Parameter", WFS_NAMESPACE):
                    newparam = Parameter(elem.get("name"), elem.get("type"))
                    params.append(newparam)
            tempdict2[id] = (abstract, params)  # store in another temporary dictionary

        # now group the results into StoredQuery objects:
        for key in list(tempdict.keys()):
            sqs.append(
                StoredQuery(
                    key,
                    tempdict[key][0],
                    tempdict[key][1],
                    tempdict2[key][0],
                    tempdict2[key][1],
                )
            )
        return sqs

    storedqueries = property(_getStoredQueries, None)

    def getOperationByName(self, name):
        """Return a named content item."""
        for item in self.operations:
            if item.name == name:
                return item
        raise KeyError("No operation named %s" % name)


class StoredQuery(object):
    """' Class to describe a storedquery """

    def __init__(self, id, title, returntype, abstract, parameters):
        self.id = id
        self.title = title
        self.returnfeaturetype = returntype
        self.abstract = abstract
        self.parameters = parameters


class Parameter(object):
    def __init__(self, name, type):
        self.name = name
        self.type = type


class ContentMetadata(AbstractContentMetadata):
    """Abstraction for WFS metadata.

    Implements IMetadata.
    """

    def __init__(
        self, elem, parent, parse_remote_metadata=False, timeout=30, headers=None, auth=None
    ):
        """."""
        super(ContentMetadata, self).__init__(headers=headers, auth=auth)
        self.id = elem.find(nspath("Name", ns=WFS_NAMESPACE)).text
        self.title = elem.find(nspath("Title", ns=WFS_NAMESPACE)).text
        abstract = elem.find(nspath("Abstract", ns=WFS_NAMESPACE))
        if abstract is not None:
            self.abstract = abstract.text
        else:
            self.abstract = None
        self.keywords = [
            f.text for f in elem.findall(nspath("Keywords/Keyword", ns=OWS_NAMESPACE))
        ]

        # bboxes
        self.boundingBoxWGS84 = None
        b = elem.find(nspath("WGS84BoundingBox", ns=OWS_NAMESPACE))
        if b is not None:
            try:
                lc = b.find(nspath("LowerCorner", ns=OWS_NAMESPACE))
                uc = b.find(nspath("UpperCorner", ns=OWS_NAMESPACE))
                ll = [float(s) for s in lc.text.split()]
                ur = [float(s) for s in uc.text.split()]
                self.boundingBoxWGS84 = (ll[0], ll[1], ur[0], ur[1])

                # there is no such think as bounding box
                # make copy of the WGS84BoundingBox
                self.boundingBox = (
                    self.boundingBoxWGS84[0],
                    self.boundingBoxWGS84[1],
                    self.boundingBoxWGS84[2],
                    self.boundingBoxWGS84[3],
                    Crs("epsg:4326"),
                )
            except AttributeError:
                self.boundingBoxWGS84 = None
        # crs options
        self.crsOptions = [
            Crs(srs.text) for srs in elem.findall(nspath("OtherCRS", ns=WFS_NAMESPACE))
        ]
        defaultCrs = elem.findall(nspath("DefaultCRS", ns=WFS_NAMESPACE))
        if len(defaultCrs) > 0:
            self.crsOptions.insert(0, Crs(defaultCrs[0].text))

        # verbs
        self.verbOptions = [
            op.tag for op in parent.findall(nspath("Operations/*", ns=WFS_NAMESPACE))
        ]
        self.verbOptions + [
            op.tag
            for op in elem.findall(nspath("Operations/*", ns=WFS_NAMESPACE))
            if op.tag not in self.verbOptions
        ]

        # others not used but needed for iContentMetadata harmonisation
        self.styles = None
        self.timepositions = None
        self.defaulttimeposition = None

        # MetadataURLs
        self.metadataUrls = []
        for m in elem.findall(nspath("MetadataURL", ns=WFS_NAMESPACE)):
            metadataUrl = {
                "url": testXMLValue(
                    m.attrib["{http://www.w3.org/1999/xlink}href"], attrib=True
                )
            }
            self.metadataUrls.append(metadataUrl)

        if parse_remote_metadata:
            self.parse_remote_metadata(timeout)

    def parse_remote_metadata(self, timeout=30):
        """Parse remote metadata for MetadataURL and add it as metadataUrl['metadata']"""
        for metadataUrl in self.metadataUrls:
            if metadataUrl["url"] is not None:
                try:
                    content = openURL(metadataUrl["url"], timeout=timeout, headers=self.headers, auth=self.auth)
                    doc = etree.fromstring(content.read())

                    mdelem = doc.find(".//metadata")
                    if mdelem is not None:
                        metadataUrl["metadata"] = Metadata(mdelem)
                        continue

                    mdelem = doc.find(
                        ".//" + util.nspath_eval("gmd:MD_Metadata", n.get_namespaces(["gmd"]))
                    ) or doc.find(
                        ".//" + util.nspath_eval("gmi:MI_Metadata", n.get_namespaces(["gmi"]))
                    )
                    if mdelem is not None:
                        metadataUrl["metadata"] = MD_Metadata(mdelem)
                        continue
                    else:  # ISO 19115 Part 3 XML
                        mdelem = MD_Metadata3.find_start(doc)
                        if mdelem is not None:
                            metadataUrl["metadata"] = MD_Metadata3(mdelem)
                        else:
                            metadataUrl["metadata"] = None
                        continue
                except Exception:
                    metadataUrl["metadata"] = None

1	# =============================================================================
2	# OWSLib. Copyright (C) 2005 Sean C. Gillies
3	#
4	# Contact email: sgillies@frii.com
5	#
6	# $Id: wfs.py 503 2006-02-01 17:09:12Z dokai $
7	# =============================================================================
8
9	from io import BytesIO	1✔
10	import logging	1✔
11	from urllib.parse import urlencode	1✔
12
13	# owslib imports:
14	from owslib import util	1✔
15	from owslib.fgdc import Metadata	1✔
16	from owslib.iso import MD_Metadata	1✔
17	from owslib.iso3 import MD_Metadata as MD_Metadata3 # ISO 19115 Part 3 XML	1✔
18	from owslib.ows import Constraint, ServiceIdentification, ServiceProvider, OperationsMetadata	1✔
19	from owslib.etree import etree	1✔
20	from owslib.util import nspath, testXMLValue, openURL, Authentication	1✔
21	from owslib.crs import Crs	1✔
22	from owslib.feature import WebFeatureService_	1✔
23	from owslib.feature.common import (	1✔
24	WFSCapabilitiesReader,
25	AbstractContentMetadata
26	)
27	from owslib.namespaces import Namespaces	1✔
28
29	LOGGER = logging.getLogger(__name__)	1✔
30
31	n = Namespaces()	1✔
32	WFS_NAMESPACE = n.get_namespace("wfs20")	1✔
33	OWS_NAMESPACE = n.get_namespace("ows110")	1✔
34	OGC_NAMESPACE = n.get_namespace("ogc")	1✔
35	GML_NAMESPACE = n.get_namespace("gml")	1✔
36	FES_NAMESPACE = n.get_namespace("fes")	1✔
37
38
39	class ServiceException(Exception):	1✔
40	pass	1✔
41
42
43	class WebFeatureService_2_0_0(WebFeatureService_):	1✔
44	"""Abstraction for OGC Web Feature Service (WFS).
45
46	Implements IWebFeatureService.
47	"""
48
49	def __new__(	1✔
50	self,
51	url,
52	version,
53	xml,
54	parse_remote_metadata=False,
55	timeout=30,
56	headers=None,
57	username=None,
58	password=None,
59	auth=None,
60	):
61	""" overridden __new__ method
62
63	@type url: string
64	@param url: url of WFS capabilities document
65	@type xml: string
66	@param xml: elementtree object
67	@type parse_remote_metadata: boolean
68	@param parse_remote_metadata: whether to fully process MetadataURL elements
69	@param headers: HTTP headers to send with requests
70	@param timeout: time (in seconds) after which requests should timeout
71	@param username: service authentication username
72	@param password: service authentication password
73	@param auth: instance of owslib.util.Authentication
74	@return: initialized WebFeatureService_2_0_0 object
75	"""
76	obj = object.__new__(self)	1✔
77	obj.__init__(	1✔
78	url,
79	version,
80	xml,
81	parse_remote_metadata,
82	timeout,
83	headers=headers,
84	username=username,
85	password=password,
86	auth=auth,
87	)
88	return obj	1✔
89
90	def __getitem__(self, name):	1✔
91	""" check contents dictionary to allow dict like access to service layers"""
92	if name in list(self.__getattribute__("contents").keys()):	×
93	return self.__getattribute__("contents")[name]	×
94	else:
95	raise KeyError("No content named %s" % name)	×
96
97	def __init__(	1✔
98	self,
99	url,
100	version,
101	xml=None,
102	parse_remote_metadata=False,
103	timeout=30,
104	headers=None,
105	username=None,
106	password=None,
107	auth=None,
108	):
109	"""Initialize."""
110	if auth:	1✔
111	if username:	1✔
112	auth.username = username	×
113	if password:	1✔
114	auth.password = password	×
115	else:
116	auth = Authentication()	×
117	super(WebFeatureService_2_0_0, self).__init__(auth)	1✔
118	LOGGER.debug("building WFS %s" % url)	1✔
119	self.url = url	1✔
120	self.version = version	1✔
121	self.timeout = timeout	1✔
122	self.headers = headers	1✔
123	self._capabilities = None	1✔
124	reader = WFSCapabilitiesReader(self.version, headers=self.headers, auth=self.auth)	1✔
125	if xml:	1✔
126	self._capabilities = reader.readString(xml)	×
127	else:
128	self._capabilities = reader.read(self.url, self.timeout)	1✔
129	self._buildMetadata(parse_remote_metadata)	1✔
130
131	def _buildMetadata(self, parse_remote_metadata=False):	1✔
132	"""set up capabilities metadata objects: """
133
134	self.updateSequence = self._capabilities.attrib.get("updateSequence")	1✔
135
136	# serviceIdentification metadata
137	serviceidentelem = self._capabilities.find(nspath("ServiceIdentification"))	1✔
138	if serviceidentelem is not None:	1✔
139	self.identification = ServiceIdentification(serviceidentelem)	1✔
140	# need to add to keywords list from featuretypelist information:
141	featuretypelistelem = self._capabilities.find(	1✔
142	nspath("FeatureTypeList", ns=WFS_NAMESPACE)
143	)
144	if featuretypelistelem is not None:	1✔
145	featuretypeelems = featuretypelistelem.findall(	1✔
146	nspath("FeatureType", ns=WFS_NAMESPACE)
147	)
148	if serviceidentelem is not None:	1✔
149	for f in featuretypeelems:	1✔
150	kwds = f.findall(nspath("Keywords/Keyword", ns=OWS_NAMESPACE))	1✔
151	if kwds is not None:	1✔
152	for kwd in kwds[:]:	1✔
153	if kwd.text not in self.identification.keywords:	1✔
154	self.identification.keywords.append(kwd.text)	1✔
155
156	# TODO: update serviceProvider metadata, miss it out for now
157	serviceproviderelem = self._capabilities.find(nspath("ServiceProvider"))	1✔
158	if serviceproviderelem is not None:	1✔
159	self.provider = ServiceProvider(serviceproviderelem)	1✔
160
161	# serviceOperations metadata
162	self.operations = []	1✔
163
164	for elem in self._capabilities.find(nspath("OperationsMetadata"))[:]:	1✔
165	if elem.tag != nspath("ExtendedCapabilities"):	1✔
166	self.operations.append(OperationsMetadata(elem))	1✔
167	self.constraints = {}	1✔
168	for elem in self._capabilities.findall(	1✔
169	nspath("OperationsMetadata/Constraint", ns=WFS_NAMESPACE)
170	):
171	self.constraints[elem.attrib["name"]] = Constraint(	×
172	elem, self.owscommon.namespace
173	)
174	self.parameters = {}	1✔
175	for elem in self._capabilities.findall(	1✔
176	nspath("OperationsMetadata/Parameter", ns=WFS_NAMESPACE)
177	):
178	self.parameters[elem.attrib["name"]] = Parameter(	×
179	elem, self.owscommon.namespace
180	)
181
182	# serviceContents metadata: our assumption is that services use a top-level
183	# layer as a metadata organizer, nothing more.
184
185	self.contents = {}	1✔
186	featuretypelist = self._capabilities.find(	1✔
187	nspath("FeatureTypeList", ns=WFS_NAMESPACE)
188	)
189	features = self._capabilities.findall(	1✔
190	nspath("FeatureTypeList/FeatureType", ns=WFS_NAMESPACE)
191	)
192	for feature in features:	1✔
193	cm = ContentMetadata(	1✔
194	feature, featuretypelist, parse_remote_metadata, auth=self.auth
195	)
196	self.contents[cm.id] = cm	1✔
197
198	# exceptions
199	self.exceptions = [	1✔
200	f.text for f in self._capabilities.findall("Capability/Exception/Format")
201	]
202
203	def getcapabilities(self):	1✔
204	"""Request and return capabilities document from the WFS as a
205	file-like object.
206	NOTE: this is effectively redundant now"""
207	reader = WFSCapabilitiesReader(self.version, auth=self.auth)	×
208	return openURL(	×
209	reader.capabilities_url(self.url), timeout=self.timeout,
210	headers=self.headers, auth=self.auth
211	)
212
213	def items(self):	1✔
214	"""supports dict-like items() access"""
215	items = []	×
216	for item in self.contents:	×
217	items.append((item, self.contents[item]))	×
218	return items	×
219
220	def getfeature(	1✔
221	self,
222	typename=None,
223	filter=None,
224	bbox=None,
225	featureid=None,
226	featureversion=None,
227	propertyname=None,
228	maxfeatures=None,
229	srsname=None,
230	storedQueryID=None,
231	storedQueryParams=None,
232	method="Get",
233	outputFormat=None,
234	startindex=None,
235	sortby=None,
236	):
237	"""Request and return feature data as a file-like object.
238
239	#TODO: NOTE: have changed property name from ['*'] to None - check the use of this in WFS 2.0
240
241	Parameters
242	----------
243	typename : list
244	List of typenames (string)
245	filter : string
246	XML-encoded OGC filter expression.
247	bbox : tuple
248	(left, bottom, right, top) in the feature type's coordinates == (minx, miny, maxx, maxy)
249	featureid : list
250	List of unique feature ids (string)
251	featureversion : string
252	Default is most recent feature version.
253	propertyname : list
254	List of feature property names. For Get request, '*' matches all.
255	For Post request, leave blank (None) to get all properties.
256	maxfeatures : int
257	Maximum number of features to be returned.
258	srsname: string
259	EPSG code to request the data in
260	storedQueryID : string
261	A name identifying a prepared set available in WFS-service
262	storedQueryParams : dict
263	Variable amount of extra information sent to server related to
264	storedQueryID to further define the requested data
265	{'parameter_name': parameter_value}
266	method : string
267	Qualified name of the HTTP DCP method to use.
268	outputFormat: string (optional)
269	Requested response format of the request.
270	startindex: int (optional)
271	Start position to return feature set (paging in combination with maxfeatures)
272	sortby: list (optional)
273	List of property names whose values should be used to order
274	(upon presentation) the set of feature instances that
275	satify the query.
276
277	There are 5 different modes of use
278
279	1) typename and bbox (simple spatial query)
280	2) typename and filter (==query) (more expressive)
281	3) featureid (direct access to known features)
282	4) storedQueryID and optional storedQueryParams
283	5) filter only via Post method
284
285	Raises:
286	ServiceException: If there is an error during the request
287
288	Returns:
289	BytesIO -- Data returned from the service as a file-like object
290	"""
291	storedQueryParams = storedQueryParams or {}	1✔
292	url = data = None	1✔
293	if typename and type(typename) == type(""): # noqa: E721	1✔
294	typename = [typename]	1✔
295	if method.upper() == "GET":	1✔
296	(url) = self.getGETGetFeatureRequest(	1✔
297	typename,
298	filter,
299	bbox,
300	featureid,
301	featureversion,
302	propertyname,
303	maxfeatures,
304	srsname,
305	storedQueryID,
306	storedQueryParams,
307	outputFormat,
308	"Get",
309	startindex,
310	sortby,
311	)
312	LOGGER.debug("GetFeature WFS GET url %s" % url)	1✔
313	else:
UNCOV 314	url, data = self.getPOSTGetFeatureRequest(	×
315	typename,
316	filter,
317	bbox,
318	featureid,
319	featureversion,
320	propertyname,
321	maxfeatures,
322	storedQueryID,
323	storedQueryParams,
324	outputFormat,
325	"Post",
326	startindex,
327	sortby)
328
329	u = openURL(url, data, method, timeout=self.timeout, headers=self.headers, auth=self.auth)	1✔
330
331	# check for service exceptions, rewrap, and return
332	# We're going to assume that anything with a content-length > 32k
333	# is data. We'll check anything smaller.
334	if "Content-Length" in u.info():	1✔
UNCOV 335	length = int(u.info()["Content-Length"])	×
UNCOV 336	have_read = False	×
337	else:
338	data = u.read()	1✔
339	have_read = True	1✔
340	length = len(data)	1✔
341
342	if length < 32000:	1✔
343	if not have_read:	1✔
UNCOV 344	data = u.read()	×
345
346	try:	1✔
347	tree = etree.fromstring(data)	1✔
348	except BaseException:	1✔
349	# Not XML
350	return BytesIO(data)	1✔
351	else:
352	if tree.tag == "{%s}ServiceExceptionReport" % OGC_NAMESPACE:	×
UNCOV 353	se = tree.find(nspath("ServiceException", OGC_NAMESPACE))	×
UNCOV 354	raise ServiceException(str(se.text).strip())	×
355	else:
356	return BytesIO(data)	×
357	else:
358	if have_read:	1✔
359	return BytesIO(data)	1✔
UNCOV 360	return u	×
361
362	def getpropertyvalue(	1✔
363	self,
364	query=None,
365	storedquery_id=None,
366	valuereference=None,
367	typename=None,
368	method=nspath("Get"),
369	**kwargs
370	):
371	""" the WFS GetPropertyValue method"""
UNCOV 372	try:	×
UNCOV 373	base_url = next(	×
374	(
375	m.get("url")
376	for m in self.getOperationByName("GetPropertyValue").methods
377	if m.get("type").lower() == method.lower()
378	)
379	)
UNCOV 380	except StopIteration:	×
UNCOV 381	base_url = self.url	×
UNCOV 382	request = {	×
383	"service": "WFS",
384	"version": self.version,
385	"request": "GetPropertyValue",
386	}
387	if query:	×
388	request["query"] = str(query)	×
389	if valuereference:	×
390	request["valueReference"] = str(valuereference)	×
391	if storedquery_id:	×
392	request["storedQuery_id"] = str(storedquery_id)	×
393	if typename:	×
394	request["typename"] = str(typename)	×
395	if kwargs:	×
396	for kw in list(kwargs.keys()):	×
UNCOV 397	request[kw] = str(kwargs[kw])	×
UNCOV 398	encoded_request = urlencode(request)	×
UNCOV 399	u = openURL(base_url + encoded_request, timeout=self.timeout, headers=self.headers, auth=self.auth)	×
400	return u.read()	×
401
402	def _getStoredQueries(self):	1✔
403	""" gets descriptions of the stored queries available on the server """
UNCOV 404	sqs = []	×
405	# This method makes two calls to the WFS - one ListStoredQueries, and one DescribeStoredQueries.
406	# The information is then aggregated in 'StoredQuery' objects
407	method = nspath("Get")	×
408
409	# first make the ListStoredQueries response and save the results in a dictionary
410	# if form {storedqueryid:(title, returnfeaturetype)}
UNCOV 411	try:	×
UNCOV 412	base_url = next(	×
413	(
414	m.get("url")
415	for m in self.getOperationByName("ListStoredQueries").methods
416	if m.get("type").lower() == method.lower()
417	)
418	)
UNCOV 419	except StopIteration:	×
UNCOV 420	base_url = self.url	×
421
UNCOV 422	request = {	×
423	"service": "WFS",
424	"version": self.version,
425	"request": "ListStoredQueries",
426	}
427	encoded_request = urlencode(request)	×
428	u = openURL(	×
429	base_url, data=encoded_request, timeout=self.timeout, headers=self.headers, auth=self.auth
430	)
431	tree = etree.fromstring(u.read())	×
432	tempdict = {}	×
433	for sqelem in tree[:]:	×
434	title = rft = id = None	×
435	id = sqelem.get("id")	×
436	for elem in sqelem[:]:	×
437	if elem.tag == nspath("Title", WFS_NAMESPACE):	×
UNCOV 438	title = elem.text	×
UNCOV 439	elif elem.tag == nspath("ReturnFeatureType", WFS_NAMESPACE):	×
440	rft = elem.text	×
441	tempdict[id] = (title, rft) # store in temporary dictionary	×
442
443	# then make the DescribeStoredQueries request and get the rest of the information about the stored queries
UNCOV 444	try:	×
UNCOV 445	base_url = next(	×
446	(
447	m.get("url")
448	for m in self.getOperationByName("DescribeStoredQueries").methods
449	if m.get("type").lower() == method.lower()
450	)
451	)
UNCOV 452	except StopIteration:	×
UNCOV 453	base_url = self.url	×
UNCOV 454	request = {	×
455	"service": "WFS",
456	"version": self.version,
457	"request": "DescribeStoredQueries",
458	}
459	encoded_request = urlencode(request)	×
460	u = openURL(	×
461	base_url, data=encoded_request, timeout=self.timeout, headers=self.headers, auth=self.auth
462	)
463	tree = etree.fromstring(u.read())	×
464	tempdict2 = {}	×
465	for sqelem in tree[:]:	×
466	params = [] # list to store parameters for the stored query description	×
467	id = sqelem.get("id")	×
468	for elem in sqelem[:]:	×
469	abstract = ""	×
470	if elem.tag == nspath("Abstract", WFS_NAMESPACE):	×
471	abstract = elem.text	×
UNCOV 472	elif elem.tag == nspath("Parameter", WFS_NAMESPACE):	×
UNCOV 473	newparam = Parameter(elem.get("name"), elem.get("type"))	×
474	params.append(newparam)	×
475	tempdict2[id] = (abstract, params) # store in another temporary dictionary	×
476
477	# now group the results into StoredQuery objects:
UNCOV 478	for key in list(tempdict.keys()):	×
UNCOV 479	sqs.append(	×
480	StoredQuery(
481	key,
482	tempdict[key][0],
483	tempdict[key][1],
484	tempdict2[key][0],
485	tempdict2[key][1],
486	)
487	)
UNCOV 488	return sqs	×
489
490	storedqueries = property(_getStoredQueries, None)	1✔
491
492	def getOperationByName(self, name):	1✔
493	"""Return a named content item."""
494	for item in self.operations:	1✔
495	if item.name == name:	1✔
496	return item	1✔
UNCOV 497	raise KeyError("No operation named %s" % name)	×
498
499
500	class StoredQuery(object):	1✔
501	"""' Class to describe a storedquery """
502
503	def __init__(self, id, title, returntype, abstract, parameters):	1✔
504	self.id = id	×
UNCOV 505	self.title = title	×
UNCOV 506	self.returnfeaturetype = returntype	×
UNCOV 507	self.abstract = abstract	×
UNCOV 508	self.parameters = parameters	×
509
510
511	class Parameter(object):	1✔
512	def __init__(self, name, type):	1✔
UNCOV 513	self.name = name	×
UNCOV 514	self.type = type	×
515
516
517	class ContentMetadata(AbstractContentMetadata):	1✔
518	"""Abstraction for WFS metadata.
519
520	Implements IMetadata.
521	"""
522
523	def __init__(	1✔
524	self, elem, parent, parse_remote_metadata=False, timeout=30, headers=None, auth=None
525	):
526	"""."""
527	super(ContentMetadata, self).__init__(headers=headers, auth=auth)	1✔
528	self.id = elem.find(nspath("Name", ns=WFS_NAMESPACE)).text	1✔
529	self.title = elem.find(nspath("Title", ns=WFS_NAMESPACE)).text	1✔
530	abstract = elem.find(nspath("Abstract", ns=WFS_NAMESPACE))	1✔
531	if abstract is not None:	1✔
532	self.abstract = abstract.text	1✔
533	else:
UNCOV 534	self.abstract = None	×
535	self.keywords = [	1✔
536	f.text for f in elem.findall(nspath("Keywords/Keyword", ns=OWS_NAMESPACE))
537	]
538
539	# bboxes
540	self.boundingBoxWGS84 = None	1✔
541	b = elem.find(nspath("WGS84BoundingBox", ns=OWS_NAMESPACE))	1✔
542	if b is not None:	1✔
543	try:	1✔
544	lc = b.find(nspath("LowerCorner", ns=OWS_NAMESPACE))	1✔
545	uc = b.find(nspath("UpperCorner", ns=OWS_NAMESPACE))	1✔
546	ll = [float(s) for s in lc.text.split()]	1✔
547	ur = [float(s) for s in uc.text.split()]	1✔
548	self.boundingBoxWGS84 = (ll[0], ll[1], ur[0], ur[1])	1✔
549
550	# there is no such think as bounding box
551	# make copy of the WGS84BoundingBox
552	self.boundingBox = (	1✔
553	self.boundingBoxWGS84[0],
554	self.boundingBoxWGS84[1],
555	self.boundingBoxWGS84[2],
556	self.boundingBoxWGS84[3],
557	Crs("epsg:4326"),
558	)
UNCOV 559	except AttributeError:	×
UNCOV 560	self.boundingBoxWGS84 = None	×
561	# crs options
562	self.crsOptions = [	1✔
563	Crs(srs.text) for srs in elem.findall(nspath("OtherCRS", ns=WFS_NAMESPACE))
564	]
565	defaultCrs = elem.findall(nspath("DefaultCRS", ns=WFS_NAMESPACE))	1✔
566	if len(defaultCrs) > 0:	1✔
567	self.crsOptions.insert(0, Crs(defaultCrs[0].text))	1✔
568
569	# verbs
570	self.verbOptions = [	1✔
571	op.tag for op in parent.findall(nspath("Operations/*", ns=WFS_NAMESPACE))
572	]
573	self.verbOptions + [	1✔
574	op.tag
575	for op in elem.findall(nspath("Operations/*", ns=WFS_NAMESPACE))
576	if op.tag not in self.verbOptions
577	]
578
579	# others not used but needed for iContentMetadata harmonisation
580	self.styles = None	1✔
581	self.timepositions = None	1✔
582	self.defaulttimeposition = None	1✔
583
584	# MetadataURLs
585	self.metadataUrls = []	1✔
586	for m in elem.findall(nspath("MetadataURL", ns=WFS_NAMESPACE)):	1✔
587	metadataUrl = {	1✔
588	"url": testXMLValue(
589	m.attrib["{http://www.w3.org/1999/xlink}href"], attrib=True
590	)
591	}
592	self.metadataUrls.append(metadataUrl)	1✔
593
594	if parse_remote_metadata:	1✔
595	self.parse_remote_metadata(timeout)	1✔
596
597	def parse_remote_metadata(self, timeout=30):	1✔
598	"""Parse remote metadata for MetadataURL and add it as metadataUrl['metadata']"""
599	for metadataUrl in self.metadataUrls:	1✔
600	if metadataUrl["url"] is not None:	1✔
601	try:	1✔
602	content = openURL(metadataUrl["url"], timeout=timeout, headers=self.headers, auth=self.auth)	1✔
603	doc = etree.fromstring(content.read())	1✔
604
605	mdelem = doc.find(".//metadata")	1✔
606	if mdelem is not None:	1✔
UNCOV 607	metadataUrl["metadata"] = Metadata(mdelem)	×
UNCOV 608	continue	×
609
610	mdelem = doc.find(	1✔
611	".//" + util.nspath_eval("gmd:MD_Metadata", n.get_namespaces(["gmd"]))
612	) or doc.find(
613	".//" + util.nspath_eval("gmi:MI_Metadata", n.get_namespaces(["gmi"]))
614	)
615	if mdelem is not None:	1✔
616	metadataUrl["metadata"] = MD_Metadata(mdelem)	1✔
617	continue	1✔
618	else: # ISO 19115 Part 3 XML
619	mdelem = MD_Metadata3.find_start(doc)	1✔
620	if mdelem is not None:	1✔
621	metadataUrl["metadata"] = MD_Metadata3(mdelem)	1✔
622	else:
UNCOV 623	metadataUrl["metadata"] = None	×
624	continue	1✔
625	except Exception:	1✔
626	metadataUrl["metadata"] = None	1✔

geopython / OWSLib / 11282283562

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous