API Reference

Core Classes

LuckyRobots

LuckyEngineClient

class luckyrobots.LuckyEngineClient(host: str = '127.0.0.1', port: int = 50051, timeout: float = 5.0, *, robot_name: str | None = None)

Client for connecting to the LuckyEngine gRPC server.

Provides access to gRPC services for RL training: - AgentService: stepping, resets - SceneService: simulation mode control - MujocoService: health checks, joint state

Usage:

client = LuckyEngineClient(host=”127.0.0.1”, port=50051) client.connect() client.wait_for_server()

schema = client.get_agent_schema() obs = client.step(actions=[0.0] * 12)

client.close()

__init__(host: str = '127.0.0.1', port: int = 50051, timeout: float = 5.0, *, robot_name: str | None = None) → None

Initialize the LuckyEngine gRPC client.

Parameters:

• host – gRPC server host address.

• port – gRPC server port.

• timeout – Default timeout for RPC calls in seconds.

• robot_name – Default robot name for calls that require it.

connect() → None

Connect to the LuckyEngine gRPC server.

Opens the gRPC channel. Service stubs are created lazily on first access so importers can skip unused services without cost.

Raises:

GrpcConnectionError – If connection fails.

close() → None

Close the gRPC channel.

is_connected() → bool

Check if the client is connected.

health_check(timeout: float | None = None) → bool

Perform a health check by calling GetMujocoInfo.

Parameters:

timeout – Timeout in seconds (uses default if None).

Returns:

True if server responds, False otherwise.

wait_for_server(timeout: float = 30.0, poll_interval: float = 0.5) → bool

Wait for the gRPC server to become available.

Parameters:

• timeout – Maximum time to wait in seconds.

• poll_interval – Time between connection attempts.

Returns:

True if server became available, False if timeout.

property pb: Any

Access protobuf modules grouped by domain (e.g., client.pb.scene).

property robot_name: str | None

Default robot name used by calls that accept an optional robot_name.

set_robot_name(robot_name: str) → None

Set the default robot name used by calls that accept an optional robot_name.

property channel: Channel

The underlying gRPC channel.

Exposed so users can attach their own service stubs using the same connection (shared keep-alive, auth, etc.):

client.connect() my_stub = my_pb2_grpc.MyServiceStub(client.channel)

property scene: Any

SceneService stub (lazy).

property mujoco: Any

MujocoService stub (lazy) — agent-scoped joint state.

For the full, engine-wide MuJoCo model (every joint, every actuator), use mujoco_scene instead.

property mujoco_scene: Any

MujocoSceneService stub (lazy) — engine-wide MuJoCo access.

Exposes every joint, actuator, and the full qpos/qvel/ctrl vectors in the loaded model. Use the higher-level helpers get_model_info(), get_full_state(), and set_ctrl() for common operations.

property agent: Any

AgentService stub (lazy).

property debug: Any

DebugService stub (lazy).

property camera: Any

CameraService stub (lazy).

property telemetry: Any

TelemetryService stub (lazy) — lightweight qpos + ctrl streaming.

property viewport: Any

ViewportService stub (lazy) — editor viewport pixel streaming.

property lidar: Any

LidarService stub (lazy) — material-aware lidar scans for training clients.

register_stub(name: str, stub_class: Any) → Any

Attach a third-party stub class to this client’s channel.

Useful when an engine build exposes a service that the shipped SDK doesn’t know about:

client.register_stub(“my_svc”, my_pb2_grpc.MyServiceStub) client.my_svc.DoThing(…)

Parameters:

• name – Attribute name under which the stub is exposed.

• stub_class – Generated *Stub class from a _pb2_grpc module.

Returns:

The instantiated stub (also available as client.<name>).

discover_services() → list[str]

Ask the server which gRPC services it advertises.

Uses grpc.reflection.v1alpha.ServerReflection. Requires the engine to be built with reflection enabled (default in the v0.2.0+ LuckyEngine server). The standard reflection service itself is filtered out of the result.

configure_cameras(cameras: list[dict]) → None

Configure cameras to capture on every Step RPC.

Parameters:

cameras –

List of camera configs. Each dict has keys: name: Camera entity name in the scene. width: Desired image width (0 = native resolution). height: Desired image height (0 = native resolution). kind: “color” (default) or “depth”. Depth frames carry live

metric depth packed as gray16le; convert to metres with metres = code * depth_scale. Requires a camera configured for depth capture.

format: “raw” (default) or “jpeg” (color only).

set_lidar_live(live: bool = True, timeout: float | None = None)

Keep the lidar firing every step, even when you’re not recording.

Call this once before polling scans with get_lidar_scan(); the setting persists across resets. Like the other scene-inspection calls (list_cameras, get_full_state, …), run it while the simulation is idle — not from inside an active step() loop.

get_lidar_scan(sensor: int = 0, material: bool = False, want_secondary: bool = False, timeout: float | None = None)

Read a lidar scan for the given sensor index (0-based).

Returns a LidarScanResponse: beams (channels * azimuth_bins), ranges (metres per beam, -1.0 = no return), and secondary_ranges when want_secondary. Call set_lidar_live(True) once first, then read scans while the simulation is idle — like the other scene-inspection calls, not from inside an active step() loop.

get_lidar_beam_count(sensor: int = 0, timeout: float | None = None) → int

Number of beams (channels * azimuth_bins) for the given lidar sensor.

list_cameras(timeout: float | None = None) → list[dict]

List available cameras in the scene.

Returns:

List of dicts with ‘name’ and ‘id’ keys for each camera.

set_action_group(group_name: str, actions: list[float], action_indices: list[int], agent_name: str = '', timeout: float | None = None) → bool

Preload actions for a named group without triggering a physics step.

Call this for each policy/controller, then call step() to fire them all atomically in one physics tick.

Parameters:

• group_name – Name for this action group (e.g., “lower_body”, “right_arm”).

• actions – Action values for this group.

• action_indices – Which indices in the action vector these map to.

• agent_name – Agent name (empty = default agent).

• timeout – RPC timeout in seconds.

Returns:

True if the group was preloaded successfully.

get_joint_state(robot_name: str = '', timeout: float | None = None)

Get current joint state (positions and velocities).

Parameters:

• robot_name – Robot entity name (uses default if empty).

• timeout – RPC timeout in seconds.

Returns:

GetJointStateResponse with state.positions (qpos) and state.velocities (qvel).

get_mujoco_info(robot_name: str = '', timeout: float | None = None)

Get MuJoCo model information (joint names, limits, etc.).

get_model_info(timeout: float | None = None)

Introspect the full loaded MuJoCo model.

Returns a GetModelInfoResponse whose joints and actuators lists cover every entry in mjModel — not just those declared by the registered RL agent. Use this to discover fingers, non-agent joints, and actuator names the agent API doesn’t expose.

get_full_state(*, include_qpos: bool = True, include_qvel: bool = True, include_ctrl: bool = True, timeout: float | None = None)

Snapshot the complete mjData state (qpos, qvel, ctrl, time).

stream_full_state(*, target_fps: int = 30, include_qpos: bool = True, include_qvel: bool = True, include_ctrl: bool = True)

Iterate GetFullStateResponse messages at approximately target_fps.

Example

for resp in client.stream_full_state(target_fps=60):

qpos = list(resp.state.qpos)

set_ctrl(values: Any, *, skip_range_clamp: bool = False, wait_for_next_step: bool = False, timeout: float | None = None)

Write actuator control values.

Accepts three input shapes:

• A flat sequence of floats: bulk write starting at index 0.

• A dict[str, float]: write by actuator name.

• A dict[int, float]: write by actuator index.

Actuators currently owned by an active RL agent are refused and returned in SetControlResponse.rejected_actuators.

list_all_joints(timeout: float | None = None) → list[dict]

Return lightweight dicts for every joint in the loaded MuJoCo model.

Convenience wrapper around get_model_info(). Each entry:

{“index”: int, “name”: str, “type”: int, “qpos_adr”: int,

“qvel_adr”: int, “limited”: bool, “range”: (lo, hi)}

list_all_actuators(timeout: float | None = None) → list[dict]

Return lightweight dicts for every actuator in the loaded MuJoCo model.

get_agent_schema(agent_name: str = '', timeout: float | None = None)

Get agent schema (observation/action sizes and names).

The schema is cached for subsequent step() calls to enable named access to observation values.

Parameters:

• agent_name – Agent name (empty = default agent).

• timeout – RPC timeout.

Returns:

GetAgentSchemaResponse with schema containing observation_names, action_names, observation_size, and action_size.

reset_agent(agent_name: str = '', randomization_cfg: Any | None = None, timeout: float | None = None)

Reset a specific agent.

Parameters:

• agent_name – Agent logical name. Empty string means default agent.

• randomization_cfg – Optional simulation contract config for this reset.

• timeout – Timeout in seconds (uses default if None).

Returns:

ResetAgentResponse with success and message fields.

step(actions: list[float] | None = None, agent_name: str = '', step_timeout_s: float = 0.0, timeout: float | None = None, action_groups: list[dict] | None = None) → ObservationResponse

Synchronous RL step: apply action, wait for physics, return observation.

Parameters:

• actions – Action vector to apply for this step (optional when using action_groups).

• agent_name – Agent name (empty = default agent).

• step_timeout_s – Server-side timeout for waiting for the physics step (seconds). 0 means use server default.

• timeout – RPC timeout in seconds.

• action_groups – Optional list of action group dicts, each with keys: group_name: str, actions: list[float], action_indices: list[int]. Groups are applied on top of actions (if provided) or default positions.

Returns:

ObservationResponse with observation after physics step.

report_progress(*, run_id: str = '', task_name: str = '', policy_name: str = '', phase: str = '', current_episode: int = 0, total_episodes: int = 0, current_step: int = 0, max_steps: int = 0, elapsed_s: float = 0.0, status_text: str = '', finished: bool = False) → None

Report evaluation/training progress to the engine for UI display.

Fire-and-forget: errors are logged but never raised.

get_capability_manifest(robot_name: str = '', scene: str = '', timeout: float | None = None) → dict

Discover what MDP components the engine supports.

Parameters:

• robot_name – Filter by robot (empty = all).

• scene – Filter by scene (empty = all).

• timeout – RPC timeout in seconds.

Returns:

Dict with observations, rewards, terminations, randomizations lists.

validate_task_contract(contract: dict, timeout: float | None = None) → dict

Dry-run a task contract against the engine’s capability registry.

Validates without configuring anything — useful for CLI tooling and pre-flight checks. Errors include actionable suggestions (“did you mean track_angular_velocity?”).

Parameters:

• contract – Task contract dict (same shape as negotiate_task).

• timeout – RPC timeout in seconds.

Returns:

is_valid (bool), errors (list of dicts), warnings (list of dicts), resolved_optionals (list of term names that were resolved), unresolved_optionals.

Return type:

Dict with keys

negotiate_task(contract: dict, timeout: float | None = None) → dict

Validate and configure engine for a task contract.

Parameters:

• contract – Task contract dict with observations, rewards, terminations, etc.

• timeout – RPC timeout in seconds.

Returns:

Dict with session_id, reward_terms, termination_terms on success.

Raises:

RuntimeError – If contract validation fails.

set_simulation_mode(mode: str = 'fast', timeout: float | None = None)

Set simulation timing mode.

Parameters:

• mode – “realtime”, “deterministic”, or “fast” - realtime: Physics runs at 1x wall-clock speed - deterministic: Physics runs at fixed rate - fast: Physics runs as fast as possible (for RL training)

• timeout – RPC timeout in seconds.

Returns:

SetSimulationModeResponse with success and current mode.

get_simulation_mode(timeout: float | None = None) → str

Query the engine’s current simulation timing mode.

Returns:

One of "realtime", "deterministic", "fast", or "unknown" if the engine returned an unrecognized enum value.

enter_play_mode(timeout: float | None = None)

Trigger the editor’s Edit -> Play transition over gRPC.

These RPCs are session boundaries, NOT pause/resume — entering Play recompiles MuJoCo and may take a moment to become ready. Returns immediately; poll get_agent_schema() or get_model_info() to detect when the simulation is steppable. In standalone (dist) builds there is no Edit/Play distinction and this RPC is a no-op.

Active recordings will be torn down when ExitPlayMode is later called (Play->Edit ends the recording session).

exit_play_mode(timeout: float | None = None)

Trigger the editor’s Play -> Edit transition over gRPC.

See enter_play_mode() for the session-boundary semantics. Any in-flight recording is closed out as part of the transition.

reset_scene(preserve_time: bool = False, timeout: float | None = None)

Soft-reset the active MuJoCo scene back to its authored initial state.

