bundle.website.builtin.components.websocket.base ================================================ .. py:module:: bundle.website.builtin.components.websocket.base Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/bundle/website/builtin/components/websocket/base/backend/index /autoapi/bundle/website/builtin/components/websocket/base/component/index /autoapi/bundle/website/builtin/components/websocket/base/gpx/index /autoapi/bundle/website/builtin/components/websocket/base/message_router/index /autoapi/bundle/website/builtin/components/websocket/base/messages/index Classes ------- .. autoapisummary:: bundle.website.builtin.components.websocket.base.WebSocketBaseComponent bundle.website.builtin.components.websocket.base.WebSocketComponentParams bundle.website.builtin.components.websocket.base.GPXComponentParams bundle.website.builtin.components.websocket.base.GPXWebSocketBaseComponent bundle.website.builtin.components.websocket.base.MessageRouter bundle.website.builtin.components.websocket.base.AckMessage bundle.website.builtin.components.websocket.base.ErrorMessage bundle.website.builtin.components.websocket.base.KeepAliveMessage Functions --------- .. autoapisummary:: bundle.website.builtin.components.websocket.base.drain_text bundle.website.builtin.components.websocket.base.every bundle.website.builtin.components.websocket.base.keepalive_loop bundle.website.builtin.components.websocket.base.receive_json bundle.website.builtin.components.websocket.base.run_websocket Package Contents ---------------- .. py:function:: drain_text(websocket: fastapi.WebSocket) -> None :async: Consume and discard incoming text frames until disconnect. .. py:function:: every(seconds: float, tick: TaskFactory) -> TaskFactory Return a task factory that runs `tick` periodically every `seconds`. .. py:function:: keepalive_loop(websocket: fastapi.WebSocket) -> None :async: Default keepalive protocol: `keepalive` -> `keepalive_ack`. .. py:function:: receive_json(handle: collections.abc.Callable[[fastapi.WebSocket, dict], collections.abc.Awaitable[None]]) -> TaskFactory Return a task factory that reads JSON objects and forwards them to `handle`. .. py:function:: run_websocket(websocket: fastapi.WebSocket, *task_factories: TaskFactory) -> None :async: Run one websocket session with composable async tasks. The function accepts the connection, starts all tasks, waits for completion, and performs robust cancellation cleanup when the socket closes. .. py:class:: WebSocketBaseComponent(/, **data: Any) Bases: :py:obj:`bundle.website.core.component.Component` Base websocket component with default params and shared assets. .. py:attribute:: shared_assets :type: ClassVar[tuple[str, Ellipsis]] :value: ('websocket/base/component.css',) .. py:attribute:: params :type: WebSocketComponentParams :value: None .. py:method:: shared_asset_paths() -> list[str] :classmethod: .. py:method:: component_asset_paths_for(component_file: str | pathlib.Path, *, asset_filenames: collections.abc.Iterable[str] | None = None) -> list[str] :classmethod: .. py:method:: handle_websocket(websocket: fastapi.WebSocket) -> None :async: Default websocket handler (keepalive protocol). .. py:method:: build_routers() Attach the component websocket route using the configured endpoint. .. py:class:: WebSocketComponentParams(/, **data: Any) Bases: :py:obj:`bundle.core.data.Data` Shared websocket parameters for component instances. .. py:attribute:: endpoint :type: str :value: '/ws/default' .. py:class:: GPXComponentParams(/, **data: Any) Bases: :py:obj:`bundle.website.builtin.components.graphic.threeD.GraphicThreeDComponentParams`, :py:obj:`bundle.website.builtin.components.websocket.base.component.WebSocketComponentParams` Shared params for GPX websocket components. .. py:attribute:: graph_id :type: str :value: 'gpx' .. py:class:: GPXWebSocketBaseComponent(/, **data: Any) Bases: :py:obj:`bundle.website.builtin.components.websocket.base.component.WebSocketBaseComponent`, :py:obj:`bundle.website.builtin.components.graphic.threeD.GraphicThreeDComponent` Base class for graph-oriented websocket components. .. py:attribute:: params :type: GPXComponentParams :value: None .. py:class:: MessageRouter Typed dispatcher that routes websocket payloads by their `type` field. .. py:method:: on(message_type: type[bundle.website.core.ws_messages.MessageT], handler: collections.abc.Callable[[fastapi.WebSocket, bundle.website.core.ws_messages.MessageT], collections.abc.Awaitable[None]]) -> MessageRouter Register a callback for a `Data` message model. .. py:method:: dispatch(websocket: fastapi.WebSocket, payload: dict) -> None :async: Deserialize payload and execute the registered typed callback. .. py:class:: AckMessage(/, **data: Any) Bases: :py:obj:`bundle.core.data.Data`, :py:obj:`bundle.website.core.ws_messages.WebSocketDataMixin` Outgoing keepalive acknowledgement. .. py:attribute:: type :type: Literal['keepalive_ack'] :value: 'keepalive_ack' .. py:attribute:: sent_at :type: int | None :value: None .. py:attribute:: received_at :type: int .. py:attribute:: server_rx_packets :type: int :value: 0 .. py:attribute:: server_tx_packets :type: int :value: 0 .. py:attribute:: server_rx_bytes :type: int :value: 0 .. py:attribute:: server_tx_bytes :type: int :value: 0 .. py:attribute:: request_frame_bytes :type: int :value: 0 .. py:attribute:: request_payload_bytes :type: int :value: 0 .. py:attribute:: ack_frame_bytes :type: int :value: 0 .. py:class:: ErrorMessage(/, **data: Any) Bases: :py:obj:`bundle.core.data.Data`, :py:obj:`bundle.website.core.ws_messages.WebSocketDataMixin` Outgoing protocol error message. .. py:attribute:: type :type: Literal['error'] :value: 'error' .. py:attribute:: message :type: str .. py:class:: KeepAliveMessage(/, **data: Any) Bases: :py:obj:`bundle.core.data.Data`, :py:obj:`bundle.website.core.ws_messages.WebSocketDataMixin` Incoming keepalive ping message. .. py:attribute:: type :type: Literal['keepalive'] :value: 'keepalive' .. py:attribute:: sent_at :type: int | None :value: None .. py:attribute:: payload :type: str | None :value: None