Callbacks

SRToolkit.evaluation.callbacks

Event-driven callback system for monitoring and controlling SR evaluation.

Provides event dataclasses fired during evaluation, the SRCallbacks base class for implementing custom callbacks, a CallbackDispatcher for managing multiple callbacks, and built-in implementations for progress display, early stopping, and logging.

ExprEvaluated dataclass

ExprEvaluated(expression: str, error: float, evaluation_number: int, experiment_id: str, is_new_best: bool)

Fired after each expression is evaluated by evaluate_expr.

Attributes:

Name	Type	Description
expression	str	String representation of the evaluated expression.
error	float	Error value returned by the ranking function (RMSE or BED).
evaluation_number	int	Total number of evaluate_expr calls made so far, including cache hits.
experiment_id	str	Identifier of the current experiment.
is_new_best	bool	True if this expression achieved a lower error than all previous ones.

BestExpressionFound dataclass

BestExpressionFound(experiment_id: str, expression: str, error: float, evaluation_number: int)

Fired when a new best expression is found during evaluation.

Attributes:

Name	Type	Description
experiment_id	str	Identifier of the current experiment.
expression	str	String representation of the new best expression.
error	float	Error value of the new best expression.
evaluation_number	int	Total number of evaluate_expr calls made at the time this event is fired.

ExperimentEvent dataclass

ExperimentEvent(dataset_name: str, approach_name: str, max_evaluations: Optional[int], success_threshold: Optional[float], seed: Optional[int])

Fired at experiment start and end.

Attributes:

Name	Type	Description
dataset_name	str	Name of the dataset being evaluated.
approach_name	str	Name of the SR approach being run.
max_evaluations	Optional[int]	Maximum number of evaluations allowed for this experiment.
success_threshold	Optional[float]	Error threshold for success, or None if not set.
seed	Optional[int]	Random seed used for this experiment, or None if not set.

SRCallbacks

Bases: ABC

Abstract base class for SR evaluation callbacks.

Implement only the methods you need. Return False from on_expr_evaluated or on_best_expression to request early stopping; return True or None to continue.

Examples:

>>> class PrintBestCallback(SRCallbacks):
...     def on_best_expression(self, event):
...         print(f"New best: {event.expression} (error={event.error:.4g})")
>>> cb = PrintBestCallback()
>>> cb.on_best_expression(BestExpressionFound("", "X_0+C", 0.01, 5))
New best: X_0+C (error=0.01)

on_expr_evaluated

on_expr_evaluated(event: ExprEvaluated) -> Optional[bool]

Called after each expression is evaluated.

Parameters:

Name	Type	Description	Default
event	ExprEvaluated	Data about the evaluated expression.	required

Returns:

Type	Description
Optional[bool]	False to stop the search early, True or None to continue.

Source code in SRToolkit/evaluation/callbacks.py

def on_expr_evaluated(self, event: ExprEvaluated) -> Optional[bool]:
    """
    Called after each expression is evaluated.

    Args:
        event: Data about the evaluated expression.

    Returns:
        ``False`` to stop the search early, ``True`` or ``None`` to continue.
    """
    return None

on_best_expression

on_best_expression(event: BestExpressionFound) -> Optional[bool]

Called when a new best expression is found.

Parameters:

Name	Type	Description	Default
event	BestExpressionFound	Data about the new best expression.	required

Returns:

Type	Description
Optional[bool]	False to stop the search early, True or None to continue.

Source code in SRToolkit/evaluation/callbacks.py

def on_best_expression(self, event: BestExpressionFound) -> Optional[bool]:
    """
    Called when a new best expression is found.

    Args:
        event: Data about the new best expression.

    Returns:
        ``False`` to stop the search early, ``True`` or ``None`` to continue.
    """
    return None

on_experiment_start

on_experiment_start(event: ExperimentEvent) -> None

Called before an experiment starts.

Parameters:

Name	Type	Description	Default
event	ExperimentEvent	Data about the experiment that is about to begin.	required

