Source code for smoke_forcing.utils.conf_tools

from functools import wraps
from inspect import signature
import os
import yaml


[docs]class InvalidConfigException(Exception): def __init__(self): self.message = "Config must be `dict` or path as `str` or `os.path`"
[docs]def check_config(func): """ Wrapper function that checks whether the `config` parameter passed to the function wrapped is: 1) A `dict` 2) `str` or `Path`. In this case, it will check if the path is a directory, and if it exists. If the config can be opened, it is loaded into a `dict`, and fed to the function wrapped in place of the `str` or `Path` Caveats: -------- 1) The function wrapped must have `config` as a positional argument. The position of the parameter itself is of no consequence. :param func: python function to be wrapped :return: whatever the wrapped function returns """ @wraps(func) def wrapper(*args, **kwargs): try: loc = list(signature(func).parameters).index("config") _config = args[loc] except ValueError: raise AttributeError(f"No argument config passed to {func.__name__}") _conf_type = type(_config) if _conf_type == dict: pass elif (_conf_type == str) or (_conf_type == os.path): conf_dir = os.path.expandvars(os.path.expanduser(os.path.join(_config))) if not os.path.exists(conf_dir): raise FileNotFoundError("Config file not found") if not os.path.isfile(conf_dir): raise IsADirectoryError("Path to config is a directory") # replace config with a dict with open(conf_dir) as f: _list_args = list(args) _list_args[loc] = yaml.safe_load(f) args = tuple(_list_args) else: raise InvalidConfigException result = func(*args, **kwargs) return result return wrapper