rb_sdk

Rainbow Robotics 통합 Python SDK. AMR · Manipulator · RB-Y1 양팔로봇 제어와 Rainbow Robot Service(common) 연동을 단일 API로 제공합니다.

진입점

내부적으로 Zenoh pub/sub/query 메시 위에서 동작하며, 같은 PC(로봇 본체)에서 실행할 때는 별도 설정 없이 즉시 사용 가능합니다.

1. 설치

pip install rainbow-rb-sdk

PyPI 패키지는 rainbow 네임스페이스로 배포됩니다. 이 문서의 예제는 모노레포 워크스페이스 이름(rb_sdk, rb_zenoh, rb_flat_buffers …)으로 적혀 있으니, PyPI 로 설치했다면 앞에 rainbow. 를 붙여 import 합니다.

from rainbow.rb_sdk import RBManipulateSDK                      # 워크스페이스: from rb_sdk import ...
from rainbow.rb_flat_buffers.IPC.State_Core import State_CoreT   # 워크스페이스: from rb_flat_buffers...

wheel 은 linux(x86_64/aarch64)·Windows 용만 배포됩니다. macOS 에서는 에디터 자동완성·타입체크용 스텁만 설치할 수 있습니다 (런타임 import 불가 — 실행은 Docker/Linux 에서):

pip install rainbow-rb-sdk-stubs

RB-Y1(RBRby1SDK)은 공식 rby1-sdk(C++)가 rb_sdk.rb_sdk_cpp에 직접 벤더링되어 있어 별도 설치가 필요 없습니다.

PyPI 패키지는 Python 3.11–3.13 wheel 을 제공합니다 (모노레포 개발/빌드는 3.12 전용).

2. 기본 사용 (로봇 본체에서 실행하는 경우)

common 서비스와 같은 PC에서 실행되는 스크립트라면 추가 설정 없이 그대로 사용 가능합니다.

from rb_sdk import RBAmrSDK, RBManipulateSDK

amr = RBAmrSDK()
manipulate = RBManipulateSDK()

# 이 시점에 ZenohClient가 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. 외부 PC에서 사용 (Remote Connect)

개발용 노트북 등 로봇 본체와 다른 머신에서 로봇과 통신하고 싶다면 ConnectOptions 를 채워 sdk.connect()를 호출합니다. connect() 는 동기 호출(등록·mesh 확인이 끝날 때까지 블록)이고, disconnect() 는 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 = "서비스 Token"
opts.host = "192.168.1.100"
# opts.ttl = 30
# opts.zenoh_port = None
# opts.heartbeat = 15.0

# ⚠️  subscribe는 반드시 connect() 이후에 등록해야 한다.
# connect() 가 zenoh 세션을 다시 열기 때문에 그 전에 등록한 subscribe는 세션과 함께 사라진다.
peer_id = sdk.connect(opts)  # 로봇 본체(같은 PC)면 아무것도 안 하고 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:
    # heartbeat 는 SDK 내부 스레드가 돌린다 — 프로세스만 살아 있으면 된다.
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    pass
finally:
    asyncio.run(sdk.disconnect())

주요 특징

• 로컬 IP 자동 감지: host 방향의 라우팅 인터페이스 IP를 자동으로 찾아 로봇에 알림
• 포트 자동 배정: 별도 지정 없이 OS가 빈 포트를 잡아주므로 포트 충돌 없음
• 자동 heartbeat: 백그라운드 스레드가 주기적으로 TTL 연장. 스크립트 종료 / 크래시 시 TTL 만료로 서버 측 정리
• 같은 PC 자동 감지: host가 localhost거나 자기 IP인 경우 connect()는 아무것도 하지 않고 None 반환 (이미 메시 안에 있음)
• 이미 등록된 상태에서 다시 부르면 기존 peer id 를 반환. 등록·재연결 실패 시 예외

ConnectOptions 필드

4. use_directly 모드

RBBaseSDK 생성자에 use_directly=True를 전달하면, 메서드 호출 시 {name}_directly 메서드가 존재하면 자동으로 그쪽으로 라우팅됩니다. 없으면 원래 메서드가 그대로 호출됩니다.

