test_types.py 8.96 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
# coding: utf-8
# /*##########################################################################
#
# Copyright (c) 2017-2019 European Synchrotron Radiation Facility
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
#
# ###########################################################################*/
__authors__ = ["H. Payno"]
__license__ = "MIT"
__date__ = "06/26/2019"


import numpy
import os
32
import unittest
33
34
import tempfile
import h5py
35
import shutil
payno's avatar
PEP8    
payno committed
36
from est.core.types import Spectrum, XASObject
payno's avatar
payno committed
37
38
from est.core.utils import spectra as spectra_utils
from est.core.io import read as read_xas
39
from silx.io.url import DataUrl
payno's avatar
payno committed
40
from est.units import ur
41
import json
42
import silx.io.utils
payno's avatar
fix CI    
payno committed
43
from est.core.types import Dim
payno's avatar
payno committed
44

45
try:
payno's avatar
PEP8    
payno committed
46
    import PyMca5  # noqa
47
48
49
50
51
except ImportError:
    has_pymca = False
else:
    has_pymca = True
    from PyMca5.PyMcaDataDir import PYMCA_DATA_DIR
52
    from est.io.utils.spec import read_spectrum
53
54
55


class TestSpectrum(unittest.TestCase):
56
    """Test the spectrum class"""
payno's avatar
payno committed
57
58

    @unittest.skipIf(has_pymca is False, "PyMca5 is not installed")
59
    def test_from_dat(self):
60
        """check that we can create a Spectrum from a pymca .dat file"""
61
        data_file = os.path.join(PYMCA_DATA_DIR, "EXAFS_Cu.dat")
62
        energy, mu = read_spectrum(
63
64
65
            data_file,
            energy_col_name="Column 1",
            absorption_col_name="Column 2",
66
        )
67
68
69
70
71
        spectrum = Spectrum(energy=energy, mu=mu)
        self.assertTrue(spectrum.energy is not None)
        self.assertTrue(spectrum.mu is not None)

    def test_from_numpy_array(self):
72
        """check that we can create a Spectrum from numpy arrays"""
73
74
75
        energy = numpy.arange(10, 20)
        mu = numpy.arange(10)
        spectrum = Spectrum(energy=energy, mu=mu)
76
        numpy.testing.assert_array_equal(spectrum.energy, (energy * ur.eV).m)
77
78
        numpy.testing.assert_array_equal(spectrum.mu, mu)
        mu_2 = numpy.arange(30, 40)
payno's avatar
payno committed
79
        spectrum["Mu"] = mu_2
80
        numpy.testing.assert_array_equal(spectrum.mu, mu_2)
81
82
83


class TestXASObject(unittest.TestCase):
84
    """test construction of XAS object from a single spectra"""
payno's avatar
payno committed
85
86

    @unittest.skipIf(has_pymca is False, "PyMca5 is not installed")
87
    def test_create_from_single_spectrum(self):
88
        """check that we can create a XASObject from a pymca .dat file"""
89
90
        data_file = os.path.join(PYMCA_DATA_DIR, "EXAFS_Cu.dat")
        spectrum = {}
91
        spectrum["Energy"], spectrum["Mu"] = read_spectrum(
92
93
94
            data_file,
            energy_col_name="Column 1",
            absorption_col_name="Column 2",
95
        )
payno's avatar
payno committed
96
        self.spectrum = Spectrum(energy=spectrum["Energy"], mu=spectrum["Mu"])
97
        self.configuration = {
payno's avatar
payno committed
98
99
            "FT": {"KWeight": 1},
            "EXAFS": {"EXAFSNormalized": numpy.array([1, 2, 3])},
100
        }
payno's avatar
payno committed
101
102
103
104
105
106
107
        obj = XASObject(
            spectra=(self.spectrum,),
            energy=self.spectrum.energy,
            configuration=self.configuration,
            dim1=1,
            dim2=1,
        )
108
        self.assertEqual(obj.n_spectrum, 1)
109
        ddict = obj.to_dict()
payno's avatar
payno committed
110
        obj2 = XASObject.from_dict(ddict)
111
        self.assertEqual(obj2, obj)
112
113
114
        # insure the XASObject is serializable
        # import json
        # json.dumps(obj2.to_dict())
115

116
    def test_create_from_several_spectrums(self):
117
        """check that we can create a XASObject from numpy arrays"""
118
        self.energy, self.spectra = spectra_utils.create_dataset(shape=(256, 20, 10))
119
        self.output_dir = tempfile.mkdtemp()
payno's avatar
payno committed
120
121
122
123
        spectra_path = "/data/NXdata/data"
        channel_path = "/data/NXdata/Channel"
        filename = os.path.join(self.output_dir, "myfile.h5")
        with h5py.File(filename, "a") as f:
