Source code for pywps.inout.outputs

##################################################################
# Copyright 2018 Open Source Geospatial Foundation and others    #
# licensed under MIT, Please consult LICENSE.txt for details     #
##################################################################
"""
WPS Output classes
"""

import lxml.etree as etree
import os
from pywps.inout import basic
from pywps.inout.storage import FileStorage
from pywps.validator.mode import MODE
from pywps import configuration as config


[docs]class BoundingBoxOutput(basic.BBoxOutput): """ :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='', keywords=[], dimensions=2, metadata=[], min_occurs='1', max_occurs='1', as_reference=False, mode=MODE.NONE): basic.BBoxOutput.__init__(self, identifier, title=title, abstract=abstract, keywords=keywords, crss=crss, dimensions=dimensions, mode=mode) self.metadata = metadata self.min_occurs = min_occurs self.max_occurs = max_occurs self.as_reference = as_reference @property def json(self): """Get JSON representation of the output """ return { 'identifier': self.identifier, 'title': self.title, 'abstract': self.abstract, 'keywords': self.keywords, 'min_occurs': self.min_occurs, 'max_occurs': self.max_occurs, 'metadata': self.metadata, 'type': 'bbox', 'crs': self.crs, 'crss': self.crss, 'dimensions': self.dimensions, 'bbox': (self.ll, self.ur), 'workdir': self.workdir, 'mode': self.valid_mode, }
[docs]class ComplexOutput(basic.ComplexOutput): """ :param identifier: The name of this output. :param title: Readable form of the output name. :param supported_formats: List of supported formats. The first format in the list will be used as the default. :type supported_formats: (pywps.inout.formats.Format, ) :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='', keywords=[], metadata=None, as_reference=False, mode=MODE.NONE): if metadata is None: metadata = [] basic.ComplexOutput.__init__(self, identifier, title=title, abstract=abstract, keywords=keywords, supported_formats=supported_formats, mode=mode) self.metadata = metadata self.as_reference = as_reference self.storage = None @property def json(self): """Get JSON representation of the output """ data = { "identifier": self.identifier, "title": self.title, "abstract": self.abstract, 'keywords': self.keywords, 'type': 'complex', 'supported_formats': [frmt.json for frmt in self.supported_formats], 'asreference': self.as_reference, 'data_format': self.data_format.json if self.data_format else None, 'file': self.file if self.prop == 'file' else None, 'workdir': self.workdir, 'mode': self.valid_mode, 'min_occurs': self.min_occurs, 'max_occurs': self.max_occurs } if self.as_reference: data = self._json_reference(data) else: data = self._json_data(data) if self.data_format: if self.data_format.mime_type: data['mimetype'] = self.data_format.mime_type if self.data_format.encoding: data['encoding'] = self.data_format.encoding if self.data_format.schema: data['schema'] = self.data_format.schema return data def _json_reference(self, data): """Return Reference node """ data["type"] = "reference" # get_url will create the file and return the url for it if self.prop == 'url': data["href"] = self.url elif self.prop is not None: self.storage = FileStorage() data["href"] = self.get_url() return data def _json_data(self, data): """Return Data node """ data["type"] = "complex" if self.data: if self.data_format.mime_type in ["application/xml", "application/gml+xml", "text/xml"]: # Note that in a client-server round trip, the original and returned file will not be identical. data_doc = etree.parse(self.file) data["data"] = etree.tostring(data_doc, pretty_print=True).decode('utf-8') else: if self.data_format.encoding == 'base64': data["data"] = self.base64.decode('utf-8') else: # Otherwise we assume all other formats are unsafe and need to be enclosed in a CDATA tag. if isinstance(self.data, bytes): out = self.data.encode(self.data_format.encoding or 'utf-8') else: out = self.data data["data"] = u'<![CDATA[{}]]>'.format(out) return data
[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='', keywords=[], metadata=[], uoms=None, mode=MODE.SIMPLE): if uoms is None: uoms = [] basic.LiteralOutput.__init__(self, identifier, title=title, abstract=abstract, keywords=keywords, data_type=data_type, uoms=uoms, mode=mode) self.metadata = metadata @property def json(self): """Get JSON representation of the output """ data = { "identifier": self.identifier, "title": self.title, "abstract": self.abstract, "keywords": self.keywords, "data": self.data, "data_type": self.data_type, "type": "literal", "uoms": [u.json for u in self.uoms] } if self.uom: data["uom"] = self.uom.json return data
class MetaFile: """MetaFile object.""" def __init__(self, identity=None, description=None, fmt=None): """Create a `MetaFile` object. :param str identity: human readable identity. :param str description: human readable file description. :param pywps.FORMAT fmt: file mime type. The content of each metafile is set like `ComplexOutputs`, ie using either the `data`, `file`, `stream` or `url` properties. The metalink document is created by a `MetaLink` instance, which holds a number of `MetaFile` instances. """ self._output = ComplexOutput( identifier=identity or '', title=description or '', as_reference=True, supported_formats=[fmt, ], ) def _set_workdir(self, workdir): self._output.workdir = workdir @property def hash(self): """Text construct that conveys a cryptographic hash for a file. All hashes are encoded in lowercase hexadecimal format. Hashes are used to verify the integrity of a complete file or portion of a file to determine if the file has been transferred without any errors. """ import hashlib m = hashlib.sha256() with open(self.file, "rb") as f: for chunk in iter(lambda: f.read(4096), b""): m.update(chunk) return m.hexdigest() @property def identity(self): """Human-readable identity.""" return self._output.identifier @property def name(self): """Indicate a specific file in a document describing multiple files.""" js = self._output.json (_, name) = os.path.split(js.get('href', 'http:///')) return name @property def size(self): """Length of the linked content in octets.""" return os.stat(self.file).st_size @property def urls(self): js = self._output.json return [js.get('href', ''), ] @property def mediatype(self): """Multipurpose Internet Mail Extensions (MIME) media type [RFC4288] of the metadata file available at the IRI.""" return self._output.data_format.mime_type @property def data(self): return self._output.data @data.setter def data(self, value): self._output.data = value @property def file(self): return self._output.file @file.setter def file(self, value): self._output.file = value @property def url(self): return self._output.url @url.setter def url(self, value): self._output.url = value @property def stream(self): return self._output.stream @stream.setter def stream(self, value): self._output.stream = value def __str__(self): out = "MetaFile {}:".format(self.name) for url in self.urls: out += "\n\t{}".format(url) return out def __repr__(self): return "<pywps.inout.outputs.MetaFile {}>".format(self.name) class MetaLink: _xml_template = 'metalink/3.0/main.xml' # Specs: https://www.metalinker.org/Metalink_3.0_Spec.pdf def __init__(self, identity=None, description=None, publisher=None, files=(), workdir=None, checksums=False): """Create a MetaLink v3.0 instance. :param str identity: human readable identity. :param str description: human readable file description. :param str publisher: The name of the file's publisher. :param tuple files: Sequence of files to include in Metalink. Can also be added using `append`. :param str workdir: Work directory to store temporary files. :param bool checksums: Whether to compute checksums on files. To use, first append `MetaFile` instances, then write the metalink using the `xml` property. Methods: - `append`: add a `MetaFile` instance """ self.identity = identity self.description = description self.workdir = workdir self.publisher = publisher self.files = [] self.checksums = checksums for file in files: self.append(file) self._load_template() def append(self, file): """Append a `MetaFile` instance.""" if not isinstance(file, MetaFile): raise ValueError("file must be a MetaFile instance.") file._set_workdir(self.workdir) self.files.append(file) @property def xml(self): return self._template.render(meta=self) @property def origin(self): """IRI where the Metalink Document was originally published. If the dynamic attribute of metalink:origin is "true", then updated versions of the Metalink can be found at this IRI. """ return "" @property def published(self): """Date construct indicating an instant in time associated with an event early in the life cycle of the entry.""" import datetime return datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ") @property def generator(self): """Generating agent name and version used to generate a Metalink Document, for debugging and other purposes.""" import pywps return "PyWPS/{}".format(pywps.__version__) @property def url(self): """Return the server URL.""" return config.get_config_value('server', 'url') def _load_template(self): from pywps.response import RelEnvironment from jinja2 import PackageLoader template_env = RelEnvironment( loader=PackageLoader('pywps', 'templates'), trim_blocks=True, lstrip_blocks=True, autoescape=True, ) self._template = template_env.get_template(self._xml_template) class MetaLink4(MetaLink): _xml_template = 'metalink/4.0/main.xml' # Specs: https://tools.ietf.org/html/rfc5854