File size: 4,470 Bytes
3de7bf6 |
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 |
"""Get configurable parameters."""
# Copyright (C) 2022-2024 Intel Corporation
# SPDX-License-Identifier: Apache-2.0
import logging
from collections.abc import Sequence
from pathlib import Path
from typing import Any, cast
from jsonargparse import Namespace
from jsonargparse import Path as JSONArgparsePath
from omegaconf import DictConfig, ListConfig, OmegaConf
logger = logging.getLogger(__name__)
def _convert_nested_path_to_str(config: Any) -> Any: # noqa: ANN401
"""Goes over the dictionary and converts all path values to str."""
if isinstance(config, dict):
for key, value in config.items():
config[key] = _convert_nested_path_to_str(value)
elif isinstance(config, list):
for i, item in enumerate(config):
config[i] = _convert_nested_path_to_str(item)
elif isinstance(config, Path | JSONArgparsePath):
config = str(config)
return config
def to_yaml(config: Namespace | ListConfig | DictConfig) -> str:
"""Convert the config to a yaml string.
Args:
config (Namespace | ListConfig | DictConfig): Config
Returns:
str: YAML string
"""
_config = config.clone() if isinstance(config, Namespace) else config.copy()
if isinstance(_config, Namespace):
_config = _config.as_dict()
_config = _convert_nested_path_to_str(_config)
return OmegaConf.to_yaml(_config)
def to_tuple(input_size: int | ListConfig) -> tuple[int, int]:
"""Convert int or list to a tuple.
Args:
input_size (int | ListConfig): input_size
Example:
>>> to_tuple(256)
(256, 256)
>>> to_tuple([256, 256])
(256, 256)
Raises:
ValueError: Unsupported value type.
Returns:
tuple[int, int]: Tuple of input_size
"""
ret_val: tuple[int, int]
if isinstance(input_size, int):
ret_val = cast(tuple[int, int], (input_size,) * 2)
elif isinstance(input_size, ListConfig | Sequence):
if len(input_size) != 2:
msg = "Expected a single integer or tuple of length 2 for width and height."
raise ValueError(msg)
ret_val = cast(tuple[int, int], tuple(input_size))
else:
msg = f"Expected either int or ListConfig, got {type(input_size)}"
raise TypeError(msg)
return ret_val
def update_config(config: DictConfig | ListConfig | Namespace) -> DictConfig | ListConfig | Namespace:
"""Update config.
Args:
config: Configurable parameters.
Returns:
DictConfig | ListConfig | Namespace: Updated config.
"""
_show_warnings(config)
return _update_nncf_config(config)
def _update_nncf_config(config: DictConfig | ListConfig) -> DictConfig | ListConfig:
"""Set the NNCF input size based on the value of the crop_size parameter in the configurable parameters object.
Args:
config (DictConfig | ListConfig): Configurable parameters of the current run.
Returns:
DictConfig | ListConfig: Updated configurable parameters in DictConfig object.
"""
if "optimization" in config and "nncf" in config.optimization:
if "input_info" not in config.optimization.nncf:
config.optimization.nncf["input_info"] = {"sample_size": None}
config.optimization.nncf.input_info.sample_size = [1, 3, 10, 10]
if config.optimization.nncf.apply and "update_config" in config.optimization.nncf:
return OmegaConf.merge(config, config.optimization.nncf.update_config)
return config
def _show_warnings(config: DictConfig | ListConfig | Namespace) -> None:
"""Show warnings if any based on the configuration settings.
Args:
config (DictConfig | ListConfig | Namespace): Configurable parameters for the current run.
"""
if "clip_length_in_frames" in config.data and config.data.init_args.clip_length_in_frames > 1:
logger.warning(
"Anomalib's models and visualizer are currently not compatible with video datasets with a clip length > 1. "
"Custom changes to these modules will be needed to prevent errors and/or unpredictable behaviour.",
)
if (
"devices" in config.trainer
and (config.trainer.devices is None or config.trainer.devices != 1)
and config.trainer.accelerator != "cpu"
):
logger.warning("Anomalib currently does not support multi-gpu training. Setting devices to 1.")
config.trainer.devices = 1
|