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.

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.

init

__init__() -> None

Initialize the EventHandler.

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.

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.

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: 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

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.

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.

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: 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

ropt.workflow.event_handlers.Table

Bases: EventHandler

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

Tables

The functions and gradients inputs are dictionaries where each item denotes a table to generate, for FunctionsResults and GradientsResults respectively. Tables are accessed by their name in attributes, i.e. if the functions input has an entry evaluations, this table can be accessed via handler["evaluations"]. Note that the 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.

By default the functions and gradients are set to None directing the handler to generate a set of default tables:

For the functions input:
- "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 the gradients input:
- "gradients": contains a set of values of the calculated gradients.
- "perturbations": contains a set of values for all perturbations.

To change (or disable) the generation of these table, pass an appropriate (or empty) dictionary specifying the desired tables.

Table specification

Each value of each entry in the functions and gradients dictionaries is another dictionary that determines which fields in the results should be stored in the corresponding table. The keys denote the names of the fields, using attribute syntax. For instance a result.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 functions input generates a table containing the batch id, the values of all calculated objectives and the vector of variables.

{
    "functions": {
        "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
...

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 via the callback input. This callback will called anytime the tables are update, passing the handler as its only argument.

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.

init

__init__(
    *,
    functions: dict[str, dict[str, str]] | None = None,
    gradients: dict[str, dict[str, str]] | None = None,
    sep: str = ",",
    callback: Callable[[EventHandler], None] | None = None,
) -> None

Initialize a default table event handler.

Parameters:

Name	Type	Description	Default
`functions`	`dict[str, dict[str, str]] \| None`	Dictionary of tables with function results.	`None`
`gradients`	`dict[str, dict[str, str]] \| None`	Dictionary of tables with gradient results.	`None`
`sep`	`str`	Separator used in column names.	`','`
`callback`	`Callable[[EventHandler], None] \| None`	An optional callback that is called when the tables are update.	`None`

handle_event

handle_event(event: EnOptEvent) -> None

Handle incoming events.

Parameters:

Name	Type	Description	Default
`event`	`EnOptEvent`	The event object.	required

getitem

__getitem__(key: str) -> Any

Retrieve a of a table from the event handler.

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.

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.

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 Handlers

ropt.workflow.event_handlers.EventHandler

event_types abstractmethod property

__init__

handle_event abstractmethod

__getitem__

__setitem__

ropt.workflow.event_handlers.Tracker

event_types property

__init__

handle_event

ropt.workflow.event_handlers.Store

event_types property

__init__

handle_event

ropt.workflow.event_handlers.Table

event_types property

__init__

handle_event

__getitem__

add_column

ropt.workflow.event_handlers.Observer

event_types property

__init__

handle_event

event_types `abstractmethod` `property`

init

handle_event `abstractmethod`

getitem

setitem

event_types `property`

init

event_types `property`

init

event_types `property`

init

getitem

event_types `property`

init