robotrt-sdk 0.1.0b2

RobotRT Python SDK (FFI thin wrapper)

RobotRT Python SDK (Embedded + Daemon)

This directory contains the Python-first SDK implementation on top of the frozen Core FFI ABI.

Scope

• Embedded mode SDK surface for node and buffer lease lifecycle.
• Embedded mode full communication APIs: pub/sub, service, action, mission.
• Embedded mode executor APIs: SimpleExecutor, CallbackExecutor, and MultiThreadExecutor.
• Daemon mode status SDK surface for snapshot/list/info/watch query.
• Unified error mapping from C ABI status and error payload.
• Zero-copy readable view via borrowed lease pointer to Python memoryview.

Embedded Communication API (MVP)

import robotrt_sdk as sdk

with sdk.Node.create() as node:
  pub = node.create_publisher("/demo/topic")
  sub = node.create_subscriber("/demo/topic")
  pub.publish_batch([b"hello", b"robotrt", b"batch"])
  batch = sub.recv_batch(8)
  for _ in batch:
    # Reliable topics require explicit ack to release capacity.
    sub.ack()
  print(batch)
  print("backpressure:", sub.backpressure_signal().value)

  server = node.create_service_server("/demo/service")
  client = node.create_service_client("/demo/service")
  request_id = client.call(b"ping")
  rid, req = server.recv_request()
  server.reply(rid, b"pong")
  print(client.poll_response(request_id))

  action_server = node.create_action_server("/demo/action")
  action_client = node.create_action_client("/demo/action")
  goal_id, accepted = action_client.send_goal(b"goal")
  gid, goal = action_server.recv_goal()
  action_server.accept_goal(gid)
  action_server.publish_feedback(gid, b"50%")
  action_server.succeed(gid, b"done")
  print(action_client.poll_result(goal_id))

  mission = node.create_mission_session(42)
  mission.open("demo_mission")
  mission.send_command("start:demo")
  mission.reconcile("cp-1", 2)

Quick Start

1. Build the shared library:

cargo build -p core-ffi --lib

2. Run tests:

PYTHONPATH=bindings/python python3 -m unittest discover -s bindings/python/tests -p 'test_*.py'

3. Run the embedded zero-copy demo:

PYTHONPATH=bindings/python python3 bindings/python/examples/embedded_zero_copy_demo.py

4. Run the daemon status demo:

PYTHONPATH=bindings/python python3 bindings/python/examples/daemon_status_demo.py

5. Run additional examples:

PYTHONPATH=bindings/python python3 bindings/python/examples/full_comms_demo.py
PYTHONPATH=bindings/python python3 bindings/python/examples/executor_demo.py
PYTHONPATH=bindings/python python3 bindings/python/examples/status_watch_demo.py

Mode Switch API

import robotrt_sdk as sdk

# Embedded mode: read local status report.
embedded = sdk.StatusClient(mode="embedded", report_path="artifacts/introspection/local-loop-status.json")
print(embedded.node_list())

# Daemon mode: query robotrt-host status service.
daemon = sdk.StatusClient(mode="daemon", endpoint="127.0.0.1:7588", timeout_ms=1000)
print(daemon.service_list())

Environment Variables

• ROBOTRT_CORE_FFI_LIB: optional absolute path to the core-ffi dynamic library.
  • If not set, SDK auto-discovers under target/debug and target/debug/deps.