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”, -32002 for a viewer still initializing, -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.

property viewer_not_ready: bool

Whether an attached viewer is still initializing getDimensions.

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.

A viewer-attached event also renews this controller’s active subscriptions before it is returned. 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.