Commit f3be7257 authored by payno's avatar payno
Browse files

Merge branch 'integrate_h5tonx' into 'master'

Integrate dicttinexus

See merge request workflow/est!39
parents e3effb06 9f149455
Pipeline #56148 failed with stages
in 6 minutes and 44 seconds
......@@ -31,7 +31,7 @@ import logging
from datetime import datetime
import h5py
import numpy
from import dicttoh5, h5todict
from import dicttoh5, h5todict, dicttonx
from import DataUrl
from silx.utils.enum import Enum
from est.units import ur
......@@ -42,6 +42,7 @@ from import get_data
from import File as HDF5File
from est import settings
import json
from import read_ascii as larch_read_ascii
......@@ -444,8 +445,6 @@ def write_xas_proc(
with HDF5File(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
......@@ -478,7 +477,6 @@ def write_xas_proc("{} is not available".format(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:
......@@ -487,7 +485,7 @@ def write_xas_proc("{} is not available".format(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:
......@@ -499,17 +497,24 @@ def write_xas_proc(
link_dataset(dataset_to_link=aux_sig_dataset, name=aux_sig)
plot_group.attrs["auxiliary_signals"] = plot.auxiliary_signals
nexus_dict = {
"@NX_class": "NXdata",
"@interpretation": "spectrum",
"@signal": plot.signal,
"@axes": plot.axes,
# handle title(s)
if plot.title is not None:
plot_group.attrs["title"] = plot.title
nexus_dict["@title"] = plot.title
if plot.title_latex is not None:
plot_group.attrs["title_latex"] = plot.title_latex
nexus_dict["@title_latex"] = plot.title_latex
# handle silx style
if plot.silx_style is not None:
import json
nexus_dict["@SILX_style"] = json.dumps(plot.silx_style)
plot_group.attrs["SILX_style"] = json.dumps(plot.silx_style)
dicttonx(nexus_dict, h5_file, mode="a", h5path=plot_path, update_mode="replace")
# save plots
for i_plot, plot in enumerate(plots):
......@@ -519,10 +524,13 @@ def write_xas_proc(
# default plot will always be the first one
if len(plots) > 0:
plots_path = "/".join((entry, process_name, "plots"))
with HDF5File(h5_file, "a") as h5f:
plots_group = h5f.require_group(plots_path)
plots_group.attrs["NX_class"] = "NXdata"
plots_group.attrs["default"] = "plot_0"
nexus_dict = {
"@NX_class": "NXdata",
"@default": "plot_0",
nexus_dict, h5_file, mode="a", h5path=plots_path, update_mode="replace"
if process.getConfiguration() is not None:
h5_path = "/".join((nx_process_path, "configuration"))
......@@ -563,58 +571,38 @@ def write_xas(
:param str title: experiment title
:param str definition: experiment definition
with HDF5File(h5_file, "w") as h5f:
nx_entry = h5f.require_group("/".join((data_path, entry)))
nx_entry.attrs["NX_class"] = "NXentry"
# store energy
nx_monochromator = nx_entry.require_group("monochromator")
nx_monochromator.attrs["NX_class"] = "NXmonochromator"
if overwrite and "energy" in nx_monochromator:
del nx_monochromator["energy"]
nx_monochromator["energy"] = energy
nx_monochromator["energy"].attrs["interpretation"] = "spectrum"
nx_monochromator["energy"].attrs["NX_class"] = "NXdata"
nx_monochromator["energy"].attrs["unit"] = "eV"
# store absorbed beam
nx_absorbed_beam = nx_entry.require_group("absorbed_beam")
nx_absorbed_beam.attrs["NX_class"] = "NXdetector"
if overwrite and "data" in nx_absorbed_beam:
del nx_absorbed_beam["data"]
nx_absorbed_beam["data"] = mu
nx_absorbed_beam["data"].attrs["interpretation"] = "image"
nx_absorbed_beam["data"].attrs["NX_class"] = "NXdata"
if sample:
nx_sample = nx_entry.require_group("sample")
nx_sample.attrs["NX_class"] = "NXsample"
if overwrite and "name" in nx_sample:
del nx_sample["name"]
nx_sample["name"] =
nx_data = nx_entry.require_group("data")
nx_data.attrs["NX_class"] = "NXdata"
# create some link on data
if overwrite and "energy" in nx_data:
del nx_data["energy"]
nx_data["energy"] = h5py.SoftLink(nx_monochromator["energy"].name)
if overwrite and "absorbed_beam" in nx_data:
del nx_data["absorbed_beam"]
nx_data["absorbed_beam"] = h5py.SoftLink(nx_absorbed_beam["data"].name)
if start_time is not None:
if overwrite and "start_time" in nx_entry:
del nx_entry["start_time"]
nx_entry["start_time"] = start_time
if title is not None:
if overwrite and "title" in nx_entry:
del nx_entry["title"]
nx_entry["title"] = title
if definition is not None:
if overwrite and "definition" in nx_entry:
del nx_entry["definition"]
nx_entry["definition"] = definition
h5path = "/".join((data_path, entry))
nx_dict = {
"@NX_class": "NXentry",
"monochromator": {
"@NX_class": "NXmonochromator",
"energy": energy,
"energy@interpretation": "spctrum",
"energy@NX_class": "NXdata",
"energy@unit": "eV",
"absorbed_beam": {
"@NX_class": "NXdetector",
"data": mu,
"data@interpretation": "image",
"data@NX_class": "NXdata",
"data": {
"@NX_class": "NXdata",
">energy": "../monochromator/energy",
">absorbed_beam": "../absorbed_beam/data",
"start_time": start_time,
"title": title,
"definition": definition,
if overwrite:
mode = "w"
update_mode = "replace"
mode = "a"
update_mode = "add"
dicttonx(nx_dict, h5_file, h5path=h5path, mode=mode, update_mode=update_mode)
def write_spectrum_saving_pt(h5_file, entry, obj, overwrite=True):
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