Language

unilab.visualization.interactive_playback¶

Shared core for interactive policy playback entrypoints.

Functions

`available_backends_for_task`(task_name)	Return the registered backends for a task, or `()` when unknown.
`build_offpolicy_env_cfg_override`(algo_name, ...)	Build the task env override for off-policy training.
`build_offpolicy_play_env_cfg_override`(...)	Build the off-policy play override, including backend render intent.
`build_play_actor`(algo_name, cfg, *, obs_dim, ...)	Build the policy actor selected by an off-policy owner config.
`build_play_backend_adapter`(cfg, *, root_dir)	Build the BackendAdapter used by play entrypoints to derive env cfg overrides.
`build_playback_config`(args, *[, num_envs])	Build an `RslRlPlaybackConfig` from a play-args-like object.
`create_appo_playback_session`(*, ...[, log])	Create an APPO interactive playback session.
`create_rsl_rl_playback_session`(*, ...[, ...])	Create a playback session and load the selected policy checkpoint.
`create_sac_playback_session`(*, playback_cfg, ...)	Create an interactive playback session for off-policy actors.
`default_device`(torch_module[, preferred])	Resolve runtime device with optional user override.
`extract_play_obs`(obs_dict)
`extract_reset_obs`(reset_result)	Extract obs_dict from env.reset(...) using the current (obs_dict, info_dict) contract.
`infer_checkpoint_actor_input_dim`(ckpt_path)	Infer the actor MLP input dim from an rsl_rl checkpoint, if detectable.
`load_play_actor`(algo_name, actor, ...)	Restore an off-policy play actor and its optional observation normalizer.
`make_sim2sim_preflight`(cfg, *, algo_name)	Build a sim2sim contract preflight closure for interactive playback entrypoints.
`normalize_checkpoint_value`(value)	Normalize a raw checkpoint selector value; sentinel values map to `None`.
`prepare_motion_overlay_selection`(env, *, ...)	Resolve body indices used by motion-target and reward-debug overlays.
`resolve_play_actor_spec`(algo_name, cfg, *, ...)	Resolve the actor implementation and model kwargs used by off-policy play.
`resolve_play_obs_dim`(obs_groups_spec)
`resolve_play_obs_dims`(obs_groups_spec)
`select_torch_device`()

Classes

`KeyboardCommander`	Mutable `[vx, vy, vyaw]` velocity command driven by keyboard nudges.
`MotionOverlaySelection`	Cold-path selection of task bodies used by playback overlays.
`OffPolicyPlaybackSession`	Direct env stepping session for SAC-style off-policy actors.
`PlayInteractiveArgs`	Scalar play arguments shared by interactive playback entrypoints.
`PlaybackControls`	Viewer-independent playback control state.
`PlaybackSession`	Viewer-facing session contract shared by all policy families.
`RslRlPlaybackConfig`	Configuration needed to bootstrap an RSL-RL interactive playback session.
`RslRlPlaybackSession`	Policy/action stepping core shared by native and web viewers.

class unilab.visualization.interactive_playback.KeyboardCommander[source]¶

Bases: object

Mutable [vx, vy, vyaw] velocity command driven by keyboard nudges.

Per-axis nudges stack and are clamped to the task’s commands.vel_limit.

Parameters:

low (ndarray)
high (ndarray)
step_lin (float)
step_ang (float)

low: ndarray¶

high: ndarray¶

step_lin: float = 0.1¶

step_ang: float = 0.2¶

command: ndarray¶

AXIS_VX: ClassVar[int] = 0¶

AXIS_VY: ClassVar[int] = 1¶

AXIS_VYAW: ClassVar[int] = 2¶

classmethod from_vel_limit(vel_limit, *, step_lin=0.1, step_ang=0.2)[source]¶

Parameters:

vel_limit (Any)
step_lin (float)
step_ang (float)

Return type:

KeyboardCommander

nudge(axis, sign)[source]¶

Parameters:

axis (int)
sign (float)

Return type:

None

zero()[source]¶

Return type:: None

describe()[source]¶

Return type:: str

__init__(low, high, step_lin=0.1, step_ang=0.2)¶

Parameters:

low (ndarray)
high (ndarray)
step_lin (float)
step_ang (float)

class unilab.visualization.interactive_playback.MotionOverlaySelection[source]¶

Bases: object

Cold-path selection of task bodies used by playback overlays.

Parameters:

enabled (bool)
selected_indices (ndarray)

