''' A module of MEX/ELS flux.
The flux is dumped from CL, and located at thor:/irf/data/aspera3/els/mat.
.. note::
I am not sure the CL really implement the ELS calibration properly.
So using this module should be careful and your own risk.
'''
import os
import logging
#import urllib.request, urllib.parse, urllib.error
import bisect
import datetime
import warnings
import scipy.io
from irfpy.util.utc import num2date
import irfpy.util.timeseriesdb
from irfpy.util.irfpyrc import Rc
from irfpy.asperacommon.excepts import DbPathNotFoundError, DatafileNotFoundError
[docs]def get_uribase_path():
''' Return the database path.
Database of ElS flux can be specified by [mels]/fluxmaturibase in irfpyrc.
If no path is found, :class:`irfpy.asperacommon.excepts.DbPathNotFoundError` is raised.
The path may include the 4-digit subdirectories expressing the year.
'''
rc = Rc()
path = rc.get('mels', 'fluxmaturibase')
if path is None:
raise DbPathNotFoundError('No entry of [mels]/fluxmaturibase in irfpyrc')
return path
[docs]class ElsFluxDatabase(irfpy.util.timeseriesdb.DB):
''' The ELS flux database folder.
The data base of the ELS flux.
First, it searches through the file name 'elecYYYYMMDDHH.mat' under the given *dbpath*.
Then, the start time is calculated from the file name, and registered into this database.
:keyword dbpath: Path of the database. Default is read by :meth:`get_uribase_path`.
# First, instance it.
>>> samplepath = os.path.join(os.path.dirname(__file__), 'sample', 'elsflux')
>>> db = ElsFluxDatabase(dbpath=samplepath)
# Then, you can check the size of the data base. In the `sample/elsflux` folder, seven files can be found.
>>> print(len(db))
7
# To get the file name of the data, you can use :meth:`get` method.
>>> file = db.get(datetime.datetime(2005, 1, 2, 15, 0, 0))
>>> print(os.path.basename(file))
elec2005010200.mat
# In case you want to load the file one before, you can use :meth:`previousof` method.
>>> file_prev = db.previousof(file)
>>> print(os.path.basename(file_prev))
elec2005010100.mat
'''
def __init__(self, dbpath=None):
''' Initialize the database.
'''
irfpy.util.timeseriesdb.DB.__init__(self)
if dbpath is None:
dbpath = get_uribase_path()
self.logger = logging.getLogger(self.__class__.__name__)
# self.logger.setLevel(logging.DEBUG)
self.logger.debug('DB path = %s' % dbpath)
if not os.path.isdir(dbpath):
raise DbPathNotFoundError('DB path %s does not exist. Check the [mels]/fluxmaturibase in irfpyrc' % dbpath)
for p, d, f in os.walk(dbpath):
for file in f:
if file.startswith('elec') and file.endswith('.mat'):
self.logger.debug('Processing %s' % file)
tstr = file[4:-4]
self.logger.debug(' -> T = %s' % tstr)
try:
yr = int(tstr[0:4])
mo = int(tstr[4:6])
dy = int(tstr[6:8])
hr = int(tstr[8:10])
start_time = datetime.datetime(yr, mo, dy, hr)
self.logger.debug(' -> T = %s' % start_time)
except ValueError as e:
self.logger.warning('Not a good file name (%s). Skipped.' % os.path.join(p, file))
self.append(os.path.join(p, file), start_time)
[docs]class ElsFluxFile:
[docs] @classmethod
def get_sample_filename(cls):
p = get_uribase_path()
fn = os.path.join(p, '2010', 'elec201006031800.mat')
logging.debug(fn)
return fn
[docs] @classmethod
def get_uribase_path(cls):
global get_uribase_path
return get_uribase_path()
[docs] @classmethod
def get_uri(cls, yr, mo, dy, hr):
ttry = datetime.datetime(yr, mo, dy, hr, 0, 0)
subpath = os.path.join(ttry.strftime('%Y'), ttry.strftime('elec%Y%m%d%H00.mat'))
return cls.get_uribase_path() + '/' + subpath
def __init__(self, uri):
''' Load the matlab flux file
:param uri: The URI or filename of the file.
'''
self.logger = logging.getLogger(self.__class__.__name__)
self.file = self.uri = uri
# self.mat contains all the matlab data.
try:
self.mat = scipy.io.loadmat(self.file)
except IOError as e:
emsg = 'Failed to open the matlab file: %s' % self.uri
self.logger.error(emsg)
raise DatafileNotFoundError(emsg)
[docs] def getobstime(self):
''' Return the observation time as array of ``datetime.datetime``
>>> sample = ElsFluxFile.get_sample_filename()
>>> s = ElsFluxFile(sample)
>>> otime = s.getobstime()
>>> print(len(otime))
784
'''
tElec = self.mat.get('tElec').flatten()
obstime_mat = list(set(tElec))
obstime_mat.sort()
obstime_mpl = [num2date(i - 366) for i in obstime_mat]
obstime_mpl = [i.replace(tzinfo=None) for i in obstime_mpl]
return obstime_mpl
[docs] def gettimerange(self):
''' Return the time range stored in this file.
The start time is the time of the first packet.
The end time is the time of the last packet.
This means that the end time does not agree with the end of the observation time.
The end of the observation time is the end time plus data acquisition time (typically 4 sec).
>>> sample = ElsFluxFile.get_sample_filename()
>>> s = ElsFluxFile(sample)
>>> print(s.gettimerange())
[datetime.datetime(2010, 6, 3, 18, 0, 4), datetime.datetime(2010, 6, 3, 18, 59, 57)]
'''
tlist = self.getobstime()
return [min(tlist), max(tlist)]
[docs] def get_data(self, t_c):
''' Return the flux data obtained at *t_c*.
Note that the time of the returned data, *t_r*, is generally different from *t_c*.
*t_r* sastisfies *t_r* < *t_c* and (*t_r* - *t_c*) is minimum.
Sometimes the *t_r* is very far from *t_c*, as no error check or data gap check is done.
It is the user's responsible to check whether *t_r* is acceptable or not.
Thus, it is recommended that the *t_c* is taken from the returned value of :meth:`getobstime()`.
:param t_c: The time of the observation.
:type t_c: ``datetime.datetime``
:returns: A tuple, (t_r, flux). *flux* should have 16x128 shaped ndarray.
>>> sample = ElsFluxFile.get_sample_filename()
>>> s = ElsFluxFile(sample)
>>> t_c = s.getobstime()[300]
>>> t_r, flux = s.get_data(t_c)
>>> t_r == t_c
True
>>> print('%.5e' % flux.max())
2.45800e+10
'''
warnings.warn('I am not sure whether the CL really implement the ELS calibration properly. Using this module should be careful, and your own risk.')
tlist = self.getobstime()
idx = bisect.bisect(tlist, t_c)
if idx == 0:
raise IndexError('Failed to load the data at *t_c* = %s.' % t_c)
idx = idx - 1
t_r = tlist[idx]
return (t_r, self.mat['fElec'][:, :, idx])
import unittest,doctest
[docs]def doctests():
return unittest.TestSuite((
doctest.DocTestSuite(),
))
if __name__=='__main__':
unittest.main(defaultTest='doctests')