gRPC 例程(强化学习)
本例程针对强化学习(RL)模式,介绍如何获取 gRPC 客户端工程、配置运行环境,并通过交互式命令行或程序集成方式控制机器人。RL 客户端位于仓库的 rl_client/ 目录,服务端口固定为 50051,包名为 pnd.robot。
获取客户端
完整客户端工程(含 Proto 协议、生成文件与示例脚本)将发布于 GitHub:
🔗 仓库地址:pnd_grpc
工程结构
仓库同时提供传统控制(MPC)与强化学习(RL)两套客户端,分别位于 mpc_client/ 与 rl_client/:
pnd_grpc/
├── mpc_client/ # 传统控制(MPC)客户端
│ ├── proto/ # gRPC 协议定义(adam_control.proto,权威来源)
│ ├── include/ # protoc 生成的 C++ 头文件与源文件
│ ├── src/ # C++ API 封装 + 交互式命令行客户端
│ ├── python/ # Python 交互式客户端 + 生成的桩代码
│ ├── ip_config.json # 机器人服务端 IP / 端口配置
│ ├── build.sh / run.sh / clean.sh
│ └── README.md # 传统控制模式详细使用说明
│
├── rl_client/ # 强化学习(RL)模式客户端 ← 本例程
│ ├── comm/
│ │ ├── proto/ # gRPC 协议定义(robot_control.proto)
│ │ └── grpc/ # protoc 生成的 Python 桩代码
│ ├── tools/grpc_client.py# Python 交互式客户端示例
│ └── README.md # 强化学习模式详细使用说明
│
└── README.md # 项目总览
本例程仅涉及 rl_client/。 其中 comm/grpc/ 内的桩代码以 from comm.grpc import robot_control_pb2 方式引用消息模块,因此须保持 rl_client/comm/grpc/ 目录结构;传统控制(MPC)客户端的使用方式见 传统控制 gRPC 接口说明。
环境准备
系统要求
推荐在 Ubuntu 22.04 x86_64 下进行客户端开发与调试。客户端可在任意能访问机器人网络的计算机上运行,与控制程序分离部署。
安装依赖
客户端最小依赖为 grpcio:
网络连接
使用网线将您的计算机与机器人连接至同一网段,网络配置方式见快速开发(真机)。
部署方式
根据使用场景,有两种方式组织客户端文件:
方式一:使用完整客户端工程(推荐)
克隆 pnd_grpc 仓库后,进入 rl_client/ 并保持默认目录结构即可。tools/grpc_client.py 会自动将 rl_client/ 根目录加入 sys.path,无需修改 import 路径。
适用于:调试、演示、基于示例脚本二次开发。
方式二:独立打包 API 文件
若只需在自有项目中集成 gRPC 调用,可将 rl_client/ 下的以下文件拷贝到您的工程:
comm/__init__.pycomm/grpc/robot_control_pb2.pycomm/grpc/robot_control_pb2_grpc.py
将包含 comm/ 的目录加入 PYTHONPATH,或在代码中通过 sys.path.insert 添加路径:
适用于:自动化脚本、高层应用集成、最小化依赖部署。
启动客户端
连接前请确认机器人控制程序已启动。通过 --addr 指定机器人地址:
| 场景 | 连接地址 |
|---|---|
| 本机 | localhost:50051 |
| 远程真机 | <机器人 IP>:50051 |
# 在仓库根目录下运行(本机)
python3 rl_client/tools/grpc_client.py
# 远程机器人(将 IP 替换为实际地址)
python3 rl_client/tools/grpc_client.py --addr 192.168.1.100:50051
若连接失败,终端会提示:
请确认控制程序已启动,且 IP 与网络连通正常。服务端口固定为 50051,请勿修改。
交互式 CLI 例程
rl_client/tools/grpc_client.py 提供交互式命令行客户端,支持 Tab 补全、命令历史,并根据机器人当前 FSM 状态动态显示可用命令。
Connected: 192.168.1.100:50051 | FSM: STOP
╔══════════════════════════════════════════╗
║ PND Robot Control Client v1.0 ║
║ Type 'help' for commands, Tab to complete ║
╚══════════════════════════════════════════╝
robot> help
Current FSM: STOP
Global commands:
state Query current robot state
mode [STATE] Switch FSM mode (Tab for options)
controlmode <0|1> Set control paradigm (0=Traditional, 1=RL)
controlstate Query control paradigm from DDS state
shutdown Shutdown controller
clear Clear screen
quit / exit Exit client
robot> state
FSM State: STOP
Velocity: vx=0.000 vy=0.000 vyaw=0.000
Height: 0.000
Motion File: (none)
Motion Playing: False
Tracking Motion: (none)
Tracking Playing: False
Switchable: ZERO
Available Actions: (none)
robot> mode ZERO
[OK] ok (current=STOP)
robot> mode STAND_WALK
[OK] ok (current=ZERO)
# 命令已接受;待零位归零动作完成后,FSM 才会自动切到 STAND_WALK
robot> mode MULTI_AGENT
[OK] ok (current=STAND_WALK)
robot> motion play Sources/motion/Greeting.txt
[OK] ok (file=Sources/motion/Greeting.txt, playing=True)
Observed: motion_file=Sources/motion/Greeting.txt, playing=True
robot> motion stop
[OK] ok
Observed: motion_file=(none), playing=False
robot> controlmode 0
[OK] ok, Traditional (0) queued for control_mode_cmd
robot> controlstate
[OK] ok, current control mode Traditional domain_id=0
robot> quit
Bye.
CLI 命令对照
| 命令 | 对应 RPC | 说明 |
|---|---|---|
state |
GetRobotState |
查询并显示当前状态 |
mode <STATE> |
SetMode |
切换 FSM,Tab 补全可选状态 |
motion play <path> |
SetMotion(PLAY) |
播放上半身动作文件 |
motion stop |
SetMotion(STOP) |
停止动作播放 |
tracking <path> |
SetTrackingMotion |
切换轨迹跟踪文件 |
controlmode <0\|1> |
SetControlMode |
切换控制范式 |
controlstate |
GetControlState |
查询控制范式 |
shutdown |
Shutdown |
关闭控制器 |
help 与 Tab 补全会根据 available_actions 动态显示当前状态下可用的命令。
典型控制流程
以下流程演示从急停状态到播放上半身动作的完整操作:
操作步骤:
state— 确认当前 FSM 与可切换状态mode ZERO— 进入零位校准mode STAND_WALK— 命令被接受后,轮询state直至fsm_state变为STAND_WALKmode MULTI_AGENT— 进入上半身动作叠加模式motion play Sources/motion/Greeting.txt— 播放动作(文件须存在于机器人侧Sources/motion/目录)
动作文件路径
预设动作位于 Sources/motion/,相对路径以控制程序根目录 /etc/pndbotics/pnd_adam_dds/ 为基准。完整动作列表见 手臂动作服务接口。
motion stop— 停止播放
安全提示
模式切换与动作播放前,请确保机器人处于安全悬挂或开阔场地,并遵循操作指南中的安全规范。
程序集成例程
交互式 CLI 适合手动调试。若需在自己的程序中集成,可参考以下封装。
封装类
import grpc
from comm.grpc import robot_control_pb2 as pb2
from comm.grpc import robot_control_pb2_grpc as pb2_grpc
class PndRobotClient:
def __init__(self, addr: str = "localhost:50051"):
self._stub = pb2_grpc.RobotControlStub(grpc.insecure_channel(addr))
def get_state(self):
return self._stub.GetRobotState(pb2.GetRobotStateRequest())
def set_mode(self, target_state: str):
r = self._stub.SetMode(pb2.SetModeRequest(target_state=target_state))
return r.success, r.message
def play_motion(self, file_path: str):
r = self._stub.SetMotion(pb2.SetMotionRequest(
command=pb2.SetMotionRequest.PLAY, motion_file=file_path))
return r.success, r.message
def stop_motion(self):
r = self._stub.SetMotion(pb2.SetMotionRequest(
command=pb2.SetMotionRequest.STOP))
return r.success, r.message
def set_tracking_motion(self, file_path: str):
r = self._stub.SetTrackingMotion(pb2.SetTrackingMotionRequest(motion_file=file_path))
return r.success, r.message
def set_control_mode(self, domain_id: int):
r = self._stub.SetControlMode(pb2.SetControlModeRequest(domain_id=domain_id))
return r.success, r.message
脚本化调用(一次性命令)
通过命令行参数执行单条指令,便于自动化集成:
import argparse
def main():
parser = argparse.ArgumentParser(description="PND Robot gRPC one-shot client")
parser.add_argument("--addr", default="localhost:50051", help="机器人地址 host:port")
parser.add_argument("--state", action="store_true", help="查询机器人状态")
parser.add_argument("--mode", help="切换 FSM 状态,如 ZERO / STAND_WALK")
parser.add_argument("--motion-play", metavar="PATH", help="播放上半身动作文件")
parser.add_argument("--motion-stop", action="store_true", help="停止动作播放")
parser.add_argument("--tracking", metavar="PATH", help="切换轨迹跟踪文件")
parser.add_argument("--control-mode", type=int, choices=[0, 1], help="0=传统, 1=RL")
args = parser.parse_args()
client = PndRobotClient(args.addr)
if args.state:
s = client.get_state()
print(f"fsm_state={s.fsm_state}")
print(f"switchable_states={list(s.switchable_states)}")
print(f"available_actions={list(s.available_actions)}")
if args.mode:
print("set_mode:", client.set_mode(args.mode))
if args.motion_play:
print("play_motion:", client.play_motion(args.motion_play))
if args.motion_stop:
print("stop_motion:", client.stop_motion())
if args.tracking:
print("set_tracking_motion:", client.set_tracking_motion(args.tracking))
if args.control_mode is not None:
print("set_control_mode:", client.set_control_mode(args.control_mode))
if __name__ == "__main__":
main()
调用示例
# 查询状态
python3 my_client.py --addr 192.168.1.100:50051 --state
# 切换到 ZERO(零位校准)
python3 my_client.py --addr 192.168.1.100:50051 --mode ZERO
# 在 MULTI_AGENT 状态下播放动作
python3 my_client.py --addr 192.168.1.100:50051 --motion-play Sources/motion/Greeting.txt
| 参数 | 说明 | 赋值示例 |
|---|---|---|
--addr |
机器人地址 host:port |
192.168.1.100:50051 |
--state |
查询机器人状态 | (无值) |
--mode |
切换 FSM 状态 | ZERO、STAND_WALK |
--motion-play |
播放上半身动作文件 | Sources/motion/Greeting.txt |
--motion-stop |
停止动作播放 | (无值) |
--tracking |
切换轨迹跟踪文件 | <trajectory_file_path> |
--control-mode |
切换控制范式 | 0 或 1 |
代码解析
rl_client/tools/grpc_client.py 的核心逻辑如下:
- 路径设置:将
rl_client/根目录(tools/的上一级)加入sys.path,以 importcomm.grpc模块。 - 状态刷新:每次命令执行前调用
GetRobotState,缓存fsm_state、switchable_states、available_actions。 - 状态感知 UI:
help和 Tab 补全根据available_actions过滤命令;mode的 Tab 补全来自switchable_states。 - 动作确认:
motion play/stop成功后轮询状态,确认motion_playing与current_motion_file已更新。
常见问题
| 现象 | 可能原因 | 处理建议 |
|---|---|---|
UNAVAILABLE - failed to connect |
控制程序未启动或网络不通 | 确认控制程序运行中,检查 IP 与网段 |
SetMode 返回失败 |
目标状态不在 switchable_states 中 |
先执行 state 查看可切换状态 |
SetMotion 返回失败 |
不在 MULTI_AGENT 状态,或文件不存在 |
确认 FSM 与 available_actions,检查机器人侧文件路径 |
mode STAND_WALK 后 FSM 仍为 ZERO |
零位归零动作尚未完成 | 轮询 state,等待 fsm_state 变化 |