Source code for irfpy.mexop.eventfile
''' A module to read the aspera-3 event file
ASPERA-3 event file is a list of the command
that is important for data analysis.
After the OBCP commanding in late 2011, no
more reasonable event file is generated (TBC).
Slight difference is found with :mod:`VEX event file <irfpy.vexop.eventfile>`
but the idea behind is the same.
'''
import logging
logger = logging.getLogger(__name__)
import datetime
import dateutil.parser
import irfpy.util.irfpyrc
import irfpy.util.filepairv
[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.
Each event is stored as an instance of :class:`Event`.
Usually, one can use direct access to the member named :attr:`list`.
Below is an sample.
.. code-block:: py
>> event = EventFile(filename)
>> for ev in event.list:
.. print(ev)
.. # Some processing
'''
logger = logging.getLogger(__name__)
def __init__(self, filename):
self.logger.debug('Instance with %s' % filename)
self.filename = filename
self.version = None
''' No version supported for A3 event file'''
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]
# self.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
try:
event = Event.parse(line)
self.list.append(event)
except ValueError as e:
self.logger.warn('Skipped string ``%s``.' % line.strip())
fp.close()
[docs]class Event:
''' Class of an event.
The class behave as a structure. Member variables are:
:eventname: Name of the event. ``string``
:orbnr: Orbit nubmer. ``int``
:orbtime: Time relative to the pericenter. ``datetime.timedelta``
:utc: Time in UTC. ``datetime.datetime``
:cmdname: Name of the command. ``string``
:parameters: Dictionary of parameters.
``dict`` of ``string`` to ``string``.
'''
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 = 'A3_ON 00023 -03:47:18 2004-01-15T22:19:22 AASF01A2'
>>> event = Event.parse(str)
>>> print(event.eventname)
A3_ON
>>> print(event.orbnr)
23
>>> print(event.orbtime)
-1 day, 20:12:42
>>> print(list(event.parameters.keys()))
[]
>>> print(event)
<Event: A3_ON 0023 2004-01-15T22:19:22 AASF01A2 (0 prms)>
>>> str = 'IMA_M_24 00041 00:23:17 2004-01-22T22:31:01 AASF60A1,VAS04054=006,VAS05010=024'
>>> event = Event.parse(str)
>>> print(event.eventname)
IMA_M_24
>>> print(event.orbnr)
41
>>> print(event.orbtime)
0:23:17
>>> print(sorted(event.parameters.keys()))
['VAS04054', 'VAS05010']
>>> print(event.parameters['VAS04054'])
006
>>> print(event)
<Event: IMA_M_24 0041 2004-01-22T22:31:01 AASF60A1 (2 prms)>
'''
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)
if spl[2][0] == '-':
pm = '-'
tstr = spl[2][1:]
elif spl[2][0] == '+':
pm = '+'
tstr = spl[2][1:]
else:
pm = '+'
tstr = spl[2]
hr, mn, se = tstr.split(':')
hr = int(hr) * 3600
mn = int(mn) * 60
se = int(se)
se = hr + mn + se
if pm == "-":
se = -se
orbtime = datetime.timedelta(seconds=se)
utc = dateutil.parser.parse(spl[3])
parsed_cmd = spl[4].split(',')
cmdname = parsed_cmd[0]
prms = {}
if len(parsed_cmd) > 1: # if parameter is there.
for prms_elem in parsed_cmd:
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_reader(fn):
return EventFile(fn)
[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 can be specified by the .irfpyrc file.::
[mexop]
default_eventfile = http://aspera-3.irf.se/archive/commanding/pors/EV_ASPERA.MEX
It is usually recommended to download the default file to local,
and set it as a new default.::
[mexop]
default_eventfile = /Volumes/scidata/data/mars/support/EV_ASPERA.MEX
'''
rc = irfpy.util.irfpyrc.Rc()
fn = rc.get('mexop', 'default_eventfile')
if fn is None:
import urllib.request, urllib.parse, urllib.error
uri = "http://aspera-3.irf.se/archive/commanding/pors/EV_ASPERA.MEX"
try:
fn, stat = urllib.request.urlretrieve(uri)
except IOError as e:
logger.error('Failed to open the default URI %s.' % uri)
raise
eventfile = irfpy.util.filepairv.filepair(fn, fn+'.filepair', converter=default_eventfile_reader)
return eventfile
[docs]def isdb():
try:
evf = default_eventfile()
return True
except IOError:
return False