Commit a261fd59 authored by payno's avatar payno
Browse files

[io][write_xas] replace writing xas using HDF5File by using dicttonx

parent 91c2e6bd
Pipeline #42094 failed with stages
in 61 minutes and 58 seconds
......@@ -40,6 +40,7 @@ from est.core.types import Spectra
from import read_spectrum as read_spec_spectrum
from est.thirdparty.silx.hdf5file import get_data, HDF5File
import json
from import read_ascii as larch_read_ascii
except ImportError:
......@@ -538,58 +539,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"
existing = "overwrite"
mode = "a"
existing = "add"
dicttonx(nx_dict, h5_file, h5path=h5path, mode=mode, existing=existing)
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