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.

get_message_buffer() → AsyncMessageBuffer[source]

Get client message buffer.

Return type:

AsyncMessageBuffer

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.

Parameters:

message (Message)

Return type:

None

register_handler(message_cls: type[TMessage], callback: Callable[[ClientId, TMessage], None | Coroutine]) → None

Register a handler for a particular message type.

Parameters:

  • message_cls (type[TMessage])

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

Return type:

None

start_recording(filter: Callable[[Message], bool]) → RecordHandle

Start recording messages that are sent. Sent messages will be serialized and can be used for playback.

Parameters:

filter (Callable[[Message], bool])

Return type:

RecordHandle

unregister_handler(message_cls: type[TMessage], callback: Callable[[ClientId, TMessage], None | Coroutine] | None = None)

Unregister a handler for a particular message type.

Parameters:

  • message_cls (type[TMessage])

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

class viser.infra.WebsockMessageHandler[source]

Bases: object

Mix-in for adding message handling to a class.

start_recording(filter: Callable[[Message], bool]) → RecordHandle[source]

Start recording messages that are sent. Sent messages will be serialized and can be used for playback.

Parameters:

filter (Callable[[Message], bool])

Return type:

RecordHandle

register_handler(message_cls: type[TMessage], callback: Callable[[ClientId, TMessage], None | Coroutine]) → None[source]

Register a handler for a particular message type.

Parameters:

  • message_cls (type[TMessage])

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

Return type:

None

unregister_handler(message_cls: type[TMessage], callback: Callable[[ClientId, TMessage], None | Coroutine] | None = None)[source]

Unregister a handler for a particular message type.

Parameters:

  • message_cls (type[TMessage])

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

abstract get_message_buffer() → AsyncMessageBuffer[source]
Return type:

AsyncMessageBuffer

queue_message(message: Message) → None[source]

Wrapped method for sending messages.

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], None | Coroutine]) → None[source]

Attach a callback to run for newly connected clients.

Parameters:

cb (Callable[[WebsockClientConnection], None | Coroutine])

Return type:

None

on_client_disconnect(cb: Callable[[WebsockClientConnection], None | Coroutine]) → None[source]

Attach a callback to run when clients disconnect.

Parameters:

cb (Callable[[WebsockClientConnection], None | Coroutine])

Return type:

None

get_message_buffer() → AsyncMessageBuffer[source]

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

Return type:

AsyncMessageBuffer

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.

Parameters:

message (Message)

Return type:

None

register_handler(message_cls: type[TMessage], callback: Callable[[ClientId, TMessage], None | Coroutine]) → None

Register a handler for a particular message type.

Parameters:

  • message_cls (type[TMessage])

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

Return type:

None

start_recording(filter: Callable[[Message], bool]) → RecordHandle

Start recording messages that are sent. Sent messages will be serialized and can be used for playback.

Parameters:

filter (Callable[[Message], bool])

Return type:

RecordHandle

unregister_handler(message_cls: type[TMessage], callback: Callable[[ClientId, TMessage], None | Coroutine] | None = None)

Unregister a handler for a particular message type.

Parameters:

  • message_cls (type[TMessage])

  • callback (Callable[[ClientId, TMessage], None | Coroutine] | 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