Commit 9115280a authored by payno's avatar payno
Browse files

[refactoring] remove import of core from io.

This require the creation of some helpers to link io and XasObject, this is the goal of xas.core.io.
parent 4d3bdd36
Pipeline #12658 failed with stage
in 4 minutes and 27 seconds
......@@ -34,7 +34,7 @@ from silx.gui import qt
from silx.io.url import DataUrl
from xas.core.types import XASObject
from xas.gui.xas_object_definition import XASObjectDialog
import xas.io
import xas.core.io
from Orange.widgets.settings import Setting
import os
import logging
......@@ -66,7 +66,7 @@ class XASInputOW(OWWidget):
_energy_url_setting = Setting(str())
_configuration_url_setting = Setting(str())
process_function = xas.io.read_pymca_xas_from_file
process_function = xas.core.io.read_frm_file
def __init__(self):
super().__init__()
......
......@@ -33,7 +33,7 @@ from Orange.widgets.widget import OWWidget
from silx.gui import qt
from Orange.widgets.settings import Setting
from xas.core.types import XASObject
import xas.io
import xas.core.io
import logging
_logger = logging.getLogger(__file__)
......@@ -56,7 +56,7 @@ class XASOutputOW(OWWidget):
inputs = [("spectra", XASObject, 'process')]
_output_file_setting = Setting(str())
process_function = xas.io.XASWriter
process_function = xas.core.io.XASWriter
def __init__(self):
super().__init__()
......
# coding: utf-8
# /*##########################################################################
#
# Copyright (c) 2017-2019 European Synchrotron Radiation Facility
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
#
# ###########################################################################*/
"""simple helper function to link io module and XASObject"""
__authors__ = ["H. Payno"]
__license__ = "MIT"
__date__ = "07/16/2019"
from xas.io import read_xas, write_xas, get_xasproc
from xas.core.types import XASObject
from silx.io.url import DataUrl
import h5py
import logging
_logger = logging.getLogger(__name__)
DEFAULT_SPECTRA_PATH = '/data/NXdata/data'
DEFAULT_CHANNEL_PATH = '/data/NXdata/Channel'
DEFAULT_CONF_PATH = '/configuration'
def read(spectra_url, channel_url, config_url=None):
"""
:param DataUrl spectra_url: data url to the spectra
:param DataUrl channel_url: data url to the channel / energy
:param DataUrl config_url: data url to the process configuration
:return:
:rtype: XASObject
"""
reader = XASReader()
return reader.read_frm_url(spectra_url=spectra_url, channel_url=channel_url,
config_url=config_url)
def read_frm_file(file_path):
"""
:param str file_path: path to the file containing the spectra. Must ba a
.dat file that pymca can handle or a .h5py with
default path
:rtype: XASObject
"""
reader = XASReader()
return reader.read_from_file(file_path=file_path)
class XASReader(object):
"""Simple reader of a xas file"""
def read_frm_url(self, spectra_url, channel_url, config_url=None):
sp, en, conf = read_xas(spectra_url=spectra_url,
channel_url=channel_url,
config_url=config_url)
return XASObject(spectra=sp, energy=en, configuration=conf)
def read_from_file(self, file_path):
"""
:param str file_path:
:return: `.XASObject`
"""
if file_path.endswith('.dat'):
return self.read_frm_url(
spectra_url=DataUrl(file_path=file_path,
scheme='PyMca'),
channel_url=DataUrl(file_path=file_path,
scheme='PyMca'))
elif h5py.is_hdf5(file_path):
return self.read_frm_url(
spectra_url=DataUrl(file_path=file_path,
scheme='silx',
data_path=DEFAULT_SPECTRA_PATH),
channel_url=DataUrl(file_path=file_path,
scheme='silx',
data_path=DEFAULT_CHANNEL_PATH),
config_url=DataUrl(file_path=file_path,
scheme='silx',
data_path='configuration'))
else:
raise ValueError('file type not managed, unable to load')
__call__ = read_from_file
class XASWriter(object):
"""
class to write the output file. In this case we need a class in order to
setup the output file before
"""
def __init__(self):
self._output_file = None
@property
def output_file(self):
return self._output_file
@output_file.setter
def output_file(self, file_):
self._output_file = file_
def setProperties(self, properties):
if '_output_file_setting' in properties:
self._output_file = properties['_output_file_setting']
def dump_xas(self, xas_obj, write_process=True):
"""
write the XASObject into a hdf5 file.
:param :class:`.XASObject` xas_obj: object to be stored
:param bool write_process: if True then store the process flow in the
same file.
"""
if isinstance(xas_obj, dict):
_xas_obj = XASObject.from_dict(xas_obj)
else:
_xas_obj = xas_obj
if not self._output_file:
_logger.warning('no output file defined, please give path to the output file')
self._output_file = input()
_logger.info(('dump xas obj to', self._output_file))
# write raw data
write_xas(h5_file=self._output_file, energy=_xas_obj.energy,
mu=_xas_obj.absorbed_beam(), entry=_xas_obj.entry)
if write_process is True:
if len(get_xasproc(self._output_file, entry=_xas_obj.entry)) > 0:
_logger.warning('removing previous process registred. They are no '
'more accurate')
_xas_obj.clean_process_flow()
# write process flow
_xas_obj.copy_process_flow_to(self._output_file)
__call__ = dump_xas
......@@ -31,7 +31,7 @@ import unittest
from xas.core.process.exafs import pymca_exafs
from xas.core.process.normalization import pymca_normalization
from xas.core.types import Spectrum, XASObject
from xas.core.utils.pymca import read_spectrum
from xas.io.utils.pymca import read_spectrum
from PyMca5.PyMcaDataDir import PYMCA_DATA_DIR
import os
......
......@@ -30,7 +30,7 @@ __date__ = "06/11/2019"
import unittest
from xas.core.process.ft import pymca_ft
from PyMca5.PyMcaDataDir import PYMCA_DATA_DIR
from xas.core.utils.pymca import read_spectrum
from xas.io.utils.pymca import read_spectrum
from xas.core.process.normalization import pymca_normalization
from xas.core.process.exafs import pymca_exafs
from xas.core.types import XASObject, Spectrum
......
......@@ -28,11 +28,12 @@ __date__ = "06/11/2019"
import unittest
from xas.io import read_pymca_xas, XASWriter
from silx.io.url import DataUrl
from PyMca5.PyMcaDataDir import PYMCA_DATA_DIR
from xas.core.io import XASWriter
from xas.core.io import read as read_xas
from xas.core.utils import spectra as spectra_utils
from xas.core.types import XASObject
from silx.io.url import DataUrl
from PyMca5.PyMcaDataDir import PYMCA_DATA_DIR
import shutil
import tempfile
import h5py
......@@ -51,9 +52,9 @@ class TestReadWrite(unittest.TestCase):
def testReadSpectrum(self):
data_file = os.path.join(PYMCA_DATA_DIR, "EXAFS_Cu.dat")
res = read_pymca_xas(spectra_url=DataUrl(file_path=data_file,
scheme='PyMca'),
channel_url=DataUrl(file_path=data_file,
res = read_xas(spectra_url=DataUrl(file_path=data_file,
scheme='PyMca'),
channel_url=DataUrl(file_path=data_file,
scheme='PyMca'))
self.assertTrue(isinstance(res, XASObject))
self.assertEqual(res.n_spectrum, 1)
......
......@@ -32,7 +32,7 @@ import os
from PyMca5.PyMcaDataDir import PYMCA_DATA_DIR
from xas.core.process.k_weight import pymca_k_weight
from xas.core.types import Spectrum, XASObject
from xas.core.utils.pymca import read_spectrum
from xas.io.utils.pymca import read_spectrum
class TestKWeightSingleSpectrum(unittest.TestCase):
......
......@@ -30,9 +30,9 @@ __date__ = "06/11/2019"
import unittest
from xas.core.process.normalization import pymca_normalization
from xas.core.types import Spectrum, XASObject
from xas.core.utils.pymca import read_spectrum
from xas.io.utils.pymca import read_spectrum
from xas.core.utils import spectra as spectra_utils
from xas.io import read_pymca_xas
from xas.core.io import read as read_xas
from PyMca5.PyMcaDataDir import PYMCA_DATA_DIR
from silx.io.url import DataUrl
import os
......@@ -82,14 +82,13 @@ class TestNormalizationMultipleSpectrum(unittest.TestCase):
f[spectra_path] = self.spectra
f[channel_path] = self.energy
self.xas_obj = read_pymca_xas(
spectra_url=DataUrl(file_path=filename,
data_path=spectra_path,
scheme='silx'),
channel_url=DataUrl(file_path=filename,
data_path=channel_path,
scheme='silx'),
)
self.xas_obj = read_xas(spectra_url=DataUrl(file_path=filename,
data_path=spectra_path,
scheme='silx'),
channel_url=DataUrl(file_path=filename,
data_path=channel_path,
scheme='silx'),
)
def tearDown(self):
shutil.rmtree(path=self.output_dir)
......
......@@ -30,7 +30,7 @@ __date__ = "06/11/2019"
import unittest
from xas.core.process.roi import xas_roi
from xas.core.utils import spectra as spectra_utils
from xas.io import read_pymca_xas
from xas.core.io import read as read_xas
from silx.io.url import DataUrl
import os
import numpy
......@@ -51,7 +51,7 @@ class TestRoi(unittest.TestCase):
f[spectra_path] = self.spectra
f[channel_path] = self.energy
self.xas_obj = read_pymca_xas(
self.xas_obj = read_xas(
spectra_url=DataUrl(file_path=filename,
data_path=spectra_path,
scheme='silx'),
......
......@@ -30,11 +30,11 @@ __date__ = "06/11/2019"
import unittest
import os
from PyMca5.PyMcaDataDir import PYMCA_DATA_DIR
from xas.io import XASWriter
from xas.core.io import XASWriter
from silx.io.url import DataUrl
from xas.core.process.exafs import pymca_exafs, PyMca_exafs
from xas.core.process.ft import pymca_ft, PyMca_ft
from xas.io import read_pymca_xas
from xas.core.io import read as read_xas
from xas.core.process.k_weight import pymca_k_weight, PyMca_k_weight
from xas.core.process.normalization import pymca_normalization, PyMca_normalization
from xas.core.process.roi import xas_roi, _ROI as XASROI
......@@ -50,9 +50,9 @@ class TestStreamSingleSpectrum(unittest.TestCase):
"""Make sure the process have valid io"""
def test_pymca_process(self):
data_file = os.path.join(PYMCA_DATA_DIR, "EXAFS_Cu.dat")
out = read_pymca_xas(spectra_url=DataUrl(file_path=data_file,
scheme='PyMca'),
channel_url=DataUrl(file_path=data_file,
out = read_xas(spectra_url=DataUrl(file_path=data_file,
scheme='PyMca'),
channel_url=DataUrl(file_path=data_file,
scheme='PyMca'))
out = pymca_normalization(xas_obj=out)
out = pymca_exafs(xas_obj=out)
......@@ -62,9 +62,9 @@ class TestStreamSingleSpectrum(unittest.TestCase):
def test_pymca_process_with_dict(self):
data_file = os.path.join(PYMCA_DATA_DIR, "EXAFS_Cu.dat")
out = read_pymca_xas(spectra_url=DataUrl(file_path=data_file,
scheme='PyMca'),
channel_url=DataUrl(file_path=data_file,
out = read_xas(spectra_url=DataUrl(file_path=data_file,
scheme='PyMca'),
channel_url=DataUrl(file_path=data_file,
scheme='PyMca'))
out = pymca_normalization(xas_obj=out.to_dict())
out = pymca_exafs(xas_obj=out.to_dict())
......@@ -74,9 +74,9 @@ class TestStreamSingleSpectrum(unittest.TestCase):
def test_pymca_process_with_cls(self):
data_file = os.path.join(PYMCA_DATA_DIR, "EXAFS_Cu.dat")
out = read_pymca_xas(spectra_url=DataUrl(file_path=data_file,
scheme='PyMca'),
channel_url=DataUrl(file_path=data_file,
out = read_xas(spectra_url=DataUrl(file_path=data_file,
scheme='PyMca'),
channel_url=DataUrl(file_path=data_file,
scheme='PyMca'))
out = PyMca_normalization()(xas_obj=out)
out = PyMca_exafs()(xas_obj=out)
......
......@@ -33,9 +33,9 @@ import unittest
import tempfile
import h5py
from xas.core.types import Spectrum, XASObject
from xas.core.utils.pymca import read_spectrum
from xas.io.utils.pymca import read_spectrum
from xas.core.utils import spectra as spectra_utils
from xas.io import read_pymca_xas
from xas.core.io import read as read_xas
from PyMca5.PyMcaDataDir import PYMCA_DATA_DIR
from silx.io.url import DataUrl
import silx.io.utils
......@@ -95,11 +95,11 @@ class TestXASObject(unittest.TestCase):
url_spectra = DataUrl(file_path=filename,
data_path=spectra_path,
scheme='silx')
self.xas_obj = read_pymca_xas(spectra_url=url_spectra,
channel_url=DataUrl(file_path=filename,
self.xas_obj = read_xas(spectra_url=url_spectra,
channel_url=DataUrl(file_path=filename,
data_path=channel_path,
scheme='silx')
)
)
self.assertEqual(self.xas_obj.dim1, 20)
self.assertEqual(self.xas_obj.dim2, 10)
self.assertEqual(self.xas_obj.n_spectrum, 20*10)
......
......@@ -30,6 +30,7 @@ __date__ = "06/11/2019"
from silx.io.dictdump import dicttoh5
from silx.io.url import DataUrl
import xas.io
import copy
import numpy
import logging
......@@ -276,7 +277,6 @@ class XASObject(object):
def from_file(h5_file, entry='scan1', spectra_path='data/absorbed_beam',
energy_path='data/energy', configuration_path='configuration'):
# load only mu and energy from the file
import xas.io
spectra_url = DataUrl(file_path=h5_file,
data_path='/'.join((entry, spectra_path)),
scheme='silx')
......@@ -289,9 +289,10 @@ class XASObject(object):
config_url = DataUrl(file_path=h5_file,
data_path='/'.join((entry, configuration_path)),
scheme='silx')
return xas.io.read_pymca_xas(spectra_url=spectra_url,
channel_url=energy_url,
config_url=config_url)
sp, en, conf = xas.io.read_xas(spectra_url=spectra_url,
channel_url=energy_url,
config_url=config_url)
return XASObject(spectra=sp, energy=en, configuration=conf)
def dump(self, h5_file):
"""dump the XAS object to a file_path within the Nexus format"""
......@@ -366,7 +367,6 @@ class XASObject(object):
should be saved and a numpy.ndarray
:type: Union[numpy.ndarray, dict]
"""
import xas.io
xas.io.write_xas_proc(self.linked_h5_file, entry=self.__entry_name,
processing_order=self.get_next_processing_index(),
process=process, data=data)
......@@ -377,8 +377,6 @@ class XASObject(object):
:return: the dict of process information
:rtype: dict
"""
import xas.io
if not self.linked_h5_file:
_logger.warning('process flow is store in the linked .h5 file. If'
'no link is defined then this information is not'
......
......@@ -31,7 +31,7 @@ __date__ = "04/07/2019"
from xas.core.types import XASObject
from xas.io import read_pymca_xas
from xas.core.io import read as read_xas
from silx.gui import qt
from silx.gui.plot import Plot1D
from silx.gui.plot.StackView import StackViewMainWindow
......@@ -303,7 +303,9 @@ if __name__ == '__main__':
data_path='/data/NXdata/Channel',
scheme='silx')
xas_obj = read_pymca_xas(spectra_url=url_spectra, channel_url=url_channel)
sp, en, conf = read_xas(spectra_url=url_spectra,
channel_url=url_channel)
xas_obj = XASObject(spectra=sp, energy=en, configuration=conf)
# import numpy
# from xas.core.utils import spectra as spectra_utils
......
......@@ -31,7 +31,8 @@ __date__ = "03/07/2019"
from silx.gui.dialog.DataFileDialog import DataFileDialog
from silx.io.url import DataUrl
from silx.gui import qt
from xas.io import read_pymca_xas, read_pymca_xas_from_file, InputType
from xas.core.io import read as read_xas, read_frm_file
from xas.io import InputType
import logging
_logger = logging.getLogger(__name__)
......@@ -111,7 +112,7 @@ class XASObjectDialog(qt.QWidget):
def buildXASObject(self):
if self.getCurrentType() == InputType.single_spectrum:
return read_pymca_xas_from_file(file_path=self._datDialog.getFileSelected())
return read_frm_file(file_path=self._datDialog.getFileSelected())
elif self.getCurrentType() == InputType.multiple_spectra:
spectra_url = self._h5Dialog.getSpectraUrl()
energy_url = self._h5Dialog.getEnergyUrl()
......@@ -125,9 +126,9 @@ class XASObjectDialog(qt.QWidget):
check_url(spectra_url, 'spectra')
check_url(energy_url, 'energy / channel')
return read_pymca_xas(spectra_url=self._h5Dialog.getSpectraUrl(),
channel_url=self._h5Dialog.getEnergyUrl(),
config_url=self._h5Dialog.getConfigurationUrl())
return read_xas(spectra_url=self._h5Dialog.getSpectraUrl(),
channel_url=self._h5Dialog.getEnergyUrl(),
config_url=self._h5Dialog.getConfigurationUrl())
else:
raise ValueError('unmanaged input type')
......
......@@ -31,63 +31,27 @@ import logging
import h5py
from silx.io import utils
from silx.io.url import DataUrl
from xas.core.utils.pymca import read_spectrum
from xas.core.types import XASObject
from xas.core.process.process import Process
from silx.io.dictdump import dicttoh5
from silx.utils.enum import Enum
from xas.io.utils.pymca import read_spectrum
from datetime import datetime
import numpy
_logger = logging.getLogger(__name__)
DEFAULT_SPECTRA_PATH = '/data/NXdata/data'
DEFAULT_CHANNEL_PATH = '/data/NXdata/Channel'
DEFAULT_CONF_PATH = '/configuration'
class InputType(Enum):
single_spectrum = '*.dat'
multiple_spectra = '*.h5'
def read_pymca_xas_from_file(file_path):
"""
:param str file_path:
:return: `.XASObject`
"""
if file_path.endswith('.dat'):
return read_pymca_xas(
spectra_url=DataUrl(file_path=file_path,
scheme='PyMca'),
channel_url=DataUrl(file_path=file_path,
scheme='PyMca'),
)
elif h5py.is_hdf5(file_path):
return read_pymca_xas(
spectra_url=DataUrl(file_path=file_path,
scheme='silx',
data_path=DEFAULT_SPECTRA_PATH),
channel_url=DataUrl(file_path=file_path,
scheme='silx',
data_path=DEFAULT_CHANNEL_PATH),
config_url=DataUrl(file_path=file_path,
scheme='silx',
data_path='configuration'))
else:
raise ValueError('file type not managed, unable to load')
def read_pymca_xas(spectra_url, channel_url, config_url=None):
def read_xas(spectra_url, channel_url, config_url=None):
"""
Read the given spectra url and the config url if any
:param Union[DataUrl, str] spectra_url:
:param DataUrl config_url:
:return: PyMcaSpectrum
:return: spectra, energy, configuration
"""
def get_url(original_url, name):
url_ = original_url
......@@ -132,60 +96,7 @@ def read_pymca_xas(spectra_url, channel_url, config_url=None):
energy = load_data(_energy_url, name='energy')
configuration = load_data(_config_url, name='configuration')
return XASObject(spectra=spectra, energy=energy, configuration=configuration)
class XASWriter(object):
"""
class to write the output file. In this case we need a class in order to
setup the output file before
"""
def __init__(self):
self._output_file = None
@property
def output_file(self):
return self._output_file
@output_file.setter
def output_file(self, file_):
self._output_file = file_
def setProperties(self, properties):
if '_output_file_setting' in properties:
self._output_file = properties['_output_file_setting']
def dump_xas(self, xas_obj, write_process=True):
"""
write the XASObject into a hdf5 file.
:param :class:`.XASObject` xas_obj: object to be stored
:param bool write_process: if True then store the process flow in the
same file.
"""
if not self._output_file:
_logger.warning('no output file defined, please give path to the output file')
self._output_file = input()
if isinstance(xas_obj, dict):
_xas_obj = XASObject.from_dict(xas_obj)
else:
_xas_obj = xas_obj
_logger.info(('dump xas obj to', self._output_file))
# write raw data
write_xas(h5_file=self._output_file, energy=_xas_obj.energy,
mu=_xas_obj.absorbed_beam(), entry=xas_obj.entry)
if len(get_xasproc(self._output_file, entry=xas_obj.entry)) > 0:
_logger.warning('removing previous process registred. They are no '
'more accurate')
xas_obj.clean_process_flow()