Commit fad58939 authored by payno's avatar payno
Browse files

[tmp commit] start creating plot group in the output file for bm23 requirements.

parent 9165e9b0
......@@ -99,9 +99,6 @@ def process_spectr_autobk(
if opt_name in _conf:
opts[opt_name] = _conf[opt_name]
# print(opts)
# print(opts)
if _DEBUG is True:
assert isinstance(spectrum, Group)
if overwrite:
......@@ -25,7 +25,7 @@
"""wrapper to the larch mback process"""
from est.core.types import Spectrum, XASObject
from est.core.process.process import Process
from est.core.process.process import Process, _NexusSpectrumDef
from est.core.process.process import _input_desc
from est.core.process.process import _output_desc
from larch.xafs.pre_edge import pre_edge
......@@ -159,7 +159,20 @@ class Larch_pre_edge(Process):
# "mu_ref",
# "I0",
# "I1",
# "I2"
_NexusSpectrumDef(signal="Mu", axes=("energy",), auxiliary_signals=None),
# _plot_definition(signal="energy", axes="mu", auxiliary_signal="I0, I1, I2"),
# _plot_definition(signal="energy", axes="mu",
# auxiliary_signal="mu_ref"),
_NexusSpectrumDef(signal="flat", axes=("energy",), auxiliary_signals=None),
return _xas_obj
......@@ -43,6 +43,51 @@ _input_desc = namedtuple("_input_desc", ["name", "type", "handler", "doc"])
_output_desc = namedtuple("_output_desc", ["name", "type", "doc"])
class _NexusSpectrumDef:
"""Util function to define a Nexus plot"""
def __init__(self, signal, axes, auxiliary_signals):
self.__signal = None
self.__axes = None
self.__auxiliary_signals = None
self.signal = signal
self.axes = axes
self.auxiliary_signals = auxiliary_signals
def signal(self):
return self.__signal
def signal(self, signal):
if not isinstance(signal, str):
raise TypeError("signal should be an instance of str")
self.__signal = signal
def axes(self):
return self.__axes
def axes(self, axes):
if not isinstance(axes, tuple):
raise TypeError("axes should be an instance of tuple")
self.__axes = axes
def auxiliary_signals(self):
return self.__auxiliary_signals
def auxiliary_signals(self, auxiliary_signals):
if not isinstance(auxiliary_signals, (tuple, type(None))):
raise TypeError("auxiliary_signals should be an instance of tuple")
self.__auxiliary_signals = auxiliary_signals
class Process(object):
def __init__(self, name):
assert type(name) is str
......@@ -122,7 +167,8 @@ class Process(object):
self._settings = configuration
def register_process(self, xas_obj: XASObject, data_keys: Iterable):
def register_process(self, xas_obj: XASObject, data_keys: Iterable,
plots: Iterable = tuple()):
Store the current process in the linked h5 file if any,
output data stored will be the one defined by the data_keys
......@@ -130,6 +176,7 @@ class Process(object):
:param xas_obj: object for which we want to save the treatment
:type: :class:`.XASObject`
:param tuple data_keys: keys of the id to save
:param plots:
if xas_obj.has_linked_file():
_data = {}
......@@ -150,6 +197,7 @@ class Process(object):
......@@ -193,7 +241,8 @@ class Process(object):
_data[key] = res
raise ValueError()
xas_obj.register_processing(process=self, results=_data)
xas_obj.register_processing(process=self, results=_data,
def addCallback(self, callback):
......@@ -458,7 +458,7 @@ class XASObject(object):
self.__processing_index += 1
return self.__processing_index
def register_processing(self, process, results) -> None:
def register_processing(self, process, results, plots) -> None:
Register one process for the current xas object. This require to having
link a h5file to this object
......@@ -476,6 +476,7 @@ class XASObject(object):
......@@ -287,7 +287,14 @@ def read_xas(
def write_xas_proc(
h5_file, entry, process, results, processing_order, data_path="/", overwrite=True
Write a xas :class:`.Process` into .h5
......@@ -303,6 +310,8 @@ def write_xas_proc(
:param data_path: path to store the data
:type: str
if plots is None:
plots = tuple()
process_name = "xas_process_" + str(processing_order)
# write the xasproc default information
with h5py.File(h5_file, "a") as h5f:
......@@ -348,6 +357,14 @@ def write_xas_proc(
return "image"
return None
def get_path_to_result(res_name):
res_name = res_name.replace(".", "/")
path = "/".join((entry, process_name, "results", res_name))
if path.startswith("/"):
path = "/" + path
return path
# save results
def save_key(key_path, value):
"""Save the given value to the associated path. Manage numpy arrays
and dictionaries"""
......@@ -377,6 +394,62 @@ def write_xas_proc(
key_path = "/".join(("results", key))
save_key(key_path=key_path, value=value)
def save_plot(plot_name, plot):
"""save the given plot to an hdf5 group"""
plot_name = plot_name.replace(".", "/")
plot_path = "/".join((entry, process_name, "plots", plot_name))
with h5py.File(h5_file, "a") as h5f:
plot_group = h5f.require_group(plot_path)
plot_group.attrs["NX_class"] = "NXdata"
plot_group.attrs["interpretation"] = "spectrum"
assert plot.signal is not None
assert plot.axes is not None
def link_dataset(dataset_to_link, name):
# to insure silx isplotting it we should have curve as a 1D object
# but by default we are handling a map of spectra. This is why we
# need to duplicate data here
if dataset_to_link.ndim == 1:
plot_group[name] = h5py.SoftLink(
elif dataset_to_link.ndim == 2:
plot_group[name] = dataset_to_link[:, 0]
elif dataset_to_link.ndim == 3:
plot_group[name] = dataset_to_link[:, 0, 0]
raise ValueError("Unable to handle dataset {}".format(
# handle signal
# plot is only handling 1D data
signal_dataset = h5f[get_path_to_result(plot.signal)]
link_dataset(dataset_to_link=signal_dataset, name=plot.signal)
plot_group.attrs["signal"] = plot.signal
# handle axes
for axe in plot.axes:
axe_dataset = h5f[get_path_to_result(axe)]
link_dataset(dataset_to_link=axe_dataset, name=axe)
plot_group.attrs["axes"] = plot.axes
# handle auxiliary signals
if plot.auxiliary_signals is not None:
for aux_sig in plot.auxiliary_signals:
aux_sig_dataset = h5f[get_path_to_result(aux_sig)]
link_dataset(dataset_to_link=aux_sig_dataset, name=axe)
plot_group.attrs["auxiliary_signal"] = plot.auxiliary_signals
# save plots
for i_plot, plot in enumerate(plots):
print('try to create plot', plot)
plot_name = "plot_{}".format(i_plot)
save_plot(plot_name=plot_name, plot=plot)
# default plot will always be the first one
if len(plots) > 0:
plots_path = "/".join((entry, process_name, "plots"))
with h5py.File(h5_file, "a") as h5f:
plots_group = h5f.require_group(plots_path)
plots_group.attrs["NX_class"] = "NXdata"
plots_group.attrs["default"] = "plot_0"
if process.getConfiguration() is not None:
h5_path = "/".join((nx_process_path, "configuration"))
......@@ -505,6 +578,7 @@ def get_xasproc(h5_file, entry):
from import h5todict
......@@ -514,6 +588,11 @@ def get_xasproc(h5_file, entry):
if key == "configuration":
config_path = "/".join((, "configuration"))
res[key] = h5todict(h5_file, config_path, asarray=False)
elif key == "plots":
plots_grp = h5_group["plots"]
res[key] = {}
for plot_key in plots_grp.keys():
res[key][plot_key] = dict(plots_grp[plot_key].attrs.items())
res[key] = h5_group[key][...]
return res
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