rabo-dev-kit 0.1.11

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

From PyPI

pip install rabo-dev-kit

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

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

Running Tests

# Install development dependencies
pip install -e .[dev]

# Run tests
pytest tests/

Building from Source

# Build distribution packages
python -m build

# The built packages will be in dist/

Version History

0.1.3 (Current)

• Added SimpleSyncBridge for ROS2 topic synchronization
• Improved .gitignore configuration
• Enhanced documentation

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.