File: confighandler.py

package info (click to toggle)
python-nxtomomill 1.1.0-4
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 2,564 kB
  • sloc: python: 15,970; makefile: 13; sh: 3
file content (302 lines) | stat: -rw-r--r-- 10,909 bytes parent folder | download
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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
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
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
# coding: utf-8

"""
contains the HDF5ConfigHandler
"""

from __future__ import annotations

import logging
import os

from nxtomomill import utils
from nxtomomill.io.utils import filter_str_def
from nxtomomill.converter.hdf5.utils import get_default_output_file

from .hdf5config import TomoHDF5Config

_logger = logging.getLogger(__name__)

__all__ = [
    "SETTABLE_PARAMETERS_UNITS",
    "SETTABLE_PARAMETERS_TYPE",
    "SETTABLE_PARAMETERS",
    "BaseHDF5ConfigHandler",
    "TomoHDF5ConfigHandler",
]

SETTABLE_PARAMETERS_UNITS = {
    "energy": "kev",
    "x_pixel_size": "m",
    "y_pixel_size": "m",
    "detector_sample_distance": "m",
}

SETTABLE_PARAMETERS_TYPE = {
    "energy": float,
    "x_pixel_size": float,
    "y_pixel_size": float,
    "detector_sample_distance": float,
}

SETTABLE_PARAMETERS = SETTABLE_PARAMETERS_UNITS.keys()


def _extract_param_value(key_values):
    """extract all the key / values elements from the str_list. Expected
    format is `param_1_name param_1_value param_2_name param_2_value ...`

    :param str_list: raw input string as `param_1_name param_1_value
                         param_2_name param_2_value ...`
    :return: dict of tuple (param_name, param_value)
    """
    if len(key_values) % 2 != 0:
        raise ValueError(
            "Expect a pair `param_name, param_value` for each " "parameters"
        )

    def pairwise(it):
        it = iter(it)
        while True:
            try:
                yield next(it), next(it)
            except StopIteration:
                # no more elements in the iterator
                return

    res = {}
    for name, value in pairwise(key_values):
        if name not in SETTABLE_PARAMETERS:
            raise ValueError(f"parameters {name} is not managed")
        if name in SETTABLE_PARAMETERS_TYPE:
            type_ = SETTABLE_PARAMETERS_TYPE[name]
            if type_ is not None:
                res[name] = type_(value)
                continue
        res[name] = value
    return res


class BaseHDF5ConfigHandler:
    """
    Class taking as input the CLI options and the user configuration file.
    Raises error if there is incoherence between the two (like if the CLI provides image_keys values AND if the configuration file provides the same input).

    This class will produce the instance of TomoHDF5Config to be used for the conversion
    """

    @staticmethod
    def get_tuple_of_keys_from_cmd(cmd_value):
        return utils.get_tuple_of_keys_from_cmd(cmd_value)

    @staticmethod
    def conv_str_to_bool(bstr):
        return bstr in ("True", True)

    @staticmethod
    def conv_log_level(bool_debug):
        if bool_debug is True:
            return "debug"
        else:
            return "warning"

    def __init__(self, argparse_options, raise_error=True):
        self._argparse_options = argparse_options
        self._config = None
        self.build_configuration(raise_error=raise_error)

    @property
    def configuration(self) -> TomoHDF5Config | None:
        return self._config

    @property
    def argparse_options(self):
        return self._argparse_options

    def _check_argparse_options(self, raise_error):
        raise NotImplementedError("BaseClass")

    def _create_HDF5_config(self):
        raise NotImplementedError("Base class")

    def build_configuration(self, raise_error) -> bool:
        """
        :param raise_error: raise error if encounter some errors. Else
                                 display a log message
        :return: True if the settings are valid
        """
        self._check_argparse_options(raise_error=raise_error)
        options = self.argparse_options
        config = self._create_HDF5_config()

        # check input and output file
        if config.input_file is None:
            config.input_file = options.input_file
        elif options.input_file is not None and config.input_file != options.input_file:
            raise ValueError(
                "Two different input files provided from "
                "command line and from the configuration file"
            )
        if config.input_file is None:
            err = "No input file provided"
            if raise_error:
                raise ValueError(err)
            else:
                _logger.error(err)

        if config.output_file is None:
            config.output_file = options.output_file
        elif (
            options.output_file is not None
            and config.output_file != options.output_file
        ):
            raise ValueError(
                "Two different output files provided from "
                "command line and from the configuration file"
            )
        if config.output_file is None:
            if config.file_extension is None:
                err = "If no output file provided you should provide the extension"
                raise ValueError(err)

            config.output_file = get_default_output_file(
                input_file=config.input_file,
                output_file_extension=config.file_extension.value,
            )
        elif os.path.isdir(config.output_file):
            # if the output file is only a directory: consider we want the same file basename with default extension
            # in this directory
            input_file, _ = os.path.splitext(config.input_file)
            config.output_file = os.path.join(
                os.path.abspath(config.output_file),
                os.path.basename(input_file) + config.file_extension.value,
            )

        # set parameter from the arg parse options
        # key is the name of the argparse option.
        # value is a tuple: (name of the setter in the HDF5Config,
        # function to format the input)
        self._config = self._map_option_to_config_param(config, options)

    def _map_option_to_config_param(self, config, options):
        raise NotImplementedError("Base class")

    def __str__(self):
        raise NotImplementedError("")


