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

Bases: object

Handle for serializing messages. In Viser, this is used to save the scene state so it can be shared/embedded in static webpages.

insert_sleep(duration: float) → None

Insert a sleep into the recorded file. This can be useful for dynamic 3D data.

Parameters:

duration (float)

Return type:

None

serialize() → bytes

Serialize saved messages. Should only be called once. Our convention is to write this binary format to a file with a .viser extension, for example via pathlib.Path("file.viser").write_bytes(...).

Returns:

The recording as bytes.

Return type:

bytes

as_html(dark_mode: bool = False) → str

Get a standalone HTML string for the serialized scene.

Returns a self-contained HTML document that can be saved to a file or embedded in other contexts.

Parameters:

dark_mode (bool) – Use dark color scheme.

Returns:

A complete HTML document as a string.

Return type:

str

show(height: int = 400, dark_mode: bool = False) → None

Display the serialized scene in a Jupyter notebook or web browser.

In Jupyter notebooks/labs, displays an inline IFrame. When running as a script, opens the visualization in the default web browser.

See also viser.ViserServer.show().

Parameters:

• height (int) – Height of the embedded viewer in pixels.

• dark_mode (bool) – Use dark color scheme.

Return type:

None

class viser.infra.WebsockClientConnection

Bases: WebsockMessageHandler

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

get_message_buffer() → AsyncMessageBuffer

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]

get_message_serializer(filter: Callable[[Message], bool]) → StateSerializer

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

Parameters:

filter (Callable[[Message], bool])

Return type:

StateSerializer

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

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

Bases: object

Mix-in for adding message handling to a class.

get_message_serializer(filter: Callable[[Message], bool]) → StateSerializer

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

Parameters:

filter (Callable[[Message], bool])

Return type:

StateSerializer

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

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)

abstract get_message_buffer() → AsyncMessageBuffer

Return type:

AsyncMessageBuffer

queue_message(message: Message) → None

Wrapped method for sending messages.

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]

class viser.infra.WebsockServer

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

Start the server.

Return type:

None

stop() → None

Stop the server.

Return type:

None

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

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

Attach a callback to run when clients disconnect.

Parameters:

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

Return type:

None

get_message_buffer() → AsyncMessageBuffer

Get the broadcast queue. Message will be sent to all clients.

Return type:

AsyncMessageBuffer

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

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]

get_message_serializer(filter: Callable[[Message], bool]) → StateSerializer

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

Parameters:

filter (Callable[[Message], bool])

Return type:

StateSerializer

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

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

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(binary_buffers: List[memoryview] | None = None) → Dict[str, Any]

Convert a Python Message object into a serializable dict.

If binary_buffers is provided, numpy arrays are extracted into it and replaced with tagged placeholder dicts for the hybrid wire format. Otherwise, arrays are inlined as memoryviews (used by API v0).

Parameters:

binary_buffers (List[memoryview] | None)

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