rb_sdk

Unified Rainbow Robotics Python SDK. Controls AMRs, manipulators and the RB-Y1 dual-arm robot, and integrates with Rainbow Robot Service (common), through a single API.

Entry points

Internally it runs on a Zenoh pub/sub/query mesh. When running on the same PC as the robot, it works out of the box with no extra configuration.

1. Installation

pip install rainbow-rb-sdk

The PyPI package is published under the rainbow namespace. The examples in this document use the monorepo workspace names (rb_sdk, rb_zenoh, rb_flat_buffers …), so if you installed from PyPI, prefix imports with rainbow..

from rainbow.rb_sdk import RBManipulateSDK                      # workspace: from rb_sdk import ...
from rainbow.rb_flat_buffers.IPC.State_Core import State_CoreT   # workspace: from rb_flat_buffers...

Wheels are published for Linux (x86_64/aarch64) and Windows only. On macOS you can install only the stubs for editor autocompletion and type checking (no runtime import — run on Docker/Linux):

pip install rainbow-rb-sdk-stubs

For RB-Y1 (RBRby1SDK), the official rby1-sdk (C++) is vendored directly into rb_sdk.rb_sdk_cpp, so no separate installation is needed.

The PyPI package provides wheels for Python 3.11–3.13 (monorepo development/build is 3.12 only).

2. Basic usage (running on the robot)

Scripts running on the same PC as the common service can use the SDK as-is with no extra configuration.

from rb_sdk import RBAmrSDK, RBManipulateSDK

amr = RBAmrSDK()
manipulate = RBManipulateSDK()

# At this point ZenohClient automatically connects to 127.0.0.1:7447 (common)
manipulate.move.call_move_j(
    robot_model="C500920",
    target={"tar_values": [0, 0, 0, 0, 0, 0], "tar_frame": 0, "tar_unit": 0},
)

3. Using from an external PC (Remote Connect)

To talk to the robot from a machine other than the robot (e.g. a development laptop), fill in ConnectOptions and call sdk.connect(). connect() is a synchronous call (blocks until registration and mesh check complete), and disconnect() is awaitable.

import asyncio
import time

from rb_sdk import RBManipulateSDK
from rb_sdk.rb_sdk_cpp import ConnectOptions
from rb_flat_buffers.IPC.State_Core import State_CoreT

sdk = RBManipulateSDK()

opts = ConnectOptions()
opts.token = "service token"
opts.host = "192.168.1.100"
# opts.ttl = 30
# opts.zenoh_port = None
# opts.heartbeat = 15.0

# ⚠️  Register subscriptions only after connect().
# connect() reopens the zenoh session, so subscriptions registered before it are lost with the old session.
peer_id = sdk.connect(opts)  # On the robot itself (same PC) this does nothing and returns None


@sdk.zenoh_subscribe("C500920/state_core", flatbuffer_obj_t=State_CoreT)
def on_state_core(*, obj_payload: State_CoreT | None, **kwargs):
    if obj_payload is None:
        return
    print("motion mode:", obj_payload.motionMode)


try:
    # The heartbeat runs on an SDK-internal thread — the process just needs to stay alive.
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    pass
finally:
    asyncio.run(sdk.disconnect())

Key features

• Local IP auto-detection: finds the IP of the interface routing toward host and announces it to the robot
• Automatic port assignment: the OS picks a free port, so no port conflicts
• Automatic heartbeat: a background thread periodically extends the TTL. On script exit / crash, the server cleans up when the TTL expires
• Same-PC detection: if host is localhost or the machine's own IP, connect() does nothing and returns None (already in the mesh)
• Calling it again while registered returns the existing peer id. Raises on registration/reconnection failure

ConnectOptions fields

4. use_directly mode

Passing use_directly=True to the RBBaseSDK constructor routes a method call to {name}_directly if that method exists. Otherwise the original method is called.

class MySDK(RBManipulateSDK):
    def call_move_j_directly(self, *, robot_model, target, **kwargs):
        # Implementation that controls hardware directly, bypassing Zenoh
        ...

# use_directly=True → calling call_move_j routes to call_move_j_directly
sdk = MySDK(use_directly=True)
sdk.move.call_move_j(robot_model="C500920", target={...})  # → calls call_move_j_directly

# use_directly=False (default) or no _directly method → original method is used
sdk2 = MySDK(use_directly=False)
sdk2.move.call_move_j(...)  # → calls call_move_j

use_directly is updated on every constructor call, even if the singleton instance already exists.

5. AMR SDK (RBAmrSDK)

from rb_sdk import RBAmrSDK

sdk = RBAmrSDK()

Submodules

Move examples

# Move to a goal node
sdk.move.send_move_goal(
    robot_model="AMR001",
    goal_type=0,
    goal_node_name="node_A",
)

# Wait for the move result (async)
result = await sdk.move.get_move_result(robot_model="AMR001")

