voevent.py
14.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
"""Routines for handling etrees representing VOEvent packets."""
from __future__ import absolute_import
from __future__ import unicode_literals
from six import string_types
from lxml import objectify, etree
import collections
import voeventparse.definitions
voevent_v2_0_schema = etree.XMLSchema(
etree.fromstring(voeventparse.definitions.v2_0_schema_str))
def Voevent(stream, stream_id, role):
"""Create a new VOEvent element tree, with specified IVORN and role.
Args:
stream (string): 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 (string): See above.
role (string): 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: '
+ 'https://github.com/timstaley/voevent-parse')
return v
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:
voevent (:py:class:`Voevent`): Root-node of the etree.
Raises:
exceptions.ValueError
"""
# .. 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
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 (file): 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:
voevent (:py:class:`Voevent`): Root-node of the etree.
"""
s = file.read()
return loads(s, check_version)
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 containing raw XML representation of VOEvent.
"""
_return_to_standard_xml(voevent)
s = etree.tostring(voevent, pretty_print=pretty_print,
xml_declaration=xml_declaration,
encoding=encoding)
_remove_root_tag_prefix(voevent)
return s
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 (file): 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))
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 (VOEvent is valid?)
"""
_return_to_standard_xml(voevent)
valid_bool = voevent_v2_0_schema.validate(voevent)
_remove_root_tag_prefix(voevent)
return valid_bool
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.
Returns: None. NB 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)
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(string): 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()
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
def add_where_when(voevent, coords, obs_time, observatory_location):
"""Add details of an observation to the WhereWhen section.
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.
observatory_location(string): Telescope locale, e.g. 'La Palma'.
May be a generic location as listed under
:class:`voeventparse.definitions.observatory_location`.
"""
# .. todo:: Implement TimeError using datetime.timedelta
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 = 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
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(string): 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))
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))
def add_citations(voevent, citations):
"""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.
citation(:class:`voeventparse.misc.Citation`): Citation or list of citations.
"""
if not voevent.xpath('Citations'):
etree.SubElement(voevent, 'Citations')
voevent.Citations.extend(_listify(citations))
# ###################################################
# 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]