Control Package

Synchronous Python control of a running Luxar viewer through its control hub.

Driving a running Luxar viewer from Python.

A viewer launched with ?control attaches to the hub that luxar serve --control exposes; this package is the other end of that hub. The method names are the viewer’s own embedder API, so anything the browser’s LuxarApp can do, a script here can ask for:

from luxar.control import Viewer

with Viewer("ws://kiosk.local:5173/control") as viewer:
    story = viewer.dimension_index("story")  # by name, never a position
    viewer.set_dimension_value(story, 3)  # fly to the fourth chapter

What it is for: scripted demos, reproducible screenshots, driving a kiosk from a cron job, and prototyping an agent. What it is not: a second API. If a call is not in the viewer’s embedder surface it is not here either — see packages/luxar-viewer/src/core/app/control/method-policy.ts for the list and docs/guides/specs/REMOTE_CONTROL_SPEC.md for the contract.

exception luxar.control.ControlError(code: int, message: str)

Bases: RuntimeError

A viewer (or the hub) refused a call.

code is the JSON-RPC error code — -32601 for a method the viewer’s policy does not expose, -32001 for “no viewer attached”, -32603 for a method that threw inside the viewer.

__init__(code: int, message: str) → None

property no_viewer_attached: bool

Whether this means “nothing is listening” rather than “that failed”.

Worth distinguishing in a kiosk script: a display that has not booted yet is something to wait for, not an error to abort on.

class luxar.control.Viewer(url: str, *, token: str | None = None, timeout_s: float = 30.0)

Bases: object

A connected controller for one hub.

Use as a context manager, or call close() yourself:

with Viewer("ws://kiosk.local:5173/control") as viewer:
    viewer.recenter_camera()

Parameters:

• url – The hub’s WebSocket URL, as luxar serve --control prints it.

• token – Shared secret, when the hub was started with --control-token.

• timeout_s – Per-call reply timeout. See DEFAULT_TIMEOUT_S.

__init__(url: str, *, token: str | None = None, timeout_s: float = 30.0) → None

close() → None

Close the socket. Idempotent.

call(method: str, *params: Any) → Any

Invoke an embedder method and return its result.

The escape hatch as much as the engine room: every named method below is one line of this, and a method with no wrapper yet is reachable as viewer.call("setLayer", "/points", {"visible": False}).

Raises:

• ControlError – the viewer or the hub refused the call.

• TimeoutError – no reply arrived within the configured timeout.

• websockets.exceptions.ConnectionClosed – the hub connection closed.

notify(method: str, *params: Any) → None

Invoke a method without waiting for a reply.

Right for a fire-and-forget nudge (notify("recenterCamera")) where the round trip is the only cost. An error is not reported back — the viewer logs it instead — so prefer call() when it matters.

recv_event(timeout_s: float | None = None) → Tuple[str, Any] | None

Return the next event as (name, payload).

Not only the events this controller subscribed to: the hub fans a viewer’s events out to every attached controller, so a script that subscribed to nothing still sees whatever another controller — a touch panel, say — asked for. Match on name rather than assuming.

Events received while call() waits for its reply are buffered and returned first. That buffer holds at most MAX_BUFFERED_EVENTS; past the cap the OLDEST are dropped, so a long-lived controller that never reads events cannot grow without limit. timeout_s=None waits indefinitely; a finite timeout returns None when no event arrives before its deadline.

Raises:

ConnectionClosed – If the socket closed while waiting. This reads the same socket as call() and fails the same way.

get_viewer_state() → Dict[str, Any]

Everything at once: src, camera, dimensions, rendering, layers, audio.

get_dimensions() → Dict[str, Any]

Dimension metadata, including categories for a labelled axis.

set_dimension_value(index: int, value: float) → None

Move one dimension — how a chapter jump is expressed.

The viewer’s waypoint driver does the rest: the camera flies, the dimension-bound overlays swap, and waypoint-arrived fires.

dimension_index(name: str) → int

The positional index of a dimension, by name.

setDimensionValue takes an index, but an index is a property of the scene’s dimension order and changes if the author reorders it. Looking it up by name is what makes a script survive that.

Raises:

KeyError – the scene has no dimension with that name.

get_camera_pose() → Dict[str, Any]

The current camera pose, in the shape fly_to accepts.

fly_to(pose: Dict[str, Any], **opts: Any) → Dict[str, Any]

Fly the camera to pose; resolves when the flight ends.

opts are the viewer’s own flight options (durationMs, easing, keepOrientation).

recenter_camera() → None

Re-frame the scene, as the keyboard’s recenter action does.

subscribe(event: str) → None

Start receiving event notifications on this socket.

unsubscribe(event: str) → None

Stop receiving event notifications.