Event Handlers

Event handlers attach to a ComputeStep and react to the events it emits. ResultsHandler keeps the best (or last) result, HistoryHandler keeps everything, TableHandler writes a structured table, CallbackHandler forwards selected events to a user callback, and EventForwardHandler forwards events to an EventServer for lock-free dispatch.

See Optimization Workflows and Working with Results for usage.

ropt.workflow.event_handlers.EventHandler

Bases: ABC

Abstract base class for event handlers.

This class defines the fundamental interface for all event handlers within an optimization workflow. Concrete handler implementations, (e.g., tracking results, storing data, logging), must inherit from this base class.

Handlers may store state using dictionary-like access ([]), allowing them to accumulate information or make data available to other components in an optimization workflow.

Subclasses must implement the abstract handle_event method to define their specific event processing logic.

Event handlers are attached to a ComputeStep using its add_event_handler method. When the compute step emits an event, the handle_event method of each attached handler is invoked, allowing it to process the event.

Warning

Event handler instances must not be called concurrently from multiple threads. Do not attach the same handler instance to compute steps that run in parallel.

__init__

__init__() -> None

Initialize the EventHandler.

event_types abstractmethod property

event_types: set[EnOptEventType]

The event types that are handled.

Returns:

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

handle_event abstractmethod

handle_event(event: EnOptEvent) -> None

Process an event.

This abstract method must be implemented by concrete EventHandler subclasses. It defines the event handler's core logic for reacting to EnOptEvent objects emitted in the optimization workflow.

Implementations should inspect the event object (its event_type and data) and perform computations accordingly, such as storing results, logging information, or updating internal state.

Parameters:

Name	Type	Description	Default
event	EnOptEvent	The event object.	required

__getitem__

__getitem__(key: str) -> Any

Retrieve a value from the event handler's internal state.

This method enables dictionary-like access (handler[key]) to the values stored within the event handler's internal state dictionary. This allows handlers to store and retrieve data accumulated during workflow execution.

Parameters:

Name	Type	Description	Default
key	str	The string key identifying the value to retrieve.	required

Returns:

Type	Description
Any	The value associated with the specified key.

Raises:

Type	Description
AttributeError	If the provided key does not exist in the event handler's stored values.

__setitem__

__setitem__(key: str, value: Any) -> None

Store or update a value in the event handler's internal state.

This method enables dictionary-like assignment (handler[key] = value) to store arbitrary data within the event handler's internal state dictionary. This allows event handlers to accumulate information or make data available to other components of the workflow.

The key must be a valid Python identifier.

Parameters:

Name	Type	Description	Default
key	str	The string key identifying the value to store (must be an identifier).	required
value	Any	The value to associate with the key.	required

Raises:

Type	Description
AttributeError	If the provided key is not a valid identifier.

ropt.workflow.event_handlers.ResultsHandler

Bases: EventHandler

Track a single optimization result based on selection criteria.

Listens for FINISHED_EVALUATION events and retains either the best (lowest weighted objective) or most recent valid result. Optionally filters by constraint tolerance.

See Optimization Workflows for full details on selection criteria and domain handling.

__init__

__init__(
    *,
    what: Literal["best", "last"] = "best",
    constraint_tolerance: float | None = None,
    domain: DomainType = "user",
    filter: Callable[[Results], bool] | None = None,
) -> None

Initialize the ResultsHandler.

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
domain	DomainType	Domain in which to store the results ('user' or 'optimizer').	'user'
filter	Callable[[Results], bool] | None	Optional callable to filter results based on custom logic.	None

event_types property

event_types: set[EnOptEventType]

The event types that are handled.

Returns:

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

ropt.workflow.event_handlers.HistoryHandler

Bases: EventHandler

Collect all optimization results into a tuple.

Listens for FINISHED_EVALUATION events and appends every Results object to a growing tuple accessible via handler["results"].

See Optimization Workflows for full details on domain handling and accumulation behavior.

__init__

__init__(*, domain: DomainType = 'user') -> None

Initialize the HistoryHandler.

Parameters:

Name	Type	Description	Default
domain	DomainType	Domain in which to store results ('user' or 'optimizer').	'user'

handle_event

handle_event(event: EnOptEvent) -> None

Handle incoming events.

Processes FINISHED_EVALUATION events, optionally transforms results to the user domain, and appends them to self["results"].

Parameters:

Name	Type	Description	Default
event	EnOptEvent	The event object.	required

event_types property

event_types: set[EnOptEventType]

The event types that are handled.

Returns:

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

ropt.workflow.event_handlers.TableHandler

Bases: EventHandler

Build pandas DataFrames from optimization results.

Collects FunctionResults and GradientResults into named tables. Tables are defined via add_table with a column specification, or registered in bulk with set_default_tables.

Access tables via dictionary syntax: handler["functions"].

Warning

