Event Handlers

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.

init

__init__() -> None

Initialize the EventHandler.

event_types `abstractmethod` `property`

event_types: set[EnOptEventType]

Return 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.Tracker

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.

init

__init__(
    *,
    what: Literal["best", "last"] = "best",
    constraint_tolerance: float | None = None,
    domain: DomainType = "user",
) -> 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.

If the domain type is "user", the result is converted from the optimizer domain to the user domain before being stored.

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`	The domain in which to store the results ('user' or 'optimizer').	`'user'`

handle_event

handle_event(event: EnOptEvent) -> 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`	`EnOptEvent`	The event object.	required

event_types `property`

event_types: set[EnOptEventType]

Return the event types that are handled.

Returns:

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

ropt.workflow.event_handlers.Store

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.

init

__init__(*, domain: DomainType = 'user') -> 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.

If the domain type is "user", 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: EnOptEvent) -> 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`	`EnOptEvent`	The event object.	required

event_types `property`

event_types: set[EnOptEventType]

Return the event types that are handled.

Returns:

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

ropt.workflow.event_handlers.Table

Bases: EventHandler

This event handler tracks results and stores them in pandas DataFrames.

Tables

Tables can be generated for FunctionsResults and GradientsResults respectively. Tables are added via the add_table method, which takes a name, a type (either "functions" or "gradients"), a column specification and an optional domain type. The column specification determines which fields of the results are stored in the table and how they are named. The domain type determines whether the results are transformed to the user domain before being stored in the table.

Tables are accessed by their name in attributes, for example, as handler["evaluations"].

Warning

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

Column specification

Columns are specified by providing a dictionary that maps field names to column titles. The keys denote the names of the fields, using attribute syntax. For instance a function.objectives key indicates the the result should contain a column with objective values that are found in the objectives field of the function field of the result. The values corresponding to the keys are used to provide the column names.

For example, passing this dictionary via the columns argument generates a table containing the batch id, the values of all calculated objectives and the vector of variables.

{
    "batch_id": "Batch",
    "functions.objectives": "Objective",
    "evaluations.variables": "Variables",
}

Some fields may result in multiple columns in the DataFrame if their values are vectors or matrices. For example, evaluations.variables will generate a separate column for each variable. The table specification above may generate a pandas dataframe looking something like this:

    Batch   Objective,0  Variables,v0  Variables,v1  Variables,v2
0       0  1.309826e+02      0.500000      0.900000      1.300000
1       0  4.362553e+12    120.900265     20.698539    -90.578972
...

Here, because the variables are vectors of length 2, there are two variable columns generated. The corresponding column names consist of the column title and the name of the variable vector, separated by a comma. Note that the function.objectives column also contains a comma followed by a 0 value. This is because the functions.objectives is also a vector of values, there just happens to be only one objective. Its index is used instead of a name, because no name was provided in the configuration of the optimization. Fields may even have matrix values, in which case the column names may be contain two item names or indices separated by commas.

Changing the column name separator.

By default a comma is used to separate fields in the column names if needed. The sep input can be used to provide an alternative separator.

You can exploit this by specifying a newline as the separator and display a nicely formatted table using the tabulate package:

from tabulate import tabulate

print(tabulate(table["functions"], headers="keys", showindex=False))

which will show something like this using multi-line headers:

  Batch         Objective    Variables    Variables     Variables
                        0           v0           v1            v2
-------  ----------------  -----------  -----------  ------------
      0           130.983          0.5          0.9           1.3
...

Default tables

The set_default_tables method can be used to add a set of default tables:

For functions results it generates these tables:
- "functions": contains a set of values of the calculated functions.
- "evaluations": contains a set of values for all evaluations.
- "constraints": contains a set of values for all constraints.
For gradients results it generates these tables:
- "gradients": contains a set of values of the calculated gradients.
- "perturbations": contains a set of values for all perturbations.

Callback functionality

The tables are updated anytime a result is processed. To be able to do something with the tables each time they are updated a callback can be provided set using set_callback. This callback will called anytime the tables are updated, passing the event that caused the tables to be updated.

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.

This callback will called anytime the tables are updated, passing the event that caused the tables to be updated.

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]

Return 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.Observer

Bases: EventHandler

The default event handler for observing events.

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

init

__init__(
    *,
    event_types: set[EnOptEventType],
    callback: Callable[[EnOptEvent], 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[EnOptEventType]`	The set of event types to respond to.	required
`callback`	`Callable[[EnOptEvent], None]`	The callable to call.	required

handle_event

handle_event(event: EnOptEvent) -> 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`	`EnOptEvent`	The event object emitted from the workflow.	required

event_types `property`

event_types: set[EnOptEventType]

Return the event types that are handled.

Returns:

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

Event Handlers

ropt.workflow.event_handlers.EventHandler

__init__

event_types abstractmethod property

handle_event abstractmethod

__getitem__

__setitem__

ropt.workflow.event_handlers.Tracker

__init__

handle_event

event_types property

ropt.workflow.event_handlers.Store

__init__

handle_event

event_types property

ropt.workflow.event_handlers.Table

__init__

set_default_tables

set_callback

add_table

get_tables

handle_event

event_types property

__getitem__

add_column

ropt.workflow.event_handlers.Observer

__init__

handle_event

event_types property

init

event_types `abstractmethod` `property`

handle_event `abstractmethod`

getitem

setitem

init

event_types `property`

init

event_types `property`

init

event_types `property`

getitem

init

event_types `property`