Event Handlers

ropt.plugins.event_handler.default

This module provides the default plugin implementations for event handlers.

Supported handlers:

• tracker: Tracks the 'best' or 'last' valid result based on objective value and constraints (DefaultTrackerHandler).
• store: Accumulates all results from specified sources (DefaultStoreHandler).
• observer: Listens for events from specified sources, and calls a callback for each event (DefaultObserverHandler).

ropt.plugins.event_handler.default.DefaultEventHandlerPlugin

Bases: EventHandlerPlugin

The default plugin for creating built-in event handlers.

This plugin acts as a factory for the standard EventHandler implementations provided by ropt.

Supported Handlers:

• tracker: Creates a DefaultTrackerHandler instance, which tracks either the 'best' or 'last' valid result based on objective value and constraints.
• store: Creates a DefaultStoreHandler instance, which accumulates all results received from specified sources.
• observer: Creates a DefaultObserverHandler instance, which calls a callback for each event received from specified sources.

ropt.plugins.event_handler._tracker.DefaultTrackerHandler

Bases: EventHandler

The default event handler for tracking optimization results.

This event handler listens for FINISHED_EVALUATION events emitted from within an optimization workflow. It processes the Results objects contained within these events and selects a single FunctionResults object to retain based on defined criteria.

The criteria for selection are:

• what='best' (default): Tracks the result with the lowest weighted objective value encountered so far.
• what='last': Tracks the most recently received valid result.

Optionally, results can be filtered based on constraint violations using the constraint_tolerance parameter. If provided, any result violating constraints beyond this tolerance is ignored.

The selected result (in the optimizer domain) is stored internally. The result accessible via dictionary access (handler["results"]) is the selected result, potentially transformed to the user domain.

event_types property

event_types: set[EventType]

Return the event types that are handled.

Returns:

Type	Description
set[EventType]	A set of event types that are handled.

__init__

__init__(
    *,
    what: Literal["best", "last"] = "best",
    constraint_tolerance: float | None = None,
) -> None

Initialize a default tracker event handler.

This event handler monitors Results objects and selects a single FunctionResults object to retain based on the what criterion ('best' or 'last').

The 'best' result is the one with the lowest weighted objective value encountered so far. The 'last' result is the most recently received valid result. Results can optionally be filtered by constraint_tolerance to ignore those violating constraints beyond the specified threshold.

Tracking logic (comparing 'best' or selecting 'last') operates on the results in the optimizer's domain. However, the final selected result that is made accessible via dictionary access (handler["results"]) is transformed to the user's domain.

Parameters:

Name	Type	Description	Default
what	Literal['best', 'last']	Criterion for selecting results ('best' or 'last').	'best'
constraint_tolerance	float | None	Optional threshold for filtering constraint violations.	None

handle_event

handle_event(event: Event) -> None

Handle incoming events.

This method processes incoming FINISHED_EVALUATION events.

If a relevant event containing results is received, this method updates the tracked result (self["results"]) based on the what criterion ('best' or 'last') and the optional constraint_tolerance.

Parameters:

Name	Type	Description	Default
event	Event	The event object.	required

ropt.plugins.event_handler._store.DefaultStoreHandler

Bases: EventHandler

The default event handler for storing optimization results.

This event handler listens for FINISHED_EVALUATION events emitted by specified compute steps from within an optimization workflow. It collects all Results objects contained within these events and stores them sequentially in memory.

The accumulated results are stored as a tuple and can be accessed via dictionary access using the key "results" (e.g., handler["results"]). Each time new results are received from a valid source, they are appended to this tuple.

event_types property

event_types: set[EventType]

Return the event types that are handled.

Returns:

Type	Description
set[EventType]	A set of event types that are handled.

__init__

__init__() -> None

Initialize a default store event handler.

This event handler collects and stores all Results objects it receives. It listens for FINISHED_EVALUATION events and appends the results contained within them to an internal tuple.

The results are converted from the optimizer domain to the user domain before being stored. The accumulated results are stored as a tuple and can be accessed via dictionary access using the key "results" (e.g., handler["results"]). Initially, handler["results"] is None.

handle_event

handle_event(event: Event) -> None

Handle incoming events.

This method processes events it receives. It specifically listens for FINISHED_EVALUATION events.

If a relevant event containing results is received, this method retrieves the results, optionally transforms them to the user domain and appends them to the tuple stored in self["results"].

Parameters:

Name	Type	Description	Default
event	Event	The event object.	required

ropt.plugins.event_handler._observer.DefaultObserverHandler

Bases: EventHandler

The default event handler for observing events.

This event handler listens for events and forwards them to one or more callback functions.

event_types property

event_types: set[EventType]

Return the event types that are handled.

Returns:

Type	Description
set[EventType]	A set of event types that are handled.

__init__

__init__(
    *,
    event_types: set[EventType],
    callback: Callable[[Event], None],
) -> None

Initialize a default event handler.

This event handler responds to events by calling callback if the event type matches event_types.

Parameters:

Name	Type	Description	Default
event_types	set[EventType]	The set of event types to respond to.	required
callback	Callable[[Event], None]	The callable to call.	required

handle_event

handle_event(event: Event) -> None

Handle incoming events.

This method processes events emitted from within the workflow.

If a event containing results is received, and its type equals the stored event type, the stored callback is called.

Parameters:

Name	Type	Description	Default
event	Event	The event object emitted from the workflow.	required