Документация логгеров

loggers

Logger integrations for embeddings_squeeze.

Classes

ClearMLLogger

ClearMLLogger(task)

Wrapper for ClearML logging compatible with PyTorch Lightning. Supports scalar metrics, plots, images, and text logging.

Source code in embeddings_squeeze\loggers\clearml_logger.py

def __init__(self, task: Task):
    self.task = task
    self.logger = task.get_logger() if task else None

Functions

log_metrics

log_metrics(metrics, step=None)

Log metrics to ClearML.

Source code in embeddings_squeeze\loggers\clearml_logger.py

def log_metrics(self, metrics: dict, step: int = None):
    """Log metrics to ClearML."""
    if self.logger is None:
        return

    for key, value in metrics.items():
        # Split key into title and series (e.g., "train/loss" -> title="train", series="loss")
        if '/' in key:
            title, series = key.split('/', 1)
        else:
            title = 'metrics'
            series = key

        self.logger.report_scalar(
            title=title,
            series=series,
            value=value,
            iteration=step
        )

log_scalar

log_scalar(title, series, value, iteration)

Log a single scalar value to ClearML.

Parameters:

Name	Type	Description	Default
title	str	Graph title (e.g., "loss", "accuracy")	required
series	str	Series name within the graph (e.g., "train", "val")	required
value	float	Scalar value to log	required
iteration	int	Iteration/step number	required

Example

logger.log_scalar("loss", "train", 0.5, iteration=100) logger.log_scalar("loss", "val", 0.3, iteration=100)

Source code in embeddings_squeeze\loggers\clearml_logger.py

def log_scalar(self, title: str, series: str, value: float, iteration: int):
    """
    Log a single scalar value to ClearML.

    Args:
        title: Graph title (e.g., "loss", "accuracy")
        series: Series name within the graph (e.g., "train", "val")
        value: Scalar value to log
        iteration: Iteration/step number

    Example:
        logger.log_scalar("loss", "train", 0.5, iteration=100)
        logger.log_scalar("loss", "val", 0.3, iteration=100)
    """
    if self.logger is None:
        return

    self.logger.report_scalar(
        title=title,
        series=series,
        value=value,
        iteration=iteration
    )

log_image

log_image(title, series, image, iteration)

Log an image to ClearML.

Parameters:

Name	Type	Description	Default
title	str	Image title/group	required
series	str	Series name (e.g., "predictions", "ground_truth")	required
image		Image as numpy array (H, W) or (H, W, C) for grayscale/RGB Supports uint8 (0-255) or float (0-1)	required
iteration	int	Iteration/step number	required

Example

Grayscale image

img = np.eye(256, 256, dtype=np.uint8) * 255 logger.log_image("predictions", "epoch_1", img, iteration=0)

RGB image

img_rgb = np.zeros((256, 256, 3), dtype=np.uint8) img_rgb[:, :, 0] = 255 # Red channel logger.log_image("predictions", "epoch_1_rgb", img_rgb, iteration=0)

Source code in embeddings_squeeze\loggers\clearml_logger.py

def log_image(self, title: str, series: str, image, iteration: int):
    """
    Log an image to ClearML.

    Args:
        title: Image title/group
        series: Series name (e.g., "predictions", "ground_truth")
        image: Image as numpy array (H, W) or (H, W, C) for grayscale/RGB
               Supports uint8 (0-255) or float (0-1)
        iteration: Iteration/step number

    Example:
        # Grayscale image
        img = np.eye(256, 256, dtype=np.uint8) * 255
        logger.log_image("predictions", "epoch_1", img, iteration=0)

        # RGB image
        img_rgb = np.zeros((256, 256, 3), dtype=np.uint8)
        img_rgb[:, :, 0] = 255  # Red channel
        logger.log_image("predictions", "epoch_1_rgb", img_rgb, iteration=0)
    """
    if self.logger is None:
        return

    self.logger.report_image(
        title=title,
        series=series,
        iteration=iteration,
        image=image
    )

