Commit adbb7b3a authored by payno's avatar payno
Browse files

[io] rework h5 output to use Nexus format

parent 0273e3ca
......@@ -56,7 +56,7 @@ class XASOutputOW(OWWidget):
inputs = [("spectra", XASObject, 'process')]
_output_file_setting = Setting(str())
process_function = xas.io.Writer_pymca_cas
process_function = xas.io.Writer_pymca_xas
def __init__(self):
super().__init__()
......
......@@ -179,4 +179,17 @@ class PyMca_exafs(Process):
for spectrum, res in zip(xas_obj.spectra, res_list):
spectrum.update(res)
def definition(self):
return "exafs calculation"
def program_version(self):
import PyMca5
return PyMca5.version()
def program_name(self):
return 'exafs'
def getConfiguration(self):
return self._settings
__call__ = process
......@@ -169,5 +169,17 @@ class PyMca_normalization(Process):
for spectrum, res in zip(xas_obj.spectra, res_list):
spectrum.update(res)
def definition(self):
return "Normalization of the spectrum"
def program_version(self):
import PyMca5
return PyMca5.version()
def program_name(self):
return 'normalization'
def getConfiguration(self):
return self._settings
__call__ = process
......@@ -67,3 +67,23 @@ class Process(object):
assert isinstance(spectrum.mu, numpy.ndarray)
assert isinstance(_xas_obj, XASObject)
return _xas_obj
def program_name(self):
"""Name of the program used for this processing"""
raise NotImplementedError('Base class')
def program_version(self):
"""version of the program used for this processing"""
raise NotImplementedError('Base class')
def definition(self):
"""definition of the process"""
raise NotImplementedError('Base class')
def getConfiguration(self):
"""
:return: configuration of the process
:rtype: dict
"""
raise NotImplementedError('Base class')
......@@ -34,6 +34,7 @@ from PyMca5.PyMcaDataDir import PYMCA_DATA_DIR
from xas.core.types import XASObject
import shutil
import tempfile
import h5py
import numpy
import os
......@@ -65,12 +66,26 @@ class TestReadWrite(unittest.TestCase):
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 suite():
test_suite = unittest.TestSuite()
for ui in (TestReadWrite,):
......
......@@ -237,7 +237,7 @@ class XASObject(object):
return XASObject.from_dict(ddict=ddict)
def dump(self, h5_file):
"""dump the XAS object to a file_path"""
"""dump the XAS object to a file_path within the Nexus format"""
dicttoh5(treedict=self.to_dict(with_process_details=False),
h5file=h5_file)
......@@ -567,3 +567,9 @@ class _FT(object):
else:
return missing
class Sample(object):
"""Description of the sample. Needed for writing valid nx file"""
def __init__(self, name='undefined sample', description=None):
self.name = name
self.description = description
......@@ -33,7 +33,9 @@ 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.utils.enum import Enum
from datetime import datetime
_logger = logging.getLogger(__name__)
......@@ -129,7 +131,7 @@ def read_pymca_xas(spectra_url, channel_url, config_url=None):
return XASObject(spectra=spectra, energy=energy, configuration=configuration)
class Writer_pymca_cas(object):
class Writer_pymca_xas(object):
"""
class to write the output file. In this case we need a class in order to
setup the output file before
......@@ -149,7 +151,145 @@ class Writer_pymca_cas(object):
_xas_obj = XASObject.from_dict(xas_obj)
else:
_xas_obj = xas_obj
_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)
__call__ = dump_pymca_xas
def str_to_utf8(text):
"""Convert str or sequence of str to type compatible with h5py
:param Union[str,List[str]] text:
:rtype: numpy.ndarray
"""
return numpy.array(text, dtype=h5py.special_dtype(vlen=str))
# TODO: process, sample, energy and data can be embed in a 'result' / output object
def write_xas_proc(h5_file, entry, process, data, processing_order, data_path='/'):
"""
Write a xas :class:`.Process` into .h5
:param str h5_file: path to the hdf5 file
:param str entry: entry name
:param :class:`.Process` process: process executed
:param numpy.ndarray data: process result data
:param int processing_order: processing order of treatment
:param str data_path: path to store the data
"""
assert isinstance(process, Process)
# write the xasproc
with h5py.File(h5_file) as h5f:
nx_entry = h5f.require_group('/'.join((data_path, entry)))
nx_entry.attrs["NX_class"] = "NXentry"
nx_process = nx_entry.require_group(process.name)
nx_process.attrs['NX_class'] = "NXprocess"
nx_process['program'] = process.program_name()
nx_process['version'] = process.program_version()
nx_process['date'] = datetime.now().replace(microsecond=0).isoformat()
nx_process['processing_order'] = numpy.int32(processing_order)
nx_data = nx_entry.require_group('data')
nx_data.attrs['NX_class'] = "NXdata"
nx_data.attrs['signal'] = 'data'
nx_process['data'] = data
nx_process['data'].attrs['interpretation'] = 'image'
if process.getConfiguration() is not None:
from silx.io.dictdump import dicttoh5
dicttoh5(process.getConfiguration(),
h5file=h5_file,
h5path=''.join((nx_process.name, 'parameters')),
overwrite_data=True)
def write_xas(h5_file, entry, sample, energy, mu, start_time=None,
data_path='/', title=None, definition=None):
"""
Write raw date in nexus format
:param str h5_file: path to the hdf5 file
:param str entry: entry name
:param :class:`.Sample` sample: definition of the sample
:param energy: beam energy
:type: numpy.ndarray, one dim
:param mu: beam absorption
:type: numpy.ndarray, two dimensions
:param start_time:
:param data_path:
:param str title: experiment title
:param str definition: experiment definition
"""
with h5py.File(h5_file) as h5f:
nx_entry = h5f.require_group('/'.join((data_path, entry)))
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"
# 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['NX_class'] = "NXdata"
if sample:
nx_sample = nx_entry.require_group('sample')
nx_sample.attrs['NX_class'] = "NXsample"
nx_sample['name'] = sample.name
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['absorbed_beam'] = h5py.SoftLink(nx_absorbed_beam['data'].name)
if start_time is not None:
nx_entry['start_time'] = start_time
# TODO: add title and definition from a class `setup` or `experimentation`
if title is not None:
nx_entry['title'] = title
if definition is not None:
nx_entry['definition'] = definition
if __name__ == '__main__':
import os
from xas.core.process.normalization import PyMca_normalization
from xas.core.process.exafs import PyMca_exafs
from xas.core.types import Sample
import numpy
h5_file = 'test_xas_123.h5'
if os.path.exists(h5_file):
os.remove(h5_file)
sample = Sample(name='mysample')
data = numpy.random.rand(256*20*10)
data = data.reshape((256, 20, 10))
process_data = numpy.random.rand(256*20*10).reshape((256, 20, 10))
energy = numpy.linspace(start=3.25, stop=3.69, num=256)
write_xas(h5_file=h5_file, entry='scan1', sample=sample, energy=energy,
mu=data, )
process_norm = PyMca_normalization()
write_xas_proc(h5_file=h5_file, entry='scan1', process=process_norm,
data=process_data, processing_order=1)
process_exafs = PyMca_exafs()
process_data2 = numpy.random.rand(256*20*10).reshape((256, 20, 10))
write_xas_proc(h5_file=h5_file, entry='scan1', process=process_exafs,
data=process_data2, processing_order=2)
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