##################################################################
# 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
from pywps import xml_util 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