import configparser import getpass import os import tempfile from typing import Any, Optional from wandb import env from wandb.old import core from wandb.sdk.lib import filesystem from wandb.sdk.lib.runid import generate_id class Settings: """Global W&B settings stored under $WANDB_CONFIG_DIR/settings.""" DEFAULT_SECTION = "default" _UNSET = object() def __init__( self, load_settings: bool = True, root_dir: Optional[str] = None ) -> None: self._global_settings = Settings._settings() self._local_settings = Settings._settings() self.root_dir = root_dir if load_settings: global_path = Settings._global_path() if global_path is not None: self._global_settings.read([global_path]) # Only attempt to read if there is a directory existing if os.path.isdir(core.wandb_dir(self.root_dir)): self._local_settings.read([Settings._local_path(self.root_dir)]) def get(self, section: str, key: str, fallback: Any = _UNSET) -> Any: # Try the local settings first. If we can't find the key, then try the global settings. # If a fallback is provided, return it if we can't find the key in either the local or global # settings. try: return self._local_settings.get(section, key) except configparser.NoOptionError: try: return self._global_settings.get(section, key) except configparser.NoOptionError: if fallback is not Settings._UNSET: return fallback else: raise def _persist_settings(self, settings, settings_path) -> None: # write a temp file and then move it to the settings path target_dir = os.path.dirname(settings_path) with tempfile.NamedTemporaryFile( "w+", suffix=".tmp", delete=False, dir=target_dir ) as fp: path = os.path.abspath(fp.name) with open(path, "w+") as f: settings.write(f) try: os.replace(path, settings_path) except AttributeError: os.rename(path, settings_path) def set(self, section, key, value, globally=False, persist=False) -> None: """Persist settings to disk if persist = True""" def write_setting(settings, settings_path, persist): if not settings.has_section(section): Settings._safe_add_section(settings, Settings.DEFAULT_SECTION) settings.set(section, key, str(value)) if persist: self._persist_settings(settings, settings_path) if globally: global_path = Settings._global_path() if global_path is not None: write_setting(self._global_settings, global_path, persist) else: write_setting( self._local_settings, Settings._local_path(self.root_dir), persist ) def clear(self, section, key, globally=False, persist=False) -> None: def clear_setting(settings, settings_path, persist): settings.remove_option(section, key) if persist: self._persist_settings(settings, settings_path) if globally: global_path = Settings._global_path() if global_path is not None: clear_setting(self._global_settings, global_path, persist) else: clear_setting( self._local_settings, Settings._local_path(self.root_dir), persist ) def items(self, section=None): section = section if section is not None else Settings.DEFAULT_SECTION result = {"section": section} try: if section in self._global_settings.sections(): for option in self._global_settings.options(section): result[option] = self._global_settings.get(section, option) if section in self._local_settings.sections(): for option in self._local_settings.options(section): result[option] = self._local_settings.get(section, option) except configparser.InterpolationSyntaxError: core.termwarn("Unable to parse settings file") return result @staticmethod def _safe_add_section(settings, section): if not settings.has_section(section): settings.add_section(section) @staticmethod def _settings(default_settings={}): settings = configparser.ConfigParser() Settings._safe_add_section(settings, Settings.DEFAULT_SECTION) for key, value in default_settings.items(): settings.set(Settings.DEFAULT_SECTION, key, str(value)) return settings @staticmethod def _global_path() -> Optional[str]: def try_create_dir(path) -> bool: try: os.makedirs(path, exist_ok=True) if os.access(path, os.W_OK): return True except OSError: pass return False def get_username() -> str: try: return getpass.getuser() except (ImportError, KeyError): return generate_id() try: home_config_dir = os.path.join(os.path.expanduser("~"), ".config", "wandb") if os.getenv(env.CONFIG_DIR): try_create_dir(os.getenv(env.CONFIG_DIR)) return os.path.join(os.getenv(env.CONFIG_DIR), "settings") if not try_create_dir(home_config_dir): temp_config_dir = os.path.join( tempfile.gettempdir(), ".config", "wandb" ) if not try_create_dir(temp_config_dir): username = get_username() config_dir = os.path.join( tempfile.gettempdir(), username, ".config", "wandb" ) try_create_dir(config_dir) else: config_dir = temp_config_dir else: config_dir = home_config_dir return os.path.join(config_dir, "settings") except Exception: return None @staticmethod def _local_path(root_dir=None): filesystem.mkdir_exists_ok(core.wandb_dir(root_dir)) return os.path.join(core.wandb_dir(root_dir), "settings")