Event System

EventEmitter

emit_sync(event_type, data)

Synchronous version of emit that properly awaits all callbacks. This is necessary because callbacks are fire-and-forget in async contexts, but in sync contexts wrapped with async_to_sync, the event loop may close before the tasks complete.

Event

add_callback(callback)

Add a callback to be executed after the event handler.

remove_callback(callback)

Remove a callback from the event.

EventRegistry

Bases: RegistryBase

Registry for managing events and their handlers. Allows registering event handlers and callbacks.

register_callback(event_type, callback)

Register a callback for an existing event.

register_event(event_type, handler, callbacks=None)

Register an event with its handler and optional callbacks.

unregister_callback(event_type, callback)

Unregister a callback for an existing event.

unregister_event(event_type)

Unregister an event by its type.

EventHandler

Bases: ABC

Abstract class for handling events. Each event handler should implement this class.

get_default_callbacks()

Return the default callbacks for this event handler. This method can be overridden by subclasses to provide custom callbacks.

RequestResponseEventHandler

Bases: EventHandler

Abstract class for handlers that produce a response payload the emitting WebSocket consumer should forward back to the originating client.

Subclasses override response() to return either a single message dict or a list of message dicts. The base BaseWebsocketEventConsumer._send_handler_message invokes response() after the handler runs and serialises the result back over the WebSocket.

broadcasts = False so commands echo only to the caller — EventEmitter._auto_broadcast skips this class. Without this opt-out, any emit_event("command.system.shutdown", ...) would fan out to every WS subscriber on the command.system.shutdown group, which is the wrong behaviour for a command-and-response pattern (the response itself is what the caller wants, not a blanket broadcast).

response() async

Return a message or a list of messages to be sent back over the WebSocket to the originating client. Subclasses must implement this method.

return_message() async

DEPRECATED. Use :meth:response instead.

Kept as a thin alias so external code that subclassed the old WebsocketEventHandler and called return_message() still works symbolically. New code MUST override response — subclasses that override return_message will NOT be invoked by the core consumers, which now call response.

noop_handler(data) async

A no-operation handler that does nothing and returns an empty dictionary. This can be used as a placeholder for events that do not require handling.

EventException

Bases: Exception

Base class for all exceptions raised by the WhiteboxEvent system.

Broadcasting to WebSocket subscribers

The event system has two paths for getting an event onto a Squawk consumer's WebSocket:

Auto-broadcast (default)

emit_event invokes the registered handler, runs all callbacks, then fans out the handler's context out to all clients subscribed to that event.

class StatusUpdateHandler(EventHandler):
    async def handle(self, data):
        return {"data": data}
        # EventEmitter fans {"data": ..., "type": "observation.status.update"}
        # out to every WS subscriber on the "observation.status.update" group.

Handlers that want request/response semantics (respond only to the originating WS client) should extend RequestResponseEventHandler instead.

broadcast_event() (ad-hoc)

whitebox.events.broadcast_event(event_type, data) is a kernel-level helper for one-shot broadcasts that don't have a registered handler. It stores an EventLog row and fans the payload to the bare event-name Channels group. Useful when you want to publish an observation without a full handler lifecycle:

from whitebox.events import broadcast_event

await broadcast_event(
    "observation.device.connection_status.update",
    {"device_id": 42, "status": "connected"},
)

The plugin-side wrapper WhiteboxStandardAPI.broadcast_event is the canonical entry point for plugin code:

class MyPlugin(Plugin):
    async def report_status(self, payload):
        await self.whitebox.broadcast_event(
            "observation.device.connection_status.update",
            payload,
        )

Event name contract

In all cases, the event name must match a regex ^[a-zA-Z\d\-_.]+$, so event names are restricted to letters, digits, -, _, and ..