"""Writes configuration to a config file in the home directory.""" import collections import configparser import datetime import pathlib import logging import typing from . import worker HOME_CONFIG_FILE = pathlib.Path('~/.flamenco-worker.cfg').expanduser() GLOBAL_CONFIG_FILE = pathlib.Path('./flamenco-worker.cfg').absolute() CONFIG_SECTION = 'flamenco-worker' DEFAULT_CONFIG = { 'flamenco-worker': collections.OrderedDict([ ('manager_url', ''), # The 'video-encoding' tasks require ffmpeg to be installed, so it's not enabled by default. ('task_types', 'sleep blender-render file-management exr-merge'), ('task_update_queue_db', 'flamenco-worker.db'), ('subprocess_pid_file', 'flamenco-worker-subprocess.pid'), ('may_i_run_interval_seconds', '5'), ('worker_id', ''), ('worker_secret', ''), # All intervals in seconds ('push_log_max_interval_seconds', str(worker.PUSH_LOG_MAX_INTERVAL.total_seconds())), ('push_log_max_entries', str(worker.PUSH_LOG_MAX_ENTRIES)), ('push_act_max_interval_seconds', str(worker.PUSH_ACT_MAX_INTERVAL.total_seconds())), ]), 'pre_task_check': collections.OrderedDict([]), } # type: typing.Mapping[str, typing.Mapping[str, typing.Any]] # Will be assigned to the config key 'task_types' when started with --test CLI arg. TESTING_TASK_TYPES = 'test-blender-render' log = logging.getLogger(__name__) class ConfigParser(configparser.ConfigParser): """ConfigParser that can easily get values from our default config section.""" _DEFAULT_INTERPOLATION = configparser.ExtendedInterpolation() def value(self, key, valtype: type=str): return valtype(self.get(CONFIG_SECTION, key)) def setvalue(self, key, value): self.set(CONFIG_SECTION, key, value) def interval_secs(self, key) -> datetime.timedelta: """Returns the configuration value as timedelta.""" secs = self.value(key, float) return datetime.timedelta(seconds=secs) def erase(self, key: str) -> None: self.set(CONFIG_SECTION, key, '') def merge_with_home_config(new_conf: dict): """Updates the home configuration file with the given config dict.""" confparser = ConfigParser() confparser.read_dict({CONFIG_SECTION: {}}) confparser.read(str(HOME_CONFIG_FILE), encoding='utf8') for key, value in new_conf.items(): confparser.set(CONFIG_SECTION, key, value) tmpname = HOME_CONFIG_FILE.with_name(HOME_CONFIG_FILE.name + '~') log.debug('Writing configuration file to %s', tmpname) with tmpname.open(mode='wt', encoding='utf8') as outfile: confparser.write(outfile) log.debug('Moving configuration file to %s', HOME_CONFIG_FILE) tmpname.replace(HOME_CONFIG_FILE) log.info('Updated configuration file %s', HOME_CONFIG_FILE) def load_config(config_file: pathlib.Path = None, show_effective_config: bool = False, enable_test_mode=False) -> ConfigParser: """Loads one or more configuration files.""" # Logging and the default interpolation of configparser both use the # same syntax for variables. To make it easier to work with, we use # another interpolation for config files, so they now use ${loglevel} # whereas logging still uses %(levelname)s. confparser = ConfigParser() confparser.read_dict(DEFAULT_CONFIG) if config_file: log.info('Loading configuration from %s', config_file) if not config_file.exists(): log.error('Config file %s does not exist', config_file) raise SystemExit(47) loaded = confparser.read(str(config_file), encoding='utf8') else: if not GLOBAL_CONFIG_FILE.exists(): log.error('Config file %s does not exist', GLOBAL_CONFIG_FILE) raise SystemExit(47) config_files = [GLOBAL_CONFIG_FILE, HOME_CONFIG_FILE] filenames = [str(f.absolute()) for f in config_files] log.info('Loading configuration from %s', ', '.join(filenames)) loaded = confparser.read(filenames, encoding='utf8') log.info('Succesfully loaded: %s', loaded) if enable_test_mode: confparser.setvalue('task_types', TESTING_TASK_TYPES) if show_effective_config: import sys log.info('Effective configuration:') to_show = configparser.ConfigParser( interpolation=configparser.ExtendedInterpolation() ) to_show.read_dict(confparser) if to_show.get(CONFIG_SECTION, 'worker_secret'): to_show.set(CONFIG_SECTION, 'worker_secret', '-hidden-') to_show.write(sys.stderr) return confparser def configure_logging(confparser: configparser.ConfigParser, enable_debug: bool): import logging.config logging.config.fileConfig(confparser, disable_existing_loggers=True) logging.captureWarnings(capture=True) if enable_debug: logging.getLogger('flamenco_worker').setLevel(logging.DEBUG) log.debug('Enabling debug logging')