/usr/lib/python3/dist-packages/owslib/feature/__init__.py is in python3-owslib 0.16.0-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 | # =============================================================================
# OWSLib. Copyright (C) 2012 Jachym Cepicky
#
# Contact email: jachym.cepicky@gmail.com
#
# =============================================================================
from __future__ import (absolute_import, division, print_function)
from owslib.crs import Crs
try:
from urllib import urlencode
except ImportError:
from urllib.parse import urlencode
import logging
from owslib.util import log
from owslib.feature.schema import get_schema
class WebFeatureService_(object):
"""Base class for WebFeatureService implementations"""
def getBBOXKVP (self,bbox,typename):
"""Formate bounding box for KVP request type (HTTP GET)
@param bbox: (minx,miny,maxx,maxy[,srs])
@type bbox: List
@param typename: feature name
@type typename: String
@returns: String properly formated according to version and
coordinate reference system
"""
srs = None
# srs of the bbox is specified in the bbox as fifth paramter
if len(bbox) == 5:
srs = self.getSRS(bbox[4],typename[0])
# take default srs
else:
srs = self.contents[typename[0]].crsOptions[0]
# 1.1.0 and 2.0.0 have same encoding
if self.version in ["1.1.0","2.0.0"]:
# format bbox parameter
if srs.encoding == "urn" :
if srs.axisorder == "yx":
return "%s,%s,%s,%s,%s" % \
(bbox[1],bbox[0],bbox[3],bbox[2],srs.getcodeurn())
else:
return "%s,%s,%s,%s,%s" % \
(bbox[0],bbox[1],bbox[2],bbox[3],srs.getcodeurn())
else:
return "%s,%s,%s,%s,%s" % \
(bbox[0],bbox[1],bbox[2],bbox[3],srs.getcode())
# 1.0.0
else:
return "%s,%s,%s,%s,%s" % \
(bbox[0],bbox[1],bbox[2],bbox[3],srs.getcode())
def getSRS(self, srsname, typename):
"""Returns None or Crs object for given name
@param typename: feature name
@type typename: String
"""
if not isinstance(srsname, Crs):
srs = Crs(srsname)
else:
srs = srsname
try:
index = self.contents[typename].crsOptions.index(srs)
# Return the Crs string that was pulled directly from the
# GetCaps document (the 'id' attribute in the Crs object).
return self.contents[typename].crsOptions[index]
except ValueError:
options = ", ".join(map(lambda x: x.id, self.contents[typename].crsOptions))
log.warning("Requested srsName '%s' not available for requested typename '%s'. \
Options are: %s. " % (srs.getcode(), typename, options))
return None
def getGETGetFeatureRequest(self, typename=None, filter=None, bbox=None, featureid=None,
featureversion=None, propertyname=None, maxfeatures=None,storedQueryID=None, storedQueryParams=None,
outputFormat=None, method='Get', startindex=None, sortby=None):
"""Formulate proper GetFeature request using KVP encoding
----------
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. '*' matches all.
maxfeatures : int
Maximum number of features to be returned.
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 3 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)
"""
storedQueryParams = storedQueryParams or {}
base_url = next((m.get('url') for m in self.getOperationByName('GetFeature').methods if m.get('type').lower() == method.lower()))
base_url = base_url if base_url.endswith("?") else base_url+"?"
request = {'service': 'WFS', 'version': self.version, 'request': 'GetFeature'}
# check featureid
if featureid:
request['featureid'] = ','.join(featureid)
elif bbox:
request['bbox'] = self.getBBOXKVP(bbox,typename)
elif filter:
request['query'] = str(filter)
if typename:
typename = [typename] if type(typename) == type("") else typename
if int(self.version.split('.')[0]) >= 2:
request['typenames'] = ','.join(typename)
else:
request['typename'] = ','.join(typename)
if propertyname:
request['propertyname'] = ','.join(propertyname)
if sortby:
request['sortby'] = ','.join(sortby)
if featureversion:
request['featureversion'] = str(featureversion)
if maxfeatures:
if int(self.version.split('.')[0]) >= 2:
request['count'] = str(maxfeatures)
else:
request['maxfeatures'] = str(maxfeatures)
if startindex:
request['startindex'] = str(startindex)
if storedQueryID:
request['storedQuery_id']=str(storedQueryID)
for param in storedQueryParams:
request[param]=storedQueryParams[param]
if outputFormat is not None:
request["outputFormat"] = outputFormat
data = urlencode(request, doseq=True)
return base_url+data
def get_schema(self, typename):
"""
Get layer schema compatible with :class:`fiona` schema object
"""
return get_schema(self.url, typename, self.version)
|