rabo-dev-kit 1.0

A ROS2 development toolkit for data recording, topic synchronization and utilities

Rabo Dev Kit

A lightweight Python development toolkit for ROS2 (Robot Operating System 2) projects, providing utilities for data recording, topic synchronization, and configuration management.

Features

• DataRecorder: Text-based logging utility for debugging and data collection
• SimpleSyncBridge: Flexible ROS2 topic message routing and transformation
• Chat: LLM 对话客户端，支持同步和流式调用
• Configuration Utilities: Helper functions for environment-based configuration

Installation

pip install rabo-dev-kit

首次 import rabo_dev_kit 时自动从 OSS 下载核心代码，后续每次 import 检查更新。

Requirements

• Python >= 3.6
• ROS2 (rclpy)

Usage

1. DataRecorder

Record text data to temporary directories for debugging:

from rabo_dev_kit import DataRecorder

# Initialize recorder
recorder = DataRecorder()

# Log some text data
recorder.log_text("sensor_data.txt", "Temperature: 25.3°C")

# Append to existing file
recorder.log_text("sensor_data.txt", "Humidity: 60%", mode='a')

# Get file path
path = recorder.get_file_path("sensor_data.txt")
print(f"Data logged to: {path}")

Environment Variables:

• RDTP_SIM_LOG_PATH: Custom log directory (default: /tmp)

2. SimpleSyncBridge

Route and transform ROS2 topic messages with frequency control:

import rclpy
from rabo_dev_kit.sync_bridge import SimpleSyncBridge, SyncRoute
from std_msgs.msg import String, Int32

# Initialize ROS2
rclpy.init()

# Create bridge node
bridge = SimpleSyncBridge()

# Define synchronization routes
routes = [
    SyncRoute(
        source_topic='/input/data',
        source_msg_type=String,
        target_topic='/output/data',
        target_msg_type=String,
        frequency=10.0,  # 10 Hz
        handler=lambda msg: process_message(msg)  # Optional transformation
    ),
    SyncRoute(
        source_topic='/sensor/raw',
        source_msg_type=Int32,
        target_topic='/sensor/filtered',
        target_msg_type=Int32,
        frequency=5.0,  # 5 Hz
    )
]

# Register routes
bridge.batch_register(routes)

# Start synchronization
bridge.setup()

# Spin
rclpy.spin(bridge)

# Cleanup
bridge.destroy_node()
rclpy.shutdown()

SyncRoute Parameters:

• source_topic: Source ROS2 topic to subscribe to
• source_msg_type: Message type of the source topic
• target_topic: Target ROS2 topic to publish to (optional if handler is provided)
• target_msg_type: Message type of the target topic (defaults to source type)
• frequency: Publishing frequency in Hz (default: 10.0)
• handler: Optional callable for message transformation
• enabled: Enable/disable the route (default: True)

Message Transformation Example:

def transform_message(msg):
    """Transform a String message to uppercase"""
    new_msg = String()
    new_msg.data = msg.data.upper()
    return new_msg

route = SyncRoute(
    source_topic='/input',
    source_msg_type=String,
    target_topic='/output',
    target_msg_type=String,
    handler=transform_message
)

3. Chat (LLM 对话)

调用 LLM Gateway 进行 AI 对话，支持同步和流式输出：

from rabo_dev_kit import Chat

# 初始化（容器内自动从 RABO_API_BASE 环境变量获取地址）
llm = Chat()

# 设置系统提示词
llm = Chat(system_prompt="你是一个ROS2机器人专家，用简短的中文回答")

# 同步调用
reply = llm.chat([{"role": "user", "content": "你好"}])
print(reply)

# 流式调用
for chunk in llm.chat([{"role": "user", "content": "写一首诗"}], stream=True):
    print(chunk, end="", flush=True)

# 指定模型
reply = llm.chat(
    [{"role": "user", "content": "解释量子力学"}],
    model="qwen-max"
)

