rich

Rich console callbacks for learner progress feedback.

Example

from flowcean.core.callbacks import RichCallback from flowcean.sklearn import RandomForestRegressorLearner learner = RandomForestRegressorLearner(callbacks=[RichCallback()])

RichCallback(console=None, *, show_metrics=True)

Bases: LearnerCallback

Rich console callback with adaptive progress display.

Initialize the Rich callback.

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

def __init__(
    self,
    console: Console | None = None,
    *,
    show_metrics: bool = True,
) -> None:
    """Initialize the Rich callback."""
    self.console = console or Console()
    self.show_metrics = show_metrics
    self._live: Live | None = None
    self._progress: Progress | None = None
    self._task_id: TaskID | None = None
    self._has_progress_updates = False

on_learning_start(learner, context=None)

Display learning start message.

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

def on_learning_start(
    self,
    learner: Named,
    context: dict[str, Any] | None = None,
) -> None:
    """Display learning start message."""
    message = Text()
    message.append("⠿ ", style="bold blue")
    message.append(f"[{learner.name}] Learning", style="bold blue")
    if context and self.show_metrics:
        context_str = ", ".join(f"{k}={v}" for k, v in context.items())
        message.append(f" ({context_str})", style="dim")
    message.append("...", style="dim")

    self.console.print(message)
    self._has_progress_updates = False

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

Update progress display with current progress and metrics.

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

def on_learning_progress(
    self,
    learner: Named,
    progress: float | None = None,
    metrics: dict[str, Any] | None = None,
) -> None:
    """Update progress display with current progress and metrics."""
    if not self._has_progress_updates:
        self._has_progress_updates = True
        self._progress = Progress(
            SpinnerColumn(),
            TextColumn("[bold blue]{task.description}"),
            BarColumn(),
            TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
            MofNCompleteColumn(),
            TimeElapsedColumn(),
            console=self.console,
        )

        description = f"[{learner.name}] Learning"
        self._task_id = self._progress.add_task(
            description,
            total=100 if progress else None,
        )

        self._live = Live(
            self._progress,
            console=self.console,
            refresh_per_second=10,
        )
        self._live.start()

    if self._progress and self._task_id is not None:
        description = f"[{learner.name}] Learning"
        if metrics and self.show_metrics:
            metrics_str = ", ".join(f"{k}={v}" for k, v in metrics.items())
            description += f" ({metrics_str})"

        if progress is not None:
            if self._progress.tasks[self._task_id].total is None:
                self._progress.update(self._task_id, total=100)
            self._progress.update(
                self._task_id,
                completed=progress * 100,
                description=description,
            )
        else:
            self._progress.update(
                self._task_id,
                description=description,
                advance=0.1,
            )

        if self._live:
            self._live.refresh()

on_learning_end(learner, model, metrics=None)

Display learning completion message.

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

def on_learning_end(
    self,
    learner: Named,
    model: Model,  # noqa: ARG002
    metrics: dict[str, Any] | None = None,
) -> None:
    """Display learning completion message."""
    if self._progress and self._task_id is not None:
        if self._progress.tasks[self._task_id].total is None:
            self._progress.update(self._task_id, total=100)
        self._progress.update(self._task_id, completed=100)

    if self._live:
        self._live.stop()

    message = Text(
        f"✓ [{learner.name}] Learning finished",
        style="bold green",
    )
    if metrics and self.show_metrics:
        metrics_str = ", ".join(f"{k}={v}" for k, v in metrics.items())
        message.append(f" ({metrics_str})", style="green")

    self.console.print(message)
    self._live = None
    self._progress = None
    self._task_id = None
    self._has_progress_updates = False

on_learning_error(learner, error)

Display error message.

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

def on_learning_error(
    self,
    learner: Named,
    error: Exception,
) -> None:
    """Display error message."""
    if self._live:
        self._live.stop()

    self.console.print(
        f"✗ [{learner.name}] Learning failed: {error}",
        style="bold red",
    )

    self._live = None
    self._progress = None
    self._task_id = None
    self._has_progress_updates = False

RichSpinnerCallback(console=None)

Bases: LearnerCallback

Simplified Rich callback with just a spinner.

Initialize the Rich spinner callback.

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

def __init__(
    self,
    console: Console | None = None,
) -> None:
    """Initialize the Rich spinner callback."""
    self.console = console or Console()
    self._live: Live | None = None

on_learning_start(learner, context=None)

Display learning start message with spinner.

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

def on_learning_start(
    self,
    learner: Named,
    context: dict[str, Any] | None = None,  # noqa: ARG002
) -> None:
    """Display learning start message with spinner."""
    text = Text()
    text.append("⠋", style="bold blue")
    text.append(f" [{learner.name}] Learning", style="bold")

    self._live = Live(text, console=self.console, refresh_per_second=10)
    self._live.start()

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

Update spinner state.

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

def on_learning_progress(
    self,
    learner: Named,
    progress: float | None = None,
    metrics: dict[str, Any] | None = None,
) -> None:
    """Update spinner state."""

on_learning_end(learner, model, metrics=None)

Display learning completion message.

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

def on_learning_end(
    self,
    learner: Named,
    model: Model,  # noqa: ARG002
    metrics: dict[str, Any] | None = None,  # noqa: ARG002
) -> None:
    """Display learning completion message."""
    if self._live:
        self._live.stop()

    self.console.print(
        f"✓ [{learner.name}] Learning finished",
        style="bold green",
    )
    self._live = None

on_learning_error(learner, error)

Display error message.

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

def on_learning_error(
    self,
    learner: Named,
    error: Exception,
) -> None:
    """Display error message."""
    if self._live:
        self._live.stop()

    self.console.print(
        f"✗ [{learner.name}] Learning failed: {error}",
        style="bold red",
    )
    self._live = None