124
125
126
            f[spectra_path] = self.spectra
            f[channel_path] = self.energy

payno's avatar
payno committed
127
128
129
130
131
132
        url_spectra = DataUrl(file_path=filename, data_path=spectra_path, scheme="silx")
        self.xas_obj = read_xas(
            spectra_url=url_spectra,
            channel_url=DataUrl(
                file_path=filename, data_path=channel_path, scheme="silx"
            ),
133
            dimensions=(Dim.DIM_2, Dim.DIM_1, Dim.DIM_0),
payno's avatar
payno committed
134
        )
payno's avatar
payno committed
135
136
        self.assertEqual(self.xas_obj.spectra.shape[0], 20)
        self.assertEqual(self.xas_obj.spectra.shape[1], 10)
payno's avatar
payno committed
137
        self.assertEqual(self.xas_obj.n_spectrum, 20 * 10)
138
        ddict = self.xas_obj.to_dict(with_process_details=False)
payno's avatar
payno committed
139
140
141
        original_spectra = silx.io.utils.get_data(
            DataUrl(file_path=filename, data_path=spectra_path, scheme="silx")
        )
payno's avatar
payno committed
142
        obj2 = XASObject.from_dict(ddict)
payno's avatar
payno committed
143
        self.assertEqual(self.xas_obj.n_spectrum, obj2.n_spectrum)
144
        obj2_mu_spectra = obj2.spectra.map_to(data_info="mu")
145

payno's avatar
payno committed
146
        numpy.testing.assert_array_equal(original_spectra, obj2_mu_spectra)
147
148
        self.assertEqual(obj2, self.xas_obj)

149

150
151
class TestXASObjectSerialization(unittest.TestCase):
    def setUp(self) -> None:
payno's avatar
payno committed
152
        self.energy, self.spectra = spectra_utils.create_dataset(shape=(256, 20, 10))
153
        self.output_dir = tempfile.mkdtemp()
payno's avatar
payno committed
154
155
156
157
        self.spectra_path = "/data/NXdata/data"
        self.channel_path = "/data/NXdata/Channel"
        self.filename = os.path.join(self.output_dir, "myfile.h5")
        with h5py.File(self.filename, "a") as f:
158
159
160
            f[self.spectra_path] = self.spectra
            f[self.channel_path] = self.energy

161
        self.dimensions = (Dim.DIM_2, Dim.DIM_1, Dim.DIM_0)
162

payno's avatar
payno committed
163
164
165
166
167
168
169
        self.url_spectra = DataUrl(
            file_path=self.filename, data_path=self.spectra_path, scheme="silx"
        )
        self.url_energy = DataUrl(
            file_path=self.filename, data_path=self.channel_path, scheme="silx"
        )
        self.process_flow_file = os.path.join(self.output_dir, "process_flow.h5")
170
171
172
173
174
175
176

    def tearDown(self) -> None:
        shutil.rmtree(self.output_dir)

    def test_serialization_url_auto(self):
        """Make sure the `to_dict` and `from_dict` functions are working
        if no url are provided"""
payno's avatar
payno committed
177
178
179
180
181
182
183
        xas_obj = XASObject(
            spectra=self.spectra,
            energy=self.energy,
            dim1=20,
            dim2=10,
            keep_process_flow=False,
        )
184
185
186
187
188
189
190
191
192
193
194
195
        # if no h5 file defined, should fail to copy it to a dictionary
        with self.assertRaises(ValueError):
            xas_obj.to_dict()

        xas_obj.link_to_h5(self.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)

payno's avatar
payno committed
196
        numpy.testing.assert_array_equal(xas_obj.energy, xas_obj_2.energy)
197
198
199
200
201
202
203
204
205
206
        self.assertEqual(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(self):
        """Make sure the `to_dict` and `from_dict` functions are working
        if url are provided"""
payno's avatar
payno committed
207
208
209
210
211
212
213
214
215
        xas_obj = XASObject(
            spectra=self.spectra,
            energy=self.energy,
            dim1=20,
            dim2=10,
            keep_process_flow=False,
            energy_url=self.url_energy,
            spectra_url=self.spectra_path,
        )
payno's avatar
payno committed
216
        self.assertTrue(isinstance(xas_obj.energy, numpy.ndarray))
217
218
219
220
221
222
223
224
225
226
227
        # if no h5 file defined, should fail to copy it to a dictionary
        with self.assertRaises(ValueError):
            xas_obj.to_dict()

        xas_obj.link_to_h5(self.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)
payno's avatar
payno committed
228
        self.assertTrue(isinstance(xas_obj_2.energy, numpy.ndarray))
229

payno's avatar
payno committed
230
        numpy.testing.assert_array_equal(xas_obj.energy, xas_obj_2.energy)
231
232
233
234
235
236
        self.assertEqual(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)