class TomoHDF5ConfigHandler(BaseHDF5ConfigHandler):
    def _create_HDF5_config(self):
        if self.argparse_options.config:
            return TomoHDF5Config.from_cfg_file(self.argparse_options.config)
        else:
            return TomoHDF5Config()

    def _map_option_to_config_param(self, config, options):
        mapping = {
            "valid_camera_names": (
                "valid_camera_names",
                self.get_tuple_of_keys_from_cmd,
            ),
            "overwrite": ("overwrite", self.conv_str_to_bool),
            "file_extension": ("file_extension", filter_str_def),
            "single_file": ("single_file", self.conv_str_to_bool),
            "debug": ("log_level", self.conv_log_level),
            "entries": ("entries", self.get_tuple_of_keys_from_cmd),
            "ignore_sub_entries": (
                "sub_entries_to_ignore",
                self.get_tuple_of_keys_from_cmd,
            ),
            "duplicate_data": ("default_copy_behavior", self.conv_str_to_bool),
            "raises_error": ("raises_error", self.conv_str_to_bool),
            "field_of_view": ("field_of_view", filter_str_def),
            "request_input": ("request_input", self.conv_str_to_bool),
            "sample_x_keys": ("sample_x_keys", self.get_tuple_of_keys_from_cmd),
            "sample_y_keys": ("sample_y_keys", self.get_tuple_of_keys_from_cmd),
            "translation_z_keys": (
                "translation_z_keys",
                self.get_tuple_of_keys_from_cmd,
            ),
            "rot_angle_keys": ("rotation_angle_keys", self.get_tuple_of_keys_from_cmd),
            "sample_detector_distance_paths": (
                "sample_detector_distance_paths",
                self.get_tuple_of_keys_from_cmd,
            ),
            "acq_expo_time_keys": (
                "exposition_time_keys",
                self.get_tuple_of_keys_from_cmd,
            ),
            "x_pixel_size_key": ("x_pixel_size_paths", self.get_tuple_of_keys_from_cmd),
            "y_pixel_size_key": ("y_pixel_size_paths", self.get_tuple_of_keys_from_cmd),
            "init_titles": ("init_titles", self.get_tuple_of_keys_from_cmd),
            "init_zserie_titles": (
                "zserie_init_titles",
                self.get_tuple_of_keys_from_cmd,
            ),
            "init_pcotomo_titles": (
                "pcotomo_init_titles",
                self.get_tuple_of_keys_from_cmd,
            ),
            "dark_titles": ("dark_titles", self.get_tuple_of_keys_from_cmd),
            "flat_titles": ("flat_titles", self.get_tuple_of_keys_from_cmd),
            "proj_titles": ("projections_titles", self.get_tuple_of_keys_from_cmd),
            "align_titles": ("alignment_titles", self.get_tuple_of_keys_from_cmd),
            "set_params": ("param_already_defined", _extract_param_value),
        }
        for argparse_name, (config_name, format_fct) in mapping.items():
            argparse_value = getattr(options, argparse_name)
            if argparse_value is not None:
                value = format_fct(argparse_value)  # pylint: disable=E1102
                setattr(config, config_name, value)
        return config

    def _check_argparse_options(self, raise_error):
        if self.argparse_options is None:
            err = "No argparse options provided"
            if raise_error:
                raise ValueError(err)
            else:
                _logger.error(err)
            return False

        options = self.argparse_options
        if options.config is not None:
            # check no other option are provided
            duplicated_inputs = []
            for opt in (
                "set_params",
                "align_titles",
                "proj_titles",
                "flat_titles",
                "dark_titles",
                "init_zserie_titles",
                "init_titles",
                "init_pcotomo_titles",
                "x_pixel_size_key",
                "y_pixel_size_key",
                "acq_expo_time_keys",
                "rot_angle_keys",
                "valid_camera_names",
                "translation_z_keys",
                "sample_y_keys",
                "sample_x_keys",
                "request_input",
                "raises_error",
                "ignore_sub_entries",
                "entries",
                "debug",
                "overwrite",
                "single_file",
                "file_extension",
                "field_of_view",
                "duplicate_data",
            ):
                if getattr(options, opt):
                    duplicated_inputs.append(opt)
            if len(duplicated_inputs) > 0:
                err = f"You provided a configuration file and inputs for {duplicated_inputs}"
                if raise_error:
                    raise ValueError(err)
                else:
                    _logger.error(err)
                return False