cpdbench.utils.BenchConfig

  1"""
  2The global configuration Singleton module.
  3Contains all configurable parameters for the bench and functions
  4to read a given config.yml file.
  5
  6To be used correctly the function load_config(config_file) has to be called first.
  7After this the other functions can be used.
  8"""
  9
 10import yaml
 11import logging
 12
 13from cpdbench.exception.ConfigurationException import ConfigurationFileNotFoundException, ConfigurationException
 14from cpdbench.exception.ValidationException import UserConfigValidationException
 15from cpdbench.utils import Logger
 16from cpdbench.utils.UserConfig import UserConfig
 17
 18_complete_config = None
 19
 20# LOGGING
 21logging_file_name: str = 'cpdbench-log.txt'
 22logging_level: int = logging.INFO
 23logging_console_level: int = logging.ERROR
 24
 25# MULTIPROCESSING
 26multiprocessing_enabled = True
 27
 28# RESULT
 29result_file_name: str = 'cpdbench-result.json'
 30
 31# USER PARAMETERS
 32_user_config = None
 33
 34_config_error_list = []
 35
 36
 37def get_user_config() -> UserConfig:
 38    """Returns the UserConfig object if the BenchConfig was already initialized.
 39    :return the UserConfig object
 40    """
 41    return _user_config
 42
 43
 44def get_complete_config() -> dict:
 45    """Returns the complete bench configuration including the user config as python dict.
 46    :return the config as dict
 47    """
 48    return {
 49        'logging': {
 50            'logging_file_name': logging_file_name,
 51            'logging_level': logging_level,
 52            'logging_console_level': logging_console_level
 53        },
 54        'multiprocessing': {
 55            'multiprocessing_enabled': multiprocessing_enabled
 56        },
 57        'result': {
 58            'result_file_name': result_file_name
 59        },
 60        'user_config': _user_config.get_param_dict()
 61    }
 62
 63
 64def load_config(config_file='config.yml') -> bool:
 65    """Initializes the BenchConfig object with the given config.yml file.
 66    If the config_file param is None or the file does not exist, the bench will use the default parameters and this
 67    function returns false.
 68    :param config_file: The path to the config file
 69    :return: True if the config could be loaded correctly, false otherwise.
 70    """
 71    global _complete_config
 72    global _user_config
 73    if config_file is None:
 74        _user_config = UserConfig()
 75        _throw_config_errors()
 76        return False
 77    _complete_config = _load_config_from_file(config_file)
 78    if _complete_config is None:
 79        _user_config = UserConfig()
 80        _throw_config_errors()
 81        return False
 82
 83    # logging
 84    _load_logging_config(_complete_config.get('logging'))
 85
 86    # multiprocessing enabled
 87    global multiprocessing_enabled
 88    multiprocessing_enabled = False if str(_complete_config.get('multiprocessing')).upper() == 'FALSE' else True
 89
 90    # result
 91    global result_file_name
 92    result = _complete_config.get('result')
 93    if result is not None:
 94        res_filename = result.get('filename')
 95        if res_filename is not None:
 96            result_file_name = res_filename
 97
 98    # user variables
 99    _user_config = UserConfig(_complete_config.get('user'))
100    try:
101        _user_config.validate_user_config()
102    except Exception as e:
103        raise UserConfigValidationException(str(e))
104
105    _throw_config_errors()
106
107    return True
108
109
110def _load_logging_config(logging_config: dict) -> None:
111    if logging_config is None:
112        return
113    # filename
114    global logging_file_name
115    filename = logging_config.get('filename')
116    if filename is not None:
117        logging_file_name = filename
118
119    # log-level
120    global logging_level
121    logging_level = _get_log_level(logging_config, 'log-level')
122
123    # log-level console
124    global logging_console_level
125    logging_console_level = _get_log_level(logging_config, 'log-level-console')
126
127
128def _get_log_level(config, param_name):
129    if config.get(param_name) is None:
130        return logging.INFO
131    if config[param_name].upper() == 'DEBUG':
132        return logging.DEBUG
133    elif config[param_name].upper() == 'INFO':
134        return logging.INFO
135    elif config[param_name].upper() == 'WARNING' or config[param_name].upper() == "WARN":
136        return logging.WARNING
137    elif config[param_name].upper() == 'ERROR':
138        return logging.ERROR
139    elif config[param_name].upper() == 'CRITICAL':
140        return logging.CRITICAL
141    else:
142        _add_config_error(ConfigurationException(param_name))
143        return logging.INFO
144
145
146def _load_config_from_file(config_file: str):
147    try:
148        with open(config_file, 'r') as config_file_stream:
149            yaml_config = yaml.safe_load(config_file_stream)
150    except OSError:
151        _add_config_error(ConfigurationFileNotFoundException(config_file))
152        return None
153    else:
154        return yaml_config
155
156
157def _add_config_error(exc: Exception) -> None:
158    _config_error_list.append(exc)
159
160
161def _throw_config_errors() -> None:
162    logger = Logger.get_application_logger()
163    for exc in _config_error_list:
164        logger.warning(exc)