aboutsummaryrefslogtreecommitdiffstats
path: root/fg21sim/configs/manager.py
blob: 0809126935b679edae9758421d65c993d59996cd (plain)
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
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
# Copyright (c) 2016 Weitian LI <liweitianux@live.com>
# MIT license
#
# References:
# [1] https://configobj.readthedocs.io/en/latest/configobj.html
# [2] https://github.com/pazz/alot/blob/master/alot/settings/manager.py

"""
Configuration manager.
"""

import os
import sys
import logging
from logging import FileHandler, StreamHandler
from functools import reduce
import operator
import pkg_resources

from configobj import ConfigObj, ConfigObjError, flatten_errors
from validate import Validator

from .checkers import check_configs
from ..errors import ConfigError


logger = logging.getLogger(__name__)


def _get_configspec():
    """Found and read all the configuration specifications"""
    files = sorted(pkg_resources.resource_listdir(__name__, ""))
    # NOTE:
    # Explicit convert the filter results to a list, since the returned
    # iterator can ONLY be used ONCE.
    specfiles = list(filter(lambda fn: fn.endswith(".conf.spec"), files))
    if os.environ.get("DEBUG_FG21SIM"):
        print("DEBUG: Found config specifications: %s" % ", ".join(specfiles),
              file=sys.stderr)
    # NOTE:
    # `resource_string()` returns the resource in *binary/bytes* string
    configspec = "\n".join([
        pkg_resources.resource_string(__name__, fn).decode("utf-8")
        for fn in specfiles
    ]).split("\n")
    return configspec


