Vuer

vuer.server

At

python
class At

Source

Proxy Object for using the @ notation. Also supports being called direction, which supports more complex arguments.

At.init

python
def __init__(self, fn)

Source

VuerSession

python
class VuerSession

Source

VuerSession.init

python
def __init__(self, vuer: 'Vuer', ws_id: int, queue_len=100)

Source

VuerSession.socket

python
def socket(self)

Source

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()

VuerSession.grab_render

python
async def grab_render(self, ttl=2.0, **kwargs) -> ClientEvent

Source

Grab a render from the client.

:param quality: The quality of the render. 0.0 - 1.0 :param subsample: The subsample of the render. :param ttl: The time to live for the handler. If the handler is not called within the time it gets removed from the handler list.

VuerSession.set

python
def set(self) -> At

Source

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=[...])

VuerSession.update

python
def update(self) -> At

Source

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, ...]

VuerSession.add

python
def add(self) -> At

Source

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

VuerSession.upsert

python
def upsert(self) -> At

Source

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

VuerSession.remove

python
def remove(self) -> At

Source

Remove elements by keys.

Example Usage::

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

or a single key: ::

app.remove @ "key1"

VuerSession.popleft

python
def popleft(self)

Source

VuerSession.pop

python
def pop(self)

Source

VuerSession.clear

python
def clear(self)

Source

clears all client messages

VuerSession.stream

python
def stream(self)

Source

VuerSession.spawn_task

python
def spawn_task(self, task, name=None)

Source

Spawn a task in the running asyncio event loop

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

.. code-block:: python :linenos:

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

.. code-block:: python :linenos:

task.cancel()

Todos

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

Vuer

python
class Vuer(PrefixProto, Server)

Source

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()

.. automethod:: bind .. automethod:: spawn .. automethod:: relay .. automethod:: bound_fn .. automethod:: spawn_task .. automethod:: get_url .. automethod:: send .. automethod:: rpc .. automethod:: rpc_stream .. automethod:: close_ws .. automethod:: uplink .. automethod:: downlink .. automethod:: add_handler .. automethod:: _ttl_handler .. automethod:: run

python
free_port: bool = Flag('Kill what is running on the requested port if True.')
python
static_root: str = Proto('.', help='root for file serving')
python
queue_len: int = Proto(100, help='use a max length to avoid the memory from blowing up.')
python
queries: Dict = Proto({}, help='query parameters to pass')
python
cert: str = Proto(None, dtype=str, help='the path to the SSL certificate')
python
key: str = Proto(None, dtype=str, help='the path to the SSL key')
python
ca_cert: str = Proto(None, dtype=str, help='the trusted root CA certificates')
python
client_root: Path = Path(__file__).parent / 'client_build'

Vuer.relay

python
async def relay(self, request)

Source

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>

:return:

  • Status 200
  • Status 400

Vuer.bound_fn

python
async def bound_fn(self, session_proxy: VuerSession)

Source

This is the default generator function in the socket connection handler

Vuer.spawn

python
def spawn(self, fn: SocketHandler=None, start=False)

Source

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

Note: this is really a misnomer.

:param fn: The function to spawn. :param start: Start server after binding :return: None

Vuer.bind

python
def bind(self, fn=None, start=False)

Source

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

:param fn: The function to bind. :return: None

Vuer.get_url

python
def get_url(self)

Source

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

Vuer.send

python
async def send(self, ws_id, event: ServerEvent=None, event_bytes=None)

Source

Vuer.rpc

python
async def rpc(self, ws_id, event: ServerRPC, ttl=2.0) -> Union[ClientEvent, None]

Source

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.

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

Vuer.rpc_stream

python
async def rpc_stream(self, ws_id, event: ServerEvent=None, event_bytes=None)

Source

This RPC offers multiple responses.

Vuer.close_ws

python
async def close_ws(self, ws_id)

Source

Vuer.uplink

python
async def uplink(self, proxy: VuerSession)

Source

Vuer.downlink

python
async def downlink(self, request: Request, ws: WebSocketResponse)

Source

The websocket handler for receiving messages from the client.

:param ws: The websocket. :param request: The request (unused). :return: None

Vuer.add_handler

python
def add_handler(self, event_type: str, fn: EventHandler=None, once: bool=False) -> Callable[[], None]

Source

Adding event handlers to the vuer server.

:param event_type: The event type to handle. :param fn: The function to handle the event. :param once: 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.

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()

Vuer.socket_index

python
async def socket_index(self, request: BaseRequest)

Source

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>

:return:

  • Status 200
  • Status 400

Vuer.add_route

python
def add_route(self, path, fn: Callable, method='GET', content_type='text/html')

Source

Vuer.run

python
def run(self, kill=None, *args, **kwargs)

Source

Inherited members

python
domain = Proto('https://vuer.ai', help='default url of web client')
python
host = Proto('localhost', help='set to 0.0.0.0 to enable remote connections')
python
port = Proto(DEFAULT_PORT, help='port to use')
python
cors = Proto('https://vuer.ai,https://staging.vuer.ai,https://dash.ml,http://localhost:8000,http://127.0.0.1:8000,*', help='domains that are allowed for cross origin requests.')
python
verbose = Flag('Print the settings if True.')

Public imports

These symbols are available from this module. Their definitions are documented in the linked modules.