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.ClientConnection

Bases: MessageHandler

Handle for interacting with a single connected client.

We can use this to read the camera state or send client-specific messages.

client_id: ClientId

send(message: Message) → None

Send a message to a specific client.

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.MessageHandler

Bases: object

Mix-in for adding message handling to a class.

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.Server

Bases: MessageHandler

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

Start the server.

Return type:

None

stop() → None

Stop the server.

Return type:

None

on_client_connect(cb: Callable[[ClientConnection], Any]) → None

Attach a callback to run for newly connected clients.

Parameters:

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

Return type:

None

on_client_disconnect(cb: Callable[[ClientConnection], Any]) → None

Attach a callback to run when clients disconnect.

Parameters:

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

Return type:

None

broadcast(message: Message) → None

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

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

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

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

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]

Convert a Python Message object into bytes.

Return type:

Dict[str, Any]

classmethod deserialize(message: bytes) → Message

Convert bytes into a Python Message object.

Parameters:

message (bytes) –

Return type:

Message

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

Recursively get message subclasses.

Return type:

List[Type[T]]

abstract redundancy_key() → str

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

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

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

Parameters:

message_cls (Type[Message]) –

Return type:

str