normalization.py 6.52 KB
Newer Older
payno's avatar
payno committed
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) 2016-2017 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/11/2019"


from PyMca5.PyMcaPhysics.xas.XASClass import XASClass
32
from xas.core.types import Spectrum, XASObject
33
from .process import Process, Progress
34
35
import multiprocessing
import functools
payno's avatar
payno committed
36
37
38
39
import logging
_logger = logging.getLogger(__name__)


40
41
def process_spectr_norm(spectrum, configuration, overwrite=True, callback=None,
                        output=None, output_dict=None):
42
43
44
45
46
    """

    :param :class:`.Spectrum` spectrum: spectrum to process
    :param dict configuration: configuration of the pymca normalization
    :param bool overwrite: False if we want to return a new Spectrum instance
47
48
49
50
51
52
    :param function pointer callback: callback to execute.
    :param output: list to store the result, needed for pool processing
    :type: multiprocessing.manager.list
    :param dict output_dict: key: input spectrum, value: index in the output
                                  list.
    :return: processed spectrum
53
54
    :rtype: tuple (configuration, spectrum)
    """
55
56
57
    if spectrum.energy is None or spectrum.mu is None:
        _logger.error('Energy and or Mu is/are not specified, unable to '
                      'compute exafs')
58
        return None, None
59
60
61
    pymca_xas = XASClass()
    pymca_xas.setSpectrum(energy=spectrum.energy,
                          mu=spectrum.mu)
62
63
    if configuration is not None:
        pymca_xas.setConfiguration(configuration)
64
    configuration = pymca_xas.getConfiguration()
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
    try:
        res = pymca_xas.normalize()
        ddict = spectrum.to_dict()
        ddict.update(res)
        spectrum_ = Spectrum.from_dict(ddict)
    except (IndexError, ValueError) as e:
        _logger.error(e)
        return None, None
    if callback:
        callback()

    def get_output(orignal_spec, res_spec):
        if overwrite:
            orignal_spec.update(res_spec)
            return orignal_spec
        else:
            return res_spec

    if output is not None:
        assert output_dict is not None
        output[output_dict[spectrum]] = get_output(spectrum, spectrum_)

    return configuration, get_output(spectrum, spectrum_)
88
89


90
def pymca_normalization(xas_obj):
payno's avatar
payno committed
91
    """
92

93
94
95
96
    :param xas_obj: object containing the configuration and spectra to process
    :type: Union[XASObject, dict]. If is a dict, should contain configuration or
                                 spectra keys. Otherwise is simply the spectra
    :return: spectra dict
97
    :rtype: dict
payno's avatar
payno committed
98
    """
99
100
    normalization_obj = PyMca_normalization()
    return normalization_obj.process(xas_obj=xas_obj)
101
102


103
104
105
106
_USE_MULTIPROCESSING_POOL = False
# note: we cannot use multiprocessing pool with push workflow for now.


107
class PyMca_normalization(Process):
108
    def __init__(self):
109
        Process.__init__(self, 'normalization')
110
        self._advancement = Progress(self.name)
111
112
113
114
115
116
117
118
119
        self._settings = None

    def setProperties(self, properties):
        if '_pymcaSettings' in properties:
            self._settings = properties['_pymcaSettings']

    def process(self, xas_obj):
        """

120
121
122
        :param xas_obj: object containing the configuration and spectra to process
        :type: Union[XASObject, dict]. If is a dict, should contain configuration or
                                     spectra keys. Otherwise is simply the spectra
123
124
        :return: updated XASObject
        :rtype: :class:`.XASObject`
125
        """
126
        _xas_obj = self.getXasObject(xas_obj)
127
128
129
130
131

        if _xas_obj.energy is None:
            _logger.error('Energy not specified, unable to normalize spectra')
            return

132
133
134
        if self._settings:
            _xas_obj.configuration['Normalization'] = self._settings

135
        self._advancement.reset(max_=_xas_obj.n_spectrum)
136
        self._advancement.startProcess()
137
        self._pool_process(xas_obj=_xas_obj)
138
139
140
141
142
        self._advancement.endProcess()
        return _xas_obj

    def _pool_process(self, xas_obj):
        """process normalization from a pool"""
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
        assert isinstance(xas_obj, XASObject)
        if not _USE_MULTIPROCESSING_POOL:
            for spectrum in xas_obj.spectra:
                process_spectr_norm(spectrum=spectrum,
                                    configuration=xas_obj.configuration,
                                    callback=self._advancement.increaseAdvancement,
                                    overwrite=True)
        else:
            from multiprocessing import Manager
            manager = Manager()
            output_dict = {}
            res_list = manager.list()
            for i_spect, spect in enumerate(xas_obj.spectra):
                res_list.append(None)
                output_dict[spect] = i_spect

            with multiprocessing.Pool(5) as p:
                partial_ = functools.partial(process_spectr_norm,
                                             configuration=xas_obj.configuration,
                                             callback=self._advancement.increaseAdvancement,
                                             overwrite=False,
                                             output=res_list,
                                             output_dict=output_dict)
                p.map(partial_, xas_obj.spectra)

            # then update local spectrum
            for spectrum, res in zip(xas_obj.spectra, res_list):
                spectrum.update(res)

172
173

    __call__ = process