# Jog
sdk.move.send_move_jog(robot_model="AMR001", vx=0.3, vy=0.0, wz=0.0)

# Multi-path
sdk.move.send_multi_path(robot_model="AMR001", path=["node_A", "node_B", "node_C"])

6. Manipulate SDK (RBManipulateSDK)

from rb_sdk import RBManipulateSDK

sdk = RBManipulateSDK()

Submodules

move module — main methods

Move examples

# Joint move
sdk.move.call_move_j(
    robot_model="C500920",
    target={"tar_values": [0, 0, 90, 0, 90, 0], "tar_frame": 0, "tar_unit": 1},
    speed={"spd_mode": 1, "spd_vel_para": 60, "spd_acc_para": 120},
)

# Linear move
sdk.move.call_move_l(
    robot_model="C500920",
    target={"tar_values": [400, 0, 500, 0, 180, 0], "tar_frame": 1, "tar_unit": 0},
)

# Move relative to a reference position
sdk.move.call_move_j(
    robot_model="C500920",
    target={"tar_values": [10, 0, 0, 0, 0, 0], "tar_frame": 0, "tar_unit": 1},
    reference_value=[0, 0, 90, 0, 90, 0],  # current reference value
)

# Power and servo control
sdk.state.call_powercontrol(robot_model="C500920", control=1)
sdk.state.call_servocontrol(robot_model="C500920", control=1)

7. RB-Y1 SDK (RBRby1SDK)

