itwinai.utils

Utilities for itwinai package.

itwinai.utils.normalize_tracking_uri(uri: str) → str[source]

Normalize a tracking URI to a valid URI. If the URI is empty, it will be treated as the current directory.

Parameters:: uri (str) – URI to normalize.
Returns:: Normalized URI.
Return type:: str

itwinai.utils.generate_random_name()[source]

itwinai.utils.load_yaml(path: str) → Dict[source]

Load YAML file as dict.

Parameters:: path (str) – path to YAML file.
Raises:: yaml.YAMLError – for loading/parsing errors.
Returns:: nested dict representation of parsed YAML file.
Return type:: Dict

itwinai.utils.dynamically_import_class(name: str) → Type[source]

Dynamically import class by module path. Adapted from https://stackoverflow.com/a/547867

Parameters:: name (str) – path to the class (e.g., mypackage.mymodule.MyClass)
Returns:: class type.
Return type:: __class__

itwinai.utils.flatten_dict(d: MutableMapping, parent_key: str = '', sep: str = '.') → MutableMapping[source]

Flatten dictionary

Parameters:

d (MutableMapping) – nested dictionary to flatten
parent_key (str, optional) – prefix for all keys. Defaults to ‘’.
sep (str, optional) – separator for nested key concatenation. Defaults to ‘.’.

Returns:

flattened dictionary with new keys.

Return type:

MutableMapping

class itwinai.utils.SignatureInspector(func: Callable)[source]

Bases: object

Provides the functionalities to inspect the signature of a function or a method.

Parameters:: func (Callable) – function to be inspected.

INFTY: int = 9223372036854775807

property has_varargs: bool: Checks if the function has *args parameter.

property has_kwargs: bool: Checks if the function has **kwargs parameter.

property required_params: Tuple[str]: Names of required parameters. Class method’s ‘self’ is skipped.

property min_params_num: int: Minimum number of arguments required.

property max_params_num: int: Max number of supported input arguments. If no limit, SignatureInspector.INFTY is returned.

itwinai.utils.clear_key(my_dict: Dict, dict_name: str, key: Hashable, complain: bool = True) → Dict[source]

Remove key from dictionary if present and complain.

Parameters:

my_dict (Dict) – Dictionary.
dict_name (str) – name of the dictionary.
key (Hashable) – Key to remove.

itwinai.utils.make_config_paths_absolute(args: List[str])[source]

Process CLI arguments to make paths specified for –config-path or -cp absolute. Returns the modified arguments list.

Parameters:

args (List[str]) – a list of system arguments

Returns:

the updated list of system arguments, where all the config path argument is: absolute.

Return type:

List(str)

itwinai.utils.get_root_cause(exception: Exception) → Exception[source]: Recursively extract the first exception in the exception chain.

itwinai.utils.to_uri(path_str: str | Path) → str[source]

Parse a path and convert it to a URI.

Parameters:: path_str (str) – path to convert.
Returns:: URI.
Return type:: str

itwinai.utils.deprecated(reason)[source]

itwinai.utils.time_and_log(func: Callable, logger: Logger, identifier: str, step: int | None = None, destroy_current_logger_context: bool = False) → Any[source]

Time and log the execution of a function (using time.monotonic()).

Parameters:

func (Callable) – function to execute, time and log, expects zero arguments. Use partial from functools if you need to add arguments.
logger
identifier (str) – identifier for the logged metric
step (int | None) – step for logging. Defaults to None.
destroy_current_logger_context (bool) – Whether to destroy the current logger context. Default is False.

Returns:

result of the function call

Return type:

result (Any)

itwinai.utils.filter_pipeline_steps(pipeline_cfg: DictConfig, pipe_steps: List[Any]) → None[source]: Filters the steps in the pipeline configuration, pipeline_cfg, using pipe_steps, and validates the provided pipe_steps. Changes the pipeline_cfg object in-place. In the event of a validation error, the program is exited, as this is expected to be used from the CLI.

itwinai.utils.retrieve_remote_omegaconf_file(url: str) → DictConfig | ListConfig[source]

Fetches and parses a remote OmegaConf configuration file.

Parameters:: url – URL to the raw configuration file (YAML/JSON format), e.g. raw GitHub link.
Returns:: Parsed OmegaConf configuration as DictConfig or ListConfig.
Raises:: typer.Exit – If the request to the URL or the parsing fails.