""" A modern implementation of CENA data retriever
This CENA data retriever uses an approach of "irfpy data center".
Refer to https://irfpy.irf.se/projects/util/tutorial/tutorial_datacenter.html
for general notes on the data center.
A path setting should be done via .irfpyrc setting.
::
[cena]
bmuopath = /path/to/data/bmu_orbit.v3.2
"""
import logging as _logging
import os as _os
import re as _re
import datetime as _datetime
from irfpy.util import datacenter as _dc
from irfpy.util import utc as _utc
from irfpy.util import irfpyrc as _rc
from irfpy.util import filepairv as _filepairv
import Sadppac as _sadppac
from irfpy.cena import cena_mass2 as _cmass
from irfpy.cy1orb import Cy1OrbitNr as _cy1onr
_logger = _logging.getLogger(__name__)
rc = _rc.Rc(raises=True)
def _search_cena_files(parent_path):
""" Search functionality of CENA file. It can be used for ``search_files`` methods.
"""
filelist = []
_logger.debug('Base: {}'.format(parent_path))
for _p, _d, _f in _os.walk(parent_path):
for _ff in _f:
_match = _re.search('orb_(\d\d\d\d).cena$', _ff)
if _match:
onr = int(_match.groups()[0])
if 364 <= onr <= 4095: # Hard coded but the maximum orbit number is 4095.
filelist.append(_os.path.join(_p, _ff))
return filelist
def _approxiate_starttime(filename):
""" Return the approxiate start time. It can be used for ``approximate_starttime`` methods.
"""
onr = int(_re.search('orb_(\d\d\d\d).cena$', filename).groups()[0])
return _cy1onr.get_start_time(onr)
def _decode_rawfile(filename):
""" Decode a raw file, decompress, and return as array of CenaPacket.
:param filename:
:return: A tuple, forming an array of datetime and an array of CenaPacket
"""
br = _sadppac.CenaBmuReader()
br.readFile(filename)
cenaarray = br.getArray()
tlist = []
dlist = []
for i in range(cenaarray.size()):
pac = cenaarray.getPacket(i)
if pac.getTMmode() == pac.TMMODE_MASS_ACC:
pac = pac.decompress()
if pac is not None:
_t = _utc.convert(pac.getUtc(), _datetime.datetime)
tlist.append(_t)
dlist.append(pac)
return tlist, dlist
def _read_rawfile(filename):
""" File pair support of decoding file.
"""
tlist, dlist = _filepairv.filepair(filename,
filename + '.filepair.gz',
converter=_decode_rawfile)
return tlist, dlist
[docs]class DataCenterCenaPacket(_dc.BaseDataCenter):
""" Data center for CENA Packet data.
This class provides data access of CENA.
The returned data is for Cena Packet data, namely,
:class:`Sadppac.CenaPacket`.
>>> dc = DataCenterCenaPacket()
>>> print(dc.t0())
2008-12-09 11:10:06.306000
>>> print(dc.t1())
2009-08-01 04:21:25.270000
>>> import datetime
>>> t = datetime.datetime(2009, 4, 3, 6, 0, 13)
>>> print(dc.nearest(t)) # doctest: +ELLIPSIS
(datetime.datetime(2009, 4, 3, 6, 0, 13, 23000), <Sadppac.CenaPacket; proxy of <Swig Object of type 'IrfPac::SaraPac::CenaPacket *' at ...
"""
def __init__(self):
self.basedir = rc.get('cena', 'bmuopath')
_dc.BaseDataCenter.__init__(self, copy=False)
[docs] def search_files(self):
return _search_cena_files(self.basedir)
[docs] def approximate_starttime(self, filename):
return _approxiate_starttime(filename)
[docs] def read_file(self, filename):
return _read_rawfile(filename)
[docs]class DataCenterCenaMassCount(_dc.BaseDataCenter):
""" A data center for CENA mass mode counts data access.
The data center provide (16, 7) shaped array containing the mass mode data.
>>> dc = DataCenterCenaMassCount()
>>> print(dc.t0())
2009-01-31 07:25:59.393000
>>> print(dc.t1())
2009-08-01 04:21:25.270000
>>> import datetime
>>> t = datetime.datetime(2009, 4, 3, 6, 25, 0)
>>> tobs, dat = dc.nearest(t)
>>> print(tobs)
2009-04-03 06:25:00.980000
>>> print(dat.sum())
17.0
"""
def __init__(self):
self.basedir = rc.get('cena', 'bmuopath')
_dc.BaseDataCenter.__init__(self)
[docs] def search_files(self):
return _search_cena_files(self.basedir)
[docs] def approximate_starttime(self, filename):
return _approxiate_starttime(filename)
[docs] def read_file(self, filename):
tlist, paclist = _read_rawfile(filename)
tlist2 = []
paclist2 = []
for _t, _p in zip(tlist, paclist):
try:
p = _cmass._packet2arrayE16(_p)
tlist2.append(_t)
paclist2.append(p)
except KeyboardInterrupt:
raise
except BaseException as e:
_logger.warning("Converting to E16 data failed.\nReason: {}".format(str(e)))
return tlist2, paclist2