Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
("STATISTICS", "MEAN"),
("STATISTICS", "MINIMUM"),
("STATISTICS", "MAXIMUM"),
("STATISTICS", "WEIGHT_SUM"),
("STATISTICS", "VARIANCE"),
("STATISTICS", "STD_DEV"),
("STATISTICS", "COUNT"),
("GROUP_BY", "STATISTIC"),
("SUMMARIZE_TIMESTEP", "false"),
("SUMMARIZE_FEATURE_ATTRIBUTE", "false"),
("FEATURE_COLLECTION", featureCollection)]
output = "OUTPUT"
# build XML request for WPS process execution
execution = WPSExecution()
requestElement = execution.buildRequest(processid, inputs, output=[(output, True)])
request = etree.tostring(requestElement)
# Compare to cached XML request
_request = open(resource_file('wps_USGSExecuteRequest3.xml'), 'rb').read()
assert compare_xml(request, _request) is True
("STATISTICS", "MAXIMUM"),
("STATISTICS", "WEIGHT_SUM"),
("STATISTICS", "VARIANCE"),
("STATISTICS", "STD_DEV"),
("STATISTICS", "COUNT"),
("GROUP_BY", "STATISTIC"),
("SUMMARIZE_TIMESTEP", "true"),
("SUMMARIZE_FEATURE_ATTRIBUTE", "true"),
("FEATURE_COLLECTION", featureCollection)
]
output = "OUTPUT"
# build XML request for WPS process execution
execution = WPSExecution()
requestElement = execution.buildRequest(processid, inputs, output=[(output, True)])
request = etree.tostring(requestElement)
# Compare to cached XML request
_request = open(resource_file('wps_USGSExecuteRequest2.xml'), 'rb').read()
assert compare_xml(request, _request) is True
def __init__(self, record):
if hasattr(record, 'getroot'): # standalone document
self.xml = etree.tostring(record.getroot())
else: # part of a larger document
self.xml = etree.tostring(record)
# check to see if Dublin Core record comes from
# rdf:RDF/rdf:Description container
# (child content model is identical)
self.rdf = False
rdf = record.find(util.nspath_eval('rdf:Description', namespaces))
if rdf is not None:
self.rdf = True
record = rdf
# some CSWs return records with multiple identifiers based on
# different schemes. Use the first dc:identifier value to set
# self.identifier, and set self.identifiers as a list of dicts
val = record.find(util.nspath_eval('dc:identifier', namespaces))
self.identifier = util.testXMLValue(val)
- dname: the value of the Parameter or Property to query
- dtype: whether to query a parameter (parameter) or property (property)
"""
# construct request
dtypename = 'ParameterName'
node0 = etree.Element(util.nspath_eval('csw:GetDomain', namespaces))
node0.set('service', self.service)
node0.set('version', self.version)
node0.set(util.nspath_eval('xsi:schemaLocation', namespaces), schema_location)
if dtype == 'property':
dtypename = 'PropertyName'
etree.SubElement(node0, util.nspath_eval('csw:%s' % dtypename, namespaces)).text = dname
self.request = util.xml2string(etree.tostring(node0))
self._invoke()
if self.exceptionreport is None:
self.results = {}
val = self._exml.find(util.nspath_eval('csw:DomainValues', namespaces)).attrib.get('type')
self.results['type'] = util.testXMLValue(val, True)
val = self._exml.find(util.nspath_eval('csw:DomainValues/csw:%s' % dtypename, namespaces))
self.results[dtype] = util.testXMLValue(val)
# get the list of values associated with the Domain
self.results['values'] = []
for f in self._exml.findall(util.nspath_eval('csw:DomainValues/csw:ListOfValues/csw:Value', namespaces)):
def getServiceXML(self):
xml = None
if self._capabilities is not None:
xml = etree.tostring(self._capabilities)
return xml
def __init__(self, md):
if hasattr(md, 'getroot'): # standalone document
self.xml = etree.tostring(md.getroot())
else: # part of a larger document
self.xml = etree.tostring(md)
self.idinfo = Idinfo(md)
self.eainfo = Eainfo(md)
self.distinfo = Distinfo(md)
self.metainfo = Metainfo(md)
if self.idinfo.datasetid:
self.identifier = self.idinfo.datasetid
reader = WPSExecuteReader(verbose=self.verbose)
if response is None:
# override status location
if url is not None:
self.statusLocation = url
log.info('\nChecking execution status... (location=%s)' % self.statusLocation)
try:
response = reader.readFromUrl( self.statusLocation, username=self.username, password=self.password, headers=self.headers, verify=self.verify, cert=self.cert)
except Exception as err:
log.error( "Could not read status document using url " + str(self.statusLocation) + ": " + str(err) + "\n" + traceback.format_exc() )
else:
response = reader.readFromString(response)
# store latest response
try:
xml = etree.tostring(response)
except Exception:
log.error("Could not parse XML response: " + str(response) )
else:
self.response = xml
# log.info(self.response)
self.parseResponse(response)
# sleep given number of seconds
if self.isComplete() is False:
log.info('Sleeping %d seconds...' % sleepSecs)
sleep(sleepSecs)
- xml_declaration (optional): whether to include xml declaration
"""
output = None
if encoding is None:
encoding = "ISO-8859-1"
if etree.__name__ == 'lxml.etree':
if xml_declaration:
if encoding in ['unicode', 'utf-8']:
output = '\n{}'.format(
etree.tostring(element, encoding='unicode'))
else:
output = etree.tostring(element, encoding=encoding, xml_declaration=True)
else:
output = etree.tostring(element)
else:
if xml_declaration:
output = '\n{}'.format(
encoding, etree.tostring(element, encoding=encoding))
else:
output = etree.tostring(element)
return output
if encoding is None:
encoding = "ISO-8859-1"
if etree.__name__ == 'lxml.etree':
if xml_declaration:
if encoding in ['unicode', 'utf-8']:
output = '\n{}'.format(
etree.tostring(element, encoding='unicode'))
else:
output = etree.tostring(element, encoding=encoding, xml_declaration=True)
else:
output = etree.tostring(element)
else:
if xml_declaration:
output = '\n{}'.format(
encoding, etree.tostring(element, encoding=encoding))
else:
output = etree.tostring(element)
return output
if url is not None:
self.statusLocation = url
log.info('\nChecking execution status... (location=%s)' %
self.statusLocation)
try:
response = reader.readFromUrl(
self.statusLocation, username=self.username, password=self.password,
headers=self.headers, verify=self.verify, cert=self.cert)
except Exception:
log.error("Could not read status document.")
else:
response = reader.readFromString(response)
# store latest response
try:
xml = etree.tostring(response)
except Exception:
log.error("Could not parse XML response.")
else:
self.response = xml
log.debug(self.response)
self.parseResponse(response)
# sleep given number of seconds
if self.isComplete() is False:
log.info('Sleeping %d seconds...' % sleepSecs)
sleep(sleepSecs)