1# -*- coding: ISO-8859-15 -*- 2# ============================================================================= 3# Copyright (c) 2008 Tom Kralidis 4# 5# Authors : Tom Kralidis <tomkralidis@gmail.com> 6# 7# Contact email: tomkralidis@gmail.com 8# ============================================================================= 9 10""" 11API for OGC Web Services Common (OWS) constructs and metadata. 12 13OWS Common: http://www.opengeospatial.org/standards/common 14 15Currently supports version 1.1.0 (06-121r3). 16""" 17 18import logging 19 20from owslib.etree import etree 21from owslib import crs, util 22from owslib.namespaces import Namespaces 23 24LOGGER = logging.getLogger(__name__) 25 26n = Namespaces() 27 28OWS_NAMESPACE_1_0_0 = n.get_namespace("ows") 29OWS_NAMESPACE_1_1_0 = n.get_namespace("ows110") 30OWS_NAMESPACE_2_0_0 = n.get_namespace("ows200") 31XSI_NAMESPACE = n.get_namespace("xsi") 32XLINK_NAMESPACE = n.get_namespace("xlink") 33 34DEFAULT_OWS_NAMESPACE = OWS_NAMESPACE_1_1_0 # Use this as default for OWSCommon objects 35 36 37class OwsCommon(object): 38 """Initialize OWS Common object""" 39 def __init__(self, version): 40 self.version = version 41 if version == '1.0.0': 42 self.namespace = OWS_NAMESPACE_1_0_0 43 elif version == '1.1.0': 44 self.namespace = OWS_NAMESPACE_1_1_0 45 else: 46 self.namespace = OWS_NAMESPACE_2_0_0 47 48 49class ServiceIdentification(object): 50 """Initialize an OWS Common ServiceIdentification construct""" 51 def __init__(self, infoset, namespace=DEFAULT_OWS_NAMESPACE): 52 self._root = infoset 53 54 val = self._root.find(util.nspath('Title', namespace)) 55 self.title = util.testXMLValue(val) 56 57 val = self._root.find(util.nspath('Abstract', namespace)) 58 self.abstract = util.testXMLValue(val) 59 60 self.keywords = [] 61 for f in self._root.findall(util.nspath('Keywords/Keyword', namespace)): 62 if f.text is not None: 63 self.keywords.append(f.text) 64 65 val = self._root.find(util.nspath('Keywords/Type', namespace)) 66 self.keywords_type = util.testXMLValue(val) 67 68 val = self._root.find(util.nspath('AccessConstraints', namespace)) 69 self.accessconstraints = util.testXMLValue(val) 70 71 val = self._root.find(util.nspath('Fees', namespace)) 72 self.fees = util.testXMLValue(val) 73 74 val = self._root.find(util.nspath('ServiceType', namespace)) 75 self.type = util.testXMLValue(val) 76 self.service = self.type # alternative? keep both?discuss 77 78 val = self._root.find(util.nspath('ServiceTypeVersion', namespace)) 79 self.version = util.testXMLValue(val) 80 81 self.versions = [] 82 for v in self._root.findall(util.nspath('ServiceTypeVersion', namespace)): 83 self.versions.append(util.testXMLValue(v)) 84 85 self.profiles = [] 86 for p in self._root.findall(util.nspath('Profile', namespace)): 87 self.profiles.append(util.testXMLValue(p)) 88 89 def __str__(self): 90 return 'Service: {}, title={}'.format(self.service, self.title or '') 91 92 def __repr__(self): 93 return '<owslib.ows.ServiceIdentification {} at {}>'.format(self.service, hex(id(self))) 94 95 96class ServiceProvider(object): 97 """Initialize an OWS Common ServiceProvider construct""" 98 def __init__(self, infoset, namespace=DEFAULT_OWS_NAMESPACE): 99 self._root = infoset 100 val = self._root.find(util.nspath('ProviderName', namespace)) 101 self.name = util.testXMLValue(val) 102 self.contact = ServiceContact(infoset, namespace) 103 val = self._root.find(util.nspath('ProviderSite', namespace)) 104 if val is not None: 105 try: 106 urlattrib = val.attrib[util.nspath('href', XLINK_NAMESPACE)] 107 self.url = util.testXMLValue(urlattrib, True) 108 except KeyError: 109 self.url = None 110 else: 111 self.url = None 112 113 114class ServiceContact(object): 115 """Initialize an OWS Common ServiceContact construct""" 116 def __init__(self, infoset, namespace=DEFAULT_OWS_NAMESPACE): 117 self._root = infoset 118 val = self._root.find(util.nspath('ProviderName', namespace)) 119 self.name = util.testXMLValue(val) 120 self.organization = util.testXMLValue( 121 self._root.find(util.nspath('ContactPersonPrimary/ContactOrganization', namespace))) 122 123 val = self._root.find(util.nspath('ProviderSite', namespace)) 124 if val is not None: 125 self.site = util.testXMLValue(val.attrib.get(util.nspath('href', XLINK_NAMESPACE)), True) 126 else: 127 self.site = None 128 129 val = self._root.find(util.nspath('ServiceContact/Role', namespace)) 130 self.role = util.testXMLValue(val) 131 132 val = self._root.find(util.nspath('ServiceContact/IndividualName', namespace)) 133 self.name = util.testXMLValue(val) 134 135 val = self._root.find(util.nspath('ServiceContact/PositionName', namespace)) 136 self.position = util.testXMLValue(val) 137 138 val = self._root.find(util.nspath('ServiceContact/ContactInfo/Phone/Voice', namespace)) 139 self.phone = util.testXMLValue(val) 140 141 val = self._root.find(util.nspath('ServiceContact/ContactInfo/Phone/Facsimile', namespace)) 142 self.fax = util.testXMLValue(val) 143 144 val = self._root.find(util.nspath('ServiceContact/ContactInfo/Address/DeliveryPoint', namespace)) 145 self.address = util.testXMLValue(val) 146 147 val = self._root.find(util.nspath('ServiceContact/ContactInfo/Address/City', namespace)) 148 self.city = util.testXMLValue(val) 149 150 val = self._root.find(util.nspath('ServiceContact/ContactInfo/Address/AdministrativeArea', namespace)) 151 self.region = util.testXMLValue(val) 152 153 val = self._root.find(util.nspath('ServiceContact/ContactInfo/Address/PostalCode', namespace)) 154 self.postcode = util.testXMLValue(val) 155 156 val = self._root.find(util.nspath('ServiceContact/ContactInfo/Address/Country', namespace)) 157 self.country = util.testXMLValue(val) 158 159 val = self._root.find(util.nspath('ServiceContact/ContactInfo/Address/ElectronicMailAddress', namespace)) 160 self.email = util.testXMLValue(val) 161 162 val = self._root.find(util.nspath('ServiceContact/ContactInfo/OnlineResource', namespace)) 163 if val is not None: 164 self.url = util.testXMLValue(val.attrib.get(util.nspath('href', XLINK_NAMESPACE)), True) 165 else: 166 self.url = None 167 168 val = self._root.find(util.nspath('ServiceContact/ContactInfo/HoursOfService', namespace)) 169 self.hours = util.testXMLValue(val) 170 171 val = self._root.find(util.nspath('ServiceContact/ContactInfo/ContactInstructions', namespace)) 172 self.instructions = util.testXMLValue(val) 173 174 175class Constraint(object): 176 def __init__(self, elem, namespace=DEFAULT_OWS_NAMESPACE): 177 self.name = elem.attrib.get('name') 178 self.values = [i.text for i in elem.findall(util.nspath('Value', namespace))] 179 self.values += [i.text for i in elem.findall(util.nspath('AllowedValues/Value', namespace))] 180 181 def __repr__(self): 182 if self.values: 183 return "Constraint: %s - %s" % (self.name, self.values) 184 else: 185 return "Constraint: %s" % self.name 186 187 188class Parameter(object): 189 def __init__(self, elem, namespace=DEFAULT_OWS_NAMESPACE): 190 self.name = elem.attrib.get('name') 191 self.values = [i.text for i in elem.findall(util.nspath('Value', namespace))] 192 self.values += [i.text for i in elem.findall(util.nspath('AllowedValues/Value', namespace))] 193 194 def __repr__(self): 195 if self.values: 196 return "Parameter: %s - %s" % (self.name, self.values) 197 else: 198 return "Parameter: %s" % self.name 199 200 201class OperationsMetadata(object): 202 """Initialize an OWS OperationMetadata construct""" 203 def __init__(self, elem, namespace=DEFAULT_OWS_NAMESPACE): 204 if 'name' not in elem.attrib: # This is not a valid element 205 return 206 self.name = elem.attrib['name'] 207 self.formatOptions = ['text/xml'] 208 parameters = [] 209 self.methods = [] 210 self.constraints = [] 211 212 for verb in elem.findall(util.nspath('DCP/HTTP/*', namespace)): 213 url = util.testXMLAttribute(verb, util.nspath('href', XLINK_NAMESPACE)) 214 if url is not None: 215 verb_constraints = [Constraint(conts, namespace) for conts in verb.findall( 216 util.nspath('Constraint', namespace))] 217 self.methods.append({'constraints': verb_constraints, 'type': util.xmltag_split(verb.tag), 'url': url}) 218 219 for parameter in elem.findall(util.nspath('Parameter', namespace)): 220 if namespace == OWS_NAMESPACE_1_1_0: 221 parameters.append((parameter.attrib['name'], {'values': [i.text for i in parameter.findall( 222 util.nspath('AllowedValues/Value', namespace))]})) 223 else: 224 parameters.append((parameter.attrib['name'], {'values': [i.text for i in parameter.findall( 225 util.nspath('Value', namespace))]})) 226 self.parameters = dict(parameters) 227 228 for constraint in elem.findall(util.nspath('Constraint', namespace)): 229 self.constraints.append(Constraint(constraint, namespace)) 230 231 def __str__(self): 232 return "Operation: {}, format={}".format(self.name, self.formatOptions) 233 234 def __repr__(self): 235 return '<owslib.ows.OperationsMetadata {} at {}>'.format(self.name, hex(id(self))) 236 237 238class BoundingBox(object): 239 """Initialize an OWS BoundingBox construct""" 240 def __init__(self, elem, namespace=DEFAULT_OWS_NAMESPACE): 241 self.minx = None 242 self.miny = None 243 self.maxx = None 244 self.maxy = None 245 self.crs = None 246 self.dimensions = 2 247 if elem is None: 248 return 249 val = elem.attrib.get('crs') or elem.attrib.get('{{{}}}crs'.format(namespace)) 250 if val: 251 try: 252 self.crs = crs.Crs(val) 253 except (AttributeError, ValueError): 254 LOGGER.warning('Invalid CRS %r. Expected integer' % val) 255 else: 256 self.crs = None 257 258 val = elem.attrib.get('dimensions') or elem.attrib.get('{{{}}}dimensions'.format(namespace)) 259 if val is not None: 260 self.dimensions = int(util.testXMLValue(val, True)) 261 else: # assume 2 262 self.dimensions = 2 263 264 val = elem.find(util.nspath('LowerCorner', namespace)) 265 tmp = util.testXMLValue(val) 266 if tmp is not None: 267 xy = tmp.split() 268 if len(xy) > 1: 269 if self.crs is not None and self.crs.axisorder == 'yx': 270 self.minx, self.miny = xy[1], xy[0] 271 else: 272 self.minx, self.miny = xy[0], xy[1] 273 274 val = elem.find(util.nspath('UpperCorner', namespace)) 275 tmp = util.testXMLValue(val) 276 if tmp is not None: 277 xy = tmp.split() 278 if len(xy) > 1: 279 if self.crs is not None and self.crs.axisorder == 'yx': 280 self.maxx, self.maxy = xy[1], xy[0] 281 else: 282 self.maxx, self.maxy = xy[0], xy[1] 283 284 285class WGS84BoundingBox(BoundingBox): 286 """WGS84 bbox, axis order xy""" 287 def __init__(self, elem, namespace=DEFAULT_OWS_NAMESPACE): 288 BoundingBox.__init__(self, elem, namespace) 289 self.dimensions = 2 290 self.crs = crs.Crs('urn:ogc:def:crs:OGC:2:84') 291 292 293class ExceptionReport(Exception): 294 """OWS ExceptionReport""" 295 296 def __init__(self, elem, namespace=DEFAULT_OWS_NAMESPACE): 297 self.exceptions = [] 298 299 if hasattr(elem, 'getroot'): 300 elem = elem.getroot() 301 302 for i in elem.findall(util.nspath('Exception', namespace)): 303 tmp = {} 304 val = i.attrib.get('exceptionCode') 305 tmp['exceptionCode'] = util.testXMLValue(val, True) 306 val = i.attrib.get('locator') 307 tmp['locator'] = util.testXMLValue(val, True) 308 val = i.find(util.nspath('ExceptionText', namespace)) 309 tmp['ExceptionText'] = util.testXMLValue(val) 310 self.exceptions.append(tmp) 311 312 # set topmost stacktrace as return message 313 self.code = self.exceptions[0]['exceptionCode'] 314 self.locator = self.exceptions[0]['locator'] 315 self.msg = self.exceptions[0]['ExceptionText'] 316 self.xml = etree.tostring(elem) 317 318 def __str__(self): 319 return repr(self.msg) 320