itwinai.loggers

itwinai wrappers for well-known ML loggers.

A logger allows to save objects of different kinds:

Table 1 Logger kinds

Object kind	Description
metric	Number, usually representing a ML metric of interest (e.g., loss, accuracy).
torch	PyTorch object (e.g., tensor).
artifact	File on the local filesystem to be stored by the logger.
figure	Matplotlib of Plotly figure
image	PIL image or numpy array storing an image.
param	Hyper-parameter (e.g., learning rate, batch size, number of layers) as a primitive Python type.
text	Running text (string).
dict	Python dictionary.
model	ML model. At the moment only torch.nn.Module is supported.
best_model	Best ML model. At the moment only torch.nn.Module is supported.
dataset	Dataset object (e.g., objects of type mlflow.data.Dataset).
watch	WandB watch: Hook into the torch model to collect gradients and the topology. More info.
flops_pb	Flops per batch, used by Prov4MLLogger.
flops_pb	Flops per batch, used by Prov4MLLogger.
flops_pe	Flops per epoch, used by Prov4MLLogger.
system	System metrics, used by Prov4MLLogger.
carbon	Carbon footprint information, used by Prov4MLLogger.
execution_time	Execution time, used by Prov4MLLogger.
prov_documents	Provenance documents, used by Prov4MLLogger.

class itwinai.loggers.LogMixin

Bases: ABC

abstract log(item: Any | List[Any], identifier: str | List[str], kind: str = 'metric', step: int | None = None, batch_idx: int | None = None, **kwargs) → None

Log item with identifier name of kind type at step time step.

Parameters:

• item (Union[Any, List[Any]]) – element to be logged (e.g., metric).

• identifier (Union[str, List[str]]) – unique identifier for the element to log(e.g., name of a metric).

• kind (str, optional) – type of the item to be logged. Must be one among the list of self.supported_kinds. Defaults to ‘metric’.

• step (Optional[int], optional) – logging step. Defaults to None.

• batch_idx (Optional[int], optional) – DataLoader batch counter (i.e., batch idx), if available. Defaults to None.

itwinai.loggers.check_initialized(method: Callable) → Callable

Decorator for logger methods to check whether the logger was correctly initialized before calling the method.

itwinai.loggers.check_not_initialized(method: Callable) → Callable

Decorator for create_logger_context method to prevent double initialization of a logger.

class itwinai.loggers.Logger(savedir: Path | str = 'mllogs', log_freq: int | Literal['epoch', 'batch'] = 'epoch', log_on_workers: int | List[int] = 0, experiment_id: str | None = None, run_id: int | str | None = None)

Bases: LogMixin

Base class for logger

Parameters:

• savedir (Union[Path, str], optional) – filesystem location where logs are stored. Defaults to ‘mllogs’.

• log_freq (Union[int, Literal['epoch', 'batch']], optional) –

  how often should the logger fulfill calls to the log() method:

  • When set to ‘epoch’, the logger logs only if batch_idx is not passed to the log method.

  • When an integer is given, the logger logs if batch_idx is a multiple of log_freq.

  • When set to 'batch', the logger logs always.

  Defaults to ‘epoch’.

• log_on_workers (Optional[Union[int, List[int]]]) – if -1, log on all workers; if int log on worker with rank equal to log_on_workers; if List[int], log on workers which rank is in the list. Defaults to 0 (the global rank of the main worker).

supported_kinds: Tuple[str]

Supported logging ‘kind’s.

worker_rank: int = 0

Current worker global rank

is_initialized: bool = False

Flag to check whether the logger was correctly initialized

savedir: Path

Location on filesystem where to store data.

property experiment_id: str | None

Return the experiment id.

property run_id: int | str | None

Return the id of the current run.

property log_freq: int | Literal['epoch', 'batch']

Get log_feq, namely how often should the logger fulfill or ignore calls to the log() method.

start_logging(rank: int = 0)

Start logging context.

Parameters:

rank (int) – global rank of current process, used in distributed environments. Defaults to 0.

Example:

>>> with my_logger.start_logging():
>>>     my_logger.log(123, 'value', kind='metric', step=0)

abstract create_logger_context(rank: int = 0, **kwargs) → Any

Initializes the logger context.

Parameters:

rank (int) – global rank of current process, used in distributed environments. Defaults to 0.

abstract destroy_logger_context() → None

Destroy logger.

abstract save_hyperparameters(params: Dict[str, Any]) → None

