Source code for federatedscope.core.configs.config

import copy
import logging
import os

from pathlib import Path

import federatedscope.register as register
from federatedscope.core.configs.yacs_config import CfgNode, _merge_a_into_b, \
    Argument

logger = logging.getLogger(__name__)


def set_help_info(cn_node, help_info_dict, prefix=""):
    for k, v in cn_node.items():
        if isinstance(v, Argument) and k not in help_info_dict:
            help_info_dict[prefix + k] = v.description
        elif isinstance(v, CN):
            set_help_info(v,
                          help_info_dict,
                          prefix=f"{k}." if prefix == "" else f"{prefix}{k}.")


[docs]class CN(CfgNode): """ An extended configuration system based on [yacs]( \ https://github.com/rbgirshick/yacs). \ The two-level tree structure consists of several internal dict-like \ containers to allow simple key-value access and management. """ def __init__(self, init_dict=None, key_list=None, new_allowed=False): init_dict = super().__init__(init_dict, key_list, new_allowed) self.__cfg_check_funcs__ = list() # to check the config values # validity self.__help_info__ = dict() # build the help dict self.is_ready_for_run = False # whether this CfgNode has checked its # validity, completeness and clean some un-useful info if init_dict: for k, v in init_dict.items(): if isinstance(v, Argument): self.__help_info__[k] = v.description elif isinstance(v, CN) and "help_info" in v: for name, des in v.__help_info__.items(): self.__help_info__[name] = des def __getattr__(self, name): if name in self: return self[name] else: raise AttributeError(name) def __delattr__(self, name): if name in self: del self[name] else: raise AttributeError(name)
[docs] def clear_aux_info(self): """ Clears all the auxiliary information of the CN object. """ if hasattr(self, "__cfg_check_funcs__"): delattr(self, "__cfg_check_funcs__") if hasattr(self, "__help_info__"): delattr(self, "__help_info__") if hasattr(self, "is_ready_for_run"): delattr(self, "is_ready_for_run") for v in self.values(): if isinstance(v, CN): v.clear_aux_info()
[docs] def print_help(self, arg_name=""): """ print help info for a specific given ``arg_name`` or \ for all arguments if not given ``arg_name`` Args: arg_name: name of specific args """ if arg_name != "" and arg_name in self.__help_info__: print(f" --{arg_name} \t {self.__help_info__[arg_name]}") else: for k, v in self.__help_info__.items(): print(f" --{k} \t {v}")
[docs] def register_cfg_check_fun(self, cfg_check_fun): """ Register a function that checks the configuration node. Args: cfg_check_fun: function for validation the correctness of cfg. """ self.__cfg_check_funcs__.append(cfg_check_fun)
[docs] def merge_from_file(self, cfg_filename, check_cfg=True): """ load configs from a yaml file, another cfg instance or a list \ stores the keys and values. Args: cfg_filename: file name of yaml file check_cfg: whether enable ``assert_cfg()`` """ cfg_check_funcs = copy.copy(self.__cfg_check_funcs__) with open(cfg_filename, "r") as f: cfg = self.load_cfg(f) self.merge_from_other_cfg(cfg) self.__cfg_check_funcs__.clear() self.__cfg_check_funcs__.extend(cfg_check_funcs) self.assert_cfg(check_cfg) set_help_info(self, self.__help_info__)
[docs] def merge_from_other_cfg(self, cfg_other, check_cfg=True): """ load configs from another cfg instance Args: cfg_other: other cfg to be merged check_cfg: whether enable ``assert_cfg()`` """ cfg_check_funcs = copy.copy(self.__cfg_check_funcs__) _merge_a_into_b(cfg_other, self, self, []) self.__cfg_check_funcs__.clear() self.__cfg_check_funcs__.extend(cfg_check_funcs) self.assert_cfg(check_cfg) set_help_info(self, self.__help_info__)
[docs] def merge_from_list(self, cfg_list, check_cfg=True): """ load configs from a list stores the keys and values. \ modified ``merge_from_list`` in ``yacs.config.py`` to allow adding \ new keys if ``is_new_allowed()`` returns True \ Args: cfg_list: list of pairs of cfg name and value check_cfg: whether enable ``assert_cfg()`` """ cfg_check_funcs = copy.copy(self.__cfg_check_funcs__) super().merge_from_list(cfg_list) self.__cfg_check_funcs__.clear() self.__cfg_check_funcs__.extend(cfg_check_funcs) self.assert_cfg(check_cfg) set_help_info(self, self.__help_info__)
[docs] def assert_cfg(self, check_cfg=True): """ check the validness of the configuration instance Args: check_cfg: whether enable checks """ if check_cfg: for check_func in self.__cfg_check_funcs__: check_func(self)
[docs] def clean_unused_sub_cfgs(self): """ Clean the un-used secondary-level CfgNode, whose ``.use`` \ attribute is ``True`` """ for v in self.values(): if isinstance(v, CfgNode) or isinstance(v, CN): # sub-config if hasattr(v, "use") and v.use is False: for k in copy.deepcopy(v).keys(): # delete the un-used attributes if k == "use": continue else: del v[k]
[docs] def check_required_args(self): """ Check required arguments. """ for k, v in self.items(): if isinstance(v, CN): v.check_required_args() if isinstance(v, Argument) and v.required and v.value is None: logger.warning(f"You have not set the required argument {k}")
[docs] def de_arguments(self): """ some config values are managed via ``Argument`` class, this function \ is used to make these values clean without the ``Argument`` class, \ such that the potential type-specific methods work correctly, \ e.g., ``len(cfg.federate.method)`` for a string config """ for k, v in copy.deepcopy(self).items(): if isinstance(v, CN): self[k].de_arguments() if isinstance(v, Argument): self[k] = v.value
[docs] def ready_for_run(self, check_cfg=True): """ Check and cleans up the internal state of cfg and save cfg. Args: check_cfg: whether enable ``assert_cfg()`` """ self.assert_cfg(check_cfg) self.clean_unused_sub_cfgs() self.check_required_args() self.de_arguments() self.is_ready_for_run = True
[docs] def freeze(self, inform=True, save=True, check_cfg=True): """ (1) make the cfg attributes immutable; (2) if ``save==True``, save the frozen cfg_check_funcs into \ ``self.outdir/config.yaml`` for better reproducibility; (3) if ``self.wandb.use==True``, update the frozen config """ self.ready_for_run(check_cfg) super(CN, self).freeze() if save: # save the final cfg Path(self.outdir).mkdir(parents=True, exist_ok=True) with open(os.path.join(self.outdir, "config.yaml"), 'w') as outfile: from contextlib import redirect_stdout with redirect_stdout(outfile): tmp_cfg = copy.deepcopy(self) tmp_cfg.clear_aux_info() print(tmp_cfg.dump()) if self.wandb.use: # update the frozen config try: import wandb import yaml cfg_yaml = yaml.safe_load(tmp_cfg.dump()) wandb.config.update(cfg_yaml, allow_val_change=True) except ImportError: logger.error( "cfg.wandb.use=True but not install the wandb " "package") exit() if inform: logger.info("the used configs are: \n" + str(tmp_cfg))
# to ensure the sub-configs registered before set up the global config from federatedscope.core.configs import all_sub_configs for sub_config in all_sub_configs: __import__("federatedscope.core.configs." + sub_config) from federatedscope.contrib.configs import all_sub_configs_contrib for sub_config in all_sub_configs_contrib: __import__("federatedscope.contrib.configs." + sub_config) # Global config object global_cfg = CN()
[docs]def init_global_cfg(cfg): """ This function sets the default config value. (1) Note that for an experiment, only part of the arguments will be used \ The remaining unused arguments won't affect anything. \ So feel free to register any argument in graphgym.contrib.config (2) We support more than one levels of configs, e.g., cfg.dataset.name """ # ---------------------------------------------------------------------- # # Basic options, first level configs # ---------------------------------------------------------------------- # cfg.backend = 'torch' # Whether to use GPU cfg.use_gpu = False # Whether to check the completeness of msg_handler cfg.check_completeness = False # Whether to print verbose logging info cfg.verbose = 1 # How many decimal places we print out using logger cfg.print_decimal_digits = 6 # Specify the device cfg.device = -1 # Random seed cfg.seed = 0 # Path of configuration file cfg.cfg_file = '' # The dir used to save log, exp_config, models, etc,. cfg.outdir = 'exp' cfg.expname = '' # detailed exp name to distinguish different sub-exp cfg.expname_tag = '' # detailed exp tag to distinguish different # sub-exp with the same expname # extend user customized configs for func in register.config_dict.values(): func(cfg) set_help_info(cfg, cfg.__help_info__)
init_global_cfg(global_cfg)