Metadata-Version: 2.4
Name: rainbow-rb-sdk
Version: 0.10.6
Summary: Unified Rainbow Robotics Python SDK — AMR, manipulator and RB-Y1 humanoid control with Rainbow Robot Service (common) integration
Author-email: Derek <dfd1123@rainbow-robotics.com>
Requires-Python: <3.14,>=3.11
Description-Content-Type: text/markdown
Requires-Dist: rainbow-rb-utils==0.10.6
Requires-Dist: rainbow-rb-zenoh==0.10.6
Requires-Dist: rainbow-rb-flat-buffers==0.10.6
Requires-Dist: rainbow-rb-schemas==0.10.6
Requires-Dist: rainbow-rb-log==0.10.6
Requires-Dist: python-socketio>=5.13.0
Requires-Dist: numpy>=2.3.3

# 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

| Class | Purpose |
|---|---|
| `RBAmrSDK` | AMR (autonomous mobile robot) control |
| `RBManipulateSDK` | Collaborative robot manipulator control |
| `RBRby1SDK` | RB-Y1 dual-arm robot control (C++ implementation, `rb_sdk.rb_sdk_cpp` binding) |
| `RBBaseSDK` | Common base (not needed unless you subclass it) |

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

```bash
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.`.

```python
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):

```bash
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.

```python
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.

```python
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

| Field | Type | Default | Description |
|---|---|---|---|
| `token` | `str` | (required) | common service JWT |
| `host` | `str` | (required) | IP of the robot running common |
| `zenoh_port` | `int \| None` | `None` | Local Zenoh listen port. `None` = automatic |
| `ttl` | `int` | `30` | Server-side session TTL (seconds) |
| `heartbeat` | `float \| None` | `None` | Heartbeat interval (seconds). `None` = `ttl/2` (min 1 second) |
| `mesh_timeout` | `float \| None` | `None` | Max time to wait for the mesh to form (seconds). `None` = wait forever |
| `mesh_probe_timeout` | `float` | `1.0` | Timeout of a single probe query (seconds) |
| `mesh_probe_interval` | `float` | `0.5` | Probe retry interval (seconds) |

---

## 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.

```python
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`)

```python
from rb_sdk import RBAmrSDK

sdk = RBAmrSDK()
```

### Submodules

| Attribute | Class | Main features |
|---|---|---|
| `sdk.move` | `AmrMoveSDK` | Move to goal, jog, stop/pause/resume, linear/arc/rotation moves |
| `sdk.control` | `AmrControlSDK` | Safety field/flag/IO settings, docking/undocking, LED/motor/jog control, obstacle box |
| `sdk.localization` | `AmrLocalizationSDK` | Pose init, random/auto/semi-auto init, start/stop localization |
| `sdk.map` | `AmrMapSDK` | Map list/load/delete, topology get/set, start/stop/save mapping |
| `sdk.setting` | `AmrSettingSDK` | Robot type, setting categories/parameters, PDU/drive/sensor settings |
| `sdk.status` | `AmrStatusSDK` | Status, move status, LiDAR, path queries |
| `sdk.file` | `AmrFileSDK` | Map and resource file management |
| `sdk.capability` | `AmrCapabilitySDK` | Robot capability queries |
| `sdk.accessory` | `AmrAccessorySDK` | Accessory control |
| `sdk.program` | `AmrProgramSDK` | AMR program execution/control |

### Move examples

```python
# 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`)

```python
from rb_sdk import RBManipulateSDK

sdk = RBManipulateSDK()
```

### Submodules

| Attribute | Class | Main features |
|---|---|---|
| `sdk.move` | `ManipulateMoveSDK` | Joint/linear/arc moves, jog, servo/online, admittance, weaving, conveyor |
| `sdk.program` | `ManipulateProgramSDK` | Program run/stop/resume, pinpoint, tool output |
| `sdk.config` | `ManipulateConfigSDK` | Tool, user frame, collision parameters, speed/impedance/freedrive settings |
| `sdk.io` | `ManipulateIOSDK` | Side/flange/ext DIO·AIO output, pulses, bit combinations, IO function settings |
| `sdk.get_data` | `ManipulateGetDataSDK` | Variable queries, relative/absolute coordinate calculation |
| `sdk.point` | `ManipulatePointSDK` | Point and coordinate registration |
| `sdk.service` | `ManipulateServiceSDK` | Mini tab features |
| `sdk.maintenance` | `ManipulateMaintenanceSDK` | Administrator settings |
| `sdk.state` | `ManipulateStateSDK` | whoami, power/servo/reference control, real-time state subscription |

### move module — main methods

| Method | Description |
|---|---|
| `call_move_j` | Joint-space move |
| `call_move_l` | Linear move |
| `call_move_cir_axis` | Axis-based arc move |
| `call_move_cir_threepoint` | Three-point arc move |
| `call_move_home` | Move to home position |
| `call_move_jb_clr/add/run` | Register/run joint blending path |
| `call_move_lb_clr/add/run` | Register/run linear blending path |
| `call_move_xb_clr/add/run` | Register/run X blending path |
| `call_smoothjog_j/l` | Smooth jog (joint/linear) |
| `call_tickjog_j/l` | Tick jog (joint/linear) |
| `call_approach_j/l` | Approach move |
| `call_servo_j/l` | Servo-mode joint/linear move |
| `call_online_j/l` | Online trajectory joint/linear |
| `call_admittance_force_on/off/change` | Admittance force control |
| `call_arc_sensing_on/off` | Arc sensing control |
| `call_tcp_weaving_on/off` | TCP weaving control |
| `call_base_conveyor_on/off/speed` | Base conveyor control |

### Move examples

```python
# 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>`.

```python
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.

```python
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).

```bash
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.

```python
# 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.

```python
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

```python
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)

```python
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

| Argument | Type | Default | Description |
|---|---|---|---|
| `topic` | `str` | (required) | Zenoh key expression (e.g. `"amr/{model}/status"`) |
| `flatbuffer_obj_t` | FlatBuffer T class \| `None` | `None` | Set to deserialize as FlatBuffer. If `None`, the callback receives `mv: memoryview` |
| `opts` | `SubscribeOptions \| None` | `None` | Dispatch / overflow options |

#### `SubscribeOptions` fields

| Field | Default | Description |
|---|---|---|
| `dispatch` | `"immediate"` | `"immediate"`: callback runs on the receive thread. `"queue"`: C++ bounded queue + dedicated dispatch thread |
| `parse_dict_payload` | `True` | Set to `False` when receiving FlatBuffers |
| `overflow` | `DROP_OLDEST` | `DROP_OLDEST` / `DROP_NEW` / `LATEST_ONLY` |
| `maxsize` | `50` | Max size of the C++ queue in `"queue"` mode |
| `rate_limit_per_sec` | `None` | Max callback invocations per second |
| `sample_every` | `1` | Deliver only 1 of every N samples (1 = all) |
| `max_async_inflight` | `1` | Number of async callbacks allowed to run concurrently |

#### Direct subscribe (when you need to keep the handle)

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

```python
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.

```python
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

| Argument | Type | Default | Description |
|---|---|---|---|
| `keyexpr` | `str` | (required) | Zenoh key expression (e.g. `"my_service/status"`) |
| `flatbuffer_req_T_class` | FlatBuffer T class \| `None` | `None` | Set to deserialize the request payload as FlatBuffer |
| `flatbuffer_res_buf_size` | `int \| None` | `None` | FlatBuffer reply serialization buffer size (bytes). Required for FlatBuffer replies |

#### 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](https://github.com/rainbow-mobile/rainbow-release-apps/issues)
