lekit-robot

Small Python SDK for a running Lekit Robot Node. Discover a robot by name, read joint/pose feedback and RGB/depth images, and send absolute motion targets. The client does not install LeRobot, Torch, OpenCV, robot drivers, or camera drivers. Python 3.10+ is required.

Install

After publication:

pip install lekit-robot

From this repository before publication:

pip install ./packages/lekit_robot

Robot Node must already be running on the robot computer. Clients need network access to its Zenoh peer; the browser's HTTP port 8080 is not the SDK endpoint. LAN multicast discovery is the default. If discovery is blocked, configure both Node (lekit-robot --zenoh-router tcp/ROUTER_IP:7447) and client (connect(name="NERO Dual", router="tcp/ROUTER_IP:7447")) to use an existing Zenoh router. This package does not install the Robot Node command-line server.

Observe (does not acquire control)

from lekit_robot import connect

with connect(name="NERO Dual", timeout=5) as robot:
    obs = robot.get_observation(images=True, timeout=5)
    print(obs.robot.status, obs.robot.state)
    camera = obs.camera(name="head_view")
    if camera.color is not None:
        print(camera.color.shape)  # H x W x 3, RGB uint8
    print(robot.descriptor.joint_fields)

images=False omits images. robot.observations(images=False, timeout=5) aligns independently published Robot and Camera samples with bounded latest-only buffers. get_observation() is a one-shot convenience over the same stream alignment path; the Node has no separate snapshot publication or observation query protocol. For alignment time T, each source contributes its newest sample captured no later than T; network arrival time is never used as capture time. Camera samples remain independent, so an observation is not a hardware-synchronized frame set. Check each sample's capture time and monotonic timestamp. Depth is a uint16 array in the server's camera units, not automatically metres. The Node's orbbec plugin outputs millimetres (0 means invalid), with depth aligned to RGB by default. Read it with obs.camera(name="head_view").depth; convert to metres with depth.astype("float32") / 1000. The Orbbec SDK is only needed on the server, not on this client. On the wire, aligned depth is always losslessly compressed using the camera's configured LZ4 or Zstandard codec. The SDK transparently decompresses it for images=True; use images="encoded" to retain the compressed payload and call camera.decode_depth() only when needed.

Each observation also carries a small calibration reference that is timestamped with the robot state. Fetch the complete, versioned bundle only when the reference changes, then keep it in the recorder or application cache:

with connect(name="NERO Dual", timeout=5) as robot:
    obs = robot.get_observation(images="encoded", timeout=5)
    ref = obs.calibration
    if ref is not None and ref.bundle_id is not None:
        bundle = robot.get_calibration(ref.bundle_id, timeout=5)
        camera = next(binding for binding in bundle.camera_bindings
                      if binding.camera_name == "head_view")
        print(ref.state, camera.intrinsics, camera.sensor_frame)
        print(bundle.edges)  # Check source_frame, target_frame, status and transform.

robot.get_calibration() returns the current bundle. Passing a bundle_id returns that immutable historical bundle, so recorded observations remain reproducible after a later recalibration. A reference state of degraded or invalid means the bundle is available but one or more transform edges no longer match the active camera bindings; downstream code decides whether those edges are usable. For the geometry of a delivered image, read obs.camera(name="head_view").geometry. If no validated bundle is active, get_calibration() raises CalibrationNotFoundError.

Control

Control requires enabled motors, allow_control=True, fresh feedback, normal mode, and no emergency stop. Close Teleop before acquiring external control. The context manager holds and renews an exclusive control lease and releases it on exit. Multiple clients may observe; only one controller may own the lease.

The following function sends a real motion command when called:

from lekit_robot import connect

def move_left_joint3(delta_rad):
    with connect(name="NERO Dual") as robot:
        with robot.control():
            obs = robot.get_observation(images=False)
            state = obs.robot.state
            if obs.robot.status != "connected" or not state or not state.get("left.feedback_fresh"):
                raise RuntimeError("Fresh left arm feedback is required")
            joints = [state[f"left.joint_{i}"] for i in range(1, 8)]
            joints[2] += delta_rad
            return robot.set_joints(joints, side="left")

