vuer.frame

class vuer.Vuer

Bases: PrefixProto, Server

Vuer Server

This is the server for the Vuer client.

Usage:

app = Vuer()

@app.spawn
async def main(session: VuerSession):
     session.set @ Scene(children=[...])

app.run()

bind(fn=None, start=False)

Bind an asynchronous generator function for use in socket connection handler. The function should be a generator that yields Page objects.

Parameters:

fn – The function to bind.

Returns:

None

spawn(fn: Callable[[VuerSession], Coroutine] | None = None, start=False)

bind the socket handler function fn to vuer, and start the event loop if start is True.

Note: this is really a misnomer.

Parameters:

• fn (Callable[[VuerSession], Coroutine] | None) – The function to spawn.

• start – Start server after binding

Returns:

None

async relay(request)

This is the relay object for sending events to the server.

Todo: add API for specifying the websocket ID. Or just broadcast to all. Todo: add type hint

Interface:

<uri>/relay?sid=<websocket_id>

Returns:

• Status 200

• Status 400

async bound_fn(session_proxy: VuerSession)

This is the default generator function in the socket connection handler

Parameters:

session_proxy (VuerSession) –

get_url()

Get the URL for the Tassa. :return: The URL for the Tassa.

async send(ws_id, event: ServerEvent | None = None, event_bytes=None)

Parameters:

event (ServerEvent | None) –

async rpc(ws_id, event: ServerRPC, ttl=2.0) → ClientEvent | None

RPC only takes a single response. For multi-response streaming, we need to build a new one

Question is whether we want to make this RPC an awaitable funciton.

Parameters:

• ttl – The time to live for the handler. If the handler is not called within the time it gets removed from the handler list.

• event (ServerRPC) –

Return type:

ClientEvent | None

async rpc_stream(ws_id, event: ServerEvent | None = None, event_bytes=None)

This RPC offers multiple responses.

Parameters:

event (ServerEvent | None) –

async close_ws(ws_id)

async uplink(proxy: VuerSession)

Parameters:

proxy (VuerSession) –

async downlink(request: aiohttp.web_request.Request, ws: aiohttp.web_ws.WebSocketResponse)

The websocket handler for receiving messages from the client.

Parameters:

• ws (aiohttp.web_ws.WebSocketResponse) – The websocket.

• request (aiohttp.web_request.Request) – The request (unused).

Returns:

None

add_handler(event_type: str, fn: Callable[[ClientEvent, VuerSession], None] | None = None, once: bool = False) → Callable[[], None]

Adding event handlers to the vuer server.

Parameters:

• event_type (str) – The event type to handle.

• fn (Callable[[ClientEvent, VuerSession], None] | None) – The function to handle the event.

• once (bool) – Whether to remove the handler after the first call. This is useful for RPC, which cleans up after itself. The issue is for RPC, the key also needs to match. So we hack it here to use a call specific event_type to enforce the cleanup.

Return type:

Callable[[], None]

# Usage:

As a decorator:

app = Vuer()
@app.add_handler("CAMERA_MOVE")
def on_camera(event: ClientEvent, session: VuerSession):
    print("camera event", event.etype, event.value)

As a function:

app = Vuer()
def on_camera(event: ClientEvent, session: VuerSession):
    print("camera event", event.etype, event.value)

app.add_handler("CAMERA_MOVE", on_camera)
app.run()

_ttl_handler(ttl, cleanup)

run(kill=None, *args, **kwargs)

domain = 'https://vuer.ai'

host = 'localhost'

port = 8012

free_port = None

static_root = '.'

todo: we want to support multiple paths.

queue_len = 100

cors = 'https://vuer.ai,https://staging.vuer.ai,https://dash.ml,http://localhost:8000,http://127.0.0.1:8000,*'

queries = {}

cert = None

key = None

ca_cert = None

client_root = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/vuer-py/checkouts/latest/vuer/client_build')

verbose = None

REQUEST_MAX_SIZE: int = 268435456

WEBSOCKET_MAX_SIZE: int = 268435456

class vuer.VuerSession

Bases: object

property socket

Getter for the websocket object.

this is useful for closing the socket session from the client side.

Example Usage:

@app.spawn(start=True):
async def main(session: VuerSession):
    print("doing something...")
    await sleep(1.0)

    print("I am done! closing the socket.")
    session.socket.close()

async grab_render(ttl=2.0, **kwargs) → ClientEvent

Grab a render from the client.

Parameters:

• quality – The quality of the render. 0.0 - 1.0

• subsample – The subsample of the render.

• ttl – The time to live for the handler. If the handler is not called within the time it gets removed from the handler list.

Return type:

ClientEvent

property set: At

Used exclusively to set the scene.

the @SET operator is responsible for setting the root node of the scene.

Examples:

proxy @ Set(Scene(children=[…]))

or

app.set @ Scene(children=[…])

property update: At

Used to update existing elements. NOOP if an element does not exist.

Supports passing in a list of elements. (Thank God I implemented this… so handy! - Ge)

Example Usage:

app.update @ [element1, element2, ...]

property add: At

Used to add elements to a specific parent.

Requires a parentKey, or treats the Scene root node as the default parent.

Example Usage:

app.add(element1, element2, ..., to=parentKey.)

or using the Scene root node as the default parent:

app.add @ element1

property upsert: At

Upsert elements to a specific parent.

Requires a parentKey, or treats the Scene root node as the default parent.

Example Usage:

app.upsert(element1, element2, ..., to=parentKey.)

or using the Scene root node as the default parent:

app.upsert @ element1

property remove: At

Remove elements by keys.

Example Usage:

app.remove @ ["key1", "key2", ...]

or a single key:

app.remove @ "key1"

popleft()

pop()

clear()

clears all client messages

stream()

spawn_task(task, name=None)

Spawn a task in the running asyncio event loop

Useful for background tasks. Returns an asyncio task that can be canceled.

async background_task():
    print('\rthis ran once')

async long_running_bg_task():
    while True:
        await asyncio.sleep(1.0)
        print("\rlong running background task is still running")

@app.spawn_task
async def main_fn(sess: VuerSession):
    # Prepare background tasks here:
    task = sess.spawn_task(background_task)
    long_running_task = sess.spawn_task(long_running_bg_task)

Now to cancel a running task, simply

task.cancel()

Todos

▫️ Add a way to automatically clean up when exiting the main_fn.