Communication#

viser.infra provides WebSocket-based communication infrastructure.

We implement abstractions for:

  • Launching a WebSocket+HTTP server on a shared port.

  • Registering callbacks for connection events and incoming messages.

  • Asynchronous message sending, both broadcasted and to individual clients.

  • Defining dataclass-based message types.

  • Translating Python message types to TypeScript interfaces.

These are what viser runs on under-the-hood, and generally won’t be useful unless you’re building a web-based application from scratch.

class viser.infra.WebsockClientConnection[source]#

Bases: WebsockMessageHandler

Handle for sending messages to and listening to messages from a single connected client.

unsafe_send_message(message: Message) → None[source]#

Send a message to a specific client.

Parameters:

message (Message) –

Return type:

None

atomic() → Generator[None, None, None]#

Returns a context where: all outgoing messages are grouped and applied by clients atomically.

This should be treated as a soft constraint that’s helpful for things like animations, or when we want position and orientation updates to happen synchronously.

Returns:

Context manager.

Return type:

Generator[None, None, None]

queue_message(message: Message) → None#

Wrapped method for sending messages safely.

Parameters:

message (Message) –

Return type:

None

register_handler(message_cls: Type[TMessage], callback: Callable[[ClientId, TMessage], Any]) → None#

Register a handler for a particular message type.

Parameters:

  • message_cls (Type[TMessage]) –

  • callback (Callable[[ClientId, TMessage], Any]) –

Return type:

None

unregister_handler(message_cls: Type[TMessage], callback: Callable[[ClientId, TMessage], Any] | None = None)#

Unregister a handler for a particular message type.

Parameters:

  • message_cls (Type[TMessage]) –

  • callback (Callable[[ClientId, TMessage], Any] | None) –

class viser.infra.WebsockMessageHandler[source]#

Bases: object

Mix-in for adding message handling to a class.

register_handler(message_cls: Type[TMessage], callback: Callable[[ClientId, TMessage], Any]) → None[source]#

Register a handler for a particular message type.

Parameters:

  • message_cls (Type[TMessage]) –

  • callback (Callable[[ClientId, TMessage], Any]) –

Return type:

None

unregister_handler(message_cls: Type[TMessage], callback: Callable[[ClientId, TMessage], Any] | None = None)[source]#

Unregister a handler for a particular message type.

Parameters:

  • message_cls (Type[TMessage]) –

  • callback (Callable[[ClientId, TMessage], Any] | None) –

abstract unsafe_send_message(message: Message) → None[source]#
Parameters:

message (Message) –

Return type:

None

queue_message(message: Message) → None[source]#

Wrapped method for sending messages safely.

Parameters:

message (Message) –

Return type:

None

atomic() → Generator[None, None, None][source]#

Returns a context where: all outgoing messages are grouped and applied by clients atomically.

This should be treated as a soft constraint that’s helpful for things like animations, or when we want position and orientation updates to happen synchronously.

Returns:

Context manager.

Return type:

Generator[None, None, None]

class viser.infra.WebsockServer[source]#

Bases: WebsockMessageHandler

Websocket server abstraction. Communicates asynchronously with client applications.

By default, all messages are broadcasted to all connected clients.

To send messages to an individual client, we can use on_client_connect() to retrieve client handles.

Parameters:

  • host – Host to bind server to.

  • port – Port to bind server to.

  • message_class – Base class for message types. Subclasses of the message type should have unique names. This argument is optional currently, but will be required in the future.

  • http_server_root – Path to root for HTTP server.

  • verbose – Toggle for print messages.

  • client_api_version – Flag for backwards compatibility. 0 sends individual messages. 1 sends windowed messages.

start() → None[source]#

Start the server.

Return type:

None

stop() → None[source]#

Stop the server.

Return type:

None

on_client_connect(cb: Callable[[WebsockClientConnection], Any]) → None[source]#

Attach a callback to run for newly connected clients.

Parameters:

cb (Callable[[WebsockClientConnection], Any]) –

Return type:

None

on_client_disconnect(cb: Callable[[WebsockClientConnection], Any]) → None[source]#

Attach a callback to run when clients disconnect.

Parameters:

cb (Callable[[WebsockClientConnection], Any]) –

Return type:

None

unsafe_send_message(message: Message) → None[source]#

Pushes a message onto the broadcast queue. Message will be sent to all clients.

Broadcasted messages are persistent: if a new client connects to the server, they will receive a buffered set of previously broadcasted messages. The buffer is culled using the value of message.redundancy_key().

Parameters:

message (Message) –

Return type:

None

flush() → None[source]#

Flush the outgoing message buffer for broadcasted messages. Any buffered messages will immediately be sent. (by default they are windowed)

Return type:

None

flush_client(client_id: int) → None[source]#

Flush the outgoing message buffer for a particular client. Any buffered messages will immediately be sent. (by default they are windowed)

Parameters:

client_id (int) –

Return type:

None

atomic() → Generator[None, None, None]#

Returns a context where: all outgoing messages are grouped and applied by clients atomically.

This should be treated as a soft constraint that’s helpful for things like animations, or when we want position and orientation updates to happen synchronously.

Returns:

Context manager.

Return type:

Generator[None, None, None]

queue_message(message: Message) → None#

Wrapped method for sending messages safely.

Parameters:

message (Message) –

Return type:

None

register_handler(message_cls: Type[TMessage], callback: Callable[[ClientId, TMessage], Any]) → None#

Register a handler for a particular message type.

Parameters:

  • message_cls (Type[TMessage]) –

  • callback (Callable[[ClientId, TMessage], Any]) –

Return type:

None

unregister_handler(message_cls: Type[TMessage], callback: Callable[[ClientId, TMessage], Any] | None = None)#

Unregister a handler for a particular message type.

Parameters:

  • message_cls (Type[TMessage]) –

  • callback (Callable[[ClientId, TMessage], Any] | None) –

class viser.infra.Message[source]#

Bases: ABC

Base message type for server/client communication.

excluded_self_client: Any | None = None#

Don’t send this message to a particular client. Useful when a client wants to send synchronization information to other clients.

as_serializable_dict() → Dict[str, Any][source]#

Convert a Python Message object into bytes.

Return type:

Dict[str, Any]

classmethod deserialize(message: bytes) → Message[source]#

Convert bytes into a Python Message object.

Parameters:

message (bytes) –

Return type:

Message

classmethod get_subclasses() → List[Type[T]][source]#

Recursively get message subclasses.

Return type:

List[Type[T]]

abstract redundancy_key() → str[source]#

Returns a unique key for this message, used for detecting redundant messages.

For example: if we send 1000 “set value” messages for the same GUI element, we should only keep the latest message.

Return type:

str

class viser.infra.TypeScriptAnnotationOverride[source]#

Bases: object

Use with typing.Annotated[] to override the automatically-generated TypeScript annotation corresponding to a dataclass field.

annotation: str#
viser.infra.generate_typescript_interfaces(message_cls: Type[Message]) → str[source]#

Generate TypeScript definitions for all subclasses of a base message class.

Parameters:

message_cls (Type[Message]) –

Return type:

str