Source code for pyvo.io.uws.tree

# Licensed under a 3-clause BSD style license - see LICENSE.rst
"""
This file contains xml element classes as defined in the VOResource standard.
"""
from functools import partial

from astropy.utils.collections import HomogeneousList
from astropy.time import Time, TimeDelta

from ...utils.xml.elements import (
    xmlattribute, xmlelement, Element, ContentMixin)

uwselement = partial(xmlelement, ns='uws')


def XSInDate(val):
    if not val:
        return None

    try:
        return Time(val, format='iso')
    except ValueError:
        pass

    try:
        return Time(val, format='isot')
    except ValueError:
        pass

    raise ValueError('Cannot parse datetime {}'.format(val))


InDuration = partial(TimeDelta, format='sec')
XSOutDate = partial(Time, out_subfmt='date')


__all__ = [
    'UWSElement', 'Reference', 'JobSummary', 'Parameters', 'Parameter',
    'Results', 'Result']


def _convert_boolean(value, default=None):
    return {
        'false': False,
        '0': False,
        'true': True,
        '1': True
    }.get(value, default)


[docs]class UWSElement(Element): def __init__(self, config=None, pos=None, _name='', _ns='uws', **kwargs): super().__init__(config, pos, _name, 'uws', **kwargs)
[docs]class Reference(UWSElement): """standard xlink references""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._type = kwargs.get('xlink:type') self._href = kwargs.get('xlink:href') @xmlattribute(name='xlink:type') def type(self): """the type of the result""" return self._type @type.setter def type(self, type_): self._type = type_ @xmlattribute(name='xlink:href') def href(self): """the url the result can be retrieved""" return self._href @href.setter def href(self, href): self._href = href
[docs]class JobSummary(Element): def __init__(self, config=None, pos=None, _name='job', **kwargs): super().__init__(config, pos, _name, **kwargs) self._jobid = kwargs.get('id') self._runid = None self._ownerid = None self._phase = None self._quote = None self._creationtime = None self._starttime = None self._endtime = None self._executionduration = None self._destruction = None self._parameters = Parameters() self._results = Results() @uwselement(name='jobId', plain=True) def jobid(self): """ The identifier for the job """ return self._jobid @jobid.setter def jobid(self, jobid): self._jobid = jobid @uwselement(name='runId', plain=True) def runid(self): """client supplied identifier""" return self._runid @runid.setter def runid(self, runid): self._runid = runid @uwselement(name='ownerId', plain=True) def ownerid(self): """the owner (creator) of the job""" return self._ownerid @ownerid.setter def ownerid(self, ownerid): self._ownerid = ownerid @uwselement(plain=True) def phase(self): """the execution phase""" return self._phase @phase.setter def phase(self, phase): self._phase = phase @uwselement(plain=True) def quote(self): """estimated completion time""" return self._quote @quote.setter def quote(self, quote): self._quote = XSInDate(quote) @quote.formatter def quote(self): try: return str(XSOutDate(self._quote)) except ValueError: return None @uwselement(name='creationTime', plain=True) def creationtime(self): """The instant at which the job was created.""" return self._creationtime @creationtime.setter def creationtime(self, creationtime): self._creationtime = XSInDate(creationtime) @creationtime.formatter def creationtime(self): try: return str(XSOutDate(self._creationtime)) except ValueError: return None @uwselement(name='startTime', plain=True) def starttime(self): """The instant at which the job started execution.""" return self._starttime @starttime.setter def starttime(self, starttime): self._starttime = XSInDate(starttime) @starttime.formatter def starttime(self): try: return str(XSOutDate(self._starttime)) except ValueError: return None @uwselement(name='endTime', plain=True) def endtime(self): """The instant at which the job finished execution""" return self._endtime @endtime.setter def endtime(self, endtime): self._endtime = XSInDate(endtime) @endtime.formatter def endtime(self): try: return str(XSOutDate(self._endtime)) except ValueError: return None @uwselement(name='executionDuration', plain=True) def executionduration(self): """ The duration (in seconds) for which the job should be allowed to run - a value of 0 is intended to mean unlimited """ return self._executionduration @executionduration.setter def executionduration(self, executionduration): if not isinstance(executionduration, TimeDelta): executionduration = InDuration(float(executionduration)) self._executionduration = executionduration @executionduration.formatter def executionduration(self): if self.executionduration: return str(int(self._executionduration.value)) @uwselement(plain=True) def destruction(self): """The time at which the whole job will be destroyed""" return self._destruction @destruction.setter def destruction(self, destruction): self._destruction = XSInDate(destruction) @destruction.formatter def destruction(self): try: return str(XSOutDate(self._destruction)) except ValueError: return None @uwselement def parameters(self): """The parameters to the job""" return self._parameters @parameters.adder def parameters(self, iterator, tag, data, config, pos): parameters = Parameters(config, pos, 'parameters', **data) parameters.parse(iterator, config) self._parameters = parameters @uwselement def results(self): """The results for the job""" return self._results @results.adder def results(self, iterator, tag, data, config, pos): results = Results(config, pos, 'results', **data) results.parse(iterator, config) self._results = results
class JobList(UWSElement, HomogeneousList): def __init__(self, config=None, pos=None, _name='jobs', **kwargs): HomogeneousList.__init__(self, JobSummary) UWSElement.__init__(self, config, pos, _name, **kwargs) @uwselement(name='jobref') def jobs(self): return self @jobs.adder def jobs(self, iterator, tag, data, config, pos): job = JobSummary(config, pos, 'jobref', **data) job.parse(iterator, config) self.append(job)
[docs]class Parameters(UWSElement, HomogeneousList): """ Parameters element of a job """ def __init__(self, config=None, pos=None, _name='parameters', **kwargs): """ """ # Note: Above is a load-bearing empty comment. # Do not remove, or else the Sphinx build may fail (see PR #193). HomogeneousList.__init__(self, Parameter) UWSElement.__init__(self, config, pos, _name, **kwargs) @uwselement(name='parameter') def parameters(self): return self @parameters.adder def parameters(self, iterator, tag, data, config, pos): parameter = Parameter(config, pos, 'parameter', **data) parameter.parse(iterator, config) self.append(parameter)
[docs]class Parameter(ContentMixin, UWSElement): def __init__(self, config=None, pos=None, _name='parameter', **kwargs): super().__init__(config, pos, _name, **kwargs) self._byreference = _convert_boolean(kwargs.get('byReference')) self._id = kwargs.get('id') @xmlattribute def byreference(self): """ if this attribute is true then the content of the parameter represents a URL to retrieve the actual parameter value. """ return self._byreference @byreference.setter def byreference(self, byreference): self._byreference = byreference @xmlattribute(name='id') def id_(self): """the identifier for the parameter""" return self._id @id_.setter def id_(self, id_): self._id = id_
[docs]class Results(UWSElement, HomogeneousList): def __init__(self, config=None, pos=None, _name='results', **kwargs): HomogeneousList.__init__(self, Result) UWSElement.__init__(self, config, pos, _name, **kwargs) @uwselement(name='result') def results(self): return self @results.adder def results(self, iterator, tag, data, config, pos): result = Result(config, pos, 'result', **data) result.parse(iterator, config) self.append(result)
[docs]class Result(Reference, UWSElement): """A reference to a UWS result.""" def __init__(self, config=None, pos=None, _name='result', **kwargs): super().__init__(config, pos, _name, **kwargs) self._id = kwargs.get('id') self._size = int(kwargs.get('size') or 0) self._mimetype = kwargs.get('mime-type') @xmlattribute(name='id') def id_(self): """the identifier for the result""" return self._id @id_.setter def id_(self, id_): self._id = id_ @xmlattribute def size(self): """the size of the result""" return self._size @size.setter def size(self, size): self._size = size @xmlattribute def mimetype(self): """the mimetype of the result""" return self._mimetype @mimetype.setter def mimetype(self, mimetype): self._mimetype = mimetype