An implementation that binds the official rby1-sdk (C++) directly with pybind11. The pure-Python wrappers (rby1.py, rby1_sdk/*) are deprecated; the same entry point name (RBRby1SDK) now comes from rb_sdk.rb_sdk_cpp (C++ extension). Official types (RobotCommandBuilder, etc.) are accessed via from rb_sdk import rby1.

It is a process singleton. Every public method is an async def by default, with robot_model as the first argument and flow_manager_args as the last, and each has a synchronous counterpart at sdk.sync.<method>.

from rb_sdk import RBRby1SDK, rby1

sdk = RBRby1SDK()

async def main():
    # If endpoint is omitted, env RBY1_HOST is used (default 192.168.30.1:50051). Re-calling with the same endpoint/model is idempotent.
    await sdk.connect_robot(endpoint="192.168.100.51:50051", model="a")
    print(sdk.model, sdk.is_robot_connected())

    await sdk.power_on(robot_model="a", dev_name=".*")
    await sdk.servo_on(robot_model="a", dev_name=".*")

    state = await sdk.get_state(robot_model="a")  # rby1.RobotState_A

    # State stream: a 2-arg callback receives (state, control_manager_state). Coroutine callbacks run on the event loop, latest value only
    async def on_state(state, cms):
        ...
    await sdk.start_state_update(robot_model="a", callback=on_state, rate=20)

    builder = rby1.RobotCommandBuilder()  # build commands with the official builder
    handle = await sdk.send_command(robot_model="a", builder=builder)
    await handle.wait()  # cancelling the await also stops the robot command via the official SDK Cancel

In synchronous contexts (PyFM Steps, etc.), call sdk.sync.<method> directly.

sdk.sync.connect_robot(model="a")
sdk.sync.power_on(robot_model="a", dev_name=".*")
state = sdk.sync.get_state(robot_model="a")

Errors are raised as RBY1Error / RBY1ConnectionError (both RuntimeError subclasses).

Submodule initialization (required for developers)

rby1-sdk/qdldl are vendored as git submodules (backend/packages/rb_sdk/src/rb_sdk/cpp/lib/{rby1-sdk,qdldl}), so they must be initialized after the first clone. Builds run only in the Docker build image (make backend.sdk-core; direct local builds are not supported).

git submodule update --init --recursive
# When bumping the upstream version (e.g. v0.10.0 → next tag)
cd backend/packages/rb_sdk/src/rb_sdk/cpp/lib/rby1-sdk
git fetch --tags && git checkout <tag>
cd - && git submodule update --init --recursive
make backend.sdk-core

The mingw build patch (backend/packages/rb_sdk/src/rb_sdk/cpp/cmake/rby1_sdk_mingw.patch) is applied against the upstream sources, so if the submodule version changes and the patch no longer applies, the CMake configure step fails immediately (in that case, regenerate the patch against the new version).

8. FlowManagerArgs integration (PyFM)

When calling SDK methods from a PyFM Step, passing flow_manager_args automatically handles the done callback, waiting for motion completion, variable binding, and so on.

# Inside a PyFM Step
def execute(self, flow_manager_args):
    sdk.move.call_move_j(
        robot_model="C500920",
        target={"tar_values": [0, 0, 0, 0, 0, 0], "tar_frame": 0, "tar_unit": 0},
        flow_manager_args=flow_manager_args,  # done() callback is called automatically
    )

If flow_manager_args is omitted, the call is blocking.

9. Architecture notes

Per-process singleton

RBBaseSDK (RBAmrSDK, RBManipulateSDK) is a singleton keyed by (pid, cls). Calling the constructor multiple times in the same process returns the same instance. The ZenohClient is also shared, one per PID.

a = RBManipulateSDK()
b = RBManipulateSDK()
assert a is b  # True

Automatic error wrapping

Every public instance method of a class inheriting RBBaseSDK is automatically wrapped in a common try/except by __init_subclass__. On ZenohNoReply / ZenohTransportError, it reconnects automatically and retries once.

Shutting down the SDK

sdk.close()                         # clean up the current instance
RBBaseSDK.close_all_for_pid()       # clean up all SDK instances in the current process

10. Notes

• RBRby1SDK is a process singleton, and there is one robot gRPC connection per process (swap it with connect_robot).
• If the connection targets (zenoh, rby1 endpoint, Muscat HTTP) are not ready, a runtime exception is raised.
• When using the SDK from an external PC, the firewall must allow 7447/TCP (zenoh) and 8000/TCP in both directions.
• use_directly=True is applied even on singleton re-entry. However, it does not propagate to sub-SDKs (sdk.move, etc.), so _directly methods must be defined on the top-level SDK.

11. Subscribe / Queryable

You can subclass or use the SDK directly to receive Zenoh events (subscribe) or register RPC responders (queryable).

11-1. zenoh_subscribe — topic subscription (decorator)

from rb_sdk import RBManipulateSDK
from rb_zenoh.schema import SubscribeOptions, OverflowPolicy

sdk = RBManipulateSDK()

# ── Pattern 1: dict payload (JSON topic) ──
@sdk.zenoh_subscribe("amr/AMR001/status")
def on_status(*, dict_payload: dict | None, **kwargs):
    print("status:", dict_payload)

# ── Pattern 2: Object API (T class) — recommended ──
# With flatbuffer_obj_t, a T object deserialized via InitFromPackedBuf is passed as obj_payload
# T class fields are accessed directly as attributes
from rb_flat_buffers.IPC.State_Core import State_CoreT

@sdk.zenoh_subscribe(
    "C500920/state_core",
    flatbuffer_obj_t=State_CoreT,
    opts=SubscribeOptions(parse_dict_payload=False),
)
def on_state_core(*, obj_payload: State_CoreT | None, **kwargs):
    if obj_payload is None:
        return
    print("motion mode:", obj_payload.motionMode)  # attribute access

# ── Pattern 3: Table API (mv + GetRootAs) — lazy access, when performance matters ──
# Without flatbuffer_obj_t → receive mv (memoryview) and parse manually
# GetRootAs lazily accesses only the fields you need, so it is faster than the T class
from rb_flat_buffers.IPC.State_Core import State_Core

@sdk.zenoh_subscribe(
    "C500920/state_core",
    opts=SubscribeOptions(parse_dict_payload=False),
)
def on_state_core_mv(*, mv=None, **kwargs):
    if mv is None:
        return
    obj = State_Core.GetRootAs(mv, 0)
    print("motion mode:", obj.MotionMode())  # method call

zenoh_subscribe arguments

SubscribeOptions fields

Direct subscribe (when you need to keep the handle)

Calling zenoh_client.subscribe() directly instead of the decorator lets you manage the handle yourself.

sub_handle = sdk.zenoh_client.subscribe(
    "amr/AMR001/moveStatus",
    callback,
    flatbuffer_obj_t=MoveStatusT,
    options=SubscribeOptions(dispatch="queue", overflow=OverflowPolicy.LATEST_ONLY),
)

# Unsubscribe
sub_handle.close()

11-2. zenoh_queryable — RPC responder (decorator)

When a remote get(keyexpr) request arrives, the registered handler is called and returns the reply.

from rb_sdk import RBManipulateSDK

sdk = RBManipulateSDK()

# dict request (no payload) → dict reply
# params is a dict[str, str] parsed from the URL query string (e.g. ?model=C500920)
@sdk.zenoh_queryable("my_service/robot_status")
def handle_status_query(*, params: dict[str, str]) -> dict:
    return {"status": "ok", "model": params.get("model")}

# FlatBuffer request → FlatBuffer reply
from rb_flat_buffers.manipulate.RB_Functions import RB_FunctionsT

@sdk.zenoh_queryable(
    "my_service/move_j",
    flatbuffer_req_T_class=RB_FunctionsT,
    flatbuffer_res_buf_size=256,
)
def handle_move_j(*, req: RB_FunctionsT, params: dict[str, str]) -> RB_FunctionsT:
    res = RB_FunctionsT()
    res.returnValue = 0
    return res

zenoh_queryable arguments

Duplicate registration guard

If a queryable is already registered with the same keyexpr, a warning is printed and the existing handler is kept. It is released automatically on SDK close().

12. License / Contact

© Rainbow Robotics. Issues/questions: GitHub Issues