Tables are generated on the fly from internal data when accessing them in this way. When multiple accesses are needed, it is more efficient to first store them in a variable.

See Optimization Workflows for full details on column specification format, default tables, and callback functionality.

__init__

__init__(*, sep: str = ',') -> None

Initialize a default table event handler.

Parameters:

Name	Type	Description	Default
sep	str	Separator used in column names.	','

set_default_tables

set_default_tables(*, domain: DomainType = 'user') -> None

Register a standard set of result tables.

Adds the default functions, evaluations, and constraints tables for function results, and the default gradients and perturbations tables for gradient results.

Parameters:

Name	Type	Description	Default
domain	DomainType	Domain ("user" or "optimizer") the tables are filled from. The "user" domain reports values as seen by the user; "optimizer" reports them in the optimizer's transformed space.	'user'

set_callback

set_callback(
    callback: Callable[[EnOptEvent], None],
) -> None

Set the callback function.

The callback is invoked from handle_event after the tables are updated, receiving the event that triggered the update. If the callback performs blocking operations (e.g. writing tables to disk), register this handler with run_in_thread=True on the EventServer:

event_server.add_event_handler(table_handler, run_in_thread=True)

Parameters:

Name	Type	Description	Default
callback	Callable[[EnOptEvent], None]	A function that is called when the tables are updated.	required

add_table

add_table(
    name: str,
    table_type: Literal["functions", "gradients"],
    columns: dict[str, str],
    domain: DomainType = "user",
) -> None

Register a new table to be populated from incoming results.

Parameters:

Name	Type	Description	Default
name	str	Key under which the table is stored and looked up.	required
table_type	Literal['functions', 'gradients']	Whether this table is filled from function results ("functions") or gradient results ("gradients").	required
columns	dict[str, str]	Mapping from result-field attribute names (using dotted attribute syntax) to display titles.	required
domain	DomainType	Domain ("user" or "optimizer") the table is filled from.	'user'

get_tables

get_tables() -> dict[str, pd.DataFrame]

Return the tables stored in the event handler.

Returns:

Type	Description
dict[str, DataFrame]	A dictionary mapping table names to their corresponding tables.

Warning

Tables are generated on the fly from internal data. When multiple access is needed, it is more efficient to first store them in a variable.

handle_event

handle_event(event: EnOptEvent) -> None

Handle incoming events.

Parameters:

Name	Type	Description	Default
event	EnOptEvent	The event object.	required

event_types property

event_types: set[EnOptEventType]

The event types that are handled.

Returns:

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

__getitem__

__getitem__(key: str) -> Any

Retrieve a of a table from the event handler.

Warning

The table is generated on the fly from internal data hen multiple access are needed, it is more efficient to first store them in a variable.

Parameters:

Name	Type	Description	Default
key	str	The string key identifying the table to retrieve.	required

Returns:

Type	Description
Any	The table associated with the specified key.

Raises:

Type	Description
AttributeError	If the requested table does not exist.

add_column

add_column(table: str, name: str, title: str) -> None

Add a column to a given table.

Parameters:

Name	Type	Description	Default
table	str	The name of the table to add the column to.	required
name	str	The name of the field to add as a column, using attribute syntax.	required
title	str	The title of the column to add.	required

ropt.workflow.event_handlers.CallbackHandler

Bases: EventHandler

The default event handler for observing events.

This event handler listens for events of matching types and forwards them to a callback function.

If the callback performs blocking operations (file I/O, network calls, etc.), register this handler with run_in_thread=True on the EventServer:

event_server.add_event_handler(handler, run_in_thread=True)

__init__

__init__(
    *,
    event_types: set[EnOptEventType],
    callback: Callable[[EnOptEvent], None],
) -> None

Initialize the CallbackHandler.

Parameters:

Name	Type	Description	Default
event_types	set[EnOptEventType]	The set of event types to respond to.	required
callback	Callable[[EnOptEvent], None]	The callable to invoke for matching events.	required

handle_event

handle_event(event: EnOptEvent) -> None

Handle incoming events.

Parameters:

Name	Type	Description	Default
event	EnOptEvent	The event object.	required

event_types property

event_types: set[EnOptEventType]

The event types that are handled.

Returns:

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

ropt.workflow.event_handlers.EventForwardHandler

Bases: EventHandler

Forwards events from a compute step to an EventServer.

See Optimization Workflows for usage.

__init__

__init__(
    server: EventServer, *, event_types: set[EnOptEventType]
) -> None

Initialize the EventForwardHandler.

Parameters:

Name	Type	Description	Default
server	EventServer	The EventServer to forward events to.	required
event_types	set[EnOptEventType]	The set of event types to forward.	required

handle_event

handle_event(event: EnOptEvent) -> None

Forward the event to the EventServer.

Parameters:

Name	Type	Description	Default
event	EnOptEvent	The event to forward.	required

event_types property

event_types: set[EnOptEventType]

The event types that are handled.