class MySDK(RBManipulateSDK):
    def call_move_j_directly(self, *, robot_model, target, **kwargs):
        # Zenoh를 거치지 않고 직접 하드웨어 제어하는 구현
        ...

# use_directly=True → call_move_j 호출 시 call_move_j_directly로 라우팅
sdk = MySDK(use_directly=True)
sdk.move.call_move_j(robot_model="C500920", target={...})  # → call_move_j_directly 호출

# use_directly=False (기본) 또는 _directly 메서드 없으면 원래 메서드 사용
sdk2 = MySDK(use_directly=False)
sdk2.move.call_move_j(...)  # → call_move_j 호출

싱글톤 인스턴스가 이미 존재해도 use_directly는 매 생성자 호출마다 갱신됩니다.

5. AMR SDK (RBAmrSDK)

from rb_sdk import RBAmrSDK

sdk = RBAmrSDK()

하위 모듈

이동 예시

# 목표 노드로 이동
sdk.move.send_move_goal(
    robot_model="AMR001",
    goal_type=0,
    goal_node_name="node_A",
)

# 이동 결과 대기 (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)

# 멀티 경로
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()

하위 모듈

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},
)

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

# 기준 위치로부터 상대 이동
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],  # 현재 기준값
)

# 전원·서보 제어
sdk.state.call_powercontrol(robot_model="C500920", control=1)
sdk.state.call_servocontrol(robot_model="C500920", control=1)

7. RB-Y1 SDK (RBRby1SDK)

