Source code for pywps.inout.outputs

##################################################################
# 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