Skip to content

base

Core callback protocols and callback manager.

Example

from flowcean.core.callbacks import ( ... CallbackManager, ... LoggingCallback, ... RichCallback, ... ) manager = CallbackManager([RichCallback(), LoggingCallback()]) manager.on_learning_start(learner)

LearnerCallback

Bases: Protocol

Protocol for learner callbacks.

on_learning_start(learner, context=None) abstractmethod

Called when learning starts.

Source code in src/flowcean/core/callbacks/base.py 
23
24
25
26
27
28
29
@abstractmethod
def on_learning_start(
    self,
    learner: Named,
    context: dict[str, Any] | None = None,
) -> None:
    """Called when learning starts."""

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

Called during learning with progress updates.

Source code in src/flowcean/core/callbacks/base.py 
31
32
33
34
35
36
37
38
@abstractmethod
def on_learning_progress(
    self,
    learner: Named,
    progress: float | None = None,
    metrics: dict[str, Any] | None = None,
) -> None:
    """Called during learning with progress updates."""

on_learning_end(learner, model, metrics=None) abstractmethod

Called when learning completes successfully.

Source code in src/flowcean/core/callbacks/base.py 
40
41
42
43
44
45
46
47
@abstractmethod
def on_learning_end(
    self,
    learner: Named,
    model: Model,
    metrics: dict[str, Any] | None = None,
) -> None:
    """Called when learning completes successfully."""

on_learning_error(learner, error) abstractmethod

Called if learning fails with an error.

Source code in src/flowcean/core/callbacks/base.py 
49
50
51
52
53
54
55
@abstractmethod
def on_learning_error(
    self,
    learner: Named,
    error: Exception,
) -> None:
    """Called if learning fails with an error."""

CallbackManager(callbacks=None)

Manage multiple callbacks through a single interface.

Initialize the callback manager.

Source code in src/flowcean/core/callbacks/base.py 
61
62
63
def __init__(self, callbacks: list[LearnerCallback] | None = None) -> None:
    """Initialize the callback manager."""
    self.callbacks = callbacks or []

on_learning_start(learner, context=None)

Notify all callbacks that learning has started.

Source code in src/flowcean/core/callbacks/base.py 
65
66
67
68
69
70
71
72
def on_learning_start(
    self,
    learner: Named,
    context: dict[str, Any] | None = None,
) -> None:
    """Notify all callbacks that learning has started."""
    for callback in self.callbacks:
        callback.on_learning_start(learner, context)

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

Notify all callbacks of learning progress.

Source code in src/flowcean/core/callbacks/base.py 
74
75
76
77
78
79
80
81
82
def on_learning_progress(
    self,
    learner: Named,
    progress: float | None = None,
    metrics: dict[str, Any] | None = None,
) -> None:
    """Notify all callbacks of learning progress."""
    for callback in self.callbacks:
        callback.on_learning_progress(learner, progress, metrics)

on_learning_end(learner, model, metrics=None)

Notify all callbacks that learning has completed.

Source code in src/flowcean/core/callbacks/base.py 
84
85
86
87
88
89
90
91
92
def on_learning_end(
    self,
    learner: Named,
    model: Model,
    metrics: dict[str, Any] | None = None,
) -> None:
    """Notify all callbacks that learning has completed."""
    for callback in self.callbacks:
        callback.on_learning_end(learner, model, metrics)

on_learning_error(learner, error)

Notify all callbacks that learning has failed.

Source code in src/flowcean/core/callbacks/base.py 
 94
 95
 96
 97
 98
 99
100
101
def on_learning_error(
    self,
    learner: Named,
    error: Exception,
) -> None:
    """Notify all callbacks that learning has failed."""
    for callback in self.callbacks:
        callback.on_learning_error(learner, error)

SilentCallback

A callback that intentionally produces no output.

on_learning_start(learner, context=None)

Do nothing when learning starts.

Source code in src/flowcean/core/callbacks/base.py 
107
108
109
110
111
112
def on_learning_start(
    self,
    learner: Named,
    context: dict[str, Any] | None = None,
) -> None:
    """Do nothing when learning starts."""

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

Do nothing during learning progress.

Source code in src/flowcean/core/callbacks/base.py 
114
115
116
117
118
119
120
def on_learning_progress(
    self,
    learner: Named,
    progress: float | None = None,
    metrics: dict[str, Any] | None = None,
) -> None:
    """Do nothing during learning progress."""

on_learning_end(learner, model, metrics=None)

Do nothing when learning ends.

Source code in src/flowcean/core/callbacks/base.py 
122
123
124
125
126
127
128
def on_learning_end(
    self,
    learner: Named,
    model: Model,
    metrics: dict[str, Any] | None = None,
) -> None:
    """Do nothing when learning ends."""

on_learning_error(learner, error)

Do nothing when learning errors.

Source code in src/flowcean/core/callbacks/base.py 
130
131
132
133
134
135
def on_learning_error(
    self,
    learner: Named,
    error: Exception,
) -> None:
    """Do nothing when learning errors."""