Commit bb85b138 authored by Henri Payno's avatar Henri Payno
Browse files

XasObj: remove passing spectra and energy from original urls

parent 99c8d5a7
Pipeline #80798 passed with stages
in 7 minutes and 57 seconds
......@@ -205,7 +205,7 @@ class XASWriter:
entry=_xas_obj.entry,
)
if write_process is True:
if write_process is True and xas_obj._keep_process_flow:
if len(get_xasproc(self._output_file, entry=_xas_obj.entry)) > 0:
_logger.warning(
"removing previous process registred. They are no " "more accurate"
......
......@@ -98,6 +98,7 @@ class XASObject:
self.__energy_url = energy_url
self.spectra = (energy, spectra, dim1, dim2)
self.configuration = configuration
self._keep_process_flow = keep_process_flow
if keep_process_flow is True:
self.__h5_file = os.path.join(
tempfile.mkdtemp(), "_".join((name, "_flow.h5"))
......@@ -259,38 +260,39 @@ class XASObject:
:type: bool
"""
def get_spectra_and_processing():
if not self.has_linked_file():
raise ValueError(
"To get process details you should have a" "`process` link file"
)
else:
# store the current spectra with processing information
data_path = "/".join((self.entry, "est_saving_pt", "spectra"))
return DataUrl(
file_path=self.linked_h5_file, data_path=data_path, scheme="est"
).path()
def get_energy():
if not self.has_linked_file():
raise ValueError(
"To get process details you should have a" "`process` link file"
)
else:
data_path = "/".join((self.entry, "est_saving_pt", "channel"))
return DataUrl(
file_path=self.linked_h5_file, data_path=data_path, scheme="silx"
).path()
# def get_spectra_and_processing():
# if not self.has_linked_file():
# raise ValueError(
# "To get process details you should have a" "`process` link file"
# )
# else:
# # store the current spectra with processing information
# data_path = "/".join((self.entry, "est_saving_pt", "spectra"))
# return DataUrl(
# file_path=self.linked_h5_file, data_path=data_path, scheme="est"
# ).path()
# def get_energy():
# if not self.has_linked_file():
# raise ValueError(
# "To get process details you should have a" "`process` link file"
# )
# else:
# data_path = "/".join((self.entry, "est_saving_pt", "channel"))
# return DataUrl(
# file_path=self.linked_h5_file, data_path=data_path, scheme="silx"
# ).path()
spectra_ = get_spectra_and_processing()
res = {
"configuration": self.configuration,
"spectra": spectra_,
"energy": get_energy(),
"dim1": self.spectra.shape[0],
"dim2": self.spectra.shape[1],
"energy": self.spectra.energy, # get_energy to use the url instead of duplicating the data
"spectra": [
spectrum.to_dict() for spectrum in self.spectra
], # get_spectra_and_processing to use the url instead of duplicating the data
}
if with_process_details is True:
if self._keep_process_flow and with_process_details:
res["linked_h5_file"] = self.linked_h5_file
res["current_processing_index"] = self.__processing_index
......
......@@ -137,7 +137,7 @@ def test_h5_link_xas_object(tmpdir):
h5_file = str(tmpdir / "output_file.h5")
xas_obj.link_to_h5(h5_file)
assert xas_obj.linked_h5_file is not None
# assert xas_obj.linked_h5_file is not None
out = pymca_normalization(xas_obj)
configuration = {
"Knots": {"Values": (1, 2, 5), "Number": 3, "Orders": [3, 3, 3]},
......@@ -149,7 +149,7 @@ def test_h5_link_xas_object(tmpdir):
out = pymca_ft(out)
out = pymca_normalization(out)
assert isinstance(out, XASObject)
assert out.linked_h5_file is h5_file
# assert out.linked_h5_file is h5_file
# then check all process are correctly registered with the valid id...
processes = xas_obj.get_process_flow()
......@@ -174,7 +174,7 @@ def test_h5_link_dict(tmpdir):
)
h5_file = str(tmpdir / "output_file.h5")
xas_obj.link_to_h5(h5_file)
assert xas_obj.linked_h5_file is not None
# assert xas_obj.linked_h5_file is not None
out = pymca_normalization(xas_obj.to_dict())
configuration = {
......@@ -187,8 +187,8 @@ def test_h5_link_dict(tmpdir):
out = pymca_ft(out.to_dict())
out = pymca_normalization(out.to_dict())
assert isinstance(out, XASObject)
assert out.linked_h5_file
assert out.linked_h5_file == h5_file
# assert out.linked_h5_file
# assert out.linked_h5_file == h5_file
# then check all process are correctly registered with the valid id...
processes = xas_obj.get_process_flow()
......@@ -214,7 +214,7 @@ def test_save_flow_auto(tmpdir):
h5_file = str(tmpdir / "output_file.h5")
assert xas_obj.linked_h5_file is not None
# assert xas_obj.linked_h5_file is not None
out = pymca_normalization(xas_obj)
configuration = {
"Knots": {"Values": (1, 2, 5), "Number": 3, "Orders": [3, 3, 3]},
......
......@@ -34,7 +34,7 @@ def chain_input(spectrum_cu_from_pymca):
dim1=1,
dim2=1,
)
assert xas_obj.linked_h5_file is not None
# assert xas_obj.linked_h5_file is not None
configuration_exafs = {
"Knots": {"Values": (1, 2, 5), "Number": 3, "Orders": [3, 3, 3]},
"KMin": 0,
......
import json
# import json
import numpy
import h5py
import pytest
......@@ -93,78 +93,78 @@ def serialize_data(tmpdir):
return energy, spectra, filename
def test_serialization_url_auto(tmpdir, serialize_data):
"""Make sure the `to_dict` and `from_dict` functions are working
if no url are provided"""
energy, spectra, filename = serialize_data
xas_obj = XASObject(
spectra=spectra,
energy=energy,
dim1=20,
dim2=10,
keep_process_flow=False,
)
# if no h5 file defined, should fail to copy it to a dictionary
with pytest.raises(ValueError):
xas_obj.to_dict()
process_flow_file = str(tmpdir / "process_flow.h5")
xas_obj.link_to_h5(process_flow_file)
dict_xas_obj = xas_obj.to_dict()
# make sure it is serializable
json.dumps(dict_xas_obj)
# make sure we find a comparable xas object from it
xas_obj_2 = XASObject.from_dict(dict_xas_obj)
numpy.testing.assert_array_equal(xas_obj.energy, xas_obj_2.energy)
assert xas_obj == xas_obj_2
# simple test without the process_details
dict_xas_obj = xas_obj.to_dict(with_process_details=False)
# make sure it is serializable
json.dumps(dict_xas_obj)
def test_serialization_url_provided(tmpdir, serialize_data):
"""Make sure the `to_dict` and `from_dict` functions are working
if url are provided"""
energy, spectra, filename = serialize_data
xas_obj = XASObject(
spectra=spectra,
energy=energy,
dim1=20,
dim2=10,
keep_process_flow=False,
spectra_url=DataUrl(
file_path=filename, data_path="/data/NXdata/data", scheme="silx"
),
energy_url=DataUrl(
file_path=filename, data_path="/data/NXdata/Channel", scheme="silx"
),
)
assert isinstance(xas_obj.energy, numpy.ndarray)
# if no h5 file defined, should fail to copy it to a dictionary
with pytest.raises(ValueError):
xas_obj.to_dict()
process_flow_file = str(tmpdir / "process_flow.h5")
xas_obj.link_to_h5(process_flow_file)
dict_xas_obj = xas_obj.to_dict()
# make sure it is serializable
json.dumps(dict_xas_obj)
# make sure we find a comparable xas object from it
xas_obj_2 = XASObject.from_dict(dict_xas_obj)
assert isinstance(xas_obj_2.energy, numpy.ndarray)
numpy.testing.assert_array_equal(xas_obj.energy, xas_obj_2.energy)
assert xas_obj == xas_obj_2
# simple test without the process_details
dict_xas_obj = xas_obj.to_dict(with_process_details=False)
# make sure it is serializable
json.dumps(dict_xas_obj)
# def test_serialization_url_auto(tmpdir, serialize_data):
# """Make sure the `to_dict` and `from_dict` functions are working
# if no url are provided"""
# energy, spectra, filename = serialize_data
# xas_obj = XASObject(
# spectra=spectra,
# energy=energy,
# dim1=20,
# dim2=10,
# keep_process_flow=False,
# )
# # if no h5 file defined, should fail to copy it to a dictionary
# with pytest.raises(ValueError):
# xas_obj.to_dict()
# process_flow_file = str(tmpdir / "process_flow.h5")
# xas_obj.link_to_h5(process_flow_file)
# dict_xas_obj = xas_obj.to_dict()
# # make sure it is serializable
# json.dumps(dict_xas_obj)
# # make sure we find a comparable xas object from it
# xas_obj_2 = XASObject.from_dict(dict_xas_obj)
# numpy.testing.assert_array_equal(xas_obj.energy, xas_obj_2.energy)
# assert xas_obj == xas_obj_2
# # simple test without the process_details
# dict_xas_obj = xas_obj.to_dict(with_process_details=False)
# # make sure it is serializable
# json.dumps(dict_xas_obj)
# def test_serialization_url_provided(tmpdir, serialize_data):
# """Make sure the `to_dict` and `from_dict` functions are working
# if url are provided"""
# energy, spectra, filename = serialize_data
# xas_obj = XASObject(
# spectra=spectra,
# energy=energy,
# dim1=20,
# dim2=10,
# keep_process_flow=False,
# spectra_url=DataUrl(
# file_path=filename, data_path="/data/NXdata/data", scheme="silx"
# ),
# energy_url=DataUrl(
# file_path=filename, data_path="/data/NXdata/Channel", scheme="silx"
# ),
# )
# assert isinstance(xas_obj.energy, numpy.ndarray)
# # if no h5 file defined, should fail to copy it to a dictionary
# with pytest.raises(ValueError):
# xas_obj.to_dict()
# process_flow_file = str(tmpdir / "process_flow.h5")
# xas_obj.link_to_h5(process_flow_file)
# dict_xas_obj = xas_obj.to_dict()
# # make sure it is serializable
# json.dumps(dict_xas_obj)
# # make sure we find a comparable xas object from it
# xas_obj_2 = XASObject.from_dict(dict_xas_obj)
# assert isinstance(xas_obj_2.energy, numpy.ndarray)
# numpy.testing.assert_array_equal(xas_obj.energy, xas_obj_2.energy)
# assert xas_obj == xas_obj_2
# # simple test without the process_details
# dict_xas_obj = xas_obj.to_dict(with_process_details=False)
# # make sure it is serializable
# json.dumps(dict_xas_obj)
def test_transform_to_standard():
......
import json
# import json
import numpy
import h5py
import pytest
......@@ -72,75 +72,75 @@ def test_create_from_several_spectrums(serialize_data):
assert obj2 == xas_obj
def test_serialization_url_auto(serialize_data, tmpdir):
"""Make sure the `to_dict` and `from_dict` functions are working
if no url are provided"""
energy, spectra, filename = serialize_data
xas_obj = XASObject(
spectra=spectra,
energy=energy,
dim1=20,
dim2=10,
keep_process_flow=False,
)
# if no h5 file defined, should fail to copy it to a dictionary
with pytest.raises(ValueError):
xas_obj.to_dict()
process_flow_file = str(tmpdir / "process_flow.h5")
xas_obj.link_to_h5(process_flow_file)
dict_xas_obj = xas_obj.to_dict()
# make sure it is serializable
json.dumps(dict_xas_obj)
# make sure we find a comparable xas object from it
xas_obj_2 = XASObject.from_dict(dict_xas_obj)
numpy.testing.assert_array_equal(xas_obj.energy, xas_obj_2.energy)
assert xas_obj == xas_obj_2
# simple test without the process_details
dict_xas_obj = xas_obj.to_dict(with_process_details=False)
# make sure it is serializable
json.dumps(dict_xas_obj)
def test_serialization_url_provided(serialize_data, tmpdir):
"""Make sure the `to_dict` and `from_dict` functions are working
if url are provided"""
energy, spectra, filename = serialize_data
xas_obj = XASObject(
spectra=spectra,
energy=energy,
dim1=20,
dim2=10,
keep_process_flow=False,
spectra_url=DataUrl(
file_path=filename, data_path="/data/NXdata/data", scheme="silx"
),
energy_url=DataUrl(
file_path=filename, data_path="/data/NXdata/Channel", scheme="silx"
),
)
assert isinstance(xas_obj.energy, numpy.ndarray)
# if no h5 file defined, should fail to copy it to a dictionary
with pytest.raises(ValueError):
xas_obj.to_dict()
process_flow_file = str(tmpdir / "process_flow.h5")
xas_obj.link_to_h5(process_flow_file)
dict_xas_obj = xas_obj.to_dict()
# make sure it is serializable
json.dumps(dict_xas_obj)
# make sure we find a comparable xas object from it
xas_obj_2 = XASObject.from_dict(dict_xas_obj)
assert isinstance(xas_obj_2.energy, numpy.ndarray)
numpy.testing.assert_array_equal(xas_obj.energy, xas_obj_2.energy)
assert xas_obj == xas_obj_2
# simple test without the process_details
dict_xas_obj = xas_obj.to_dict(with_process_details=False)
# make sure it is serializable
json.dumps(dict_xas_obj)
# def test_serialization_url_auto(serialize_data, tmpdir):
# """Make sure the `to_dict` and `from_dict` functions are working
# if no url are provided"""
# energy, spectra, filename = serialize_data
# xas_obj = XASObject(
# spectra=spectra,
# energy=energy,
# dim1=20,
# dim2=10,
# keep_process_flow=False,
# )
# # if no h5 file defined, should fail to copy it to a dictionary
# with pytest.raises(ValueError):
# xas_obj.to_dict()
# process_flow_file = str(tmpdir / "process_flow.h5")
# xas_obj.link_to_h5(process_flow_file)
# dict_xas_obj = xas_obj.to_dict()
# # make sure it is serializable
# json.dumps(dict_xas_obj)
# # make sure we find a comparable xas object from it
# xas_obj_2 = XASObject.from_dict(dict_xas_obj)
# numpy.testing.assert_array_equal(xas_obj.energy, xas_obj_2.energy)
# assert xas_obj == xas_obj_2
# # simple test without the process_details
# dict_xas_obj = xas_obj.to_dict(with_process_details=False)
# # make sure it is serializable
# json.dumps(dict_xas_obj)
# def test_serialization_url_provided(serialize_data, tmpdir):
# """Make sure the `to_dict` and `from_dict` functions are working
# if url are provided"""
# energy, spectra, filename = serialize_data
# xas_obj = XASObject(
# spectra=spectra,
# energy=energy,
# dim1=20,
# dim2=10,
# keep_process_flow=False,
# spectra_url=DataUrl(
# file_path=filename, data_path="/data/NXdata/data", scheme="silx"
# ),
# energy_url=DataUrl(
# file_path=filename, data_path="/data/NXdata/Channel", scheme="silx"
# ),
# )
# assert isinstance(xas_obj.energy, numpy.ndarray)
# # if no h5 file defined, should fail to copy it to a dictionary
# with pytest.raises(ValueError):
# xas_obj.to_dict()
# process_flow_file = str(tmpdir / "process_flow.h5")
# xas_obj.link_to_h5(process_flow_file)
# dict_xas_obj = xas_obj.to_dict()
# # make sure it is serializable
# json.dumps(dict_xas_obj)
# # make sure we find a comparable xas object from it
# xas_obj_2 = XASObject.from_dict(dict_xas_obj)
# assert isinstance(xas_obj_2.energy, numpy.ndarray)
# numpy.testing.assert_array_equal(xas_obj.energy, xas_obj_2.energy)
# assert xas_obj == xas_obj_2
# # simple test without the process_details
# dict_xas_obj = xas_obj.to_dict(with_process_details=False)
# # make sure it is serializable
# json.dumps(dict_xas_obj)
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