logging

Logging callbacks for learner progress feedback.

Example

from flowcean.core.callbacks import LoggingCallback callback = LoggingCallback()

LoggingCallback(logger=None, level_start=logging.INFO, level_progress=logging.DEBUG, level_end=logging.INFO, level_error=logging.ERROR)

Bases: LearnerCallback

Standard Python logging callback.

Initialize the logging callback.

Source code in src/flowcean/core/callbacks/logging.py

def __init__(
    self,
    logger: logging.Logger | None = None,
    level_start: int = logging.INFO,
    level_progress: int = logging.DEBUG,
    level_end: int = logging.INFO,
    level_error: int = logging.ERROR,
) -> None:
    """Initialize the logging callback."""
    self.logger = logger or logging.getLogger("flowcean.learner")
    self.level_start = level_start
    self.level_progress = level_progress
    self.level_end = level_end
    self.level_error = level_error

on_learning_start(learner, context=None)

Log learning start event.

Source code in src/flowcean/core/callbacks/logging.py

def on_learning_start(
    self,
    learner: Named,
    context: dict[str, Any] | None = None,
) -> None:
    """Log learning start event."""
    message = f"[{learner.name}] Learning started"
    if context:
        context_str = ", ".join(f"{k}={v}" for k, v in context.items())
        message += f" ({context_str})"
    self.logger.log(self.level_start, message)

on_learning_progress(learner, progress=None, metrics=None)

Log learning progress.

Source code in src/flowcean/core/callbacks/logging.py

def on_learning_progress(
    self,
    learner: Named,
    progress: float | None = None,
    metrics: dict[str, Any] | None = None,
) -> None:
    """Log learning progress."""
    message = f"[{learner.name}] Learning in progress"
    if progress is not None:
        message += f" ({progress * 100:.1f}%)"
    if metrics:
        metrics_str = ", ".join(f"{k}={v}" for k, v in metrics.items())
        message += f" - {metrics_str}"
    self.logger.log(self.level_progress, message)

on_learning_end(learner, model, metrics=None)

Log learning completion.

Source code in src/flowcean/core/callbacks/logging.py

def on_learning_end(
    self,
    learner: Named,
    model: Model,  # noqa: ARG002
    metrics: dict[str, Any] | None = None,
) -> None:
    """Log learning completion."""
    message = f"[{learner.name}] Learning finished"
    if metrics:
        metrics_str = ", ".join(f"{k}={v}" for k, v in metrics.items())
        message += f" ({metrics_str})"
    self.logger.log(self.level_end, message)

on_learning_error(learner, error)

Log learning error.

Source code in src/flowcean/core/callbacks/logging.py

def on_learning_error(
    self,
    learner: Named,
    error: Exception,
) -> None:
    """Log learning error."""
    self.logger.log(
        self.level_error,
        "[%s] Learning failed: %s",
        learner.name,
        error,
        exc_info=error,
    )