log_images_batch

log_images_batch(title, series, images, iteration)

Log multiple images to ClearML.

Parameters:

Name	Type	Description	Default
title	str	Image title/group	required
series	str	Series name	required
images	list	List of images (numpy arrays)	required
iteration	int	Iteration/step number	required

Example

images = [img1, img2, img3] logger.log_images_batch("samples", "batch_0", images, iteration=0)

Source code in embeddings_squeeze\loggers\clearml_logger.py

def log_images_batch(self, title: str, series: str, images: list, iteration: int):
    """
    Log multiple images to ClearML.

    Args:
        title: Image title/group
        series: Series name
        images: List of images (numpy arrays)
        iteration: Iteration/step number

    Example:
        images = [img1, img2, img3]
        logger.log_images_batch("samples", "batch_0", images, iteration=0)
    """
    if self.logger is None:
        return

    for idx, image in enumerate(images):
        self.logger.report_image(
            title=title,
            series=f"{series}_img_{idx}",
            iteration=iteration,
            image=image
        )

log_text

log_text(text, title='Info')

Log text to ClearML.

Source code in embeddings_squeeze\loggers\clearml_logger.py

def log_text(self, text: str, title: str = "Info"):
    """Log text to ClearML."""
    if self.logger is None:
        return
    self.logger.report_text(text, print_console=True)

report_text

report_text(text)

Report text to ClearML (alias for log_text with default title).

Source code in embeddings_squeeze\loggers\clearml_logger.py

def report_text(self, text: str):
    """Report text to ClearML (alias for log_text with default title)."""
    self.log_text(text)

report_plotly

report_plotly(title, series, iteration, figure)

Report a Plotly figure to ClearML.

Parameters:

Name	Type	Description	Default
title	str	Plot title/group	required
series	str	Series name	required
iteration	int	Iteration/step number	required
figure		Plotly figure object	required

Example

import plotly.graph_objects as go fig = go.Figure() fig.add_trace(go.Scatter(x=[1,2,3], y=[4,5,6], mode='lines', name='data')) fig.update_layout(title="My Plot", xaxis_title="x", yaxis_title="y") logger.report_plotly("metrics", "loss", iteration=0, figure=fig)

Source code in embeddings_squeeze\loggers\clearml_logger.py

def report_plotly(self, title: str, series: str, iteration: int, figure):
    """
    Report a Plotly figure to ClearML.

    Args:
        title: Plot title/group
        series: Series name
        iteration: Iteration/step number
        figure: Plotly figure object

    Example:
        import plotly.graph_objects as go
        fig = go.Figure()
        fig.add_trace(go.Scatter(x=[1,2,3], y=[4,5,6], mode='lines', name='data'))
        fig.update_layout(title="My Plot", xaxis_title="x", yaxis_title="y")
        logger.report_plotly("metrics", "loss", iteration=0, figure=fig)
    """
    if self.logger is None:
        return

    self.logger.report_plotly(
        title=title,
        series=series,
        iteration=iteration,
        figure=figure
    )

finalize

finalize()

Finalize logging and close task.

Source code in embeddings_squeeze\loggers\clearml_logger.py

def finalize(self):
    """Finalize logging and close task."""
    if self.task:
        self.task.close()

ClearMLUploadCallback

ClearMLUploadCallback(
    task,
    clearml_logger=None,
    checkpoint_dir="checkpoints",
    embedding_dir="embeddings",
)

Bases: Callback

PyTorch Lightning callback for logging checkpoint and embedding paths to ClearML.

Automatically logs local file paths for: - Latest checkpoint after each validation epoch - Per-epoch validation embeddings

Usage

from pytorch_lightning import Trainer from embeddings_squeeze.loggers import ClearMLUploadCallback, setup_clearml

task = setup_clearml(project_name="my_project", task_name="experiment_1") logger = ClearMLLogger(task) if task else None callback = ClearMLUploadCallback(task, logger, checkpoint_dir="checkpoints")

