Source code for config
"""Module for configuration settings."""
import logging
from merge_utils import io_utils
DEFAULT_CONFIG = ["validation.yaml", "defaults.yaml"]
# Configuration dictionaries
inputs: dict = {}
validation: dict = {}
sites: dict = {}
merging: dict = {}
output: dict = {}
initialized: bool = False
logger = logging.getLogger(__name__)
[docs]
def recursive_update(old_dict: dict, new_dict: dict) -> None:
"""
Recursively update dictionary d with values from dictionary u.
:param old_dict: Dictionary to be updated.
:param new_dict: Dictionary with new values.
:return: None
"""
for key, val in new_dict.items():
if key not in old_dict:
# If the key does not exist in the old dictionary, add it
old_dict[key] = val
continue
old_val = old_dict.get(key, None)
if isinstance(old_val, dict):
# If both are dictionaries, recursively update
if isinstance(val, dict):
recursive_update(old_dict[key], val)
elif isinstance(old_val, list):
# If the old value is a list, extend it with the new value
if isinstance(val, list):
old_dict[key].extend(val)
elif val is not None:
old_dict[key].append(val)
else:
old_dict[key] = val
[docs]
def update(cfg: dict) -> None:
"""
Update the global configuration with values from the provided dictionary.
:param cfg: Dictionary containing new configuration values.
:return: None
"""
recursive_update(inputs, cfg.get("inputs", {}))
recursive_update(validation, cfg.get("validation", {}))
recursive_update(sites, cfg.get("sites", {}))
recursive_update(merging, cfg.get("merging", {}))
recursive_update(output, cfg.get("output", {}))
[docs]
def load(files: list = None) -> None:
"""
Load the specified configuration files.
Missing keys will be filled in with the defaults in DEFAULT_CONFIG.
:param files: List of configuration files.
:return: None
"""
# Add the default configuration file to the beginning of the list
if not files:
files = DEFAULT_CONFIG
elif isinstance(files, str):
files = DEFAULT_CONFIG + [files]
else:
files = DEFAULT_CONFIG + files
for file in files:
cfg = io_utils.read_config_file(file)
logger.info("Loaded configuration file %s", file)
update(cfg)