Aluode/PerceptionLabPortable

1from __future__ import annotations2 3import configparser4import json5import os6import warnings7from typing import Any8 9conf: dict[str, dict[str, Any]] = {}10default_conf_dir = os.path.join(os.path.expanduser("~"), ".config/fsspec")11conf_dir = os.environ.get("FSSPEC_CONFIG_DIR", default_conf_dir)12 13 14def set_conf_env(conf_dict, envdict=os.environ):15    """Set config values from environment variables16 17    Looks for variables of the form ``FSSPEC_<protocol>`` and18    ``FSSPEC_<protocol>_<kwarg>``. For ``FSSPEC_<protocol>`` the value is parsed19    as a json dictionary and used to ``update`` the config of the20    corresponding protocol. For ``FSSPEC_<protocol>_<kwarg>`` there is no21    attempt to convert the string value, but the kwarg keys will be lower-cased.22 23    The ``FSSPEC_<protocol>_<kwarg>`` variables are applied after the24    ``FSSPEC_<protocol>`` ones.25 26    Parameters27    ----------28    conf_dict : dict(str, dict)29        This dict will be mutated30    envdict : dict-like(str, str)31        Source for the values - usually the real environment32    """33    kwarg_keys = []34    for key in envdict:35        if key.startswith("FSSPEC_") and len(key) > 7 and key[7] != "_":36            if key.count("_") > 1:37                kwarg_keys.append(key)38                continue39            try:40                value = json.loads(envdict[key])41            except json.decoder.JSONDecodeError as ex:42                warnings.warn(43                    f"Ignoring environment variable {key} due to a parse failure: {ex}"44                )45            else:46                if isinstance(value, dict):47                    _, proto = key.split("_", 1)48                    conf_dict.setdefault(proto.lower(), {}).update(value)49                else:50                    warnings.warn(51                        f"Ignoring environment variable {key} due to not being a dict:"52                        f" {type(value)}"53                    )54        elif key.startswith("FSSPEC"):55            warnings.warn(56                f"Ignoring environment variable {key} due to having an unexpected name"57            )58 59    for key in kwarg_keys:60        _, proto, kwarg = key.split("_", 2)61        conf_dict.setdefault(proto.lower(), {})[kwarg.lower()] = envdict[key]62 63 64def set_conf_files(cdir, conf_dict):65    """Set config values from files66 67    Scans for INI and JSON files in the given dictionary, and uses their68    contents to set the config. In case of repeated values, later values69    win.70 71    In the case of INI files, all values are strings, and these will not72    be converted.73 74    Parameters75    ----------76    cdir : str77        Directory to search78    conf_dict : dict(str, dict)79        This dict will be mutated80    """81    if not os.path.isdir(cdir):82        return83    allfiles = sorted(os.listdir(cdir))84    for fn in allfiles:85        if fn.endswith(".ini"):86            ini = configparser.ConfigParser()87            ini.read(os.path.join(cdir, fn))88            for key in ini:89                if key == "DEFAULT":90                    continue91                conf_dict.setdefault(key, {}).update(dict(ini[key]))92        if fn.endswith(".json"):93            with open(os.path.join(cdir, fn)) as f:94                js = json.load(f)95            for key in js:96                conf_dict.setdefault(key, {}).update(dict(js[key]))97 98 99def apply_config(cls, kwargs, conf_dict=None):100    """Supply default values for kwargs when instantiating class101 102    Augments the passed kwargs, by finding entries in the config dict103    which match the classes ``.protocol`` attribute (one or more str)104 105    Parameters106    ----------107    cls : file system implementation108    kwargs : dict109    conf_dict : dict of dict110        Typically this is the global configuration111 112    Returns113    -------114    dict : the modified set of kwargs115    """116    if conf_dict is None:117        conf_dict = conf118    protos = cls.protocol if isinstance(cls.protocol, (tuple, list)) else [cls.protocol]119    kw = {}120    for proto in protos:121        # default kwargs from the current state of the config122        if proto in conf_dict:123            kw.update(conf_dict[proto])124    # explicit kwargs always win125    kw.update(**kwargs)126    kwargs = kw127    return kwargs128 129 130set_conf_files(conf_dir, conf)131set_conf_env(conf)132