event bus

event_bus

Event-bus API for Femora lifecycle coordination.

This page documents the event objects that power Femora's advanced lifecycle hooks. A Model owns a per-model event bus exposed as model.events. Runtime subsystems such as the assembler emit FemoraEvent values at key stages, and subscribed callbacks receive payload data describing what just happened.

This mechanism is mainly useful for advanced extension work. Ordinary modeling flows usually create components directly through managers and do not need to subscribe to events.

Under the hood, the flow looks like this:

1. a runtime subsystem reaches a lifecycle stage such as POST_ASSEMBLE
2. it emits a FemoraEvent through the model-owned bus
3. subscribers receive keyword-argument payload data
4. advanced components inspect the payload and react

Example

from femora.core.model import Model
from femora.components.event.event_bus import FemoraEvent

model = Model()

def on_post_assemble(**payload):
    assembled_mesh = payload.get("assembled_mesh")
    if assembled_mesh is not None:
        print(f"assembled cells: {assembled_mesh.n_cells}")

model.events.subscribe(FemoraEvent.POST_ASSEMBLE, on_post_assemble)

Tip

Prefer the model-owned model.events bus for new code. The legacy EventBus wrapper is kept for compatibility with older code paths and tests.

Classes

FemoraEvent

Bases: Enum

Lifecycle signals emitted by Femora runtime subsystems.

These enum values describe when a callback is being invoked. They are not ordinary modeling commands. Instead, they mark important stages in the runtime lifecycle so advanced components can react at the right time.

The most important practical events are:

• PRE_ASSEMBLE: emitted before the final assembled mesh is built
• POST_ASSEMBLE: emitted after the final assembled mesh exists
• RESOLVE_CORE_CONFLICTS: emitted after assembly when partition/core ownership updates may need follow-up work
• PRE_EXPORT and POST_EXPORT: emitted around export-time workflows

More specialized events such as EMBEDDED_BEAM_SOLID_TCL are used by advanced exporters and interface-specific integrations.

ModelEventBus

ModelEventBus()

Per-model publish/subscribe bus owned by a Model.

A Model creates one ModelEventBus and exposes it through model.events. Managers and advanced components use that bus to coordinate lifecycle work without tightly coupling themselves to one another.

Callbacks are registered per FemoraEvent and are invoked with keyword-argument payload data supplied by the emitter.

Example

from femora.core.model import Model
from femora.components.event.event_bus import FemoraEvent

model = Model()

def on_post_assemble(**payload):
    assembled_mesh = payload.get("assembled_mesh")
    if assembled_mesh is not None:
        print(assembled_mesh.n_points)

model.events.subscribe(FemoraEvent.POST_ASSEMBLE, on_post_assemble)

Initialize an empty per-model subscriber registry.

Methods:

subscribe

subscribe(event: FemoraEvent, callback: Callable) -> None

Register a callback for one lifecycle event.

If the callback is already subscribed to the same event, it is not added a second time.

Parameters:

Name	Type	Description	Default
event	FemoraEvent	Lifecycle event to listen for.	required
callback	Callable	Callable invoked when event is emitted. The callback receives keyword arguments from the emitter.	required

unsubscribe

unsubscribe(event: FemoraEvent, callback: Callable) -> None

Remove a previously registered callback from one event.

Parameters:

Name	Type	Description	Default
event	FemoraEvent	Lifecycle event previously used during subscription.	required
callback	Callable	Callback to remove.	required

emit

emit(event: FemoraEvent, **payload) -> None

Dispatch one lifecycle event to its current subscribers.

The bus iterates over a snapshot of the subscriber list so callbacks can safely unsubscribe or adjust registrations during dispatch without corrupting the iteration order.

Parameters:

Name	Type	Description	Default
event	FemoraEvent	Lifecycle event being emitted.	required
**payload		Keyword arguments forwarded directly to each callback. Emitters commonly provide objects such as assembled_mesh.	{}

clear

clear() -> None

Remove all subscribers from this bus instance.

EventBus

Legacy process-global compatibility wrapper around ModelEventBus.

This class exists mainly to preserve older code paths and tests that still expect a process-global event bus. New code should prefer the model-owned ModelEventBus exposed through model.events.

Under the hood, this wrapper forwards operations to an internal ModelEventBus instance while preserving compatibility with older code that mutates _subscribers directly.

Methods:

subscribe classmethod

subscribe(event: FemoraEvent, callback: Callable) -> None

Register a callback through the legacy compatibility wrapper.

unsubscribe classmethod

unsubscribe(event: FemoraEvent, callback: Callable) -> None

Remove a callback through the legacy compatibility wrapper.

emit classmethod

emit(event: FemoraEvent, **payload) -> None

Emit one lifecycle event through the legacy compatibility wrapper.