Save hyperparameters.

Parameters:

params (Dict[str, Any]) – hyperparameters dictionary.

serialize(obj: Any, identifier: str) → str

Serializes object to disk and returns its path.

Parameters:

• obj (Any) – item to save.

• identifier (str) – identifier of the item to log (expected to be a path under self.savedir).

Returns:

local path of the serialized object to be logged.

Return type:

str

should_log(batch_idx: int | None = None) → bool

Determines whether the logger should fulfill or ignore calls to the log() method, depending on the log_freq property:

• When log_freq is set to ‘epoch’, the logger logs only if batch_idx is not passed to the log method.

• When log_freq is an integer is given, the logger logs if batch_idx is a multiple of log_freq.

• When log_freq is set to 'batch', the logger logs always.

It also takes into account whether logging on the current worker rank is allowed by self.log_on_workers.

Parameters:

batch_idx (Optional[int]) – the dataloader batch idx, if available. Defaults to None.

Returns:

True if the logger should log, False otherwise.

Return type:

bool

class itwinai.loggers.ConsoleLogger(savedir: Path | str = 'mllogs', log_freq: int | Literal['epoch', 'batch'] = 'epoch', log_on_workers: int | List[int] = 0)

Bases: Logger

Simplified logger.

Parameters:

• savedir (Union[Path, str], optional) – where to store artifacts. Defaults to ‘mllogs’.

• log_freq (Union[int, Literal['epoch', 'batch']], optional) – determines whether the logger should fulfill or ignore calls to the log() method. See Logger.should_log method for more details. Defaults to ‘epoch’.

• log_on_workers (Optional[Union[int, List[int]]]) – if -1, log on all workers; if int log on worker with rank equal to log_on_workers; if List[int], log on workers which rank is in the list. Defaults to 0 (the global rank of the main worker).

supported_kinds: Tuple[str] = ('torch', 'artifact', 'metric')

Supported kinds in the log method

create_logger_context(rank: int = 0, **kwargs)

Initializes the logger context.

Parameters:

rank (int) – global rank of current process, used in distributed environments. Defaults to 0.

destroy_logger_context() → None

Destroy logger. Do nothing.

save_hyperparameters(params: Dict[str, Any]) → None

Save hyperparameters. Do nothing.

Parameters:

params (Dict[str, Any]) – hyperparameters dictionary.

log(item: Any | List[Any], identifier: str | List[str], kind: str = 'metric', step: int | None = None, batch_idx: int | None = None, **kwargs) → None

Print metrics to stdout and save artifacts to the filesystem.

Parameters:

• item (Union[Any, List[Any]]) – element to be logged (e.g., metric).

• identifier (Union[str, List[str]]) – unique identifier for the element to log(e.g., name of a metric).

• kind (str, optional) – type of the item to be logged. Must be one among the list of self.supported_kinds. Defaults to ‘metric’.

• step (Optional[int], optional) – logging step. Defaults to None.

• batch_idx (Optional[int], optional) – DataLoader batch counter (i.e., batch idx), if available. Defaults to None.

• kwargs – keyword arguments to pass to the logger.

class itwinai.loggers.MLFlowLogger(savedir: Path | str = 'mllogs', experiment_name: str = 'unnamed-experiment', tracking_uri: str | None = None, run_description: str | None = None, run_name: str | None = None, log_freq: int | Literal['epoch', 'batch'] = 'epoch', log_on_workers: int | List[int] = 0)

Bases: Logger

Abstraction around MLFlow logger.

Parameters:

• savedir (Union[Path, str], optional) – path on local filesystem where logs are stored. Defaults to ‘mllogs’.

• experiment_name (str, optional) – experiment name. Defaults to itwinai.loggers.BASE_EXP_NAME.

• tracking_uri (Optional[str], optional) – MLFLow tracking URI. Overrides savedir if given. Defaults to None.

• run_description (Optional[str], optional) – run description. Defaults to None.

• run_name (Optional[str], optional) – run name. Defaults to None.

• log_freq (Union[int, Literal['epoch', 'batch']], optional) – determines whether the logger should fulfill or ignore calls to the log() method. See Logger.should_log method for more details. Defaults to ‘epoch’.

• log_on_workers (Optional[Union[int, List[int]]]) – if -1, log on all workers; if int log on worker with rank equal to log_on_workers; if List[int], log on workers which rank is in the list. Defaults to 0 (the global rank of the main worker).