공식 rby1-sdk(C++)를 pybind11 로 직접 바인딩한 구현입니다. 순수 Python 래퍼(rby1.py, rby1_sdk/*)는 폐기되었고, 동일 진입점 이름(RBRby1SDK)이 이제 rb_sdk.rb_sdk_cpp(C++ 확장)에서 옵니다. 공식 타입(RobotCommandBuilder 등)은 from rb_sdk import rby1로 접근합니다.

프로세스 싱글톤이며, 모든 public 메서드는 robot_model이 첫 인자 / flow_manager_args가 마지막 인자인 기본 async def이고 각각 sdk.sync.<method> 동기 버전이 짝으로 존재합니다.

from rb_sdk import RBRby1SDK, rby1

sdk = RBRby1SDK()

async def main():
    # endpoint 생략 시 env RBY1_HOST (기본 192.168.30.1:50051). 같은 endpoint/model 재호출은 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

    # 상태 스트림: 콜백 인자 2개면 (state, control_manager_state). 코루틴 콜백은 이벤트 루프에서, 최신값만 전달
    async def on_state(state, cms):
        ...
    await sdk.start_state_update(robot_model="a", callback=on_state, rate=20)

    builder = rby1.RobotCommandBuilder()  # 공식 빌더로 명령 구성
    handle = await sdk.send_command(robot_model="a", builder=builder)
    await handle.wait()  # await 가 취소되면 공식 SDK Cancel 로 로봇 명령도 멈춘다

동기 컨텍스트(PyFM Step 등)에서는 sdk.sync.<method>를 그대로 호출합니다.

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

에러는 RBY1Error / RBY1ConnectionError(둘 다 RuntimeError 서브클래스)로 던져집니다.

서브모듈 초기화 (개발자 필수)

rby1-sdk/qdldl은 git submodule(backend/packages/rb_sdk/src/rb_sdk/cpp/lib/{rby1-sdk,qdldl})로 벤더링되어 있어 최초 clone 후 반드시 초기화해야 하며, 빌드는 Docker 빌드 이미지에서만 수행합니다 (make backend.sdk-core, 로컬 직접 빌드 불가).

git submodule update --init --recursive
# 업스트림 버전을 올릴 때 (예: v0.10.0 → 다음 태그)
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

mingw 빌드용 패치(backend/packages/rb_sdk/src/rb_sdk/cpp/cmake/rby1_sdk_mingw.patch)는 업스트림 소스에 대해 적용되므로, 서브모듈 버전이 바뀌어 패치가 더 이상 맞지 않으면 CMake configure 단계에서 즉시 실패합니다 (그 경우 패치를 새 버전 기준으로 다시 만들어야 함).

8. FlowManagerArgs 통합 (PyFM)

PyFM Step에서 SDK 메서드를 호출할 때 flow_manager_args를 전달하면 done 콜백 처리, 이동 완료 대기, 변수 바인딩 등이 자동으로 처리됩니다.

# 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() 콜백 자동 호출
    )

flow_manager_args를 생략하면 블로킹 호출로 동작합니다.

9. 아키텍처 메모

프로세스 단위 싱글톤

RBBaseSDK(RBAmrSDK, RBManipulateSDK)는 (pid, cls) 기준 싱글톤입니다. 같은 프로세스에서 여러 번 생성자를 호출해도 동일 인스턴스가 반환됩니다. ZenohClient도 PID당 하나를 공유합니다.

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

자동 에러 래핑

RBBaseSDK를 상속한 클래스의 모든 public instance method는 __init_subclass__에 의해 자동으로 공통 try/except로 래핑됩니다. ZenohNoReply / ZenohTransportError 발생 시 자동 재연결 후 1회 재시도합니다.

SDK 종료

sdk.close()                         # 현재 인스턴스 정리
RBBaseSDK.close_all_for_pid()       # 현재 프로세스의 모든 SDK 인스턴스 정리

10. 주의사항

• RBRby1SDK는 프로세스 싱글톤이고 로봇 gRPC 연결도 프로세스에 하나입니다 (connect_robot 으로 교체).
• 연결 대상(zenoh, rby1 endpoint, Muscat HTTP)이 준비되지 않으면 런타임 예외가 발생합니다.
• 외부 PC에서 SDK 사용 시 방화벽이 7447/TCP(zenoh)와 8000/TCP을 양방향으로 허용해야 합니다.
• use_directly=True는 싱글톤 재진입 시에도 반영됩니다. 단, 하위 SDK(sdk.move 등)에는 전파되지 않으므로 _directly 메서드는 최상위 SDK에 정의해야 합니다.

11. Subscribe / Queryable

SDK를 상속하거나 직접 사용해 Zenoh 이벤트를 수신(subscribe) 하거나 RPC 응답자(queryable) 를 등록할 수 있습니다.

11-1. zenoh_subscribe — 토픽 구독 (데코레이터)

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

sdk = RBManipulateSDK()

# ── 패턴 1: dict 페이로드 수신 (JSON 토픽) ──
@sdk.zenoh_subscribe("amr/AMR001/status")
def on_status(*, dict_payload: dict | None, **kwargs):
    print("status:", dict_payload)

# ── 패턴 2: Object API (T 클래스) — 권장 ──
# flatbuffer_obj_t 지정 시 InitFromPackedBuf로 역직렬화된 T 객체가 obj_payload로 전달됨
# T 클래스 필드는 snake_case 속성으로 직접 접근 가능
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)  # snake_case 속성

# ── 패턴 3: Table API (mv + GetRootAs) — lazy access, 성능 우선 시 ──
# flatbuffer_obj_t 미지정 → mv(memoryview)를 직접 받아 수동 파싱
# GetRootAs는 필요한 필드만 lazy 접근하므로 T 클래스보다 빠름
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())  # 메서드 호출

zenoh_subscribe 인자

SubscribeOptions 필드

직접 subscribe (handle 보관이 필요한 경우)

데코레이터 대신 zenoh_client.subscribe()를 직접 호출하면 handle을 직접 관리할 수 있습니다.

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

# 구독 해제
sub_handle.close()

11-2. zenoh_queryable — RPC 응답자 (데코레이터)

원격에서 get(keyexpr) 요청이 오면 등록된 handler가 호출되어 응답을 반환합니다.

from rb_sdk import RBManipulateSDK

sdk = RBManipulateSDK()

# dict 요청 (payload 없음) → dict 반환
# params는 URL query string (예: ?model=C500920)을 파싱한 dict[str, str]
@sdk.zenoh_queryable("my_service/robot_status")
def handle_status_query(*, params: dict[str, str]) -> dict:
    return {"status": "ok", "model": params.get("model")}

# FlatBuffer 요청 수신 → FlatBuffer 응답
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 인자

중복 등록 방지

같은 keyexpr로 이미 등록된 queryable이 있으면 경고 메시지를 출력하고 기존 handler를 유지합니다. SDK close() 시 자동 해제됩니다.

12. 라이선스 / 문의

© Rainbow Robotics. 내부 이슈/문의: GitHub Issues