Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_string_interpolation_with_readonly_parent() -> None:
cfg = OmegaConf.create({"a": 10, "b": {"c": "hello_${a}"}})
OmegaConf.set_readonly(cfg, True)
assert OmegaConf.to_container(cfg, resolve=True) == {
"a": 10,
"b": {"c": "hello_10"},
}
def test_to_container(src: Any, expected: Any, expected_with_resolve: Any) -> None:
if expected is None:
expected = src
if expected_with_resolve is None:
expected_with_resolve = expected
cfg = OmegaConf.create(src)
container = OmegaConf.to_container(cfg)
assert container == expected
container = OmegaConf.to_container(cfg, resolve=True)
assert container == expected_with_resolve
def test_merge(inputs: Any, expected: Any) -> None:
configs = [OmegaConf.create(c) for c in inputs]
if isinstance(expected, (MutableMapping, MutableSequence)) or is_structured_config(
expected
):
merged = OmegaConf.merge(*configs)
assert merged == expected
# test input configs are not changed.
# Note that converting to container without resolving to avoid resolution errors while comparing
for i in range(len(inputs)):
input_i = OmegaConf.create(inputs[i])
orig = OmegaConf.to_container(input_i, resolve=False)
merged2 = OmegaConf.to_container(configs[i], resolve=False)
assert orig == merged2
else:
with expected:
OmegaConf.merge(*configs)
def test_merge(inputs: Any, expected: Any) -> None:
configs = [OmegaConf.create(c) for c in inputs]
if isinstance(expected, (MutableMapping, MutableSequence)) or is_structured_config(
expected
):
merged = OmegaConf.merge(*configs)
assert merged == expected
# test input configs are not changed.
# Note that converting to container without resolving to avoid resolution errors while comparing
for i in range(len(inputs)):
input_i = OmegaConf.create(inputs[i])
orig = OmegaConf.to_container(input_i, resolve=False)
merged2 = OmegaConf.to_container(configs[i], resolve=False)
assert orig == merged2
else:
with expected:
OmegaConf.merge(*configs)
def configure_log(log_config, verbose_config):
assert isinstance(verbose_config, (bool, str, ListConfig))
if log_config is not None:
conf = OmegaConf.to_container(log_config, resolve=True)
logging.config.dictConfig(conf)
else:
# default logging to stdout
root = logging.getLogger()
root.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stdout)
formatter = logging.Formatter(
"[%(asctime)s][%(name)s][%(levelname)s] - %(message)s"
)
handler.setFormatter(formatter)
root.addHandler(handler)
if isinstance(verbose_config, bool):
if verbose_config:
logging.getLogger().setLevel(logging.DEBUG)
else:
if isinstance(verbose_config, str):
def load_sweep_config(self, master_config, sweep_overrides):
# Recreate the config for this sweep instance with the appropriate overrides
overrides = (
OmegaConf.to_container(master_config.hydra.overrides.hydra)
+ sweep_overrides
)
sweep_config = self.load_configuration(
config_file=master_config.hydra.job.config_file,
strict=self.default_strict,
overrides=overrides,
)
with open_dict(sweep_config):
sweep_config.hydra.runtime.merge_with(master_config.hydra.runtime)
# Copy old config cache to ensure we get the same resolved values (for things like timestamps etc)
OmegaConf.copy_cache(from_config=master_config, to_config=sweep_config)
return sweep_config
Args:
config_id (str): Id of the config file to read.
should_make_dir (bool, optional): Should make dir (for saving
models, logs etc). Defaults to True.
should_make_config_immutable (bool, optional): Should the config be frozen (immutable).
Defaults to True.
Returns:
ConfigType: [description]
"""
sample_config = _read_config_file_and_load_components("sample_config")
current_config = _read_config_file_and_load_components(config_id)
config = OmegaConf.merge(sample_config, current_config)
OmegaConf.set_struct(config, True)
resolved_config = OmegaConf.create(OmegaConf.to_container(config, resolve=True))
assert isinstance(resolved_config, DictConfig)
config = _process(config=resolved_config, should_make_dir=should_make_dir)
if should_make_config_immutable:
config = make_config_immutable(config)
assert _is_valid_config(config, config_id)
return config
def to_dict(config: ConfigType) -> Dict[str, Any]:
"""Convert config to serializable dictionary
Args:
config (ConfigType):
Returns:
Dict:
"""
dict_config = cast(Dict[str, Any], OmegaConf.to_container(config))
return dict_config