supported_kinds: Tuple[str] = ('metric', 'figure', 'image', 'artifact', 'torch', 'dict', 'param', 'text', 'model', 'dataset')

Supported kinds in the log method

active_run: mlflow.ActiveRun

Current MLFLow experiment’s run.

tracking_uri: str

create_logger_context(rank: int = 0, **kwargs) → mlflow.ActiveRun | None

Initializes the logger context. Start MLFLow run.

Parameters:

• rank (int) – global rank of current process, used in distributed environments. Defaults to 0.

• kwargs –

  • run_id (Optional[str]): MLFlow run ID to attach to. Defaults to None.

  • run_name (Optional[str]): name of the MLFlow run. Defaults to None.

  • parent_run_id (Optional[str]): parent run ID to attach to.

Returns:

active MLFlow run.

Return type:

mlflow.ActiveRun

destroy_logger_context() → None

Destroy logger. End current MLFlow run.

save_hyperparameters(params: Dict[str, Any]) → None

Save hyperparameters as MLFlow parameters.

Parameters:

params (Dict[str, Any]) – hyperparameters dictionary.

log(item: Any | List[Any], identifier: str | List[str], kind: str = 'metric', step: int | None = None, batch_idx: int | None = None, **kwargs) → None

Log with MLFlow.

Parameters:

• item (Union[Any, List[Any]]) – element to be logged (e.g., metric).

• identifier (Union[str, List[str]]) – unique identifier for the element to log(e.g., name of a metric).

• kind (str, optional) – type of the item to be logged. Must be one among the list of self.supported_kinds. Defaults to ‘metric’.

• step (Optional[int], optional) – logging step. Defaults to None.

• batch_idx (Optional[int], optional) – DataLoader batch counter (i.e., batch idx), if available. Defaults to None.

• kwargs – keyword arguments to pass to the logger.

resolve_experiment_id() → str

Sets experiment id by the experiment_name of the logger if not set. Returns the experiment id.

property experiment_id: str

Return the experiment id.

class itwinai.loggers.WandBLogger(savedir: Path | str = 'mllogs', project_name: str = 'unnamed-experiment', log_freq: int | Literal['epoch', 'batch'] = 'epoch', log_on_workers: int | List[int] = 0, offline_mode: bool = False)

Bases: Logger

Abstraction around WandB logger.

Parameters:

• savedir (Union[Path, str], optional) – location on local filesystem where logs are stored. Defaults to ‘mllogs’.

• project_name (str, optional) – experiment name. Defaults to itwinai.loggers.BASE_EXP_NAME.

• log_freq (Union[int, Literal['epoch', 'batch']], optional) – determines whether the logger should fulfill or ignore calls to the log() method. See Logger.should_log method for more details. Defaults to ‘epoch’.

• log_on_workers (Optional[Union[int, List[int]]]) – if -1, log on all workers; if int log on worker with rank equal to log_on_workers; if List[int], log on workers which rank is in the list. Defaults to 0 (the global rank of the main worker).

• offline_mode (str, optional) – Use this option if working on compute node without internet access. Saves logs locally. Defaults to ‘False’.

supported_kinds: Tuple[str] = ('watch', 'metric', 'figure', 'image', 'torch', 'dict', 'param', 'text')

Supported kinds in the log method

create_logger_context(rank: int = 0, **kwargs) → None

Initializes the logger context. Init WandB run.

Parameters:

rank (int) – global rank of current process, used in distributed environments. Defaults to 0.

destroy_logger_context() → None

Destroy logger.

save_hyperparameters(params: Dict[str, Any]) → None

Save hyperparameters.

Parameters:

params (Dict[str, Any]) – hyperparameters dictionary.

log(item: Any | List[Any], identifier: str | List[str], kind: str = 'metric', step: int | None = None, batch_idx: int | None = None, **kwargs) → None

Log with WandB. Wrapper of https://docs.wandb.ai/ref/python/log

Parameters:

• item (Union[Any, List[Any]]) – element to be logged (e.g., metric).

• identifier (Union[str, List[str]]) – unique identifier for the element to log(e.g., name of a metric).

• kind (str, optional) – type of the item to be logged. Must be one among the list of self.supported_kinds. Defaults to ‘metric’.

• step (Optional[int], optional) – ignored by WandBLogger.

• batch_idx (Optional[int], optional) – DataLoader batch counter (i.e., batch idx), if available. Defaults to None.

