Utils

This module implements implements additional utilities to handle the training process.

Configuration resolvers for the topobenchmarkx package.

topobenchmarkx.utils.config_resolvers.get_default_metrics(task)

Get default metrics for a given task.

Parameters:

taskstr

Task, either “classification” or “regression”.

Returns:

list

List of default metrics.

Raises:

ValueError

If the task is invalid.

topobenchmarkx.utils.config_resolvers.get_default_transform(dataset, model)

Get default transform for a given data domain and model.

Parameters:

datasetstr

Dataset name. Should be in the format “data_domain/name”.

modelstr

Model name. Should be in the format “model_domain/name”.

Returns:

str

Default transform.

Raises:

ValueError

If the combination of data_domain and model is invalid.

topobenchmarkx.utils.config_resolvers.get_monitor_metric(task, metric)

Get monitor metric for a given task.

Parameters:

taskstr

Task, either “classification” or “regression”.

metricstr

Name of the metric function.

Returns:

str

Monitor metric.

Raises:

ValueError

If the task is invalid.

topobenchmarkx.utils.config_resolvers.get_monitor_mode(task)

Get monitor mode for a given task.

Parameters:

taskstr

Task, either “classification” or “regression”.

Returns:

str

Monitor mode, either “max” or “min”.

Raises:

ValueError

If the task is invalid.

topobenchmarkx.utils.config_resolvers.get_required_lifting(data_domain, model)

Get required transform for a given data domain and model.

Parameters:

data_domainstr

Dataset domain.

modelstr

Model name. Should be in the format “model_domain/name”.

Returns:

str

Required transform.

Raises:

ValueError

If the combination of data_domain and model is invalid.

topobenchmarkx.utils.config_resolvers.infer_in_channels(dataset, transforms)

Infer the number of input channels for a given dataset.

Parameters:

datasetDictConfig

Configuration parameters for the dataset.

transformsDictConfig

Configuration parameters for the transforms.

Returns:

list

List with dimensions of the input channels.

topobenchmarkx.utils.config_resolvers.infere_num_cell_dimensions(selected_dimensions, in_channels)

Infer the length of a list.

Parameters:

selected_dimensionslist

List of selected dimensions. If not None it will be used to infer the length.

in_channelslist

List of input channels. If selected_dimensions is None, this list will be used to infer the length.

Returns:

int

Length of the input list.

Instantiators for callbacks and loggers.

topobenchmarkx.utils.instantiators.instantiate_callbacks(callbacks_cfg: DictConfig) → list[Callback]

Instantiate callbacks from config.

Parameters:

callbacks_cfgDictConfig

A DictConfig object containing callback configurations.

Returns:

list[Callback]

A list of instantiated callbacks.

topobenchmarkx.utils.instantiators.instantiate_loggers(logger_cfg: DictConfig) → list[Logger]

Instantiate loggers from config.

Parameters:

logger_cfgDictConfig

A DictConfig object containing logger configurations.

Returns:

list[Logger]

A list of instantiated loggers.

Utilities for logging hyperparameters.

topobenchmarkx.utils.logging_utils.log_hyperparameters(object_dict: dict[str, Any]) → None

Control which config parts are saved by Lightning loggers.

Additionally saves:

• Number of model parameters

Parameters:

object_dictdict[str, Any]

A dictionary containing the following objects:

• “cfg”: A DictConfig object containing the main config.

• “model”: The Lightning model.

• “trainer”: The Lightning trainer.

A multi-GPU-friendly python command line logger.

class topobenchmarkx.utils.pylogger.RankedLogger(name: str = 'topobenchmarkx.utils.pylogger', rank_zero_only: bool = False, extra: Mapping[str, object] | None = None)

Initialize a multi-GPU-friendly python command line logger.

The logger logs on all processes with their rank prefixed in the log message.

Parameters:

namestr, optional

The name of the logger, by default __name__.

rank_zero_onlybool, optional

Whether to force all logs to only occur on the rank zero process (default: False).

extraMapping[str, object], optional

A dict-like object which provides contextual information. See logging.LoggerAdapter for more information (default: None).

log(level: int, msg: str, rank: int | None = None, *args, **kwargs) → None

Delegate a log call to the underlying logger.

The function first prefixes the message with the rank of the process it’s being logged from and then logs the message. If ‘rank’ is provided, then the log will only occur on that rank/process.

Parameters:

levelint

The level to log at. Look at logging.__init__.py for more information.

msgstr

The message to log.

rankint, optional

The rank to log at (default: None).

*argsAny

Additional args to pass to the underlying logging function.

**kwargsAny

Any additional keyword args to pass to the underlying logging function.

This module contains utility functions for printing and saving Hydra configs.

topobenchmarkx.utils.rich_utils.enforce_tags(cfg: DictConfig, save_to_file: bool = False) → None

Prompt user to input tags from terminal if no tags are provided in config.

Parameters:

cfgDictConfig

A DictConfig composed by Hydra.

save_to_filebool, optional

Whether to export tags to the hydra output folder (default: False).

topobenchmarkx.utils.rich_utils.print_config_tree(cfg: DictConfig, print_order: Sequence[str] = ('data', 'model', 'callbacks', 'logger', 'trainer', 'paths', 'extras'), resolve: bool = False, save_to_file: bool = False) → None

Print the contents of a DictConfig using the Rich library.

Parameters:

cfgDictConfig

A DictConfig object containing the config tree.

print_orderSequence[str], optional

Determines in what order config components are printed, by default (“data”, “model”, “callbacks”, “logger”, “trainer”, “paths”, “extras”).

resolvebool, optional

Whether to resolve reference fields of DictConfig, by default False.

save_to_filebool, optional

Whether to export config to the hydra output folder, by default False.

Utility functions for the TopoBenchMarkX library.

topobenchmarkx.utils.utils.extras(cfg: DictConfig) → None

Apply optional utilities before the task is started.

Utilities:

• Ignoring python warnings.

• Setting tags from command line.

• Rich config printing.

Parameters:

cfgDictConfig

A DictConfig object containing the config tree.

topobenchmarkx.utils.utils.get_metric_value(metric_dict: dict[str, Any], metric_name: str | None) → float | None

Safely retrieves value of the metric logged in LightningModule.

Parameters:

metric_dictdict

A dict containing metric values.

metric_namestr, optional

If provided, the name of the metric to retrieve.

Returns:

float, None

If a metric name was provided, the value of the metric.

topobenchmarkx.utils.utils.task_wrapper(task_func: Callable) → Callable

Optional decorator that controls the failure behavior when executing the task function.

This wrapper can be used to: - make sure loggers are closed even if the task function raises an exception (prevents multirun failure). - save the exception to a .log file. - mark the run as failed with a dedicated file in the logs/ folder (so we can find and rerun it later). - etc. (adjust depending on your needs).

Example: ``` @utils.task_wrapper def train(cfg: DictConfig) -> Tuple[Dict[str, Any], Dict[str, Any]]:

… return metric_dict, object_dict

```

Parameters:

task_funcCallable

The task function to be wrapped.

Returns:

Callable

The wrapped task function.