trainer = Trainer(callbacks=[callback], ...)

Initialize ClearML path logging callback.

Parameters:

Name	Type	Description	Default
task	Task	ClearML Task object	required
clearml_logger	ClearMLLogger	ClearML logger for text reporting (optional)	None
checkpoint_dir	str	Directory containing checkpoints	'checkpoints'
embedding_dir	str	Directory containing embeddings	'embeddings'

Source code in embeddings_squeeze\loggers\clearml_logger.py

def __init__(self, task: Task, clearml_logger: ClearMLLogger = None, 
             checkpoint_dir: str = "checkpoints", embedding_dir: str = "embeddings"):
    """
    Initialize ClearML path logging callback.

    Args:
        task: ClearML Task object
        clearml_logger: ClearML logger for text reporting (optional)
        checkpoint_dir: Directory containing checkpoints
        embedding_dir: Directory containing embeddings
    """
    super().__init__()
    self.task = task
    self.clearml_logger = clearml_logger
    self.checkpoint_dir = checkpoint_dir
    self.embedding_dir = embedding_dir

Functions

on_validation_epoch_end

on_validation_epoch_end(trainer, pl_module)

Called after validation epoch ends.

Source code in embeddings_squeeze\loggers\clearml_logger.py

def on_validation_epoch_end(self, trainer, pl_module):
    """Called after validation epoch ends."""
    if self.task is None:
        return

    # Log checkpoint path
    try:
        ckpt_path = self._find_latest_checkpoint()
        if ckpt_path:
            abs_path = os.path.abspath(ckpt_path)
            if self.clearml_logger:
                self.clearml_logger.report_text(f"Checkpoint saved: {abs_path}")
    except Exception as e:
        if self.clearml_logger:
            self.clearml_logger.report_text(f"Failed finding checkpoint: {e}")

    # Log embedding path
    try:
        emb_path = os.path.join(
            self.embedding_dir, 
            f"val_embedding_epoch{pl_module.current_epoch}.pt"
        )
        if os.path.exists(emb_path):
            abs_path = os.path.abspath(emb_path)
            if self.clearml_logger:
                self.clearml_logger.report_text(f"Embedding saved: {abs_path}")
    except Exception as e:
        if self.clearml_logger:
            self.clearml_logger.report_text(f"Failed logging embedding path: {e}")

Functions

setup_clearml

setup_clearml(project_name, task_name, auto_connect=True)

Setup ClearML with credentials from config file.

Parameters:

Name	Type	Description	Default
project_name	str	ClearML project name	required
task_name	str	ClearML task name	required
auto_connect	bool	If True, automatically connect frameworks	True

Returns:

Type	Description
	Task object

Source code in embeddings_squeeze\loggers\clearml_logger.py

def setup_clearml(project_name: str, task_name: str, auto_connect: bool = True):
    """
    Setup ClearML with credentials from config file.

    Args:
        project_name: ClearML project name
        task_name: ClearML task name
        auto_connect: If True, automatically connect frameworks

    Returns:
        Task object
    """
    # Load credentials
    config_dir = Path(__file__).parent.parent / 'configs'

    try:
        creds = load_credentials(config_dir)

        # Set credentials
        clearml.Task.set_credentials(
            api_host=creds.get('api_host', 'https://api.clear.ml'),
            web_host=creds.get('web_host', 'https://app.clear.ml'),
            files_host=creds.get('files_host', 'https://files.clear.ml'),
            key=creds['api_key'],
            secret=creds['api_secret']
        )

        # Initialize task
        task = Task.init(
            project_name=project_name,
            task_name=task_name,
            auto_connect_frameworks=auto_connect
        )
        return task
    except FileNotFoundError as e:
        print(f"Warning: {e}")
        print("ClearML logging disabled. Using TensorBoard instead.")
        return None
    except Exception as e:
        print(f"Warning: Failed to setup ClearML: {e}")
        print("ClearML logging disabled. Using TensorBoard instead.")
        return None