• kwargs – keyword arguments to pass to the logger.

class itwinai.loggers.TensorBoardLogger(savedir: Path | str = 'mllogs', log_freq: int | Literal['epoch', 'batch'] = 'epoch', framework: Literal['tensorflow', 'pytorch'] = 'pytorch', log_on_workers: int | List[int] = 0)

Bases: Logger

Abstraction around TensorBoard logger, both for PyTorch and TensorFlow.

Parameters:

• savedir (Union[Path, str], optional) – location on local filesystem where logs are stored. Defaults to ‘mllogs’.

• log_freq (Union[int, Literal['epoch', 'batch']], optional) – determines whether the logger should fulfill or ignore calls to the log() method. See Logger.should_log method for more details. Defaults to ‘epoch’.

• framework (Literal['tensorflow', 'pytorch'], optional) – whether to log PyTorch or TensorFlow ML data. Defaults to ‘pytorch’.

• log_on_workers (Optional[Union[int, List[int]]]) – if -1, log on all workers; if int log on worker with rank equal to log_on_workers; if List[int], log on workers which rank is in the list. Defaults to 0 (the global rank of the main worker).

Raises:

ValueError – when framework is not recognized.

supported_kinds: Tuple[str] = ('metric', 'image', 'text', 'figure', 'torch')

Supported kinds in the log method

create_logger_context(rank: int = 0, **kwargs) → None

Initializes the logger context. Init Tensorboard run.

Parameters:

rank (int) – global rank of current process, used in distributed environments. Defaults to 0.

destroy_logger_context() → None

Destroy logger.

save_hyperparameters(params: Dict[str, Any]) → None

Save hyperparameters.

Parameters:

params (Dict[str, Any]) – hyperparameters dictionary.

log(item: Any | List[Any], identifier: str | List[str], kind: str = 'metric', step: int | None = None, batch_idx: int | None = None, **kwargs) → None

Log with Tensorboard.

Parameters:

• item (Union[Any, List[Any]]) – element to be logged (e.g., metric).

• identifier (Union[str, List[str]]) – unique identifier for the element to log(e.g., name of a metric).

• kind (str, optional) – type of the item to be logged. Must be one among the list of self.supported_kinds. Defaults to ‘metric’.

• step (Optional[int], optional) – logging step. Defaults to None.

• batch_idx (Optional[int], optional) – DataLoader batch counter (i.e., batch idx), if available. Defaults to None.

• kwargs – keyword arguments to pass to the logger.

class itwinai.loggers.LoggersCollection(loggers: List[Logger])

Bases: Logger

Wrapper of a set of loggers, allowing to use them simultaneously.

Parameters:

• loggers (List[Logger]) – List of itwinai loggers. Only one logger of each type is

• collection. (allowed in the)

Raises:

ValueError – when multiple loggers of the same type are passed.

supported_kinds: Tuple[str]

Supported kinds are delegated to the loggers in the collection.

should_log(batch_idx: int | None = None) → bool

