"""VEX/ELS with the "data center" support.
A data center for VEX/ELS.
The data structure is in the :class:`irfpy.vels.scidata.ElsCountData` object.
"""
import os as _os
import datetime as _datetime
import logging as _logging
import scipy.io as _spio
from irfpy.util import datacenter as _datacenter
from irfpy.util import irfpyrc as _irfpyrc
from irfpy.util import exception as _ex
from irfpy.util import utc as _utc
[docs]class DataCenterElsCounts(_datacenter.BaseDataCenter):
""" ELS count rate data's data center.
The data is in numpy array with a shape of (E128, A16).
>>> from irfpy.vels import rawdata
>>> import datetime
>>> t0 = datetime.datetime(2010, 10, 30, 23, 15)
>>> velsdc = rawdata.DataCenterElsCounts() # ELS count's data center.
>>> tobs, cnts = velsdc.nearest(t0)
>>> print(tobs, cnts.shape, cnts.max())
2010-10-30 23:15:00.509000 (128, 16) 44
"""
def __init__(self):
_datacenter.BaseDataCenter.__init__(self)
[docs] def search_files(self):
rc = _irfpyrc.Rc()
basepath = rc.get('vels', 'elsmatbase')
if not _os.path.isdir(basepath):
msg = 'The path {} not found in ELS data center.'.format(basepath)
raise _ex.IrfpyException(msg)
filelist = []
for pp, dd, ff in _os.walk(basepath):
for f in ff:
pp_split = _os.path.split(pp)
bf = _os.path.basename(f)
if len(bf) != 18:
continue
if not f.startswith(pp_split[-1][-6:]):
continue
if not f.endswith('.mat'):
continue
try:
int(f[:14])
except ValueError:
continue
full_f = _os.path.join(pp, f)
filelist.append(full_f)
return filelist
[docs] def read_file(self, filename):
""" Read the file contents.
From a given file, electron data is read.
The format is in the :class:`irfpy.vels.scidata.ElsCountData` object.
"""
_logger = _logging.getLogger(self.__class__.__name__)
import scipy.io as _spio
try:
mat = _spio.loadmat(filename)
except IOError as e:
emsg = 'Failed to load the matlba file: %s' % filename
_logger.error(emsg)
_logger.error('... Continue with ignoring the file %s' % filename)
return (), ()
obstime_mat = mat['elstimes'].flatten() # (N,) data
from irfpy.util import utc as _utc
obstime_dt = [_utc.mat2dt(_t) for _t in obstime_mat]
ecnts = mat['elsmatrix'] # 128 x 16 x N
# A special case. If the number of data is only one,
# the data is stored as in 2-D data! Annoying feature
# of the dataset, but need to be checked and handled.
if len(obstime_dt) == 1 and ecnts.shape == (128, 16):
import numpy as _np
ecnts = ecnts[:, :, _np.newaxis]
ecnts = list(ecnts.transpose((2, 0, 1)))
return (obstime_dt, ecnts)
[docs] def approximate_starttime(self, filename):
f = _os.path.basename(filename)
yr, mo, dy, hr = int(f[:4]), int(f[4:6]), int(f[6:8]), int(f[8:10])
return _datetime.datetime(yr, mo, dy, hr)
# def exact_starttime(self, filename):
# pass
[docs]class DataCenterElsLevel(_datacenter.BaseDataCenter):
""" VEX/ELS level data's data center.
The data is in numpy array with a shape of (E128,).
>>> from irfpy.vels import rawdata
>>> import datetime
>>> t0 = datetime.datetime(2010, 10, 30, 23, 15)
>>> velsdc = rawdata.DataCenterElsLevel() # ELS level's data center.
>>> tobs, levels = velsdc.nearest(t0)
>>> print(tobs, levels.shape, levels.max())
2010-10-30 23:14:48.509000 (128,) 4095
"""
def __init__(self):
_datacenter.BaseDataCenter.__init__(self)
[docs] def search_files(self):
rc = _irfpyrc.Rc()
basepath = rc.get('vels', 'elsmatbase')
if not _os.path.isdir(basepath):
msg = 'The path {} not found in ELS data center.'.format(basepath)
raise _ex.IrfpyException(msg)
filelist = []
for pp, dd, ff in _os.walk(basepath):
for f in ff:
pp_split = _os.path.split(pp)
bf = _os.path.basename(f)
if len(bf) != 18:
continue
if not f.startswith(pp_split[-1][-6:]):
continue
if not f.endswith('.mat'):
continue
try:
int(f[:14])
except ValueError:
continue
full_f = _os.path.join(pp, f)
filelist.append(full_f)
return filelist
[docs] def read_file(self, filename):
""" Read the file contents.
From a given file, electron data is read.
The format is in the :class:`irfpy.vels.scidata.ElsCountData` object.
"""
_logger = _logging.getLogger(self.__class__.__name__)
try:
mat = _spio.loadmat(filename)
except IOError as e:
emsg = 'Failed to load the matlba file: %s' % filename
_logger.error(emsg)
_logger.error('... Continue with ignoring the file %s' % filename)
return (), ()
obstime_mat = mat['elsleveltimes'].flatten() # (N,) data
obstime_dt = [_utc.mat2dt(_t) for _t in obstime_mat]
elvls = mat['elslevels'] # 128 x N
elvls = list(elvls.transpose((1, 0)))
return (obstime_dt, elvls)
[docs] def approximate_starttime(self, filename):
f = _os.path.basename(filename)
yr, mo, dy, hr = int(f[:4]), int(f[4:6]), int(f[6:8]), int(f[8:10])
return _datetime.datetime(yr, mo, dy, hr)
# def exact_starttime(self, filename):
# pass
[docs]class DataCenterElsAs128(DataCenterElsCounts):
""" ELS data center.
>>> from irfpy.vels import rawdata
>>> import datetime
>>> t0 = datetime.datetime(2010, 10, 30, 23, 15)
>>> velsdc = rawdata.DataCenterElsAs128() # ELS count's data center.
>>> tobs, cnts = velsdc.nearest(t0)
>>> print(tobs)
2010-10-30 23:15:00.509000
>>> print(cnts)
<ElsCountData at 2010-10-30T23:15:00, count_max=44.0, level=(Defined (@2010-10-30T23:14:48))>
"""
def __init__(self):
DataCenterElsCounts.__init__(self)
self.level_dc = DataCenterElsLevel()
[docs] def read_file(self, filename):
tlist, dlist = DataCenterElsCounts.read_file(self, filename)
# tlist is the list of the time
# dlist is the matrix.
d2list = []
for t, d in zip(tlist, dlist):
# Get the corresponding level, which should be earlier than
# or equal to the observed time
t2, level = self.level_dc.nearest_earlier_or_at(t)
from irfpy.vels import scidata as vsci
if (t - t2).total_seconds() > 30:
level = None
level_time = None
d2list.append(vsci.ElsCountData(t, d, level, level_time=t2))
return tlist, d2list
[docs]class DataCenterEls(DataCenterElsCounts):
""" ELS data center.
>>> from irfpy.vels import rawdata
>>> import datetime
>>> t0 = datetime.datetime(2010, 10, 30, 23, 15)
>>> velsdc = rawdata.DataCenterEls() # ELS count's data center.
>>> tobs, cnts = velsdc.nearest(t0)
>>> print(tobs)
2010-10-30 23:15:00.509000
>>> print(cnts)
<ElsCountDataE128 at 2010-10-30T23:15:00, count_max=44.0, level=(Defined)>
"""
def __init__(self):
DataCenterElsCounts.__init__(self)
self.level_dc = DataCenterElsLevel()
[docs] def read_file(self, filename):
tlist, dlist = DataCenterElsCounts.read_file(self, filename)
# tlist is the list of the time
# dlist is the matrix.
t2list = []
d2list = []
for t, d in zip(tlist, dlist):
# Get the corresponding level, which should be earlier than
# or equal to the observed time
t2, level = self.level_dc.nearest_earlier_or_at(t)
from irfpy.vels import scidata as vsci
if (t - t2).total_seconds() > 30:
level = None
level_time = None
cntdat = vsci.ElsCountData(t, d, level, level_time=t2)
cntdat = vsci.interpret_rawdata(cntdat)
tcntdat = [_.starttime for _ in cntdat]
t2list = t2list + tcntdat
d2list = d2list + list(cntdat)
return t2list, d2list