##################################################################
# Copyright 2016 OSGeo Foundation, #
# represented by PyWPS Project Steering Committee, #
# licensed under MIT, Please consult LICENSE.txt for details #
##################################################################
from pywps._compat import text_type
from pywps import E, WPS, OWS, OGCTYPE, NAMESPACES
from pywps.inout import basic
from pywps.inout.storage import FileStorage
from pywps.inout.formats import Format
from pywps.validator.mode import MODE
import lxml.etree as etree
import six
[docs]class BoundingBoxOutput(basic.BBoxInput):
"""
:param identifier: The name of this input.
:param str title: Title of the input
:param str abstract: Input abstract
:param crss: List of supported coordinate reference system (e.g. ['EPSG:4326'])
:param int dimensions: number of dimensions (2 or 3)
:param int min_occurs: minimum occurence
:param int max_occurs: maximum occurence
:param pywps.validator.mode.MODE mode: validation mode (none to strict)
:param metadata: List of metadata advertised by this process. They
should be :class:`pywps.app.Common.Metadata` objects.
"""
def __init__(self, identifier, title, crss, abstract='',
dimensions=2, metadata=[], min_occurs='1',
max_occurs='1', as_reference=False,
mode=MODE.NONE):
basic.BBoxInput.__init__(self, identifier, title=title,
abstract=abstract, crss=crss,
dimensions=dimensions, mode=mode)
self.metadata = metadata
self.min_occurs = min_occurs
self.max_occurs = max_occurs
self.as_reference = as_reference
def describe_xml(self):
doc = E.Output(
OWS.Identifier(self.identifier),
OWS.Title(self.title)
)
if self.abstract:
doc.append(OWS.Abstract(self.abstract))
for m in self.metadata:
doc.append(OWS.Metadata(dict(m)))
bbox_data_doc = E.BoundingBoxOutput()
doc.append(bbox_data_doc)
default_doc = E.Default()
default_doc.append(E.CRS(self.crss[0]))
supported_doc = E.Supported()
for c in self.crss:
supported_doc.append(E.CRS(c))
bbox_data_doc.append(default_doc)
bbox_data_doc.append(supported_doc)
return doc
def execute_xml(self):
doc = E.Output(
OWS.Identifier(self.identifier),
OWS.Title(self.title)
)
if self.abstract:
doc.append(OWS.Abstract(self.abstract))
bbox_data_doc = OWS.BoundingBox()
bbox_data_doc.attrib['crs'] = self.crs
bbox_data_doc.attrib['dimensions'] = str(self.dimensions)
bbox_data_doc.append(OWS.LowerCorner('{0[0]} {0[1]}'.format(self.data)))
bbox_data_doc.append(OWS.UpperCorner('{0[2]} {0[3]}'.format(self.data)))
doc.append(bbox_data_doc)
return doc
[docs]class ComplexOutput(basic.ComplexOutput):
"""
:param identifier: The name of this output.
:param title: Readable form of the output name.
:param pywps.inout.formats.Format supported_formats: List of supported
formats. The first format in the list will be used as the default.
:param str abstract: Description of the output
:param pywps.validator.mode.MODE mode: validation mode (none to strict)
:param metadata: List of metadata advertised by this process. They
should be :class:`pywps.app.Common.Metadata` objects.
"""
def __init__(self, identifier, title, supported_formats=None,
abstract='', metadata=None,
as_reference=False, mode=MODE.NONE):
if metadata is None:
metadata = []
basic.ComplexOutput.__init__(self, identifier, title=title,
abstract=abstract,
supported_formats=supported_formats,
mode=mode)
self.metadata = metadata
self.as_reference = as_reference
self.storage = None
def describe_xml(self):
"""Generate DescribeProcess element
"""
default_format_el = self.supported_formats[0].describe_xml()
supported_format_elements = [f.describe_xml() for f in self.supported_formats]
doc = E.Output(
OWS.Identifier(self.identifier),
OWS.Title(self.title)
)
if self.abstract:
doc.append(OWS.Abstract(self.abstract))
for m in self.metadata:
doc.append(OWS.Metadata(dict(m)))
doc.append(
E.ComplexOutput(
E.Default(default_format_el),
E.Supported(*supported_format_elements)
)
)
return doc
def execute_xml_lineage(self):
doc = WPS.Output(
OWS.Identifier(self.identifier),
OWS.Title(self.title)
)
if self.abstract:
doc.append(OWS.Abstract(self.abstract))
return doc
def execute_xml(self):
"""Render Execute response XML node
:return: node
:rtype: ElementMaker
"""
self.identifier
node = None
if self.as_reference:
node = self._execute_xml_reference()
else:
node = self._execute_xml_data()
doc = WPS.Output(
OWS.Identifier(self.identifier),
OWS.Title(self.title)
)
if self.abstract:
doc.append(OWS.Abstract(self.abstract))
doc.append(node)
return doc
def _execute_xml_reference(self):
"""Return Reference node
"""
doc = WPS.Reference()
# get_url will create the file and return the url for it
self.storage = FileStorage()
doc.attrib['{http://www.w3.org/1999/xlink}href'] = self.get_url()
if self.data_format:
if self.data_format.mime_type:
doc.attrib['mimeType'] = self.data_format.mime_type
if self.data_format.encoding:
doc.attrib['encoding'] = self.data_format.encoding
if self.data_format.schema:
doc.attrib['schema'] = self.data_format.schema
return doc
def _execute_xml_data(self):
"""Return Data node
"""
doc = WPS.Data()
if self.data is None:
complex_doc = WPS.ComplexData()
else:
complex_doc = WPS.ComplexData()
try:
data_doc = etree.parse(self.file)
complex_doc.append(data_doc.getroot())
except:
if isinstance(self.data, six.string_types):
complex_doc.text = self.data
else:
complex_doc.text = etree.CDATA(self.base64)
if self.data_format:
if self.data_format.mime_type:
complex_doc.attrib['mimeType'] = self.data_format.mime_type
if self.data_format.encoding:
complex_doc.attrib['encoding'] = self.data_format.encoding
if self.data_format.schema:
complex_doc.attrib['schema'] = self.data_format.schema
doc.append(complex_doc)
return doc
[docs]class LiteralOutput(basic.LiteralOutput):
"""
:param identifier: The name of this output.
:param str title: Title of the input
:param pywps.inout.literaltypes.LITERAL_DATA_TYPES data_type: data type
:param str abstract: Input abstract
:param str uoms: units
:param pywps.validator.mode.MODE mode: validation mode (none to strict)
:param metadata: List of metadata advertised by this process. They
should be :class:`pywps.app.Common.Metadata` objects.
"""
def __init__(self, identifier, title, data_type='string', abstract='',
metadata=[], uoms=[], mode=MODE.SIMPLE):
if uoms is None:
uoms = []
basic.LiteralOutput.__init__(self, identifier, title=title,
data_type=data_type, uoms=uoms, mode=mode)
self.abstract = abstract
self.metadata = metadata
def describe_xml(self):
doc = E.Output(
OWS.Identifier(self.identifier),
OWS.Title(self.title)
)
if self.abstract:
doc.append(OWS.Abstract(self.abstract))
for m in self.metadata:
doc.append(OWS.Metadata(dict(m)))
literal_data_doc = E.LiteralOutput()
if self.data_type:
data_type = OWS.DataType(self.data_type)
data_type.attrib['{%s}reference' % NAMESPACES['ows']] = OGCTYPE[self.data_type]
literal_data_doc.append(data_type)
if self.uoms:
default_uom_element = self.uom.describe_xml()
supported_uom_elements = [u.describe_xml() for u in self.uoms]
literal_data_doc.append(
E.UOMs(
E.Default(default_uom_element),
E.Supported(*supported_uom_elements)
)
)
doc.append(literal_data_doc)
return doc
def execute_xml_lineage(self):
doc = WPS.Output(
OWS.Identifier(self.identifier),
OWS.Title(self.title)
)
if self.abstract:
doc.append(OWS.Abstract(self.abstract))
return doc
def execute_xml(self):
doc = WPS.Output(
OWS.Identifier(self.identifier),
OWS.Title(self.title)
)
if self.abstract:
doc.append(OWS.Abstract(self.abstract))
data_doc = WPS.Data()
literal_data_doc = WPS.LiteralData(text_type(self.data))
literal_data_doc.attrib['dataType'] = OGCTYPE[self.data_type]
if self.uom:
literal_data_doc.attrib['uom'] = self.uom.execute_attribute()
data_doc.append(literal_data_doc)
doc.append(data_doc)
return doc