# 多轮对话
reply = llm.chat([
    {"role": "user", "content": "ROS2 是什么？"},
    {"role": "assistant", "content": "ROS2 是新一代机器人操作系统框架。"},
    {"role": "user", "content": "它和 ROS1 有什么区别？"},
])

# 查看可用模型
models = llm.models()
print(models)  # ['qwen-turbo', 'qwen-plus', 'qwen-max', ...]

Chat 参数:

• api_base: API 基础地址，默认从环境变量 RABO_API_BASE 获取，未设置则使用 https://www.rabo.cc
• model: 默认模型，默认 "qwen-turbo"
• system_prompt: 系统提示词，自动注入到每次对话的 messages 最前面
• timeout: 请求超时时间（秒），默认 60

环境变量:

• RABO_API_BASE: API 基础地址（容器内已自动设置）
• RABO_ID: 实例 ID，用于获取配置（容器内已自动设置）

chat() 参数:

• messages: 消息列表，格式为 [{"role": "user", "content": "..."}]
• model: 模型名称，不传则使用默认模型
• stream: 是否流式输出，默认 False
• session_id: 会话 ID，用于多轮对话上下文管理
• temperature: 温度参数，控制回复的随机性
• max_tokens: 最大 token 数

4. Configuration Utilities

from rabo_dev_kit import get_parameter_topic

# Get ROS2 parameter topic from environment
param_topic = get_parameter_topic()

Environment Variables:

• CONTROLLER_PARAMETER_TOPIC: ROS2 parameter topic name

API Reference

DataRecorder

Methods

• log_text(filename: str, content: str, mode: str = 'w') -> bool

  • Log text content to a file
  • Returns: True if successful, False otherwise

• get_file_path(filename: str) -> str

  • Get the absolute path where the file will be saved
  • Returns: Full file path

SimpleSyncBridge

Methods

• register_sync_route(route: SyncRoute) -> bool

  • Register a single synchronization route
  • Returns: True if successful, False otherwise

• batch_register(routes: List[SyncRoute]) -> int

  • Register multiple routes at once
  • Returns: Number of successfully registered routes

• setup()

  • Initialize all subscriptions and publishers
  • Must be called after registering routes

Chat

Methods

• chat(messages, model=None, stream=False, session_id=None, temperature=None, max_tokens=None)

  • 发送聊天请求
  • stream=False 时返回完整回复字符串
  • stream=True 时返回生成器，逐步 yield 内容片段

• models() -> List[str]

  • 获取可用模型列表
  • Returns: 模型 ID 列表

Development

开发模式

设置环境变量 RABO_DEV=1，直接使用本地源码，跳过 OSS 热更新：

export RABO_DEV=1
python -c "from rabo_dev_kit import DataRecorder; print('dev mode')"

Running Tests

pip install -e .[dev]
pytest tests/

构建核心包

核心代码编译为 .pyc 并打包为 zip，托管在阿里云 OSS。

# 本地构建（当前 Python 版本）
python scripts/build_core.py

# 使用 Docker 构建多版本
./scripts/build_all.sh          # 构建 cp310, cp312, cp313
./scripts/build_all.sh 10 12    # 只构建指定版本

# 构建产物在 dist/ 目录

CI/CD 通过 Drone 自动构建和上传，详见 PUBLISHING.md。

Version History

1.0 (Current)

• 热更新架构：PyPI 发布 loader shell，核心代码通过 OSS 自动分发
• 移除 Cython 编译，改用 .pyc + zipimport

0.1.3

• Added SimpleSyncBridge for ROS2 topic synchronization

0.1.0

• Initial release
• DataRecorder utility
• Basic configuration helpers

License

This project is licensed under the MIT License - see the LICENSE file for details.

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

Author

origin ([e-mail])

Acknowledgments

This toolkit is designed to simplify ROS2 development workflows and provide common utilities for robotics projects.