enabled: bool¶

selected_indices: ndarray¶

__init__(enabled, selected_indices)¶

Parameters:

enabled (bool)
selected_indices (ndarray)

class unilab.visualization.interactive_playback.OffPolicyPlaybackSession[source]¶

Bases: object

Direct env stepping session for SAC-style off-policy actors.

Parameters:

env (Any)
device (str)
action_mode (str)
actor (Any | None)
actor_algo_type (str)
normalizer (Any | None)
num_envs (int)
obs_extractor (Callable[[dict[str, ndarray]], ndarray])

__init__(*, env, device, action_mode, actor, actor_algo_type, normalizer, num_envs, obs_extractor)[source]¶

Parameters:

env (Any)
device (str)
action_mode (str)
actor (Any | None)
actor_algo_type (str)
normalizer (Any | None)
num_envs (int)
obs_extractor (Callable[[dict[str, ndarray]], ndarray])

reset()[source]¶

Return type:: ndarray

step_once()[source]¶

Return type:: ndarray

advance(controls)[source]¶

Parameters:: controls (PlaybackControls)
Return type:: bool

physics_state()[source]¶

Return type:: ndarray

property info: dict[str, Any]¶

class unilab.visualization.interactive_playback.PlayInteractiveArgs[source]¶

Bases: object

Scalar play arguments shared by interactive playback entrypoints.

Parameters:

task (str)
load_run (str)
checkpoint (str | None)
action_mode (str)
policy_obs_mode (str)
algo_log_name (str)
log_root (str | None)
show_target_bodies (bool)
show_reward_debug (bool)
target_show_axes (bool)
target_body_names (str)
target_max_bodies (int)
target_marker_radius (float)
target_axis_length (float)
target_marker_alpha (float)
reward_debug_show_velocity (bool)
reward_debug_lin_vel_scale (float)
reward_debug_ang_vel_scale (float)
reward_debug_show_connectors (bool)
reward_debug_show_global_anchor (bool)
camera_follow_body (bool)
camera_focus_body_name (str)
camera_height_offset (float)
camera_distance (float | None)
camera_elevation (float | None)
camera_azimuth (float | None)
use_env_visual_model (bool)
speed (float)
start_paused (bool)
keyboard (bool)
keyboard_step_lin (float)
keyboard_step_ang (float)
require_keyboard_command_obs (bool)
algo (str)
sim (str)

task: str¶

load_run: str¶

checkpoint: str | None¶

action_mode: str¶

policy_obs_mode: str¶

algo_log_name: str¶

log_root: str | None¶

show_target_bodies: bool¶

show_reward_debug: bool¶

target_show_axes: bool¶

target_body_names: str¶

target_max_bodies: int¶

target_marker_radius: float¶

target_axis_length: float¶

target_marker_alpha: float¶

reward_debug_show_velocity: bool¶

reward_debug_lin_vel_scale: float¶

reward_debug_ang_vel_scale: float¶

reward_debug_show_connectors: bool¶

reward_debug_show_global_anchor: bool¶

camera_follow_body: bool¶

camera_focus_body_name: str¶

camera_height_offset: float¶

camera_distance: float | None¶

camera_elevation: float | None¶

camera_azimuth: float | None¶

use_env_visual_model: bool¶

speed: float¶

start_paused: bool¶

keyboard: bool = False¶

keyboard_step_lin: float = 0.1¶

keyboard_step_ang: float = 0.2¶

require_keyboard_command_obs: bool = True¶

algo: str = 'ppo'¶

sim: str = 'mujoco'¶

__init__(task, load_run, checkpoint, action_mode, policy_obs_mode, algo_log_name, log_root, show_target_bodies, show_reward_debug, target_show_axes, target_body_names, target_max_bodies, target_marker_radius, target_axis_length, target_marker_alpha, reward_debug_show_velocity, reward_debug_lin_vel_scale, reward_debug_ang_vel_scale, reward_debug_show_connectors, reward_debug_show_global_anchor, camera_follow_body, camera_focus_body_name, camera_height_offset, camera_distance, camera_elevation, camera_azimuth, use_env_visual_model, speed, start_paused, keyboard=False, keyboard_step_lin=0.1, keyboard_step_ang=0.2, require_keyboard_command_obs=True, algo='ppo', sim='mujoco')¶

Parameters:

