<s> from __future__ import ( absolute_import , division , print_function ) <EOL> import cgi <EOL> from owslib . etree import etree <EOL> from datetime import datetime <EOL> try : <EOL> from urllib . parse import urlencode <EOL> except ImportError : <EOL> from urllib import urlencode <EOL> from owslib import ows <EOL> from owslib . crs import Crs <EOL> from owslib . fes import FilterCapabilities <EOL> from owslib . util import openURL , testXMLValue , nspath_eval , nspath , extract_time <EOL> from owslib . namespaces import Namespaces <EOL> def get_namespaces ( ) : <EOL> n = Namespaces ( ) <EOL> ns = n . get_namespaces ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> ns [ "<STR_LIT>" ] = n . get_namespace ( "<STR_LIT>" ) <EOL> return ns <EOL> namespaces = get_namespaces ( ) <EOL> class SensorObservationService_1_0_0 ( object ) : <EOL> """<STR_LIT>""" <EOL> def __new__ ( self , url , version , xml = None , username = None , password = None ) : <EOL> """<STR_LIT>""" <EOL> obj = object . __new__ ( self ) <EOL> obj . __init__ ( url , version , xml , username , password ) <EOL> return obj <EOL> def __getitem__ ( self , id ) : <EOL> '''<STR_LIT>''' <EOL> if id in self . __getattribute__ ( '<STR_LIT>' ) . keys ( ) : <EOL> return self . __getattribute__ ( '<STR_LIT>' ) [ id ] <EOL> else : <EOL> raise KeyError ( "<STR_LIT>" % id ) <EOL> def __init__ ( self , url , version = '<STR_LIT>' , xml = None , username = None , password = None ) : <EOL> """<STR_LIT>""" <EOL> self . url = url <EOL> self . username = username <EOL> self . password = password <EOL> self . version = version <EOL> self . _capabilities = None <EOL> reader = SosCapabilitiesReader ( <EOL> version = self . version , url = self . url , username = self . username , password = self . password <EOL> ) <EOL> if xml : <EOL> self . _capabilities = reader . read_string ( xml ) <EOL> else : <EOL> self . _capabilities = reader . read ( self . url ) <EOL> if self . _capabilities . tag == nspath_eval ( "<STR_LIT>" , namespaces ) : <EOL> raise ows . ExceptionReport ( self . _capabilities ) <EOL> self . _build_metadata ( ) <EOL> def getOperationByName ( self , name ) : <EOL> """<STR_LIT>""" <EOL> for item in self . operations : <EOL> if item . name == name : <EOL> return item <EOL> raise KeyError ( "<STR_LIT>" % name ) <EOL> def _build_metadata ( self ) : <EOL> """<STR_LIT>""" <EOL> service_id_element = self . _capabilities . find ( nspath_eval ( '<STR_LIT>' , namespaces ) ) <EOL> self . identification = ows . ServiceIdentification ( service_id_element ) <EOL> service_provider_element = self . _capabilities . find ( nspath_eval ( '<STR_LIT>' , namespaces ) ) <EOL> self . provider = ows . ServiceProvider ( service_provider_element ) <EOL> self . operations = [ ] <EOL> for elem in self . _capabilities . findall ( nspath_eval ( '<STR_LIT>' , namespaces ) ) : <EOL> self . operations . append ( ows . OperationsMetadata ( elem ) ) <EOL> filters = self . _capabilities . find ( nspath_eval ( '<STR_LIT>' , namespaces ) ) <EOL> if filters is not None : <EOL> self . filters = FilterCapabilities ( filters ) <EOL> else : <EOL> self . filters = None <EOL> self . contents = { } <EOL> self . offerings = [ ] <EOL> for offering in self . _capabilities . findall ( nspath_eval ( '<STR_LIT>' , namespaces ) ) : <EOL> off = SosObservationOffering ( offering ) <EOL> self . contents [ off . id ] = off <EOL> self . offerings . append ( off ) <EOL> def describe_sensor ( self , outputFormat = None , <EOL> procedure = None , <EOL> method = '<STR_LIT>' , <EOL> ** kwargs ) : <EOL> try : <EOL> base_url = next ( ( m . get ( '<STR_LIT:url>' ) for m in self . getOperationByName ( '<STR_LIT>' ) . methods if m . get ( '<STR_LIT:type>' ) . lower ( ) == method . lower ( ) ) ) <EOL> except StopIteration : <EOL> base_url = self . url <EOL> request = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:version>' : self . version , '<STR_LIT>' : '<STR_LIT>' } <EOL> assert isinstance ( outputFormat , str ) <EOL> request [ '<STR_LIT>' ] = outputFormat <EOL> assert isinstance ( procedure , str ) <EOL> request [ '<STR_LIT>' ] = procedure <EOL> url_kwargs = { } <EOL> if '<STR_LIT>' in kwargs : <EOL> url_kwargs [ '<STR_LIT>' ] = kwargs . pop ( '<STR_LIT>' ) <EOL> if kwargs : <EOL> for kw in kwargs : <EOL> request [ kw ] = kwargs [ kw ] <EOL> data = urlencode ( request ) <EOL> response = openURL ( base_url , data , method , username = self . username , password = self . password , ** url_kwargs ) . read ( ) <EOL> tr = etree . fromstring ( response ) <EOL> if tr . tag == nspath_eval ( "<STR_LIT>" , namespaces ) : <EOL> raise ows . ExceptionReport ( tr ) <EOL> return response <EOL> def get_observation ( self , responseFormat = None , <EOL> offerings = None , <EOL> observedProperties = None , <EOL> eventTime = None , <EOL> procedure = None , <EOL> method = '<STR_LIT>' , <EOL> ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> base_url = next ( ( m . get ( '<STR_LIT:url>' ) for m in self . getOperationByName ( '<STR_LIT>' ) . methods if m . get ( '<STR_LIT:type>' ) . lower ( ) == method . lower ( ) ) ) <EOL> except StopIteration : <EOL> base_url = self . url <EOL> request = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:version>' : self . version , '<STR_LIT>' : '<STR_LIT>' } <EOL> assert isinstance ( offerings , list ) and len ( offerings ) > <NUM_LIT:0> <EOL> request [ '<STR_LIT>' ] = '<STR_LIT:U+002C>' . join ( offerings ) <EOL> assert isinstance ( observedProperties , list ) and len ( observedProperties ) > <NUM_LIT:0> <EOL> request [ '<STR_LIT>' ] = '<STR_LIT:U+002C>' . join ( observedProperties ) <EOL> assert isinstance ( responseFormat , str ) <EOL> request [ '<STR_LIT>' ] = responseFormat <EOL> if eventTime is not None : <EOL> request [ '<STR_LIT>' ] = eventTime <EOL> url_kwargs = { } <EOL> if '<STR_LIT>' in kwargs : <EOL> url_kwargs [ '<STR_LIT>' ] = kwargs . pop ( '<STR_LIT>' ) <EOL> if procedure is not None : <EOL> request [ '<STR_LIT>' ] = procedure <EOL> if kwargs : <EOL> for kw in kwargs : <EOL> request [ kw ] = kwargs [ kw ] <EOL> data = urlencode ( request ) <EOL> response = openURL ( base_url , data , method , username = self . username , <EOL> password = self . password , ** url_kwargs ) . read ( ) <EOL> try : <EOL> tr = etree . fromstring ( response ) <EOL> if tr . tag == nspath_eval ( "<STR_LIT>" , namespaces ) : <EOL> raise ows . ExceptionReport ( tr ) <EOL> else : <EOL> return response <EOL> except ows . ExceptionReport : <EOL> raise <EOL> except BaseException : <EOL> return response <EOL> def get_operation_by_name ( self , name ) : <EOL> """<STR_LIT>""" <EOL> for item in self . operations : <EOL> if item . name . lower ( ) == name . lower ( ) : <EOL> return item <EOL> raise KeyError ( "<STR_LIT>" % name ) <EOL> class SosObservationOffering ( object ) : <EOL> def __init__ ( self , element ) : <EOL> self . _root = element <EOL> self . id = testXMLValue ( self . _root . attrib . get ( nspath_eval ( '<STR_LIT>' , namespaces ) ) , True ) <EOL> self . description = testXMLValue ( self . _root . find ( nspath_eval ( '<STR_LIT>' , namespaces ) ) ) <EOL> self . name = testXMLValue ( self . _root . find ( nspath_eval ( '<STR_LIT>' , namespaces ) ) ) <EOL> val = testXMLValue ( self . _root . find ( nspath_eval ( '<STR_LIT>' , namespaces ) ) ) <EOL> if val is not None : <EOL> self . srs = Crs ( val ) <EOL> try : <EOL> envelope = self . _root . find ( nspath_eval ( '<STR_LIT>' , namespaces ) ) <EOL> lower_left_corner = testXMLValue ( envelope . find ( nspath_eval ( '<STR_LIT>' , namespaces ) ) ) . split ( ) <EOL> upper_right_corner = testXMLValue ( envelope . find ( nspath_eval ( '<STR_LIT>' , namespaces ) ) ) . split ( ) <EOL> self . bbox = ( float ( lower_left_corner [ <NUM_LIT:1> ] ) , float ( lower_left_corner [ <NUM_LIT:0> ] ) , float ( upper_right_corner [ <NUM_LIT:1> ] ) , float ( upper_right_corner [ <NUM_LIT:0> ] ) ) <EOL> self . bbox_srs = Crs ( testXMLValue ( envelope . attrib . get ( '<STR_LIT>' ) , True ) ) <EOL> except Exception : <EOL> self . bbox = None <EOL> self . bbox_srs = None <EOL> begin_position_element = self . _root . find ( nspath_eval ( '<STR_LIT>' , namespaces ) ) <EOL> self . begin_position = extract_time ( begin_position_element ) <EOL> end_position_element = self . _root . find ( nspath_eval ( '<STR_LIT>' , namespaces ) ) <EOL> self . end_position = extract_time ( end_position_element ) <EOL> self . result_model = testXMLValue ( self . _root . find ( nspath_eval ( '<STR_LIT>' , namespaces ) ) ) <EOL> self . procedures = [ ] <EOL> for proc in self . _root . findall ( nspath_eval ( '<STR_LIT>' , namespaces ) ) : <EOL> self . procedures . append ( testXMLValue ( proc . attrib . get ( nspath_eval ( '<STR_LIT>' , namespaces ) ) , True ) ) <EOL> self . observed_properties = [ ] <EOL> for op in self . _root . findall ( nspath_eval ( '<STR_LIT>' , namespaces ) ) : <EOL> self . observed_properties . append ( testXMLValue ( op . attrib . get ( nspath_eval ( '<STR_LIT>' , namespaces ) ) , True ) ) <EOL> self . features_of_interest = [ ] <EOL> for fot in self . _root . findall ( nspath_eval ( '<STR_LIT>' , namespaces ) ) : <EOL> self . features_of_interest . append ( testXMLValue ( fot . attrib . get ( nspath_eval ( '<STR_LIT>' , namespaces ) ) , True ) ) <EOL> self . response_formats = [ ] <EOL> for rf in self . _root . findall ( nspath_eval ( '<STR_LIT>' , namespaces ) ) : <EOL> self . response_formats . append ( testXMLValue ( rf ) ) <EOL> self . response_modes = [ ] <EOL> for rm in self . _root . findall ( nspath_eval ( '<STR_LIT>' , namespaces ) ) : <EOL> self . response_modes . append ( testXMLValue ( rm ) ) <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' % ( self . id , self . name ) <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % self . name <EOL> class SosCapabilitiesReader ( object ) : <EOL> def __init__ ( self , version = "<STR_LIT>" , url = None , username = None , password = None ) : <EOL> self . version = version <EOL> self . url = url <EOL> self . username = username <EOL> self . password = password <EOL> def capabilities_url ( self , service_url ) : <EOL> """<STR_LIT>""" <EOL> qs = [ ] <EOL> if service_url . find ( '<STR_LIT:?>' ) != - <NUM_LIT:1> : <EOL> qs = cgi . parse_qsl ( service_url . split ( '<STR_LIT:?>' ) [ <NUM_LIT:1> ] ) <EOL> params = [ x [ <NUM_LIT:0> ] for x in qs ] <EOL> if '<STR_LIT>' not in params : <EOL> qs . append ( ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> if '<STR_LIT>' not in params : <EOL> qs . append ( ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> if '<STR_LIT>' not in params : <EOL> qs . append ( ( '<STR_LIT>' , self . version ) ) <EOL> urlqs = urlencode ( tuple ( qs ) ) <EOL> return service_url . split ( '<STR_LIT:?>' ) [ <NUM_LIT:0> ] + '<STR_LIT:?>' + urlqs <EOL> def read ( self , service_url ) : <EOL> """<STR_LIT>""" <EOL> getcaprequest = self . capabilities_url ( service_url ) <EOL> spliturl = getcaprequest . split ( '<STR_LIT:?>' ) <EOL> u = openURL ( spliturl [ <NUM_LIT:0> ] , spliturl [ <NUM_LIT:1> ] , method = '<STR_LIT>' , username = self . username , password = self . password ) <EOL> return etree . fromstring ( u . read ( ) ) <EOL> def read_string ( self , st ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( st , bytes ) : <EOL> raise ValueError ( "<STR_LIT>" % type ( st ) ) <EOL> return etree . fromstring ( st ) </s>
