Events¶

The event emitter and event payloads.

events ¶

Event system for ALchemist Core.

Provides a simple event emitter pattern for progress reporting and status updates without requiring UI dependencies.

EventEmitter ¶

EventEmitter()

Simple event emitter for progress and status updates.

Allows components to emit events that can be subscribed to by UI or other systems without creating tight coupling.

Example

emitter = EventEmitter()

Subscribe to an event¶

def on_progress(data): ... print(f"Progress: {data['current']}/{data['total']}")

emitter.on("progress", on_progress)

Emit an event¶

emitter.emit("progress", {"current": 5, "total": 10}) Progress: 5/10

Unsubscribe¶

emitter.off("progress", on_progress)

Initialize the event emitter with no listeners.

on ¶

on(event_name: str, callback: Callable[[Dict[str, Any]], None]) -> None

Subscribe to an event.

Parameters:

Name	Type	Description	Default
`event_name`	`str`	Name of the event to listen for	required
`callback`	`Callable[[Dict[str, Any]], None]`	Function to call when event is emitted. Should accept a dict of event data.	required

Example

def my_handler(data): ... print(f"Received: {data}") emitter.on("my_event", my_handler)

off ¶

off(event_name: str, callback: Callable[[Dict[str, Any]], None]) -> None

Unsubscribe from an event.

Parameters:

Name	Type	Description	Default
`event_name`	`str`	Name of the event to stop listening for	required
`callback`	`Callable[[Dict[str, Any]], None]`	The callback function to remove	required

Example

emitter.off("my_event", my_handler)

once ¶

once(event_name: str, callback: Callable[[Dict[str, Any]], None]) -> None

Subscribe to an event but only fire once, then auto-unsubscribe.

Parameters:

Name	Type	Description	Default
`event_name`	`str`	Name of the event to listen for	required
`callback`	`Callable[[Dict[str, Any]], None]`	Function to call when event is emitted (only once)	required

Example

emitter.once("complete", lambda data: print("Done!"))

emit ¶

emit(event_name: str, data: Dict[str, Any] = None) -> None

Emit an event to all subscribers.

Parameters:

Name	Type	Description	Default
`event_name`	`str`	Name of the event to emit	required
`data`	`Dict[str, Any]`	Optional dictionary of data to pass to listeners	`None`

Example

emitter.emit("progress", {"current": 1, "total": 10}) emitter.emit("error", {"message": "Something went wrong"})

remove_all_listeners ¶

remove_all_listeners(event_name: str = None) -> None

Remove all listeners for a specific event, or all events if no name given.

Parameters:

Name	Type	Description	Default
`event_name`	`str`	Optional event name. If None, removes all listeners for all events.	`None`

Example

emitter.remove_all_listeners("progress") # Remove all progress listeners emitter.remove_all_listeners() # Remove all listeners

listener_count ¶

listener_count(event_name: str) -> int

Get the number of listeners for a specific event.

Parameters:

Name	Type	Description	Default
`event_name`	`str`	Name of the event	required

Returns:

Type	Description
`int`	Number of listeners registered for this event

Example

count = emitter.listener_count("progress")

event_names ¶

event_names() -> List[str]

Get list of all event names that have listeners.

Returns:

Type	Description
`List[str]`	List of event names

Example

events = emitter.event_names() print(events) # ['progress', 'complete', 'error']

ProgressEventData ¶

Standard schema for progress events.

Fields

current (int): Current step/iteration total (int): Total steps/iterations message (str): Optional progress message percentage (float): Optional percentage complete (0-100)

ErrorEventData ¶

Standard schema for error events.

Fields

message (str): Error message exception (Exception): Optional exception object recoverable (bool): Whether the error is recoverable

ModelTrainedEventData ¶

Standard schema for model trained events.

Fields

backend (str): Backend used (sklearn, botorch, etc.) hyperparameters (dict): Model hyperparameters training_time (float): Training time in seconds

NextPointSuggestedEventData ¶

Standard schema for next point suggested events.

Fields

point (dict or DataFrame): Suggested experiment point(s) strategy (str): Strategy used (acquisition, EMOC, etc.) acq_func (str): Acquisition function used acq_value (float): Acquisition function value at suggested point