task (str)
load_run (str)
checkpoint (str | None)
action_mode (str)
policy_obs_mode (str)
algo_log_name (str)
log_root (str | None)
show_target_bodies (bool)
show_reward_debug (bool)
target_show_axes (bool)
target_body_names (str)
target_max_bodies (int)
target_marker_radius (float)
target_axis_length (float)
target_marker_alpha (float)
reward_debug_show_velocity (bool)
reward_debug_lin_vel_scale (float)
reward_debug_ang_vel_scale (float)
reward_debug_show_connectors (bool)
reward_debug_show_global_anchor (bool)
camera_follow_body (bool)
camera_focus_body_name (str)
camera_height_offset (float)
camera_distance (float | None)
camera_elevation (float | None)
camera_azimuth (float | None)
use_env_visual_model (bool)
speed (float)
start_paused (bool)
keyboard (bool)
keyboard_step_lin (float)
keyboard_step_ang (float)
require_keyboard_command_obs (bool)
algo (str)
sim (str)

class unilab.visualization.interactive_playback.PlaybackControls[source]¶

Bases: object

Viewer-independent playback control state.

Parameters:

paused (bool)
speed (float)

paused: bool = False¶

speed: float = 1.0¶

pause()[source]¶

Return type:: None

resume()[source]¶

Return type:: None

toggle_pause()[source]¶

Return type:: bool

request_single_step(count=1)[source]¶

Parameters:: count (int)
Return type:: None

set_speed(value)[source]¶

Parameters:: value (float)
Return type:: None

consume_step_permission()[source]¶

Return type:: bool

target_dt(ctrl_dt)[source]¶

Parameters:: ctrl_dt (float)
Return type:: float

__init__(paused=False, speed=1.0)¶

Parameters:

paused (bool)
speed (float)

class unilab.visualization.interactive_playback.PlaybackSession[source]¶

Bases: Protocol

Viewer-facing session contract shared by all policy families.

env: Any¶

reset()[source]¶

Return type:: Any

advance(controls)[source]¶

Parameters:: controls (PlaybackControls)
Return type:: bool

physics_state()[source]¶

Return type:: ndarray

property info: dict[str, Any]¶

__init__(*args, **kwargs)¶

class unilab.visualization.interactive_playback.RslRlPlaybackConfig[source]¶

Bases: object

Configuration needed to bootstrap an RSL-RL interactive playback session.

Parameters:

task (str)
load_run (str)
checkpoint (str | None)
action_mode (str)
policy_obs_mode (str)
algo_log_name (str)
log_root (str | None)
num_envs (int)
speed (float)
start_paused (bool)

task: str¶

load_run: str¶

checkpoint: str | None¶

action_mode: str¶

policy_obs_mode: str¶

algo_log_name: str¶

log_root: str | None¶

num_envs: int = 1¶

speed: float = 1.0¶

start_paused: bool = False¶

__init__(task, load_run, checkpoint, action_mode, policy_obs_mode, algo_log_name, log_root, num_envs=1, speed=1.0, start_paused=False)¶

Parameters:

task (str)
load_run (str)
checkpoint (str | None)
action_mode (str)
policy_obs_mode (str)
algo_log_name (str)
log_root (str | None)
num_envs (int)
speed (float)
start_paused (bool)

class unilab.visualization.interactive_playback.RslRlPlaybackSession[source]¶

Bases: object

Policy/action stepping core shared by native and web viewers.

Parameters:

env (Any)
wrapped_env (Any)
device (str)
action_mode (str)
policy (Callable[[Any], Any] | None)
num_envs (int)
runner (Any | None)
actor (Any | None)

__init__(*, env, wrapped_env, device, action_mode, policy, num_envs, runner=None, actor=None)[source]¶

Parameters:

env (Any)
wrapped_env (Any)
device (str)
action_mode (str)
policy (Callable[[Any], Any] | None)
num_envs (int)
runner (Any | None)
actor (Any | None)

reset()[source]¶

Return type:: Any

step_once()[source]¶

Return type:: Any

advance(controls)[source]¶

Parameters:: controls (PlaybackControls)
Return type:: bool

physics_state()[source]¶

Return type:: ndarray

property info: dict[str, Any]¶

unilab.visualization.interactive_playback.available_backends_for_task(task_name)[source]¶

Return the registered backends for a task, or () when unknown.

Parameters:: task_name (str)
Return type:: tuple[str, ...]

unilab.visualization.interactive_playback.build_play_backend_adapter(cfg, *, root_dir, algo_name='ppo')[source]¶

