''' A module to read the aspera event file
ASPERA-4 event file is a list of the command
that is used for data analysis.
*Preparation*
Download the file, http://aspera-4.irf.se/archive/commanding/pmrqs/EV_ASPERA.VEX and
put the downloaded path to ``.irfpyrc``.
::
[vexop]
default_eventfile = /Volumes/scidata/data/venus/support/EV_ASPERA.VEX
*Test*
Run the following
>>> from irfpy.vexop import eventfile
>>> ev = eventfile.default_eventfile()
>>> print(ev)
<VEX event file: Nr_of_entry=110134>
*How to use?*
Sample 1: Find out switch on time of ASPERA-4.
Switch on of ASPERA is done by commanding "A4_ON".
>>> on_events = list(filter(lambda e: e.eventname == 'A4_ON', ev.list)) # List of ``Event`` object.
>>> print(len(on_events))
5886
>>> print(on_events[0].utc)
2006-05-14 00:43:57
*More reading*
For more details, look at the tutorial, :ref:`eventfile`.
.. todo::
To make an event catalog. It could be in a form of python code, or in a document.
'''
import logging as _logging
_logger = _logging.getLogger(__name__)
import datetime
import dateutil.parser
from irfpy.util.irfpyrc import Rc as _Rc
[docs]class EventFile:
''' The event file.
:param filename: The file name of the event file.
:type: ``string``
The event file is loaded, parsed, and stored in the memory
as an object.
Each event is stored as an instance of :class:`Event`.
Usually, one can use direct access to the member :attr:`list`.
Below is an sample.
.. code-block:: py
>> event = EventFile(filename)
>> for ev in event.list:
.. print ev
.. # Some processing
'''
def __init__(self, filename):
# self.logger.setLevel(logging.DEBUG)
_logger.debug('Instance with %s' % filename)
self.filename = filename
self.version = None
self.list = []
''' List of event
'''
self.parse()
def __len__(self):
return len(self.list)
[docs] def parse(self):
fp = open(self.filename)
# First look for VERSION information
for line in fp:
if line.find('VERSION') > 0:
lsplit = line.split()
self.version = lsplit[-1]
_logger.debug('VERSION = %s' % self.version)
break
# If VERSION info is not found...
if self.version is None:
msg = " ".join(("The file %s is not event file." % self.filename,
"No version information found.", ))
raise IOError(msg)
# Parsing
for line in fp:
if line.startswith('#'):
continue
event = Event.parse(line)
self.list.append(event)
fp.close()
def __str__(self):
return "<VEX event file: Nr_of_entry={}>".format(len(self))
[docs]class Event:
''' Class of an event.
The class behave as a structure. Member variables (attributes) are:
:eventname: Name of the event, e.g. "A4_ON". ``string``
:orbnr: Orbit nubmer, e.g., 328. ``int``
:orbtime: Time relative to the pericenter, e.g., ``datetime.timedelta(0, 8709)``. ``datetime.timedelta``
:utc: Time in UTC. ``datetime.datetime``
:cmdname: Name of the command, e.g. "AASF001A". ``string``
:parameters: Dictionary of parameters.
``dict`` of ``string`` to ``string``.
For example, ``{'VASD6561': '0', 'VASD6563': '1', 'VASD6566': '4', 'VASD6567': '116'}``
'''
def __init__(self, evtname, orbnr, orbtime, utc, cmdname, prms):
self.eventname = evtname
self.orbnr = orbnr
self.orbtime = orbtime
self.utc = utc
self.cmdname = cmdname
self.parameters = prms
def __repr__(self):
s = (self.eventname + 12 * ' ')[:12]
s = s + ' %04d ' % self.orbnr
s = s + self.utc.strftime('%FT%T')
s = s + ' ' + self.cmdname
s = s + ' ' + '(%d prms)' % len(self.parameters)
return "<%s: %s>" % (self.__class__.__name__, s)
[docs] @classmethod
def parse(self, str):
''' Parse the given string returning the :class:`Event` instance.
>>> str = 'A4_OFF 0046 +03:10:10 2006-06-06T04:53:17 AASF006A'
>>> event = Event.parse(str)
>>> print(event.eventname)
A4_OFF
>>> print(list(event.parameters.keys()))
[]
>>> str = 'ELS_ON_OFF 0081 -02:30:34 2006-07-10T23:25:56 AASF057A VASD6525=0, VASD6528=ON,'
>>> event = Event.parse(str)
>>> print(event.orbnr)
81
>>> print(event.orbtime)
-1 day, 21:29:26
>>> print(sorted(event.parameters.keys()))
['VASD6525', 'VASD6528']
'''
spl = str.split()
if len(spl) < 5:
raise ValueError('Given string "%s" is not a line of event.' % str)
evtname = spl[0]
orbnr = int(spl[1], 10)
pm = spl[2][0]
hr = int(spl[2][1:3]) * 3600
mn = int(spl[2][4:6]) * 60
se = int(spl[2][7:9])
se = hr + mn + se
if pm[0] == "-":
se = -se
orbtime = datetime.timedelta(seconds=se)
utc = dateutil.parser.parse(spl[3])
cmdname = spl[4]
prms = {}
if len(spl) > 5: # if parameter is there.
prms_str = ' '.join(spl[5:])
prms_spl = prms_str.split(',')
for prms_elem in prms_spl:
prms_pair = prms_elem.split('=')
if len(prms_pair) == 2:
prms[prms_pair[0].strip()] = prms_pair[1].strip()
return Event(evtname, orbnr, orbtime, utc, cmdname, prms)
[docs]def default_eventfile():
''' Return the :class:`EventFile` class of the default event file.
Returns the :class:`EventFile` instance corresponding
to the default event file.
The default event file should be prepared locally, but if not found,
this function tries to fetch the default URI,
http://aspera-4.irf.se/archive/commanding/pmrqs/EV_ASPERA.VEX
'''
try:
return _default_eventfile_local()
except BaseException as e:
_logger.warning('Failed to load the event file locally.')
_logger.warning(e)
_logger.warning('Try to load the event file remotelly.')
return _default_eventfile_online()
raise IOError('Error loading the VEX event file both locally and remotely')
def _default_eventfile_local():
_logger.debug('Loading the event file locally')
rc = _Rc(raises=True)
path = rc.get('vexop', 'default_eventfile')
_logger.debug('Event file={}'.format(path))
eventfile = EventFile(path)
return eventfile
def _default_eventfile_online():
import urllib.request, urllib.parse, urllib.error
uri = "http://aspera-4.irf.se/archive/commanding/pmrqs/EV_ASPERA.VEX"
_logger.debug('Loading the event file remotely from {}'.format(uri))
try:
fn, stat = urllib.request.urlretrieve(uri)
_logger.debug('Downloaded. Cached to {}'.format(fn))
eventfile = EventFile(fn)
return eventfile
except IOError as e:
_logger.error('Failed to open the default URI (%s).' % uri)
raise