Commit 04349cf7 authored by Henri Payno's avatar Henri Payno
Browse files

[test] fix warning on some test regarding pymca (bad configuration bring warnings)

parent 916c9947
Pipeline #13409 failed with stage
in 4 minutes and 41 seconds
......@@ -96,6 +96,9 @@ class Process(object):
else:
return None
def setConfiguration(self, configuration):
self._settings = configuration
def register_process(self, xas_obj, data_keys):
"""
Store the current process in the linked h5 file if any,
......
......@@ -136,7 +136,7 @@ class PyMca_exafs(Process):
def setProperties(self, properties):
if '_pymcaSettings' in properties:
self._settings = properties['_pymcaSettings']
self.setConfiguration(properties['_pymcaSettings'])
def process(self, xas_obj):
_xas_obj = self.getXasObject(xas_obj=xas_obj)
......@@ -148,7 +148,7 @@ class PyMca_exafs(Process):
self._pool_process(xas_obj=_xas_obj)
self._advancement.endProcess()
self.register_process(_xas_obj, data_keys=("EXAFSKValues",
"EXAFSSignal"))
"EXAFSSignal"))
return _xas_obj
def _pool_process(self, xas_obj):
......
......@@ -119,7 +119,10 @@ class PyMca_k_weight(Process):
def setProperties(self, properties):
if '_kWeightSetting' in properties:
self._settings['k_weight'] = properties['_kWeightSetting']
_properties = properties.copy()
_properties['k_weight'] = _properties['_kWeightSetting']
del _properties['_kWeightSetting']
self.setConfiguration(_properties)
def _k_weight(self):
if 'k_weight' in self._settings:
......
......@@ -35,7 +35,7 @@ import shutil
import tempfile
import unittest
import h5py
import numpy
from silx.io.dictdump import h5todict
from silx.io.url import DataUrl
from xas.core.process.roi import xas_roi, _ROI as XASROI
from xas.core.io import XASWriter
......@@ -59,12 +59,20 @@ else:
@unittest.skipIf(has_pymca is False, 'PyMca5 is not installed')
class TestStreamSingleSpectrum(unittest.TestCase):
"""Make sure the process have valid io"""
def setUp(self):
self.exafs_configuration = {
"Knots": {"Values": (1, 2, 5), "Number": 3, 'Orders': [3, 3, 3]},
'KMin': 0,
'KMax': 2.3
}
def test_pymca_process(self):
data_file = os.path.join(PYMCA_DATA_DIR, "EXAFS_Cu.dat")
out = read_xas(spectra_url=DataUrl(file_path=data_file,
scheme='PyMca'),
channel_url=DataUrl(file_path=data_file,
scheme='PyMca'))
out.configuration = {'EXAFS': self.exafs_configuration, 'SET_KWEIGHT': 0}
out = pymca_normalization(xas_obj=out)
out = pymca_exafs(xas_obj=out)
out = pymca_k_weight(xas_obj=out)
......@@ -77,6 +85,7 @@ class TestStreamSingleSpectrum(unittest.TestCase):
scheme='PyMca'),
channel_url=DataUrl(file_path=data_file,
scheme='PyMca'))
out.configuration = {'EXAFS': self.exafs_configuration, 'SET_KWEIGHT': 0}
out = pymca_normalization(xas_obj=out.to_dict())
out = pymca_exafs(xas_obj=out.to_dict())
out = pymca_k_weight(xas_obj=out.to_dict())
......@@ -90,8 +99,12 @@ class TestStreamSingleSpectrum(unittest.TestCase):
channel_url=DataUrl(file_path=data_file,
scheme='PyMca'))
out = PyMca_normalization()(xas_obj=out)
out = PyMca_exafs()(xas_obj=out)
out = PyMca_k_weight()(xas_obj=out)
exafs_process = PyMca_exafs()
exafs_process.setConfiguration(configuration=self.exafs_configuration)
out = exafs_process(xas_obj=out)
kweight_process = PyMca_k_weight()
kweight_process.setConfiguration({'k_weight': 0})
out = kweight_process(xas_obj=out)
out = PyMca_ft()(xas_obj=out)
assert isinstance(out, XASObject)
self.assertTrue(out.spectra[0].ft is not None)
......@@ -103,7 +116,8 @@ class TestStreamSingleSpectrum(unittest.TestCase):
xas_obj = XASObject(spectra=spectra, energy=energy,
configuration={'roi': roi.to_dict()})
out = xas_roi(xas_obj.to_dict())
out = pymca_normalization(xas_obj=out.to_dict())
out.configuration = {'EXAFS': self.exafs_configuration, 'SET_KWEIGHT': 0}
out = pymca_normalization(xas_obj=out)
out = pymca_exafs(xas_obj=out)
out = pymca_k_weight(xas_obj=out)
out = pymca_ft(xas_obj=out)
......@@ -135,8 +149,17 @@ class TestWorkflowAndH5LinkedFile(unittest.TestCase):
"""Test that the processing can be stored continuously on a .h5 file"""
self.assertTrue(self.xas_obj.linked_h5_file is not None)
out = PyMca_normalization()(xas_obj=self.xas_obj)
out = PyMca_exafs()(xas_obj=out)
out = PyMca_k_weight()(xas_obj=out)
exafs_process = PyMca_exafs()
configuration = {
"Knots": {"Values": (1, 2, 5), "Number": 3, 'Orders': [3, 3, 3]},
'KMin': 0,
'KMax': 2.3
}
exafs_process.setConfiguration(configuration=configuration)
out = exafs_process(xas_obj=out)
k_weight_process = PyMca_k_weight()
k_weight_process.setConfiguration({'k_weight': 0})
out = k_weight_process(xas_obj=out)
out = PyMca_ft()(xas_obj=out)
out = PyMca_normalization()(xas_obj=out)
assert isinstance(out, XASObject)
......@@ -156,8 +179,17 @@ class TestWorkflowAndH5LinkedFile(unittest.TestCase):
processes"""
self.assertTrue(self.xas_obj.linked_h5_file is not None)
out = PyMca_normalization()(xas_obj=self.xas_obj.to_dict())
out = PyMca_exafs()(xas_obj=out.to_dict())
out = PyMca_k_weight()(xas_obj=out.to_dict())
exafs_process = PyMca_exafs()
configuration = {
"Knots": {"Values": (1, 2, 5), "Number": 3, 'Orders': [3, 3, 3]},
'KMin': 0,
'KMax': 2.3
}
exafs_process.setConfiguration(configuration=configuration)
out = exafs_process(xas_obj=out.to_dict())
k_weight_process = PyMca_k_weight()
k_weight_process.setConfiguration({'k_weight': 0})
out = k_weight_process(xas_obj=out.to_dict())
out = PyMca_ft()(xas_obj=out.to_dict())
out = PyMca_normalization()(xas_obj=out.to_dict())
assert isinstance(out, XASObject)
......@@ -186,6 +218,7 @@ class TestSaveFlowAuto(unittest.TestCase):
assert len(energy) == spectra.shape[0]
self.xas_obj = XASObject(spectra=spectra, energy=energy, dim1=10,
dim2=10)
self.h5_file = os.path.join(self.output_dir, 'output_file.h5')
def tearDown(self):
......@@ -194,14 +227,23 @@ class TestSaveFlowAuto(unittest.TestCase):
def test(self):
self.assertTrue(self.xas_obj.linked_h5_file is not None)
out = PyMca_normalization()(xas_obj=self.xas_obj)
out = PyMca_exafs()(xas_obj=out)
out = PyMca_k_weight()(xas_obj=out)
exafs = PyMca_exafs()
configuration = {
"Knots": {"Values": (1, 2, 5), "Number": 3, 'Orders': [3, 3, 3]},
'KMin': 0,
'KMax': 2.3
}
exafs.setConfiguration(configuration)
out = exafs(xas_obj=out)
kweight_process = PyMca_k_weight()
kweight_process.setConfiguration({'k_weight': 0})
out = kweight_process(xas_obj=out)
out = PyMca_ft()(xas_obj=out)
fina_xas_obj = PyMca_normalization()(xas_obj=out)
out = PyMca_normalization()(xas_obj=out)
writer = XASWriter()
writer.output_file = self.h5_file
writer(fina_xas_obj)
writer(out)
with h5py.File(self.h5_file) as hdf:
self.assertTrue('scan1' in hdf.keys())
......@@ -209,6 +251,21 @@ class TestSaveFlowAuto(unittest.TestCase):
self.assertTrue('absorbed_beam' in hdf['scan1'].keys())
self.assertTrue('monochromator' in hdf['scan1'].keys())
self.assertTrue('xas_process_1' in hdf['scan1'].keys())
self.assertTrue('xas_process_2' in hdf['scan1'].keys())
self.assertTrue('xas_process_3' in hdf['scan1'].keys())
self.assertTrue('xas_process_4' in hdf['scan1'].keys())
self.assertTrue('xas_process_5' in hdf['scan1'].keys())
targetted_xas_process = hdf['scan1']['xas_process_2']
self.assertTrue('program' in targetted_xas_process)
print(targetted_xas_process['program'][()])
self.assertTrue(targetted_xas_process['program'][()] == 'pymca_exafs')
self.assertTrue('configuration' in targetted_xas_process)
# check one configuration
stored_config = h5todict(self.h5_file,
path='/scan1/xas_process_2/configuration')
for key in ('KMin', 'KMax'):
self.assertTrue(configuration[key] == stored_config[key])
def suite():
......
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