"""Routines for handling etrees representing VOEvent packets."""
from __future__ import absolute_import
from __future__ import unicode_literals
import copy
import collections
import pytz
from lxml import objectify, etree
from six import string_types
import voeventparse.definitions
voevent_v2_0_schema = etree.XMLSchema(
etree.fromstring(voeventparse.definitions.v2_0_schema_str))
from ._version import get_versions
__version__ = get_versions()['version']
[docs]def Voevent(stream, stream_id, role):
"""Create a new VOEvent element tree, with specified IVORN and role.
Args:
stream (str): used to construct the IVORN like so::
ivorn = 'ivo://' + stream + '#' + stream_id
(N.B. ``stream_id`` is converted to string if required.)
So, e.g. we might set::
stream='voevent.soton.ac.uk/super_exciting_events'
stream_id=77
stream_id (str): See above.
role (str): role as defined in VOEvent spec.
(See also :py:class:`.definitions.roles`)
Returns:
Root-node of the VOEvent, as represented by an lxml.objectify element
tree ('etree'). See also
http://lxml.de/objectify.html#the-lxml-objectify-api
"""
parser = objectify.makeparser(remove_blank_text=True)
v = objectify.fromstring(voeventparse.definitions.v2_0_skeleton_str,
parser=parser)
_remove_root_tag_prefix(v)
if not isinstance(stream_id, string_types):
stream_id = repr(stream_id)
v.attrib['ivorn'] = ''.join(('ivo://', stream, '#', stream_id))
v.attrib['role'] = role
# Presumably we'll always want the following children:
# (NB, valid to then leave them empty)
etree.SubElement(v, 'Who')
etree.SubElement(v, 'What')
etree.SubElement(v, 'WhereWhen')
v.Who.Description = (
'VOEvent created with voevent-parse, version {}. '
'See https://github.com/timstaley/voevent-parse for details.').format(
__version__
)
return v
[docs]def loads(s, check_version=True):
"""
Load VOEvent from bytes.
This parses a VOEvent XML packet string, taking care of some subtleties.
For Python 3 users, ``s`` should be a bytes object - see also
http://lxml.de/FAQ.html,
"Why can't lxml parse my XML from unicode strings?"
(Python 2 users can stick with old-school ``str`` type if preferred)
By default, will raise an exception if the VOEvent is not of version
2.0. This can be disabled but voevent-parse routines are untested with
other versions.
Args:
s (bytes): Bytes containing raw XML.
check_version (bool): (Default=True) Checks that the VOEvent is of a
supported schema version - currently only v2.0 is supported.
Returns:
:py:class:`Voevent`: Root-node of the etree.
Raises:
ValueError: If passed a VOEvent of wrong schema version
(i.e. schema 1.1)
"""
# .. note::
#
# The namespace is removed from the root element tag to make
# objectify access work as expected,
# (see :py:func:`._remove_root_tag_prefix`)
# so we must re-insert it when we want to conform to schema.
v = objectify.fromstring(s)
_remove_root_tag_prefix(v)
if check_version:
version = v.attrib['version']
if not version == '2.0':
raise ValueError('Unsupported VOEvent schema version:' + version)
return v
[docs]def load(file, check_version=True):
"""Load VOEvent from file object.
A simple wrapper to read a file before passing the contents to
:py:func:`.loads`. Use with an open file object, e.g.::
with open('/path/to/voevent.xml', 'rb') as f:
v = vp.load(f)
Args:
file (io.IOBase): An open file object (binary mode preferred), see also
http://lxml.de/FAQ.html :
"Can lxml parse from file objects opened in unicode/text mode?"
check_version (bool): (Default=True) Checks that the VOEvent is of a
supported schema version - currently only v2.0 is supported.
Returns:
:py:class:`Voevent`: Root-node of the etree.
"""
s = file.read()
return loads(s, check_version)
[docs]def dumps(voevent, pretty_print=False, xml_declaration=True, encoding='UTF-8'):
"""Converts voevent to string.
.. note:: Default encoding is UTF-8, in line with VOE2.0 schema.
Declaring the encoding can cause diffs with the original loaded VOEvent,
but I think it's probably the right thing to do (and lxml doesn't
really give you a choice anyway).
Args:
voevent (:class:`Voevent`): Root node of the VOevent etree.
pretty_print (bool): indent the output for improved human-legibility
when possible. See also:
http://lxml.de/FAQ.html#why-doesn-t-the-pretty-print-option-reformat-my-xml-output
xml_declaration (bool): Prepends a doctype tag to the string output,
i.e. something like ``<?xml version='1.0' encoding='UTF-8'?>``
Returns:
bytes: Bytestring containing raw XML representation of VOEvent.
"""
vcopy = copy.deepcopy(voevent)
_return_to_standard_xml(vcopy)
s = etree.tostring(vcopy, pretty_print=pretty_print,
xml_declaration=xml_declaration,
encoding=encoding)
return s
[docs]def dump(voevent, file, pretty_print=True, xml_declaration=True):
"""Writes the voevent to the file object.
e.g.::
with open('/tmp/myvoevent.xml','wb') as f:
voeventparse.dump(v, f)
Args:
voevent(:class:`Voevent`): Root node of the VOevent etree.
file (io.IOBase): An open (binary mode) file object for writing.
pretty_print
pretty_print(bool): See :func:`dumps`
xml_declaration(bool): See :func:`dumps`
"""
file.write(dumps(voevent, pretty_print, xml_declaration))
[docs]def valid_as_v2_0(voevent):
"""Tests if a voevent conforms to the schema.
Args:
voevent(:class:`Voevent`): Root node of a VOEvent etree.
Returns:
bool: Whether VOEvent is valid
"""
_return_to_standard_xml(voevent)
valid_bool = voevent_v2_0_schema.validate(voevent)
_remove_root_tag_prefix(voevent)
return valid_bool
[docs]def assert_valid_as_v2_0(voevent):
"""
Raises :py:obj:`lxml.etree.DocumentInvalid` if voevent is invalid.
Especially useful for debugging,
since the stack trace contains a reason for the invalidation.
Args:
voevent(:class:`Voevent`): Root node of a VOEvent etree.
Raises:
:py:obj:`lxml.etree.DocumentInvalid`: if VOEvent does not conform to
schema.
"""
_return_to_standard_xml(voevent)
voevent_v2_0_schema.assertValid(voevent)
_remove_root_tag_prefix(voevent)
[docs]def set_who(voevent, date=None, author_ivorn=None):
"""Sets the minimal 'Who' attributes: date of authoring, AuthorIVORN.
Args:
voevent(:class:`Voevent`): Root node of a VOEvent etree.
date(datetime.datetime): Date of authoring.
NB Microseconds are ignored, as per the VOEvent spec.
author_ivorn(str): Short author identifier,
e.g. ``voevent.4pisky.org/ALARRM``.
Note that the prefix ``ivo://`` will be prepended internally.
"""
if author_ivorn is not None:
voevent.Who.AuthorIVORN = ''.join(('ivo://', author_ivorn))
if date is not None:
voevent.Who.Date = date.replace(microsecond=0).isoformat()
[docs]def set_author(voevent, title=None, shortName=None, logoURL=None,
contactName=None, contactEmail=None, contactPhone=None,
contributor=None):
"""For setting fields in the detailed author description.
This can optionally be neglected if a well defined AuthorIVORN is supplied.
.. note:: Unusually for this library,
the args here use CamelCase naming convention,
since there's a direct mapping to the ``Author.*``
attributes to which they will be assigned.
Args:
voevent(:class:`Voevent`): Root node of a VOEvent etree.
The rest of the arguments are strings corresponding to child elements.
"""
# We inspect all local variables except the voevent packet,
# Cycling through and assigning them on the Who.Author element.
AuthChildren = locals()
AuthChildren.pop('voevent')
if not voevent.xpath('Who/Author'):
etree.SubElement(voevent.Who, 'Author')
for k, v in AuthChildren.items():
if v is not None:
voevent.Who.Author[k] = v
[docs]def add_where_when(voevent, coords, obs_time, observatory_location,
allow_tz_naive_datetime=False):
"""
Add details of an observation to the WhereWhen section.
We
Args:
voevent(:class:`Voevent`): Root node of a VOEvent etree.
coords(:class:`.Position2D`): Sky co-ordinates of event.
obs_time(datetime.datetime): Nominal DateTime of the observation. Must
either be timezone-aware, or should be carefully verified as
representing UTC and then set parameter
``allow_tz_naive_datetime=True``.
observatory_location(str): Telescope locale, e.g. 'La Palma'.
May be a generic location as listed under
:class:`voeventparse.definitions.observatory_location`.
allow_tz_naive_datetime (bool): (Default False). Accept timezone-naive
datetime-timestamps. See comments for ``obs_time``.
"""
# .. todo:: Implement TimeError using datetime.timedelta
if obs_time.tzinfo is not None:
utc_naive_obs_time = obs_time.astimezone(pytz.utc).replace(tzinfo=None)
elif not allow_tz_naive_datetime:
raise ValueError(
"Datetime passed without tzinfo, cannot be sure if it is really a "
"UTC timestamp. Please verify function call and either add tzinfo "
"or pass parameter 'allow_tz_naive_obstime=True', as appropriate",
)
else:
utc_naive_obs_time = obs_time
obs_data = etree.SubElement(voevent.WhereWhen, 'ObsDataLocation')
etree.SubElement(obs_data, 'ObservatoryLocation', id=observatory_location)
ol = etree.SubElement(obs_data, 'ObservationLocation')
etree.SubElement(ol, 'AstroCoordSystem', id=coords.system)
ac = etree.SubElement(ol, 'AstroCoords',
coord_system_id=coords.system)
time = etree.SubElement(ac, 'Time', unit='s')
instant = etree.SubElement(time, 'TimeInstant')
instant.ISOTime = utc_naive_obs_time.isoformat()
# iso_time = etree.SubElement(instant, 'ISOTime') = obs_time.isoformat()
pos2d = etree.SubElement(ac, 'Position2D', unit=coords.units)
pos2d.Name1 = 'RA'
pos2d.Name2 = 'Dec'
pos2d_val = etree.SubElement(pos2d, 'Value2')
pos2d_val.C1 = coords.ra
pos2d_val.C2 = coords.dec
pos2d.Error2Radius = coords.err
[docs]def add_how(voevent, descriptions=None, references=None):
"""Add descriptions or references to the How section.
Args:
voevent(:class:`Voevent`): Root node of a VOEvent etree.
descriptions(str): Description string, or list of description
strings.
references(:py:class:`voeventparse.misc.Reference`): A reference element
(or list thereof).
"""
if not voevent.xpath('How'):
etree.SubElement(voevent, 'How')
if descriptions is not None:
for desc in _listify(descriptions):
# d = etree.SubElement(voevent.How, 'Description')
# voevent.How.Description[voevent.How.index(d)] = desc
##Simpler:
etree.SubElement(voevent.How, 'Description')
voevent.How.Description[-1] = desc
if references is not None:
voevent.How.extend(_listify(references))
[docs]def add_why(voevent, importance=None, expires=None, inferences=None):
"""Add Inferences, or set importance / expires attributes of the Why section.
.. note::
``importance`` / ``expires`` are 'Why' attributes, therefore setting them
will overwrite previous values.
``inferences``, on the other hand, are appended to the list.
Args:
voevent(:class:`Voevent`): Root node of a VOEvent etree.
importance(float): Value from 0.0 to 1.0
expires(datetime.datetime): Expiration date given inferred reason
(See voevent spec).
inferences(:class:`voeventparse.misc.Inference`): Inference or list of
inferences, denoting probable identifications or associations, etc.
"""
if not voevent.xpath('Why'):
etree.SubElement(voevent, 'Why')
if importance is not None:
voevent.Why.attrib['importance'] = str(importance)
if expires is not None:
voevent.Why.attrib['expires'] = expires.replace(
microsecond=0).isoformat()
if inferences is not None:
voevent.Why.extend(_listify(inferences))
[docs]def add_citations(voevent, event_ivorns):
"""Add citations to other voevents.
The schema mandates that the 'Citations' section must either be entirely
absent, or non-empty - hence we require this wrapper function for its
creation prior to listing the first citation.
Args:
voevent(:class:`Voevent`): Root node of a VOEvent etree.
event_ivorns (:class:`voeventparse.misc.EventIvorn`): List of EventIvorn
elements to add to citation list.
"""
if not voevent.xpath('Citations'):
etree.SubElement(voevent, 'Citations')
voevent.Citations.extend(_listify(event_ivorns))
# ###################################################
# And finally, utility functions...
def _remove_root_tag_prefix(v):
"""
Removes 'voe' namespace prefix from root tag.
When we load in a VOEvent, the root element has a tag prefixed by
the VOE namespace, e.g. {http://www.ivoa.net/xml/VOEvent/v2.0}VOEvent
Because objectify expects child elements to have the same namespace as
their parent, this breaks the python-attribute style access mechanism.
We can get around it without altering root, via e.g
who = v['{}Who']
Alternatively, we can temporarily ditch the namespace altogether.
This makes access to elements easier, but requires care to reinsert
the namespace upon output.
I've gone for the latter option.
"""
if v.prefix:
# Create subelement without a prefix via etree.SubElement
etree.SubElement(v, 'original_prefix')
# Now carefully access said named subelement (without prefix cascade)
# and alter the first value in the list of children with this name...
# LXML syntax is a minefield!
v['{}original_prefix'][0] = v.prefix
v.tag = v.tag.replace(''.join(('{', v.nsmap[v.prefix], '}')), '')
# Now v.tag = '{}VOEvent', v.prefix = None
return
def _reinsert_root_tag_prefix(v):
"""
Returns namespace prefix to root tag, if it had one.
"""
if hasattr(v, 'original_prefix'):
original_prefix = v.original_prefix
del v.original_prefix
v.tag = ''.join(('{', v.nsmap[original_prefix], '}VOEvent'))
return
def _return_to_standard_xml(v):
# Remove lxml.objectify DataType namespace prefixes:
objectify.deannotate(v)
# Put the default namespace back:
_reinsert_root_tag_prefix(v)
etree.cleanup_namespaces(v)
# Define this for convenience in add_how:
def _listify(x):
"""Ensure x is iterable; if not then enclose it in a list and return it."""
if isinstance(x, string_types):
return [x]
elif isinstance(x, collections.Iterable):
return x
else:
return [x]