Build the BackendAdapter used by play entrypoints to derive env cfg overrides.

Parameters:

cfg (Any)
root_dir (str | Path)
algo_name (str)

Return type:

Any

unilab.visualization.interactive_playback.build_playback_config(args, *, num_envs=1)[source]¶

Build an RslRlPlaybackConfig from a play-args-like object.

Parameters:

args (Any)
num_envs (int)

Return type:

RslRlPlaybackConfig

unilab.visualization.interactive_playback.create_appo_playback_session(*, playback_cfg, cfg, rl_cfg, env_factory, root_dir, device, wrapper_cls, log=<built-in function print>)[source]¶

Create an APPO interactive playback session.

Parameters:

playback_cfg (RslRlPlaybackConfig)
cfg (Any)
rl_cfg (dict[str, Any])
env_factory (Callable[[int], Any])
root_dir (str | Path)
device (str | None)
wrapper_cls (Any)
log (Callable[[str], None])

Return type:

tuple[RslRlPlaybackSession, str, str | None]

unilab.visualization.interactive_playback.create_rsl_rl_playback_session(*, playback_cfg, env_factory, algo_config, root_dir, device, checkpoint_resolver, checkpoint_input_dim_reader, entrypoint_log_root, wrapper_cls, runner_cls, policy_obs_dims_getter, train_cfg_normalizer, sim2sim_preflight=None, runner_loader=None, guard_algo_name=None, log=<built-in function print>)[source]¶

Create a playback session and load the selected policy checkpoint.

Parameters:

playback_cfg (RslRlPlaybackConfig)
env_factory (Callable[[int], Any])
algo_config (dict[str, Any])
root_dir (str | Path)
device (str | None)
checkpoint_resolver (Callable[[str, str, str | None, str, str | None], str | None])
checkpoint_input_dim_reader (Callable[[str], int | None])
entrypoint_log_root (Callable[..., Path])
wrapper_cls (Any)
runner_cls (Any)
policy_obs_dims_getter (Callable[[Any], tuple[int, int]])
train_cfg_normalizer (Callable[[dict[str, Any]], dict[str, Any]])
sim2sim_preflight (Callable[[str | None], Any] | None)
runner_loader (Callable[[Any, str], None] | None)
guard_algo_name (str | None)
log (Callable[[str], None])

Return type:

tuple[RslRlPlaybackSession, str, str | None]

unilab.visualization.interactive_playback.create_sac_playback_session(*, playback_cfg, cfg, env_factory, root_dir, device, algo_name='sac', log=<built-in function print>)[source]¶

Create an interactive playback session for off-policy actors.

Parameters:

playback_cfg (RslRlPlaybackConfig)
cfg (Any)
env_factory (Callable[[int], Any])
root_dir (str | Path)
device (str | None)
algo_name (str)
log (Callable[[str], None])

Return type:

tuple[OffPolicyPlaybackSession, str, str | None]

unilab.visualization.interactive_playback.infer_checkpoint_actor_input_dim(ckpt_path)[source]¶

Infer the actor MLP input dim from an rsl_rl checkpoint, if detectable.

Parameters:: ckpt_path (str)
Return type:: int | None

unilab.visualization.interactive_playback.make_sim2sim_preflight(cfg, *, algo_name)[source]¶

Build a sim2sim contract preflight closure for interactive playback entrypoints.

Returns None when no Hydra config is available (legacy non-Hydra path). Old runs without a contract snapshot keep the fallback + warning semantics of resolve_sim2sim_config().

Parameters:

cfg (Any)
algo_name (str)

Return type:

Callable[[str | None], Any] | None

unilab.visualization.interactive_playback.normalize_checkpoint_value(value)[source]¶

Normalize a raw checkpoint selector value; sentinel values map to None.

Parameters:: value (object)
Return type:: str | None

unilab.visualization.interactive_playback.prepare_motion_overlay_selection(env, *, show_target_bodies, show_reward_debug, target_body_names, target_max_bodies, log=<built-in function print>)[source]¶

Resolve body indices used by motion-target and reward-debug overlays.

Parameters:

env (Any)
show_target_bodies (bool)
show_reward_debug (bool)
target_body_names (str)
target_max_bodies (int)
log (Callable[[str], None])

Return type:

MotionOverlaySelection

unilab.visualization.interactive_playback.select_torch_device()[source]¶

Return type:: str