For NERO Dual, left is can1 and right is can0. Joint/rotation units are radians; position and gripper width units are metres. Pose fields represent the flange in that arm's base frame. For other robots use the plugin-declared field semantics. side selects only that arm; the other arm receives no target. Without side, provide the complete advertised target (14 joint values or 12 pose values for NERO). A scalar gripper width requires side on a dual-arm robot. There is no implicit broadcast of one pose or gripper value to both arms.

NERO uses nero-dual-ik for Cartesian targets without an SDK or firmware IK fallback. stream_joints() and set_target(mode="stream") use the Node's latest-only stream channel and arm its loss-of-stream watchdog. Their immediate receipt has status="submitted"; it does not say whether the asynchronous device validation accepted the target. Streaming does not guarantee a hard realtime servo or a straight path. Unsupported modes, partial targets and out-of-range values are rejected.

A receipt acknowledges command submission, not arrival or successful IK. Continue observing measured state and alarms. ActionOutcomeUnknownError means a command may already have executed: do not retry it automatically. A control lease is ownership arbitration, not a hardware emergency stop or a stop-on-disconnect guarantee. Use the Robot Node emergency stop or physical stop as appropriate.

For a joint target that needs confirmed arrival, use a device-owned motion task:

import time

from lekit_robot import connect

with connect(name="NERO Dual", timeout=5) as robot:
    task = robot.start_joint_task({"left.joint_3.pos": 0.2}, timeout_s=15)
    while not task.done:
        time.sleep(0.1)
        task = robot.get_motion_task(task.task_id)
    if task.state != "succeeded":
        raise RuntimeError(f"Motion task ended as {task.state}: {task.error}")

A task acquires its own exclusive control lease; do not enter robot.control() around it. start_joint_task() returns an accepted task, not a completion guarantee. cancel_motion_task(id) requests a stop; poll until cancelled with stop_confirmed=True or inspect failed. If the start reply is lost, MotionTaskOutcomeUnknownError.task_id identifies the task to query before retrying. The Node currently supports joint tasks only.

Compatibility

The wire protocol remains lekit/v1. New clients opt into the gripper manifest extension; old clients continue to receive the original manifest shape. An old Node can still supply observations and existing targets, but gripper commands need an updated Node; unsupported grippers raise UnsupportedCommandError locally. Per-arm commands also require the updated Node validation. Update client and Node together for complete NERO Dual support. Reconnect after replacing the Node or its robot configuration; the SDK never silently rebinds an existing control session.

Existing applications may keep from lekit.robot import connect when the main project is installed: those imports re-export this package's implementation.

Build and release

From packages/lekit_robot:

python -m pip install build twine
python -m build
python -m twine check dist/*
python -m twine upload --repository testpypi dist/*
# After testing the installed wheel in a clean environment:
python -m twine upload dist/*

Publishing needs a PyPI account and an API token or configured Trusted Publisher. Never store credentials in source files. Version 0.1.0 is the initial release; subsequent releases must use a new version number.

License: Apache-2.0, matching the repository's existing standalone package convention.

A build/test and manual publishing workflow is provided at .github/workflows/lekit-robot-release.yml. It installs the built wheel in an isolated environment before testing. Pushes and pull requests only build/test. After the workflow is committed to the default branch, configure a PyPI pending Trusted Publisher for a new project:

• Project: lekit-robot
• Owner: sorelferris
• Repository: lekit
• Workflow: lekit-robot-release.yml
• Environment: pypi (or testpypi on TestPyPI)

Then run the workflow manually, selecting testpypi first, and pypi for the release. Configure the two registries separately. This avoids storing a long-lived PyPI token. See https://docs.pypi.org/trusted-publishers/creating-a-project-through-oidc/.