Commit b87f1e3a authored by payno's avatar payno
Browse files

[xas][io] add test on the xas writer + some relative fix.

parent 4438dfc1
Pipeline #12642 failed with stage
in 60 minutes
......@@ -56,7 +56,7 @@ class XASOutputOW(OWWidget):
inputs = [("spectra", XASObject, 'process')]
_output_file_setting = Setting(str())
process_function = xas.io.Writer_pymca_xas
process_function = xas.io.XASWriter
def __init__(self):
super().__init__()
......
......@@ -28,9 +28,10 @@ __date__ = "06/11/2019"
import unittest
from xas.io import read_pymca_xas
from xas.io import read_pymca_xas, XASWriter
from silx.io.url import DataUrl
from PyMca5.PyMcaDataDir import PYMCA_DATA_DIR
from xas.core.utils import spectra as spectra_utils
from xas.core.types import XASObject
import shutil
import tempfile
......@@ -59,36 +60,38 @@ class TestReadWrite(unittest.TestCase):
self.assertTrue('Mu' in res.spectra[0])
self.assertTrue('Energy' in res.spectra[0])
def testReadWriteH5(self):
"""Tset io relative to h5py"""
xas_obj = XASObject(configuration={'axis': 2},
spectra=numpy.arange(10).reshape(10, 1, 1),
energy=numpy.arange(10))
xas_obj.dump(self.outputfile)
with h5py.File(self.outputfile) as f:
self.assertTrue('scan1' in f)
self.assertTrue('title' in f['scan1'])
self.assertTrue('start_time' in f['scan1'])
self.assertTrue('start_time' in f['scan1'])
self.assertTrue('definition' in f['scan1'])
self.assertTrue('data' in f['scan1'])
self.assertTrue('absorbed_beam' in f['scan1/data'])
self.assertTrue('energy' in f['scan1/data'])
self.assertTrue(os.path.exists(self.outputfile))
xas_obj2 = XASObject.from_file(self.outputfile)
assert xas_obj2
self.assertEqual(xas_obj, xas_obj2)
class TestNxWriting(unittest.TestCase):
"""Test that the nx process is correctly store ad the output data"""
pass
def setUp(self):
self.output_dir = tempfile.mkdtemp()
spectra = spectra_utils.create_dataset(shape=(256, 20, 10))
energy = numpy.linspace(start=3.26, stop=3.96, num=256)
self.xas_obj = XASObject(spectra=spectra, energy=energy)
self.h5_file = os.path.join(self.output_dir, 'output_file.h5')
def tearDown(self):
shutil.rmtree(self.output_dir)
def testWriteRead(self):
writer = XASWriter()
writer.output_file = self.h5_file
writer(self.xas_obj)
with h5py.File(self.h5_file) as hdf:
self.assertTrue('scan1' in hdf.keys())
self.assertTrue('data' in hdf['scan1'].keys())
self.assertTrue('absorbed_beam' in hdf['scan1'].keys())
self.assertTrue('monochromator' in hdf['scan1'].keys())
loaded_xas_obj = XASObject.from_file(self.h5_file,
configuration_path=None)
numpy.testing.assert_allclose(loaded_xas_obj.energy, self.xas_obj.energy)
numpy.testing.assert_allclose(loaded_xas_obj.absorbed_beam(), self.xas_obj.absorbed_beam())
def suite():
test_suite = unittest.TestSuite()
for ui in (TestReadWrite,):
for ui in (TestReadWrite, TestNxWriting):
test_suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(ui))
return test_suite
......
......@@ -111,7 +111,7 @@ class TestWorkflowAndH5LinkedFile(unittest.TestCase):
roi = XASROI(origin=(0, 2), size=(5, 1))
self.xas_obj = XASObject(spectra=spectra, energy=energy,
configuration={'roi': roi.to_dict()})
self.h5_file = os.path.join(self.output_dir, 'putput_file.h5')
self.h5_file = os.path.join(self.output_dir, 'output_file.h5')
self.xas_obj.link_to_h5(self.h5_file)
def tearDown(self):
......@@ -130,7 +130,6 @@ class TestWorkflowAndH5LinkedFile(unittest.TestCase):
assert out.linked_h5_file is self.h5_file
# then check all process are correctly registered with the valid id...
processes = self.xas_obj.get_process_flow()
shutil.copy(self.h5_file, '/nobackup/linazimov/payno/dev/esrf/xas/test_workflow.h5')
self.assertEqual(len(processes), 5)
self.assertEqual(processes[1]['program'], 'pymca_normalization')
self.assertEqual(processes[2]['program'], 'pymca_exafs')
......@@ -153,7 +152,6 @@ class TestWorkflowAndH5LinkedFile(unittest.TestCase):
assert out.linked_h5_file == self.h5_file
# then check all process are correctly registered with the valid id...
processes = self.xas_obj.get_process_flow()
shutil.copy(self.h5_file, '/nobackup/linazimov/payno/dev/esrf/xas/test_workflow.h5')
self.assertEqual(len(processes), 5)
self.assertEqual(processes[1]['program'], 'pymca_normalization')
self.assertEqual(processes[2]['program'], 'pymca_exafs')
......
......@@ -29,6 +29,7 @@ __date__ = "06/11/2019"
from silx.io.dictdump import dicttoh5, h5todict
from silx.io.url import DataUrl
import copy
import numpy
import logging
......@@ -52,6 +53,10 @@ class XASObject(object):
self.spectra = (energy, spectra, dim1, dim2)
self.configuration = configuration
@property
def entry(self):
return self.__entry_name
@property
def spectra(self):
return self.__spectra
......@@ -179,6 +184,12 @@ class XASObject(object):
spectra_dict[str(i_spectrum) + '_spectrum'] = spectrum.to_dict()
return spectra_dict
def absorbed_beam(self):
return XASObject._spectra_volume(spectra=self.spectra,
key='Mu',
dim_1=self.dim1,
dim_2=self.dim2)
@staticmethod
def _spectra_volume(spectra, key, dim_1, dim_2):
"""Convert a list of spectra (mu) to a numpy array.
......@@ -243,9 +254,25 @@ class XASObject(object):
return XASObject().load_frm_dict(ddict=ddict)
@staticmethod
def from_file(h5_file):
ddict = h5todict(h5file=h5_file)
return XASObject.from_dict(ddict=ddict)
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')
energy_url = DataUrl(file_path=h5_file,
data_path='/'.join((entry, energy_path)),
scheme='silx')
if configuration_path is None:
config_url = None
else:
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)
def dump(self, h5_file):
"""dump the XAS object to a file_path within the Nexus format"""
......
......@@ -104,7 +104,7 @@ def read_pymca_xas(spectra_url, channel_url, config_url=None):
_spectra_url = get_url(original_url=spectra_url, name='spectra')
_energy_url = get_url(original_url=channel_url, name='energy')
_config_url = config_url
if _config_url == '':
if type(_config_url) is str and _config_url == '':
_config_url = None
if not (_config_url is None or isinstance(_config_url, DataUrl)):
raise TypeError('given input for configuration is invalid')
......@@ -121,7 +121,10 @@ def read_pymca_xas(spectra_url, channel_url, config_url=None):
return energy
else:
if data_url.is_valid():
return utils.get_data(data_url)
try:
return utils.get_data(data_url)
except ValueError as e:
_logger.error(e)
else:
_logger.warning('invalid url for', name, ', will not load it')
......@@ -132,7 +135,7 @@ def read_pymca_xas(spectra_url, channel_url, config_url=None):
return XASObject(spectra=spectra, energy=energy, configuration=configuration)
class Writer_pymca_xas(object):
class XASWriter(object):
"""
class to write the output file. In this case we need a class in order to
setup the output file before
......@@ -140,6 +143,14 @@ class Writer_pymca_xas(object):
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']
......@@ -155,11 +166,10 @@ class Writer_pymca_xas(object):
_logger.info(('dump xas obj to', self._output_file))
mu = XASObject._spectra_volume(spectra=xas_obj.spectra, key='mu',
dim_1=xas_obj.dim1, dim_2=xas_obj.dim2)
write_xas(h5_file=self._output_file, energy=xas_obj.energy, mu=mu,
entry='scan1')
_xas_obj.dump(self._output_file)
mu = XASObject._spectra_volume(spectra=_xas_obj.spectra, key='Mu',
dim_1=_xas_obj.dim1, dim_2=_xas_obj.dim2)
write_xas(h5_file=self._output_file, energy=_xas_obj.energy, mu=mu,
entry=xas_obj.entry)
__call__ = dump_pymca_xas
......@@ -229,7 +239,7 @@ def write_xas_proc(h5_file, entry, process, data, processing_order, data_path='/
overwrite_data=True)
def write_xas(h5_file, entry, sample, energy, mu, start_time=None,
def write_xas(h5_file, entry, energy, mu, sample=None, start_time=None,
data_path='/', title=None, definition=None):
"""
Write raw date in nexus format
......@@ -251,17 +261,17 @@ def write_xas(h5_file, entry, sample, energy, mu, start_time=None,
nx_entry.attrs["NX_class"] = "NXentry"
# store energy
nx_monocromator = nx_entry.require_group('monocromator')
nx_monocromator.attrs['NX_class'] = "NXmonochromator"
nx_monocromator['energy'] = energy
nx_monocromator['energy'].attrs['interpretation'] = 'spectrum'
nx_monocromator['energy'].attrs['NX_class'] = "NXdata"
nx_monochromator = nx_entry.require_group('monochromator')
nx_monochromator.attrs['NX_class'] = "NXmonochromator"
nx_monochromator['energy'] = energy
nx_monochromator['energy'].attrs['interpretation'] = 'spectrum'
nx_monochromator['energy'].attrs['NX_class'] = "NXdata"
# store absorbed beam
nx_absorbed_beam = nx_entry.require_group('absorbed_beam')
nx_absorbed_beam.attrs['NX_class'] = "NXdetector"
nx_absorbed_beam['data'] = mu
nx_absorbed_beam['data'].attrs['interpretation'] = 'spectrum'
nx_absorbed_beam['data'].attrs['interpretation'] = 'image'
nx_absorbed_beam['data'].attrs['NX_class'] = "NXdata"
if sample:
......@@ -272,7 +282,7 @@ def write_xas(h5_file, entry, sample, energy, mu, start_time=None,
nx_data = nx_entry.require_group('data')
nx_data.attrs['NX_class'] = "NXdata"
# create some link on data
nx_data['energy'] = h5py.SoftLink(nx_monocromator['energy'].name)
nx_data['energy'] = h5py.SoftLink(nx_monochromator['energy'].name)
nx_data['absorbed_beam'] = h5py.SoftLink(nx_absorbed_beam['data'].name)
if start_time is not None:
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment