API Reference

Core Classes

LuckyRobots

class luckyrobots.LuckyRobots(host: str = None, port: int = None)

Main LuckyRobots node for managing robot communication and control

__init__(host: str = None, port: int = None)

host = 'localhost'

port = 3000

robot_client = None

world_client = None

static set_host(ip_address: str) → None

Set the host address for the LuckyRobots node

static get_robot_config(robot: str = None) → dict

Get the configuration for the LuckyRobots node

register_node(node: Node) → None

Register a node with the LuckyRobots node

start(scene: str = 'ArmLevel', robot: str = 'so100', task: str = 'pickandplace', observation_type: str = 'pixels_agent_pos', game_path: str = None) → None

Start the LuckyRobots node

wait_for_world_client(timeout: float = 60.0) → bool

Wait for the world client to connect to the websocket server

async handle_reset(request: ResetRequest) → ResetResponse

Handle the reset request by forwarding to the world client

async handle_step(request: StepRequest) → StepResponse

Handle the step request by forwarding to the world client

spin() → None

Spin the LuckyRobots node to keep it running

shutdown() → None

Shutdown the LuckyRobots node and clean up resources

Node

class luckyrobots.Node(name: str, namespace: str = '', host: str = None, port: int = None)

__init__(name: str, namespace: str = '', host: str = None, port: int = None)

get_qualified_name(name: str) → str

Get the qualified name for a given name

create_publisher(message_type: Type, topic: str, queue_size: int = 10) → Publisher

Create a publisher for a given topic

create_subscription(message_type: Type, topic: str, callback: Callable[[Any], None], queue_size: int = 10) → Subscriber

Create a subscriber for a given topic

create_client(service_type: Type, service_name: str) → ServiceClient

Create a client for a given service

async create_service(service_type: Type, service_name: str, handler: Callable[[Any], Any]) → ServiceServer

Create a service for a given service name

create_service_client(service_type: Type, service_name: str, host: str = 'localhost', port: int = 3000) → ServiceClient

Create a service client for a given service name

start() → None

Start the node

spin() → None

Spin the node

shutdown() → None

Shutdown the node

Models

class luckyrobots.ObservationModel(*, ObservationState: Dict[str, float], ObservationCameras: List[CameraData] | None = None)

Observation model that matches the JSON structure

observation_state: Dict[str, float]

observation_cameras: List[CameraData] | None

process_all_cameras() → None

Process all camera images in the observation

class Config

populate_by_name = True

model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Message Types

class luckyrobots.Reset

Reset the robot

Request

alias of ResetRequest

Response

alias of ResetResponse

class luckyrobots.Step

Step the robot with an action

Request

alias of StepRequest

Response

alias of StepResponse