Source code in SRToolkit/evaluation/callbacks.py

def on_experiment_start(self, event: ExperimentEvent) -> None:
    """
    Called before an experiment starts.

    Args:
        event: Data about the experiment that is about to begin.
    """
    pass

on_experiment_end

on_experiment_end(event: ExperimentEvent, results: EvalResult) -> None

Called after an experiment completes.

Parameters:

Name	Type	Description	Default
event	ExperimentEvent	Data about the experiment that just ended.	required
results	EvalResult	Final EvalResult for this experiment.	required

Source code in SRToolkit/evaluation/callbacks.py

def on_experiment_end(self, event: ExperimentEvent, results: EvalResult) -> None:
    """
    Called after an experiment completes.

    Args:
        event: Data about the experiment that just ended.
        results: Final [EvalResult][SRToolkit.utils.types.EvalResult] for this experiment.
    """
    pass

to_dict

to_dict() -> dict

Serialise this callback to a JSON-safe dictionary.

The default implementation stores only the fully-qualified class path. Override in subclasses to include constructor parameters so that from_dict can reconstruct a functionally identical instance.

Returns:

Type	Description
dict	A JSON-safe dict with at least a "callback_class" key.

Source code in SRToolkit/evaluation/callbacks.py

def to_dict(self) -> dict:
    """
    Serialise this callback to a JSON-safe dictionary.

    The default implementation stores only the fully-qualified class path.
    Override in subclasses to include constructor parameters so that
    [from_dict][SRToolkit.evaluation.callbacks.SRCallbacks.from_dict] can
    reconstruct a functionally identical instance.

    Returns:
        A JSON-safe dict with at least a ``"callback_class"`` key.
    """
    return {"callback_class": f"{self.__class__.__module__}.{self.__class__.__qualname__}"}

from_dict classmethod

from_dict(d: dict) -> SRCallbacks

Reconstruct a callback from a serialised dictionary.

The default implementation calls cls() with no arguments. Override in subclasses that require constructor parameters.

Parameters:

Name	Type	Description	Default
d	dict	Dictionary produced by to_dict.	required

Returns:

Type	Description
SRCallbacks	A new instance of this callback class.

Source code in SRToolkit/evaluation/callbacks.py

@classmethod
def from_dict(cls, d: dict) -> "SRCallbacks":
    """
    Reconstruct a callback from a serialised dictionary.

    The default implementation calls ``cls()`` with no arguments. Override in
    subclasses that require constructor parameters.

    Args:
        d: Dictionary produced by
            [to_dict][SRToolkit.evaluation.callbacks.SRCallbacks.to_dict].

    Returns:
        A new instance of this callback class.
    """
    return cls()

CallbackDispatcher

CallbackDispatcher(callbacks: Optional[List[SRCallbacks]] = None)

Manages multiple SRCallbacks instances and dispatches events to all of them.

Examples:

>>> dispatcher = CallbackDispatcher()
>>> dispatcher.add(EarlyStoppingCallback(threshold=1e-6))
>>> len(dispatcher._callbacks)
1

Parameters:

Name	Type	Description	Default
callbacks	Optional[List[SRCallbacks]]	Initial list of callbacks. Defaults to an empty list.	None

Source code in SRToolkit/evaluation/callbacks.py

def __init__(self, callbacks: Optional[List[SRCallbacks]] = None):
    """
    Args:
        callbacks: Initial list of callbacks. Defaults to an empty list.
    """
    if callbacks is None:
        self._callbacks: List[SRCallbacks] = []
    else:
        self._callbacks = callbacks

get_callbacks

get_callbacks() -> List[SRCallbacks]

Returns the list of callbacks.

Returns:

Type	Description
List[SRCallbacks]	A list of SRCallbacks instances in this dispatcher.

Source code in SRToolkit/evaluation/callbacks.py

def get_callbacks(self) -> List[SRCallbacks]:
    """
    Returns the list of callbacks.

    Returns:
        A list of [SRCallbacks][SRToolkit.evaluation.callbacks.SRCallbacks] instances in this dispatcher.
    """
    return self._callbacks

add

add(callback: SRCallbacks) -> None

Add a callback to the dispatcher.

Parameters:

Name	Type	Description	Default
callback	SRCallbacks	The SRCallbacks instance to add.	required

Source code in SRToolkit/evaluation/callbacks.py

def add(self, callback: SRCallbacks) -> None:
    """
    Add a callback to the dispatcher.

    Args:
        callback: The [SRCallbacks][SRToolkit.evaluation.callbacks.SRCallbacks] instance to add.
    """
    self._callbacks.append(callback)

remove

remove(callback: SRCallbacks) -> None

Remove a callback from the dispatcher.

Parameters:

Name	Type	Description	Default
callback	SRCallbacks	The SRCallbacks instance to remove.	required

Raises:

Type	Description
ValueError	If callback is not currently registered.

Source code in SRToolkit/evaluation/callbacks.py

def remove(self, callback: SRCallbacks) -> None:
    """
    Remove a callback from the dispatcher.

    Args:
        callback: The [SRCallbacks][SRToolkit.evaluation.callbacks.SRCallbacks] instance to remove.

    Raises:
        ValueError: If ``callback`` is not currently registered.
    """
    self._callbacks.remove(callback)

on_expr_evaluated

on_expr_evaluated(event: ExprEvaluated) -> bool

Dispatch to all callbacks and aggregate the stop signal.

Parameters:

Name	Type	Description	Default
event	ExprEvaluated	Data about the evaluated expression.	required

Returns:

Type	Description
bool	False if any callback returned False (requesting early stop), True otherwise.

Source code in SRToolkit/evaluation/callbacks.py

def on_expr_evaluated(self, event: ExprEvaluated) -> bool:
    """
    Dispatch to all callbacks and aggregate the stop signal.

    Args:
        event: Data about the evaluated expression.

    Returns:
        ``False`` if any callback returned ``False`` (requesting early stop), ``True`` otherwise.
    """
    should_continue = True
    for cb in self._callbacks:
        cont = cb.on_expr_evaluated(event)
        if isinstance(cont, bool) and not cont:
            should_continue = False
    return should_continue

on_best_expression

on_best_expression(event: BestExpressionFound) -> bool

Dispatch to all callbacks and aggregate the stop signal.

Parameters:

Name	Type	Description	Default
event	BestExpressionFound	Data about the new best expression.	required

Returns:

Type	Description
bool	False if any callback returned False (requesting early stop), True otherwise.

Source code in SRToolkit/evaluation/callbacks.py

def on_best_expression(self, event: BestExpressionFound) -> bool:
    """
    Dispatch to all callbacks and aggregate the stop signal.

    Args:
        event: Data about the new best expression.

    Returns:
        ``False`` if any callback returned ``False`` (requesting early stop), ``True`` otherwise.
    """
    should_continue = True
    for cb in self._callbacks:
        cont = cb.on_best_expression(event)
        if isinstance(cont, bool) and not cont:
            should_continue = False
    return should_continue

on_experiment_start

on_experiment_start(event: ExperimentEvent) -> None

Dispatch to all callbacks.

Parameters:

Name	Type	Description	Default
event	ExperimentEvent	Data about the experiment that is about to begin.	required

Source code in SRToolkit/evaluation/callbacks.py

def on_experiment_start(self, event: ExperimentEvent) -> None:
    """
    Dispatch to all callbacks.

    Args:
        event: Data about the experiment that is about to begin.
    """
    for cb in self._callbacks:
        cb.on_experiment_start(event)

on_experiment_end

on_experiment_end(event: ExperimentEvent, results: EvalResult) -> None

Dispatch to all callbacks.

Parameters:

Name	Type	Description	Default
event	ExperimentEvent	Data about the experiment that just ended.	required
results	EvalResult	Final EvalResult for this experiment.	required

Source code in SRToolkit/evaluation/callbacks.py

def on_experiment_end(self, event: ExperimentEvent, results: EvalResult) -> None:
    """
    Dispatch to all callbacks.

    Args:
        event: Data about the experiment that just ended.
        results: Final [EvalResult][SRToolkit.utils.types.EvalResult] for this experiment.
    """
    for cb in self._callbacks:
        cb.on_experiment_end(event, results)

ProgressBarCallback

ProgressBarCallback(desc: Optional[str] = None)

Bases: SRCallbacks

Displays a tqdm progress bar that updates after each expression evaluation.

Examples:

>>> cb = ProgressBarCallback(desc="My search")
>>> cb.desc
'My search'

Parameters:

Name	Type	Description	Default
desc	Optional[str]	Description label shown on the progress bar. If None, the label is auto-generated as "<approach> on <dataset>" when the experiment starts.	None

Source code in SRToolkit/evaluation/callbacks.py

def __init__(self, desc: Optional[str] = None):
    """
    Args:
        desc: Description label shown on the progress bar. If ``None``, the label
            is auto-generated as ``"<approach> on <dataset>"`` when the experiment starts.
    """
    self.pbar = None
    self.desc = desc

EarlyStoppingCallback

EarlyStoppingCallback(threshold: Optional[float], max_evaluations: Optional[int] = None)

Bases: SRCallbacks

Stops the search when the best expression error falls below a threshold.

Examples:

>>> cb = EarlyStoppingCallback(threshold=1e-6)
>>> cb.on_best_expression(BestExpressionFound("", "X_0", 1e-7, 42))
False
>>> cb.on_best_expression(BestExpressionFound("", "X_0", 1e-5, 43))
True

Parameters:

Name	Type	Description	Default
threshold	Optional[float]	Error value below which the search is stopped.	required

Source code in SRToolkit/evaluation/callbacks.py

def __init__(self, threshold: Optional[float], max_evaluations: Optional[int] = None):
    """
    Args:
        threshold: Error value below which the search is stopped.
    """
    self.threshold = threshold
    self.max_evaluations = max_evaluations

LoggingCallback

LoggingCallback(log_file: Optional[str] = None)

Bases: SRCallbacks

Logs each new best expression to stdout or a file.

log_file may contain placeholders that are resolved at experiment start using fields from ExperimentEvent. Available placeholders: {dataset_name},{approach_name}, {seed}. Using per-experiment placeholders (e.g. {seed}) gives each job its own file, which is the recommended approach for parallel execution.

When multiple jobs share the same resolved file path, writes are protected by fcntl.flock (POSIX advisory locking) so concurrent processes on Linux / macOS do not corrupt each other's output. On Windows or network filesystems where flock is unavailable the lock is silently skipped.

Examples:

>>> cb = LoggingCallback()
>>> cb.on_best_expression(BestExpressionFound("Nguyen-1_ProGED_42", "X_0+C", 0.001, 10))
[Experiment Nguyen-1_ProGED_42] New best: X_0+C (error=1.000000e-03)
>>> cb = LoggingCallback(log_file="logs/{dataset_name}_{seed}.log")
>>> cb.on_experiment_start(ExperimentEvent(dataset_name="test", max_evaluations=10, seed=1,
...                                        success_threshold=0, approach_name="ta"))
>>> cb._resolved_log_file
'logs/test_1.log'

Parameters:

Name	Type	Description	Default
log_file	Optional[str]	Destination for log messages. If None, messages are printed to stdout. May be a plain path or a template string with placeholders {dataset_name}, {approach_name}, {seed} that are resolved when the experiment starts.	None

Source code in SRToolkit/evaluation/callbacks.py

def __init__(self, log_file: Optional[str] = None):
    """
    Args:
        log_file: Destination for log messages.  If ``None``, messages are
            printed to stdout.  May be a plain path or a template string with
            placeholders ``{dataset_name}``, ``{approach_name}``, ``{seed}``
            that are resolved when the experiment starts.
    """
    self.log_file = log_file
    self._resolved_log_file: Optional[str] = log_file