Restores qpos to keyframe[0] (or qpos0 if no keyframe is authored), zeroes velocities/forces/ctrl, and reseeds active PolicyRuntime PD targets so the policies don’t yank the robot back to a stale target on the next substep.

Recording behaviour: recording continues across the reset by design. The first frame captured after the reset has the post_reset bit (= 0x02) set in the new frame_flags column so consumers can drop the qpos/ctrl discontinuity if needed.

Parameters:

• preserve_time – Keep mjData.time intact across the reset. Default (False) zeroes time. Set True when in-flight RL training or data capture is tracking elapsed sim time.

• timeout – RPC timeout in seconds.

get_scene_info(timeout: float | None = None) → dict

Return the active scene’s name, path, and entity count.

list_entities(include_transforms: bool = False, include_components: bool = False, timeout: float | None = None) → list[dict]

Enumerate entities in the active scene.

Parameters:

• include_transforms – Populate per-entity transform.

• include_components – Populate per-entity components (type names).

• timeout – RPC timeout in seconds.

Returns:

List of dicts with keys id, name, optionally transform, components.

get_entity(name: str | None = None, entity_id: int | None = None, timeout: float | None = None) → dict | None

Look up one entity by name or numeric id.

Returns the same dict shape as list_entities() (always with transform and components populated), or None if not found.

set_entity_transform(entity_id: int, position: tuple[float, float, float] | None = None, rotation: tuple[float, float, float, float] | None = None, scale: tuple[float, float, float] | None = None, timeout: float | None = None) → bool

Set an entity’s transform. Unspecified fields default to identity.

Parameters:

• entity_id – The entity’s numeric id (from list_entities / get_entity).

• position – (x, y, z) world-space position.

• rotation – (x, y, z, w) quaternion.

• scale – (x, y, z) scale.

Returns:

True on success.

get_telemetry_schema(timeout: float | None = None) → dict

Return the telemetry vector schema (observation names, action names, nq, nu).

Telemetry is the lightweight streaming surface — qpos + last-applied ctrl per frame. For full mjData (qpos/qvel/ctrl with filters), use stream_full_state() on the MujocoScene wrapper instead.

stream_telemetry(target_fps: int = 30)

Iterate over server-streamed TelemetryFrame protos.

Each frame carries timestamp_ms, frame_number, observation_qpos (full mjData qpos), and action_ctrl (last-applied ctrl). Cancellation-safe — break out of the loop to terminate the stream.

get_viewport_info(timeout: float | None = None) → dict

List the viewports the engine exposes plus the current stream config.

On this engine branch the server reports a single "Main" viewport.

stream_viewport(viewport_name: str = 'Main', target_fps: int = 30, width: int = 0, height: int = 0, format: str = 'raw')

Iterate over server-streamed ImageFrame protos for a viewport.

Parameters:

• viewport_name – Viewport id ("Main" is the only one on this engine branch).

• target_fps – Desired frame rate; server may clamp.

• height (width /) – Desired resolution. 0 = native.

• format – "raw" (RGBA bytes) or "jpeg".

stream_camera(name: str | None = None, entity_id: int | None = None, target_fps: int = 30, width: int = 0, height: int = 0, format: str = 'raw')

Iterate over server-streamed ImageFrame protos for a camera.

Identify by name (camera entity tag) or by numeric entity id. For synchronous in-step capture use configure_cameras() plus step() instead.

stream_joint_state(robot_name: str | None = None)

Iterate over server-streamed agent-scoped joint state.

Returns positions/velocities only for the joints declared by the registered RobotAgent — not the full mjModel. For full-model streaming use stream_full_state() on the MujocoScene wrapper.

benchmark(duration_seconds: float = 5.0, method: str = 'step', print_results: bool = False) → BenchmarkResult

Benchmark a client method by calling it in a tight loop.

Parameters:

• duration_seconds – How long to run the benchmark.

• method – Method to benchmark. Currently supports “step”.

• print_results – Print results to stdout.

Returns:

BenchmarkResult with timing statistics.

Raises:

ValueError – If method is not recognized.

GrpcConnectionError

class luckyrobots.GrpcConnectionError(message: str)

Raised when gRPC connection fails.

__init__(message: str)

Models

Engine Management

Engine Lifecycle

Engine lifecycle management for LuckyEngine.