embodichain.lab.visualization#
Browser-based visualization of simulation scenes via Viser.
SceneExporter produces backend-neutral snapshots, VisualizationRuntime pushes them to a Viser backend, with gizmo overlays, camera preview, and CLI helpers.
Overview#
Browser-based visualization of simulation scenes. The
SceneExporter reads simulation assets on the simulation thread and
produces detached, backend-neutral CPU snapshots (SceneManifest,
SceneFrame, CameraImageFrame). A background
VisualizationRuntime owns latest-frame queues, rate limiting, health,
and telemetry, and pushes snapshots to a visualization backend - currently the
Viser server (ViserServerCfg) that publishes an interactive 3D view
to the browser. The stack supports interactive gizmos and scalar articulation
joint controls (with optional command write-back to the simulation), scene
overlays (targets, trajectories, point clouds), and live RGB camera preview.
CLI helpers
(add_viser_args_to_parser(), visualization_cfg_from_args()) wire
the standard --viser* arguments into launchers.
Configuration#
- class embodichain.lab.visualization.VisualizationCfg[source]#
Bases:
objectConfigure live scene visualization.
- Parameters:
backend (
str) – Visualization backend name. Supported values are"none"and"viser".scene_fps (
float) – Maximum scene capture rate.env_ids (
list[int] |None) – Environment indices exposed by the visualizer.Noneselects every simulation environment.max_visible_envs (
int|None) – Optional safety limit on the number of selected environments.Nonedisables the limit.point_cloud_max_points (
int) – Maximum number of points retained per point cloud.sensor_image_fps (
float|None) – Maximum camera RGB preview update rate.Nonecaptures once per visualization step instead of using wall-clock rate limiting.soft_body_fps (
float) – Maximum soft-body and cloth vertex update rate.allow_commands (
bool) – Whether simulation-mutating browser commands are allowed. This enables Viser Gizmo dragging and registered articulation joint controls. Keep it disabled for untrusted or publicly reachable browser sessions.viser_server (
ViserServerCfg) – Viser HTTP/WebSocket server binding settings.
Attributes:
-
allow_commands:
bool#
-
backend:
str#
-
env_ids:
list[int] |None#
-
max_visible_envs:
int|None#
-
point_cloud_max_points:
int#
-
scene_fps:
float#
-
sensor_image_fps:
float|None#
-
soft_body_fps:
float#
-
viser_server:
ViserServerCfg#
- class embodichain.lab.visualization.ViserServerCfg[source]#
Bases:
objectConfigure the local Viser HTTP/WebSocket server.
- Parameters:
host (
str) – Interface on which Viser listens. Server deployments should inject this value instead of accepting it from an untrusted run request.port (
int) – TCP port allocated to the worker.label (
str) – Browser application label.verbose (
bool) – Whether Viser should print server diagnostics.
Attributes:
-
host:
str#
-
label:
str#
-
port:
int#
-
verbose:
bool#
Runtime#
- class embodichain.lab.visualization.VisualizationRuntime[source]#
Bases:
objectRun scene capture and a visualization backend without blocking simulation.
The simulation thread calls
capture(). Viser creation and all handle mutations occur on one private update thread. The frame queue always keeps the newest sample, preventing visualization overload from accumulating lag.- Parameters:
exporter (
SceneExporter) – Scene exporter bound to a simulation manager.cfg (
VisualizationCfg) – Visualization, frame rate, and Viser server configuration.backend (
VisualizationBackend|None) – Optional backend injection hook used by tests and alternate UIs.
Methods:
__init__(exporter, cfg[, backend])capture(*, sim_step, sim_time[, overlays, ...])Capture a due frame and enqueue it without waiting for Viser.
Drain browser Gizmo commands for simulation-thread processing.
Drain browser joint commands for simulation-thread processing.
Drain browser click-pick commands for simulation-thread processing.
Return the newest browser replay seek, if one is pending.
publish_replay_control(*, step, max_step[, ...])Asynchronously publish trajectory replay progress to Viser.
Capture and asynchronously publish a new topology revision.
set_joint_control_provider(provider)Install a simulation-thread joint source for future scene captures.
start([timeout])Build the initial manifest and start the backend update thread.
stop([timeout])Stop the backend and reclaim its server port and update thread.
Attributes:
Local browser endpoint after
start()returns.Return runtime, endpoint, client, and worker health information.
Whether the visualization update thread is alive.
Return an immutable telemetry snapshot.
- capture(*, sim_step, sim_time, overlays=None, force=False, capture_camera_images=True)[source]#
Capture a due frame and enqueue it without waiting for Viser.
- Parameters:
sim_step (
int) – Current simulation step.sim_time (
float) – Current simulation time in seconds.overlays (
SceneOverlays|None) – Optional backend-neutral debug overlays.force (
bool) – Ignore the configured scene FPS limiter.capture_camera_images (
bool) – Whether camera images may be captured in this call. Simulation batches disable this for intermediate physics substeps.
- Return type:
bool- Returns:
Truewhen a frame was captured, otherwiseFalsewhen limited.
- drain_gizmo_commands()[source]#
Drain browser Gizmo commands for simulation-thread processing.
- Return type:
tuple[GizmoCommand,...]
- drain_joint_control_commands()[source]#
Drain browser joint commands for simulation-thread processing.
- Return type:
tuple[JointControlCommand,...]
- drain_pick_commands()[source]#
Drain browser click-pick commands for simulation-thread processing.
- Return type:
tuple[PickCommand,...]
- drain_replay_control_command()[source]#
Return the newest browser replay seek, if one is pending.
- Return type:
int|None- Returns:
Requested trajectory step, or
Nonewhen no seek is pending.
- property health: RuntimeHealth#
Return runtime, endpoint, client, and worker health information.
- property is_running: bool#
Whether the visualization update thread is alive.
- publish_replay_control(*, step, max_step, visible=True)[source]#
Asynchronously publish trajectory replay progress to Viser.
- Parameters:
step (
int) – Current trajectory step.max_step (
int) – Largest valid trajectory step.visible (
bool) – Whether the replay control should be visible.
- Raises:
RuntimeError – If the visualization runtime is not running.
ValueError – If the step range is invalid.
- Return type:
None
- set_joint_control_provider(provider)[source]#
Install a simulation-thread joint source for future scene captures.
Registering a provider does not publish a new manifest by itself. The caller must refresh the scene after registration so the backend can build its controls.
- Return type:
None
- start(timeout=10.0)[source]#
Build the initial manifest and start the backend update thread.
- Return type:
None
- property stats: RuntimeStats#
Return an immutable telemetry snapshot.
- class embodichain.lab.visualization.LatestFrameQueue[source]#
A one-slot queue where producers replace an unconsumed old frame.
Methods:
__init__()clear()Discard any queued frame.
get([timeout])Return the queued frame, waiting for up to
timeoutseconds.Return the queued frame without blocking.
put_latest(frame)Enqueue
frameand return whether an older frame was dropped.
- class embodichain.lab.visualization.GizmoCommandQueue[source]#
Bounded command queue that coalesces high-rate drag updates.
Drag lifecycle commands are retained. When the queue reaches its soft capacity, an older
updatefor the same Gizmo/client is replaced first.Methods:
__init__([maxsize])clear()Discard all queued commands.
drain()Return and clear all queued commands in arrival order.
put(command)Enqueue a command without blocking the Viser callback thread.
- drain()[source]#
Return and clear all queued commands in arrival order.
- Return type:
tuple[GizmoCommand,...]
- class embodichain.lab.visualization.JointControlCommandQueue[source]#
Bounded queue that keeps only the newest value for each joint control.
Methods:
__init__([maxsize])clear()Discard all queued commands.
drain()Return and clear all queued commands in arrival order.
put(command)Enqueue a value without blocking the Viser callback thread.
- drain()[source]#
Return and clear all queued commands in arrival order.
- Return type:
tuple[JointControlCommand,...]
- class embodichain.lab.visualization.RuntimeHealth[source]#
Current visualization runtime health and connection state.
Methods:
__init__(status, running, endpoint, ...[, ...])Attributes:
- __init__(status, running, endpoint, client_count, published_scene_revision, worker_error=None)#
-
client_count:
int#
-
endpoint:
str|None#
-
published_scene_revision:
int#
-
running:
bool#
-
status:
str#
-
worker_error:
str|None= None#
- class embodichain.lab.visualization.RuntimeStats[source]#
Snapshot of scene and camera-image capture/upload telemetry.
Methods:
__init__([captured_frames, ...])Attributes:
- __init__(captured_frames=0, published_frames=0, dropped_frames=0, rejected_frames=0, manifest_bytes=0, frame_bytes=0, capture_seconds=0.0, upload_seconds=0.0, captured_image_frames=0, published_image_frames=0, dropped_image_frames=0, rejected_image_frames=0, image_bytes=0, image_capture_seconds=0.0, image_upload_seconds=0.0)#
-
capture_seconds:
float= 0.0#
-
captured_frames:
int= 0#
-
captured_image_frames:
int= 0#
-
dropped_frames:
int= 0#
-
dropped_image_frames:
int= 0#
-
frame_bytes:
int= 0#
-
image_bytes:
int= 0#
-
image_capture_seconds:
float= 0.0#
-
image_upload_seconds:
float= 0.0#
-
manifest_bytes:
int= 0#
-
published_frames:
int= 0#
-
published_image_frames:
int= 0#
-
rejected_frames:
int= 0#
-
rejected_image_frames:
int= 0#
-
upload_seconds:
float= 0.0#
Scene Export#
- class embodichain.lab.visualization.SceneExporter[source]#
Bases:
objectCapture backend-neutral scene data from a
SimulationManager.Scene access happens only in
build_manifest()andcapture(), so callers can keep both methods on the simulation thread. Returned protocol objects own detached CPU copies and are safe to hand to a background backend.- Parameters:
sim (
SimulationManager) – Simulation manager whose public asset interfaces are exported.cfg (
VisualizationCfg) – Visualization selection and sampling configuration.run_id (
str|None) – Stable identifier for this runtime. A UUID is generated by default.
Methods:
__init__(sim, cfg[, run_id])Capture scene topology, incrementing the scene revision.
capture(*, sim_step, sim_time[, overlays, ...])Capture one dynamic scene frame on the simulation thread.
capture_camera_images(*, sim_step, sim_time)Render and detach one RGB image for every exported camera instance.
resolve_node_target(node_id)Map a published scene node id to its
(uid, kind).set_joint_control_provider(provider)Install an optional simulation-thread joint-control source.
Attributes:
Whether the current manifest contains an RGB-capable camera.
Whether the current manifest contains soft-body or cloth nodes.
Current manifest revision, or zero before the first manifest.
- build_manifest()[source]#
Capture scene topology, incrementing the scene revision.
Call this after assets are created and again after topology-changing resets. Empty articulation links are ignored.
- Return type:
- capture(*, sim_step, sim_time, overlays=None, capture_dynamic_geometry=True)[source]#
Capture one dynamic scene frame on the simulation thread.
- Parameters:
sim_step (
int) – Current simulation step.sim_time (
float) – Current simulation time in seconds.overlays (
SceneOverlays|None) – Optional debug overlays.capture_dynamic_geometry (
bool) – Whether to copy soft-body and cloth vertices.
- Return type:
- Returns:
Captured frame and producer-side copy duration.
- capture_camera_images(*, sim_step, sim_time)[source]#
Render and detach one RGB image for every exported camera instance.
- Return type:
- property has_cameras: bool#
Whether the current manifest contains an RGB-capable camera.
- property has_deformables: bool#
Whether the current manifest contains soft-body or cloth nodes.
- resolve_node_target(node_id)[source]#
Map a published scene node id to its
(uid, kind).Used by the simulation thread to turn a Viser click-pick result into the asset uid that
SimulationManager.enable_gizmo()expects.- Parameters:
node_id (
str) – Scene node id from the current manifest.- Return type:
tuple[str,str] |None- Returns:
(uid, kind)wherekindis the asset kind (for example"rigid","robot", or"articulation"), orNoneif the node id is not part of the current scene.
- property scene_revision: int#
Current manifest revision, or zero before the first manifest.
- set_joint_control_provider(provider)[source]#
Install an optional simulation-thread joint-control source.
The next
build_manifest()call snapshots the provider’s static controls. Dynamic values are then sampled bycapture().- Return type:
None
- class embodichain.lab.visualization.SceneManifest[source]#
Static scene topology and geometry for one scene revision.
Methods:
__init__(run_id, scene_revision, nodes, ...)Attributes:
- __init__(run_id, scene_revision, nodes, geometries, cameras=<factory>, gizmos=<factory>, joint_controls=<factory>, schema_version=5, up_direction='+z', length_unit='meter')#
-
cameras:
tuple[CameraSpec,...]#
-
geometries:
tuple[MeshGeometry,...]#
-
joint_controls:
tuple[JointControlSpec,...]#
-
length_unit:
str= 'meter'#
-
run_id:
str#
-
scene_revision:
int#
-
schema_version:
int= 5#
-
up_direction:
str= '+z'#
- class embodichain.lab.visualization.SceneFrame[source]#
Dynamic poses and overlays for one simulation sample.
Methods:
__init__(run_id, scene_revision, sequence, ...)Attributes:
- __init__(run_id, scene_revision, sequence, sim_step, sim_time, node_ids, positions, wxyz, visible, camera_ids=<factory>, camera_positions=<factory>, camera_wxyz=<factory>, dynamic_meshes=<factory>, gizmos=<factory>, joint_controls=<factory>, overlays=<factory>, wall_time=<factory>, schema_version=5)#
-
camera_ids:
tuple[str,...]#
-
camera_positions:
ndarray#
-
camera_wxyz:
ndarray#
-
dynamic_meshes:
tuple[DynamicMeshUpdate,...]#
-
gizmos:
tuple[GizmoState,...]#
-
joint_controls:
tuple[JointControlState,...]#
-
node_ids:
tuple[str,...]#
-
overlays:
SceneOverlays#
-
positions:
ndarray#
-
run_id:
str#
-
scene_revision:
int#
-
schema_version:
int= 5#
-
sequence:
int#
-
sim_step:
int#
-
sim_time:
float#
-
visible:
ndarray#
-
wall_time:
str#
-
wxyz:
ndarray#
- class embodichain.lab.visualization.SceneNode[source]#
One mesh-bearing logical node in a scene manifest.
Methods:
__init__(node_id, path, parent_id, env_id, ...)Attributes:
- __init__(node_id, path, parent_id, env_id, kind, geometry_id, dynamic_geometry=False, visible=True)#
-
dynamic_geometry:
bool= False#
-
env_id:
int#
-
geometry_id:
str#
-
kind:
str#
-
node_id:
str#
-
parent_id:
str|None#
-
path:
str#
-
visible:
bool= True#
- class embodichain.lab.visualization.MeshGeometry[source]#
Backend-neutral triangle mesh stored in local coordinates.
Methods:
__init__(geometry_id, vertices, faces[, color])Attributes:
- __init__(geometry_id, vertices, faces, color=(90, 200, 255))#
-
color:
tuple[int,int,int] = (90, 200, 255)#
-
faces:
ndarray#
-
geometry_id:
str#
-
vertices:
ndarray#
- class embodichain.lab.visualization.DynamicMeshUpdate[source]#
Detached vertex positions for one deformable scene node.
Methods:
__init__(node_id, vertices)Attributes:
- __init__(node_id, vertices)#
-
node_id:
str#
-
vertices:
ndarray#
- class embodichain.lab.visualization.GizmoSpec[source]#
Static description of one simulation Gizmo exposed by a backend.
Methods:
__init__(gizmo_id, target_uid, target_type, ...)Attributes:
- __init__(gizmo_id, target_uid, target_type, control_part, env_id, path, scale=0.2, line_width=2.5, visible=True)#
-
control_part:
str|None#
-
env_id:
int#
-
gizmo_id:
str#
-
line_width:
float= 2.5#
-
path:
str#
-
scale:
float= 0.2#
-
target_type:
str#
-
target_uid:
str#
-
visible:
bool= True#
- class embodichain.lab.visualization.GizmoState[source]#
Authoritative world pose and visibility of one simulation Gizmo.
Methods:
__init__(gizmo_id, position, wxyz[, visible])Attributes:
- __init__(gizmo_id, position, wxyz, visible=True)#
-
gizmo_id:
str#
-
position:
ndarray#
-
visible:
bool= True#
-
wxyz:
ndarray#
- class embodichain.lab.visualization.GizmoCommand[source]#
Immutable browser drag command consumed on the simulation thread.
Methods:
__init__(run_id, scene_revision, sequence, ...)Attributes:
- __init__(run_id, scene_revision, sequence, gizmo_id, phase, client_id, position, wxyz, schema_version=5)#
-
client_id:
str#
-
gizmo_id:
str#
-
phase:
Literal['start','update','end']#
-
position:
ndarray#
-
run_id:
str#
-
scene_revision:
int#
-
schema_version:
int= 5#
-
sequence:
int#
-
wxyz:
ndarray#
PickCommand identifies a node in a specific simulation run and scene
revision. An empty selection releases only the gizmo created by click picking;
explicitly configured gizmos retain their ownership.
- class embodichain.lab.visualization.PickCommand[source]#
Immutable browser click-pick command consumed on the simulation thread.
A non-empty
node_idrequests a Gizmo on the clicked scene node; aNonenode_id(clicking empty space) clears the picker-owned Gizmo.Methods:
__init__(run_id, scene_revision, client_id, ...)Attributes:
- __init__(run_id, scene_revision, client_id, node_id, schema_version=5)#
-
client_id:
str#
-
node_id:
str|None#
-
run_id:
str#
-
scene_revision:
int#
-
schema_version:
int= 5#
- class embodichain.lab.visualization.JointControlSpec[source]#
Static description of one scalar articulation joint control.
Values use simulation units: radians for rotational joints and meters for prismatic joints. Controls with two finite limits can be rendered as a slider; controls with one or both limits missing use a numeric input.
Methods:
__init__(control_id, articulation_uid, ...)Attributes:
- __init__(control_id, articulation_uid, env_id, joint_id, joint_name, joint_type, lower, upper, step, initial_value)#
-
articulation_uid:
str#
-
control_id:
str#
-
env_id:
int#
-
initial_value:
float#
-
joint_id:
int#
-
joint_name:
str#
-
joint_type:
Literal['revolute','continuous','prismatic']#
-
lower:
float|None#
-
step:
float#
-
upper:
float|None#
- class embodichain.lab.visualization.JointControlState[source]#
Authoritative value and command acknowledgement for one joint control.
Methods:
__init__(control_id, value[, applied_sequence])Attributes:
- __init__(control_id, value, applied_sequence=0)#
-
applied_sequence:
int= 0#
-
control_id:
str#
-
value:
float#
- class embodichain.lab.visualization.JointControlCommand[source]#
Immutable browser joint command consumed on the simulation thread.
Methods:
__init__(run_id, scene_revision, sequence, ...)Attributes:
- __init__(run_id, scene_revision, sequence, client_id, control_id, value, schema_version=5)#
-
client_id:
str#
-
control_id:
str#
-
run_id:
str#
-
scene_revision:
int#
-
schema_version:
int= 5#
-
sequence:
int#
-
value:
float#
- class embodichain.lab.visualization.JointControlProvider[source]#
Simulation-thread source of optional articulation joint controls.
Methods:
__init__(*args, **kwargs)Return the static controls to include in the next manifest.
Return current values ordered independently of backend state.
- __init__(*args, **kwargs)#
- joint_control_specs()[source]#
Return the static controls to include in the next manifest.
- Return type:
tuple[JointControlSpec,...]
- joint_control_states()[source]#
Return current values ordered independently of backend state.
- Return type:
tuple[JointControlState,...]
Overlays and Camera Preview#
- class embodichain.lab.visualization.SceneOverlays[source]#
All optional overlays attached to a dynamic frame.
Methods:
__init__([frames, trajectories, targets, ...])Attributes:
- __init__(frames=<factory>, trajectories=<factory>, targets=<factory>, point_clouds=<factory>)#
-
frames:
tuple[FrameOverlay,...]#
-
point_clouds:
tuple[PointCloudOverlay,...]#
-
targets:
tuple[TargetOverlay,...]#
-
trajectories:
tuple[TrajectoryOverlay,...]#
- class embodichain.lab.visualization.FrameOverlay[source]#
Coordinate frame overlay.
Methods:
__init__(overlay_id, position, wxyz[, ...])Attributes:
- __init__(overlay_id, position, wxyz, axes_length=0.15, axes_radius=0.006, visible=True)#
-
axes_length:
float= 0.15#
-
axes_radius:
float= 0.006#
-
overlay_id:
str#
-
position:
ndarray#
-
visible:
bool= True#
-
wxyz:
ndarray#
- class embodichain.lab.visualization.TargetOverlay[source]#
Target pose overlay rendered as a coordinate frame.
Methods:
__init__(overlay_id, position, wxyz[, ...])Attributes:
- __init__(overlay_id, position, wxyz, axes_length=0.2, visible=True)#
-
axes_length:
float= 0.2#
-
overlay_id:
str#
-
position:
ndarray#
-
visible:
bool= True#
-
wxyz:
ndarray#
- class embodichain.lab.visualization.TrajectoryOverlay[source]#
Polyline trajectory overlay.
Methods:
__init__(overlay_id, points[, color, ...])Attributes:
- __init__(overlay_id, points, color=(255, 170, 30), line_width=3.0, visible=True)#
-
color:
tuple[int,int,int] = (255, 170, 30)#
-
line_width:
float= 3.0#
-
overlay_id:
str#
-
points:
ndarray#
-
visible:
bool= True#
- class embodichain.lab.visualization.PointCloudOverlay[source]#
Point cloud overlay with per-cloud or per-point RGB colors.
Methods:
__init__(overlay_id, points[, colors, ...])Attributes:
- __init__(overlay_id, points, colors=(90, 200, 255), point_size=0.01, visible=True)#
-
colors:
ndarray|tuple[int,int,int] = (90, 200, 255)#
-
overlay_id:
str#
-
point_size:
float= 0.01#
-
points:
ndarray#
-
visible:
bool= True#
- class embodichain.lab.visualization.CameraSpec[source]#
Static pinhole-camera parameters for one environment instance.
Methods:
__init__(camera_id, sensor_uid, env_id, ...)Attributes:
- __init__(camera_id, sensor_uid, env_id, path, fov_y, aspect, near, far, role='sensor')#
-
aspect:
float#
-
camera_id:
str#
-
env_id:
int#
-
far:
float#
-
fov_y:
float#
-
near:
float#
-
path:
str#
-
role:
Literal['sensor','record'] = 'sensor'#
-
sensor_uid:
str#
- class embodichain.lab.visualization.CameraImage[source]#
One detached RGB image associated with a manifest camera.
Methods:
__init__(camera_id, image)Attributes:
- __init__(camera_id, image)#
-
camera_id:
str#
-
image:
ndarray#
- class embodichain.lab.visualization.CameraImageFrame[source]#
Low-frequency RGB images captured at one simulation timestamp.
Methods:
__init__(run_id, scene_revision, sequence, ...)Attributes:
- __init__(run_id, scene_revision, sequence, sim_step, sim_time, images, wall_time=<factory>, schema_version=5)#
-
images:
tuple[CameraImage,...]#
-
run_id:
str#
-
scene_revision:
int#
-
schema_version:
int= 5#
-
sequence:
int#
-
sim_step:
int#
-
sim_time:
float#
-
wall_time:
str#
Pose Conversion#
- embodichain.lab.visualization.pose_to_position_wxyz(pose)[source]#
Split pose arrays into positions and normalized wxyz quaternions.
The accepted layouts are
(..., 7)in EmbodiChain’s(x, y, z, qw, qx, qy, qz)convention or homogeneous(..., 4, 4)matrices. This is the single conversion boundary used by scene exporters.- Parameters:
pose (
object) – Pose or batch of poses.- Return type:
tuple[ndarray,ndarray]- Returns:
A pair of float32 arrays containing positions and wxyz quaternions.
- Raises:
ValueError – If the shape is unsupported or a quaternion is degenerate.
Command-Line Helpers#
- embodichain.lab.visualization.add_viser_args_to_parser(parser)[source]#
Add the standard EmbodiChain Viser command-line options.
- Parameters:
parser (
ArgumentParser) – Parser receiving the Viser options.- Return type:
None
- embodichain.lab.visualization.visualization_cfg_from_args(args)[source]#
Build visualization configuration from parsed CLI arguments.
- Parameters:
args (
Namespace) – Namespace populated byadd_viser_args_to_parser().- Return type:
- Returns:
Visualization configuration including Viser server settings.