class ConfigManager:
    """Manage the default configurations with specifications, as well as
    the user configurations.

    Both the default configurations and user configurations are validated
    against the bundled specifications.

    Parameters
    ----------
    userconfig: str, optional
        Filename/path to the user configuration file.
        If provided, the user configurations is also loaded, validated, and
        merged into the configurations data.
        The user configuration can also be later loaded by
        ``self.read_userconfig()``.

    Attributes
    ----------
    _config : `~configobj.ConfigObj`
        The current effective configurations.
    _configspec : `~configobj.ConfigObj`
        The configuration specifications bundled with this package.
    userconfig : str
        The filename and path to the user-provided configurations.
        NOTE:
        - This attribute has valid value only after loading the user
          configuration by ``self.read_userconfig()``;
        - This attribute is used to determine the absolute path of the
          configs specifying the input templates or data files, therefore
          allow the use of relative path for those config options.
    """
    # Path to the user provided configuration file, which indicates user
    # configurations merged if not ``None``.
    userconfig = None

    def __init__(self, userconfig=None):
        """Load the bundled default configurations and specifications.
        If the ``userconfig`` provided, the user configurations is also
        loaded, validated, and merged.
        """
        configspec = _get_configspec()
        self._configspec = ConfigObj(configspec, interpolation=False,
                                     list_values=False, _inspec=True)
        configs_default = ConfigObj(interpolation=False,
                                    configspec=self._configspec)
        # Keep a copy of the default configurations
        self._config_default = self._validate(configs_default)
        # NOTE: `_config_default.copy()` only returns a *shallow* copy.
        self._config = ConfigObj(self._config_default, interpolation=False)
        if userconfig:
            self.read_userconfig(userconfig)

    def merge(self, config):
        """Simply merge the given configurations without validation.

        Parameters
        ----------
        config : `~ConfigObj`, dict, str, or list[str]
            Supplied configurations to be merged.
        """
        if not isinstance(config, ConfigObj):
            try:
                config = ConfigObj(config, interpolation=False)
            except ConfigObjError as e:
                logger.exception(e)
                raise ConfigError(e)
        self._config.merge(config)

    def read_config(self, config):
        """Read, validate and merge the input config.

        Parameters
        ----------
        config : str, or list[str]
            Input config to be validated and merged.
            This parameter can be the filename of the config file, or a list
            contains the lines of the configs.
        """
        try:
            newconfig = ConfigObj(config, interpolation=False,
                                  configspec=self._configspec)
        except ConfigObjError as e:
            raise ConfigError(e)
        newconfig = self._validate(newconfig)
        self.merge(newconfig)
        logger.info("Loaded additional config")

    def read_userconfig(self, userconfig):
        """Read user configuration file, validate, and merge into the
        default configurations.

        Parameters
        ----------
        userconfig : str
            Filename/path to the user configuration file.

        NOTE
        ----
        If a user configuration file is already loaded, then the
        configurations are *reset* before loading the supplied user
        configuration file.
        """
        try:
            config = open(userconfig).read().split("\n")
        except IOError:
            raise ConfigError('Cannot read config from "%s"' % userconfig)
        #
        if self.userconfig is not None:
            logger.warning('User configuration already loaded from "%s"' %
                           self.userconfig)
            self.reset()
        self.read_config(config)
        self.userconfig = os.path.abspath(userconfig)
        logger.info("Loaded user config: {0}".format(self.userconfig))

    def reset(self):
        """Reset the current configurations to the copy of defaults from
        the specifications.

        NOTE: Also reset ``self.userconfig`` to ``None``.
        """
        # NOTE: `_config_default.copy()` only returns a *shallow* copy.
        self._config = ConfigObj(self._config_default, interpolation=False)
        self.userconfig = None
        logger.warning("Reset the configurations to the copy of defaults!")

    def _validate(self, config):
        """Validate the config against the specification using a default
        validator.  The validated config values are returned if success,
        otherwise, the ``ConfigError`` raised with details.
        """
        validator = Validator()
        try:
            results = config.validate(validator, preserve_errors=True)
        except ConfigObjError as e:
            raise ConfigError(e)
        if results is not True:
            error_msg = ""
            for (section_list, key, res) in flatten_errors(config, results):
                if key is not None:
                    if res is False:
                        msg = 'key "%s" in section "%s" is missing.'
                        msg = msg % (key, ", ".join(section_list))
                    else:
                        msg = 'key "%s" in section "%s" failed validation: %s'
                        msg = msg % (key, ", ".join(section_list), res)
                else:
                    msg = 'section "%s" is missing' % ".".join(section_list)
                error_msg += msg + "\n"
            raise ConfigError(error_msg)
        return config

    def check_all(self, raise_exception=True):
        """Further check the whole configurations through a set of custom
        checker functions, which may check one config option against its
        context if necessary to determine whether it has a valid value.

        Parameters
        ----------
        raise_exception : bool, optional
            Whether raise a ``ConfigError`` exception if there is any invalid
            config options?

        Returns
        -------
        result : bool
            ``True`` if the configurations pass all checker functions.
        errors : dict
            An dictionary containing the details about the invalid config
            options, with the keys identifying the config options and values
            indicating the error message.
            If above ``result=True``, then this is an empty dictionary ``{}``.

        Raises
        ------
        ConfigError
            If any config option failed to pass any of the checkers, a
            ``ConfigError`` with details is raised.
        """
        result, errors = check_configs(self, raise_exception=raise_exception)
        return (result, errors)

    def get(self, key, fallback=None, from_default=False):
        """Get config value by key."""
        if from_default:
            config = self._config_default
        else:
            config = self._config
        return config.get(key, fallback)

    def getn(self, key, from_default=False):
        """Get the value of a config option specified by the input key from
        from the configurations which is a nested dictionary.

        Parameters
        ----------
        key : str, or list[str]
            List of keys or a string of keys separated by a the ``/``
            character to specify the item in the ``self._config``, which
            is a nested dictionary.
            e.g., ``["section1", "key2"]``, ``"section1/key2"``
        from_default : bool, optional
            If True, get the config option value from the *default*
            configurations, other than the configurations merged with user
            configurations (default).

        Raises
        ------
        KeyError :
            The input key specifies a non-exist config option.

        References
        ----------
        - Stackoverflow: Checking a Dictionary using a dot notation string
          https://stackoverflow.com/q/12414821/4856091
          https://stackoverflow.com/a/12414913/4856091
        """
        if isinstance(key, str):
            key = key.split("/")
        #
        if from_default:
            config = self._config_default
        else:
            config = self._config
        #
        try:
            return reduce(operator.getitem, key, config)
        except (KeyError, TypeError):
            raise KeyError("%s: invalid key" % "/".join(key))

    def setn(self, key, value):
        """Set the value of config option specified by a list of keys or a
        "sep"-separated keys string.

        The supplied key-value config pair is first used to create a
        temporary ``ConfigObj`` instance, which is then validated against
        the configuration specifications.
        If validated to be *valid*, the input key-value pair is then *merged*
        into the configurations, otherwise, a ``ConfigError`` raised.

        NOTE/XXX
        --------
        Given a ``ConfigObj`` instance with an option that does NOT exist in
        the specifications, it will simply *pass* the validation against the
        specifications.
        There seems no way to prevent the ``Validator`` from accepting the
        config options that does NOT exist in the specification.
        Therefore, try to get the option value specified by the input key
        first, if no ``KeyError`` raised, then it is a valid key.

        Parameters
        ----------
        key : str, or list[str]
            List of keys or a string of keys separated by a the ``/``
            character to specify the item in the ``self._config``, which
            is a nested dictionary.
            e.g., ``["section1", "key2"]``, ``"section1/key2"``
        value : str, bool, int, float, list
            The value can be any acceptable type to ``ConfigObj``.

        Raises
        ------
        KeyError :
            The input key specifies a non-exist config option.
        ConfigError :
            The value fails to pass the validation against specifications.
        """
        try:
            val_old = self.getn(key)
        except KeyError as e:
            raise KeyError(e)
        if val_old == value:
            # No need to set this option value
            return
        # Create a nested dictionary from the input key-value pair
        # Credit:
        # * Stackoverflow: Convert a list into a nested dictionary
        #   https://stackoverflow.com/a/6689604/4856091
        if isinstance(key, str):
            key = key.split("/")
        d = reduce(lambda x, y: {y: x}, reversed(key), value)
        # Create the temporary ``ConfigObj`` instance and validate it
        config_new = ConfigObj(d, interpolation=False,
                               configspec=self._configspec)
        config_new = self._validate(config_new)
        # NOTE:
        # The validated ``config_new`` is populated with all other options
        # from the specifications.
        val_new = reduce(operator.getitem, key, config_new)
        d2 = reduce(lambda x, y: {y: x}, reversed(key), val_new)
        self.merge(d2)
        logger.info("Set config: {key}: {val_new} <= {val_old}".format(
            key="/".join(key), val_new=val_new, val_old=val_old))

    def get_path(self, key):
        """Return the absolute path of the file/directory specified by the
        config keys.

        Parameters
        ----------
        key : str
            "/"-separated string specifying the config name of the
            file/directory

        Returns
        -------
        path : str
            The absolute path (if user configuration loaded) or relative
            path specified by the input key, or ``None`` if specified
            config is ``None``.

        Raises
        ------
        ValueError:
            If the value of the specified config is not string.

        NOTE
        ----
        - The "~" (tilde) inside path is expanded to the user home directory.
        - The relative path (with respect to the user configuration file)
          is converted to absolute path if `self.userconfig` presents.
        """
        value = self.getn(key)
        if value is None:
            logger.warning("Specified config '%s' is None or not exist" % key)
            return None
        if not isinstance(value, str):
            msg = "Specified config '%s' is non-string: %s" % (key, value)
            logger.error(msg)
            raise ValueError(msg)
        #
        path = os.path.expanduser(value)
        if not os.path.isabs(path):
            # Got relative path, try to convert to the absolute path
            if hasattr(self, "userconfig"):
                # User configuration loaded
                path = os.path.join(os.path.dirname(self.userconfig), path)
            else:
                logger.warning("Cannot convert to absolute path: %s" % path)
        return os.path.normpath(path)

    @property
    def frequencies(self):
        """Get or calculate if ``frequency/type = custom`` the frequencies
        where to perform the simulations.

        Returns
        -------
        frequencies : list[float]
            List of frequencies where the simulations are requested.
        """
        if self.getn("frequency/type") == "custom":
            # The value is validated to be a float list
            frequencies = self.getn("frequency/frequencies")
        else:
            # Calculate the frequency values
            start = self.getn("frequency/start")
            stop = self.getn("frequency/stop")
            step = self.getn("frequency/step")
            num = int((stop - start) / step + 1)
            frequencies = [start + step*i for i in range(num)]
        return frequencies

    @property
    def logging(self):
        """Get and prepare the logging configurations for
        ``logging.basicConfig()`` to initialize the logging module.

        NOTE
        ----
        ``basicConfig()`` will automatically create a ``Formatter`` with the
        giving ``format`` and ``datefmt`` for each handlers if necessary,
        and then adding the handlers to the "root" logger.
        """
        conf = self.get("logging")
        level = conf["level"]
        if os.environ.get("DEBUG_FG21SIM"):
            print("DEBUG: Force 'DEBUG' logging level", file=sys.stderr)
            level = "DEBUG"
        # logging handlers
        handlers = []
        stream = conf["stream"]
        if stream:
            handlers.append(StreamHandler(getattr(sys, stream)))
        logfile = conf["filename"]
        filemode = conf["filemode"]
        if logfile:
            handlers.append(FileHandler(logfile, mode=filemode))
        #
        logconf = {
            "level": getattr(logging, level),
            "format": conf["format"],
            "datefmt": conf["datefmt"],
            "filemode": filemode,
            "handlers": handlers,
        }
        return logconf

    def dump(self, from_default=False):
        """Dump the configurations as plain Python dictionary.

        Parameters
        ----------
        from_default : bool, optional
            If True, dump the default configurations (as specified by the
            bundled specifications); otherwise, dump the configurations with
            user-supplied options merged (default).

        NOTE
        ----
        The original option orders are missing.
        """
        if from_default:
            config = self._config_default
        else:
            config = self._config
        return config.dict()