File size: 6,482 Bytes
9c6594c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 |
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")
|