""" Official implementation of raw data access.
A new implementation of raw data access using the
data center (:class:`irfpy.util.datacenter.BaseDataCenter`) approach.
.. seealso:: :mod:`irfpy.mima.scidata_util`
*Scipy version*
Scipy >=1.4.0 is recommended.
Prior version has a minor problem loading several specific data files.
*For developer*
This is almost identical to VEX version. So if one change the contents,
it is also recommended to change VEX version.
"""
import os as _os
import logging as _logging
from pkg_resources import parse_version as _parse_version
import warnings as _warnings
from irfpy.util import __version__ as _util_version
from irfpy.util import datacenter as _dc
from irfpy.util import timeseries as _ts
from irfpy.util import exception as _ex
from irfpy.util import irfpyrc as _irfpyrc
from irfpy.mima import SENSOR_NAME
from irfpy.mima import scidata_tree as _st
from irfpy.mima import scidata_util as _vsu
_logger = _logging.getLogger(__name__)
def _irfpy_util_version_check():
minimum_recommended = _parse_version('4.5.0a1')
irfpy_version = _parse_version(_util_version)
if irfpy_version < minimum_recommended:
_warnings.warn(f'Installed irfpy.util version is {_util_version}.\n'
'Recommended version is 4.5.0a1 or above.\n'
'See https://irfpy.irf.se/projects/util/quickinstall.html#update-irfpy-util'
)
def _version_check_scipy():
""" Scipy version check
"""
import scipy as _scipy
minimum_recommended = _parse_version('1.4.0')
scipy_version = _parse_version(_scipy.__version__)
if scipy_version < minimum_recommended:
_mesg = (f'Installed scipy version is {scipy_version}.',
f'Recommended version is >=1.4.0.',
)
_mesg = '\n'.join(_mesg)
_warnings.warn(_mesg)
[docs]class DataCenterCount2d(_dc.BaseDataCenter):
""" Raw count data center for 2D, emulated matrix.
>>> dc = DataCenterCount2d()
>>> import datetime
>>> t0 = datetime.datetime(2007, 3, 25, 5)
>>> t, d = dc.nearest(t0)
>>> print(t)
2007-03-25 04:59:57.417000
>>> print(d)
<<class 'irfpy.imacommon.imascipac.CntMatrix2D'>(MEX/IMA)@2007-03-25T04:59:57.417000:MOD=24 >>24<<:POL=08/08:CNTmax=30>
>>> t1 = datetime.datetime(2007, 3, 25, 6)
>>> tlist, dlist = dc.get_array_strict(t0, t1)
>>> from pprint import pprint
>>> pprint(tlist) # doctest: +ELLIPSIS
[datetime.datetime(2007, 3, 25, 5, 0, 9, 417000),
datetime.datetime(2007, 3, 25, 5, 0, 21, 417000),
...
datetime.datetime(2007, 3, 25, 5, 59, 57, 948180)]
>>> for t, d in dc.iter_wide(t0, t1):
... print(t) # doctest: +ELLIPSIS
2007-03-25 04:59:57.417000
2007-03-25 05:00:09.417000
...
2007-03-25 05:59:57.948180
2007-03-25 06:00:09.948180
"""
def __init__(self, emulate_full=True):
_irfpy_util_version_check()
_version_check_scipy()
rc = _irfpyrc.Rc()
self.basepath = rc.get('mima', 'imamatbase')
self.emulate_full = emulate_full
if not _os.path.isdir(self.basepath):
msg = 'The path {} not found for MIMA data center.'.format(self.basepath)
_logger.error(msg)
raise ValueError(msg)
_dc.BaseDataCenter.__init__(self)
[docs] def search_files(self):
filelist = []
for pp, dd, ff in _os.walk(self.basepath):
for f in ff:
if not f.endswith('.mat'):
_logger.info('{} is not mat file. Skipped.'.format(f))
continue
fn = _os.path.join(pp, f)
filelist.append(fn)
return filelist
[docs] def read_file(self, filename):
return read_mat_file(filename, emulate_full=self.emulate_full)
[docs] def approximate_starttime(self, filename):
return _st._guess(filename)
[docs]class DataCenterCount3d(_dc.BaseDataCenter):
""" Raw count data center for IMA 3D data.
The datacenter approach is used:
>>> dc = DataCenterCount3d()
>>> import datetime
>>> t0 = datetime.datetime(2007, 3, 25, 5)
>>> t, d = dc.nearest(t0)
>>> print(t)
2007-03-25 05:01:33.510740
>>> print(d)
<<class 'irfpy.imacommon.imascipac.CntMatrix'>(MEX/IMA)@2007-03-25T05:01:33.510740:MOD=24 >>24<<:CNTmax=19>
>>> t1 = datetime.datetime(2007, 3, 25, 6)
>>> tlist, dlist = dc.get_array_strict(t0, t1)
>>> from pprint import pprint
>>> pprint(tlist) # doctest: +ELLIPSIS
[datetime.datetime(2007, 3, 25, 5, 1, 33, 510740),
datetime.datetime(2007, 3, 25, 5, 4, 45, 479500),
...
datetime.datetime(2007, 3, 25, 5, 55, 57, 979440),
datetime.datetime(2007, 3, 25, 5, 59, 9, 948180)]
>>> for t, d in dc.iter_wide(t0, t1):
... print(t) # doctest: +ELLIPSIS
2007-03-25 04:58:21.417000
2007-03-25 05:01:33.510740
...
2007-03-25 05:59:09.948180
2007-03-25 06:02:22.010680
"""
def __init__(self, emulate_full=True):
_irfpy_util_version_check()
_version_check_scipy()
rc = _irfpyrc.Rc()
self.basepath = rc.get('mima', 'imamatbase')
self.emulate_full = emulate_full
if not _os.path.isdir(self.basepath):
msg = 'The path {} not found for MIMA data center.'.format(self.basepath)
_logger.error(msg)
raise ValueError(msg)
_dc.BaseDataCenter.__init__(self, cache_size=5)
[docs] def search_files(self):
filelist = []
for pp, dd, ff in _os.walk(self.basepath):
for f in ff:
if not f.endswith('.mat'):
_logger.info('{} is not mat file. Skipped.'.format(f))
continue
fn = _os.path.join(pp, f)
filelist.append(fn)
return filelist
[docs] def read_file(self, filename):
return read_mat_file_3d(filename, emulate_full=self.emulate_full)
[docs] def approximate_starttime(self, filename):
return _st._guess(filename)
[docs]def read_mat_file(filename, emulate_full=True):
try:
arranged_matrix = _vsu._arrange_matlab_file(filename, emulate_full=emulate_full)
except _ex.IrfpyException:
return _ts.ObjectSeries([], []) # No data to return
tlist = []
dlist = []
for t, d in arranged_matrix:
d.name = SENSOR_NAME
tlist.append(t)
dlist.append(d)
return _ts.ObjectSeries(tlist, dlist)
[docs]def read_mat_file_3d(filename, emulate_full=True):
""" Return the 3D data series.
:param filename: The matlab file name
:return: (tlist, dlist), where tlist is the observation time and dlist is for the data.
"""
try:
arranged_matrix = _vsu._arrange_matlab_file(filename, emulate_full=emulate_full)
except _ex.IrfpyException:
return _ts.ObjectSeries([], []) # No data to return
from irfpy.imacommon import imascipac as _imascipac
arranged_matrix = _imascipac.convert2to3(arranged_matrix, emulate_full=emulate_full)
tlist = []
dlist = []
for t, d in arranged_matrix:
d.name = SENSOR_NAME
tlist.append(t)
dlist.append(d)
return tlist, dlist