Skip to content

basic_bot.commons.messages

Functions from this module are used to send messages to the central_hub via websockets.

MessageTypeIn Objects

class MessageTypeIn(Enum)

Incoming message types (client → central_hub).

MessageTypeOut Objects

class MessageTypeOut(Enum)

Outgoing message types (central_hub → client).

MessageType Objects

class MessageType(Enum)

Message types for communication to/from central_hub.

WebSocketProtocol Objects

@runtime_checkable
class WebSocketProtocol(Protocol)

Protocol for websocket objects.

BaseMessage Objects

@dataclass
class BaseMessage()

Base message structure for websocket communication.

IdentityMessage Objects

@dataclass
class IdentityMessage(BaseMessage)

Identity message for service registration.

StateUpdateMessage Objects

@dataclass
class StateUpdateMessage(BaseMessage)

State update message for publishing state changes.

SubscribeMessage Objects

@dataclass
class SubscribeMessage(BaseMessage)

Subscribe message for state key subscriptions.

GetStateMessage Objects

@dataclass
class GetStateMessage(BaseMessage)

Get state message for requesting current state.

send_message

async def send_message(websocket: Any, message: Union[BaseMessage,
                                                      Dict[str, Any]]) -> None

Send a message to central_hub.

send_identity

async def send_identity(websocket: Any, name: str) -> None

Send the identity type message to central_hub. Name should uniquely identify the service.

send_subscribe

async def send_subscribe(
        websocket: Any, subscriptionNames: Union[List[str],
                                                 Literal["*"]]) -> None

Send the subscribeState message type to central_hub.

subscriptionNames should be an array of keys to subscribe or "*" to subscribe to all keys.

send_get_state

async def send_get_state(websocket: Any,
                         keys: Optional[List[str]] = None) -> None

Send the getState message type to central_hub optionally specifying a list of keys to get the state.

send_update_state

async def send_update_state(websocket: Any, stateData: Dict[str, Any]) -> None

Send the updateState message type to central_hub with the key->value state data to update.