Transparent method which delegates the Logger.should_log` to individual loggers. Always returns True.

Parameters:

• batch_idx (int, optional) – dataloader batch index.

• None. (Defaults to)

Returns:

always True.

Return type:

bool

log(item: Any | List[Any], identifier: str | List[str], kind: str = 'metric', step: int | None = None, batch_idx: int | None = None, **kwargs) → None

Log on all loggers.

Parameters:

• item (Union[Any, List[Any]]) – element to be logged (e.g., metric).

• identifier (Union[str, List[str]]) – unique identifier for the element to log(e.g., name of a metric).

• kind (str, optional) – type of the item to be logged. Must be one among the list of self.supported_kinds. Defaults to ‘metric’.

• step (Optional[int], optional) – logging step. Defaults to None.

• batch_idx (Optional[int], optional) – DataLoader batch counter (i.e., batch idx), if available. Defaults to None.

• kwargs – keyword arguments to pass to the logger.

create_logger_context(rank: int = 0, **kwargs) → None

Initializes all loggers.

Parameters:

rank (int) – global rank of current process, used in distributed environments. Defaults to 0.

destroy_logger_context() → None

Destroy all loggers.

save_hyperparameters(params: Dict[str, Any]) → None

Save hyperparameters for all loggers.

Parameters:

params (Dict[str, Any]) – hyperparameters dictionary.

class itwinai.loggers.Prov4MLLogger(prov_user_namespace: str = 'www.example.org', experiment_name: str = 'experiment_name', provenance_save_dir: str = 'mllogs/prov_logs', save_after_n_logs: int | None = 100, create_graph: bool | None = True, create_svg: bool | None = True, log_freq: int | Literal['epoch', 'batch'] = 'epoch', log_on_workers: int | List[int] = 0)

Bases: Logger

Abstraction around Prov4ML logger.

Parameters:

• prov_user_namespace (str, optional) – location to where provenance files will be uploaded. Defaults to “www.example.org”.

• experiment_name (str, optional) – experiment name. Defaults to “experiment_name”.

• provenance_save_dir (Union[Path, str], optional) – path where to store provenance files and logs. Defaults to “prov”.

• save_after_n_logs (Optional[int], optional) – how often to save logs to disk from main memory. Defaults to 100.

• create_graph (Optional[bool], optional) – whether to create a provenance graph. Defaults to True.

• create_svg (Optional[bool], optional) – whether to create an SVG representation of the provenance graph. Defaults to True.

• log_freq (Union[int, Literal['epoch', 'batch']], optional) – determines whether the logger should fulfill or ignore calls to the log() method. See Logger.should_log method for more details. Defaults to ‘epoch’.

• log_on_workers (Optional[Union[int, List[int]]]) – if -1, log on all workers; if int log on worker with rank equal to log_on_workers; if List[int], log on workers which rank is in the list. Defaults to 0 (the global rank of the main worker).

supported_kinds: Tuple[str] = ('metric', 'flops_pb', 'flops_pe', 'system', 'carbon', 'execution_time', 'model', 'best_model', 'torch')

Supported kinds in the log method

create_logger_context(rank: int = 0, **kwargs) → None

Initializes the logger context.

Parameters:

rank (int) – global rank of current process, used in distributed environments. Defaults to 0.

destroy_logger_context() → None

Destroy logger.

save_hyperparameters(params: Dict[str, Any]) → None

Save hyperparameters.

Parameters:

params (Dict[str, Any]) – hyperparameters dictionary.

log(item: Any | List[Any], identifier: str | List[str], kind: str = 'metric', step: int | None = None, batch_idx: int | None = None, context: str | None = 'training', **kwargs) → None

Logs with Prov4ML.

Parameters:

• item (Union[Any, List[Any]]) – element to be logged (e.g., metric).

• identifier (Union[str, List[str]]) – unique identifier for the element to log(e.g., name of a metric).

• kind (str, optional) – type of the item to be logged. Must be one among the list of self.supported_kinds. Defaults to ‘metric’.

• step (Optional[int], optional) – logging step. Defaults to None.

• batch_idx (Optional[int], optional) – DataLoader batch counter (i.e., batch idx), if available. Defaults to None.

• kwargs – keyword arguments to pass to the logger.

class itwinai.loggers.EmptyLogger(savedir: Path | str = 'mllogs', log_freq: int | Literal['epoch'] | Literal['batch'] = 'epoch', log_on_workers: int | List[int] = 0)

Bases: Logger

Dummy logger which can be used as a placeholder when a real logger is not available. All methods do nothing.

create_logger_context(rank: int = 0, **kwargs)

Initializes the logger context.

Parameters:

rank (int) – global rank of current process, used in distributed environments. Defaults to 0.

destroy_logger_context() → None

Destroy logger.

save_hyperparameters(params: Dict[str, Any]) → None

Save hyperparameters.

Parameters:

params (Dict[str, Any]) – hyperparameters dictionary.

log(item: Any | List[Any], identifier: str | List[str], kind: str = 'metric', step: int | None = None, batch_idx: int | None = None, **kwargs) → None

Log item with identifier name of kind type at step time step.

Parameters:

• item (Union[Any, List[Any]]) – element to be logged (e.g., metric).

• identifier (Union[str, List[str]]) – unique identifier for the element to log(e.g., name of a metric).

• kind (str, optional) – type of the item to be logged. Must be one among the list of self.supported_kinds. Defaults to ‘metric’.

• step (Optional[int], optional) – logging step. Defaults to None.

• batch_idx (Optional[int], optional) – DataLoader batch counter (i.e., batch idx), if available. Defaults to None.

itwinai.loggers.get_mlflow_logger(logger: Logger | None) → MLFlowLogger | None

Finds and returns the MLFlowLogger if present in the given logger.