# Copyright (c) 2016 Weitian LI # 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 pkg_resources from configobj import ConfigObj, ConfigObjError, flatten_errors from validate import Validator 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 only presents 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 configs. """ # 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 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._config.merge(newconfig) logger.info("Loaded additional config: {0}".format(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 ---- The user configuration file can be loaded *only once*, i.e., *only one* user configuration file supported. Since the *path* of the user configuration file is recorded, and thus allow the use of *relative path* of some input files (e.g., "galactic/synchrotron/template") within the configurations. """ if self.userconfig is not None: raise ConfigError('User configuration already loaded from "%s"' % self.userconfig) # try: config = open(userconfig).read().split("\n") except IOError: raise ConfigError('Cannot read config from "%s"' % userconfig) # 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.message) 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 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, sep="/", from_default=False): """Get the config value from the nested dictionary configs using a list of keys or a "sep"-separated keys strings. Parameters ---------- key : str, or list[str] List of keys or a string separated by a specific character (e.g., "/") to specify the item in the ``self._config``, which is a nested dictionary. e.g., ``["section1", "key2"]``, ``"section1/key2"`` sep : str (len=1), optional If the above "keys" is a string, then this parameter specify the character used to separate the multi-level keys. This parameter should be a string of length 1 (i.e., a character). from_default : bool, optional If True, get the config option value from the *default* configurations, other than the configurations merged with user configurations (default). References ---------- - Stackoverflow: Checking a Dictionary using a dot notation string https://stackoverflow.com/q/12414821/4856091 """ if len(sep) != 1: raise ValueError("Invalid parameter 'sep': %s" % sep) if isinstance(key, str): key = key.split(sep) # if from_default: config = self._config_default else: config = self._config # return reduce(dict.get, key, config) 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()