rb_sdk

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

진입점

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

1. 설치

pip install rb_sdk
# RB-Y1 양팔로봇 제어가 필요하면 함께 설치
pip install rby1-sdk

requires-python = ">=3.12,<3.13" — Python 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)

개발용 노트북 등 로봇 본체와 다른 머신에서 로봇과 통신하고 싶다면 sdk.connect()를 호출합니다.

import asyncio
from rb_sdk import RBManipulateSDK
from rb_flat_buffers.IPC.State_Core import State_CoreT
from rb_zenoh.schema import SubscribeOptions

sdk = RBManipulateSDK()


async def main():
    # ⚠️  subscribe는 반드시 connect() 이후에 등록해야 한다.
    # connect() 내부에서 ZenohClient를 close() 후 재연결하므로
    # connect() 전에 등록한 subscribe는 세션과 함께 모두 사라진다.
    await sdk.connect(
        token="서비스 Token",
        host="192.168.1.100",
        # ttl=30,
        # zenoh_port=None,
        # heartbeat_interval=15.0,
    )

    @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)

    try:
        # heartbeat task는 asyncio.run()이 살아있는 동안만 동작한다.
        # asyncio.run()이 반환되면 event loop가 닫히고 heartbeat가 취소되어
        # TTL 후 서버가 peer를 제거하므로 loop 안에서 대기해야 한다.
        while True:
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        pass
    finally:
        await sdk.disconnect()


try:
    asyncio.run(main())
except KeyboardInterrupt:
    print("Ctrl+C pressed")

주요 특징

• 로컬 IP 자동 감지: host 방향의 라우팅 인터페이스 IP를 자동으로 찾아 로봇에 알림
• 포트 자동 배정: 별도 지정 없이 OS가 빈 포트를 잡아주므로 포트 충돌 없음
• 자동 heartbeat: 주기적으로 TTL 연장. 스크립트 종료 / 크래시 시 TTL 만료로 서버 측 정리
• 같은 PC 자동 감지: host가 localhost거나 자기 IP인 경우 connect()는 아무것도 하지 않고 None 반환 (이미 메시 안에 있음)

connect() 인자

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 래퍼. 자세한 설명은 rby1_sdk/README.md 참조.

from rb_sdk import RBRby1SDK

sdk = RBRby1SDK()  # 내부: 127.0.0.1:50051 background probing

# 외부/다른 host 접근
# 생성 직후 바로 호출하면 pending local probing 결과는 폐기된다.
await sdk.connect(host="192.168.100.51")

print(sdk.models)

await sdk.control.power_on(robot_model="a")
await sdk.control.servo_on(robot_model="a")

state = await sdk.state.get_state(robot_model="a")

모든 public 메서드는 async def + 내부 asyncio.to_thread입니다 (공식 SDK가 blocking C++ pybind11이므로). sync 컨텍스트(PyFM Step 등)에서는 asyncio.run(...) 또는 RBRby1BaseSDK.run_sync(...) 사용.

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는 최초 endpoint/model 기준으로 세션이 공유됩니다.
• 연결 대상(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