rtsi 的使用

一、简介

rtsi类是Elite 机器人的实时通讯接口，可以获取机器人状态、设置IO等

二、操作流程

1、安装elite python sdk，确保sdk安装正确、解释器使用python版本正确
2、进入示例代码所在文件夹，打开cmd
3、执行这行代码：python example_rtsi_client.py --ip 192.168.0.1 --port 30004，请注意，这里example_rtsi_client.py是文件名，需要替换成真实文件名，--ip参数后输入的是机器人的ip地址，也需要根据实际ip输入

三、常见问题

1、Q:为什么sdk安装后，ide却提示找不到这个库呢？
A:首先请确认安装的python wheel基于什么版本编译的，和使用的版本是否有差异，其次，需要确认当前环境变量里优先级最高的python版本和pip版本，与ide中的解释器使用的是否是同一版本，如果不是可以修改环境变量后重新安装sdk，也可以修改解释器版本
2、只能通过控制台执行这个文件吗？能否通过ide直接执行？
A:这里使用了argparse解析参数，所以只需要将这里参数解析部分删除，直接指定ip, port即可

四、代码和附件

  1. 解析参数 --ip（必传）和 --port（默认 30004）
  2. RtsiClientInterface() 构造接口对象
  3. connect(ip, port) 连 RTSI
  4. negotiateProtocolVersion() 协商协议版本
  5. getControllerVersion() 读控制器版本（major/minor/bugfix/build）
  6. setupOutputRecipe(["actual_joint_positions", "target_joint_positions", "target_speed_fraction"], 125.0) 配置输出配方，125Hz 订阅三个变量
  7. start() 启动实时数据流
  8. receiveData(recipes, read_newest=True) 循环读 2 帧 → getRecipe() / getValue(name) / getID() 取数据
  9. pause() 暂停输出流
  10. setupInputRecipe(["speed_slider_mask", "speed_slider_fraction"]) 配置输入配方
  11. setValue("speed_slider_mask", 1) / setValue("speed_slider_fraction", 0.5) 设值 → send(in_recipe) 写入控制器
  12. start() 恢复输出流 → receiveData() 再读一帧确认倍率生效
  13. pause() → disconnect() 断开
#!/usr/bin/env python3
"""
Example script for using the elite_cs_sdk RTSI client.

中文说明：
这个示例演示如何通过 RTSI 接口协商协议版本、
配置输入输出 recipe、接收实时数据并写入速度倍率。

Usage:
    python example_rtsi_client.py --ip 192.168.0.58 [--port 30004]
"""

import argparse
import sys
from time import sleep
import elite_cs_sdk as cs

def main():
    parser = argparse.ArgumentParser(
        description="Connect to a robot's RTSI server and exchange recipes."
    )
    parser.add_argument(
        "--ip",
        required=True,
        help="IP address of the robot's RTSI server"
    )
    parser.add_argument(
        "--port",
        type=int,
        default=30004,
        help="RTSI server port (default: 30004)"
    )
    args = parser.parse_args()

    client = cs.RtsiClientInterface()

    print(f"[INFO] Connecting to RTSI at {args.ip}:{args.port}...")
    client.connect(args.ip, args.port)

    print("[INFO] Negotiating protocol version...")
    if not client.negotiateProtocolVersion():
        print("[ERROR] Protocol negotiation failed.", file=sys.stderr)
        sys.exit(1)

    # 读取控制器版本，确认当前连接到的设备信息。
    version = client.getControllerVersion()
    print(f"[INFO] Controller version: "
          f"{version.major}.{version.minor}.{version.bugfix} (build {version.build})")

    # 配置输出 recipe，周期性读取实时关节位置和目标速度倍率等变量。
    variables = ["actual_joint_positions", "target_joint_positions", "target_speed_fraction"]
    frequency = 125.0
    print(f"[INFO] Setting up output recipe {variables} @ {frequency} Hz...")
    out_recipe = client.setupOutputRecipe(variables, frequency)

    # 启动实时数据流。
    print("[INFO] Starting data stream...")
    if not client.start():
        print("[ERROR] Failed to start RTSI stream.", file=sys.stderr)
        sys.exit(1)

    # 连续读取若干帧输出数据并打印。
    for i in range(2):
        print(f"[INFO] Waiting for sample {i+1}...")
        recipes = [out_recipe]
        count = client.receiveData(recipes, read_newest=True)
        if count >= 0 and recipes:
            sample = recipes[0]
            names = sample.getRecipe()
            values = [sample.getValue(n) for n in names]
            rid = sample.getID()
            print(f"[INFO] Sample {i+1} (ID={rid}):")
            for n, v in zip(names, values):
                print(f"    {n} = {v}")
        else:
            print(f"[WARN] No data received. {count}")
        sleep(1.0)

    # 暂停输出流，准备发送输入 recipe。
    print("[INFO] Pausing data stream...")
    client.pause()

    # 配置输入 recipe，用于修改示教器速度倍率。
    in_vars = ["speed_slider_mask", "speed_slider_fraction"]
    print(f"[INFO] Setting up input recipe {in_vars}...")
    in_recipe = client.setupInputRecipe(in_vars)
    in_id = in_recipe.getID()
    print(f"[INFO] Input recipe ID: {in_id}")

    # 写入新的速度倍率并发送给控制器。
    frac = 0.5
    print(f"[INFO] Sending speed_slider_fraction = {frac}")
    in_recipe.setValue("speed_slider_mask", 1)
    in_recipe.setValue("speed_slider_fraction", frac)
    client.send(in_recipe)

    # 恢复输出流，再读一帧确认倍率设置是否生效。
    print("[INFO] Resuming data stream for one more sample...")
    client.start()
    recipes = [out_recipe]
    client.receiveData(recipes, read_newest=True)
    if recipes:
        sample = recipes[0]
        speed = sample.getValue("target_speed_fraction")
        print(f"[INFO] Confirmed target_speed_fraction = {speed}")

    print("[INFO] Pausing data stream...")
    client.pause()

    # 断开 RTSI 连接。
    print("[INFO] Disconnecting RTSI client...")
    client.disconnect()

if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print(f"[ERROR] {e}", file=sys.stderr)
        sys.exit(1)