Compare commits

..

6 Commits

Author SHA1 Message Date
e079566597 xbox controller demo
Some checks failed
Secret Leaks / trufflehog (push) Has been cancelled
2025-06-07 11:22:05 +08:00
83d6419d70 手柄控制第一次提交 2025-06-05 21:56:52 +08:00
a0ec9e1cb1 single arm test
Some checks failed
Secret Leaks / trufflehog (push) Has been cancelled
2025-06-05 15:50:26 +08:00
3eede4447d dual arm test 2025-06-05 15:50:18 +08:00
9c6a7d9701 new md 2025-06-05 15:50:11 +08:00
7b201773f3 single arm test 2025-06-05 15:49:57 +08:00
7 changed files with 552 additions and 96 deletions

View File

@@ -10,8 +10,3 @@
- local: getting_started_real_world_robot
title: Getting Started with Real-World Robots
title: "Tutorials"
- sections:
- local: smolvla
title: Use SmolVLA
title: "Policies"

View File

@@ -1,91 +0,0 @@
# Use SmolVLA
SmolVLA is designed to be easy to use and integrate—whether you're finetuning on your own data or plugging it into an existing robotics stack.
<p align="center">
<img src="https://cdn-uploads.huggingface.co/production/uploads/640e21ef3c82bd463ee5a76d/aooU0a3DMtYmy_1IWMaIM.png" alt="SmolVLA architecture." width="500"/>
<br/>
<em>Figure 2. SmolVLA takes as input a sequence of RGB images from multiple cameras, the robots current sensorimotor state, and a natural language instruction. The VLM encodes these into contextual features, which condition the action expert to generate a continuous sequence of actions.</em>
</p>
### Install
First, install the required dependencies:
```python
git clone https://github.com/huggingface/lerobot.git
cd lerobot
pip install -e ".[smolvla]"
```
### Finetune the pretrained model
Use [`smolvla_base`](https://hf.co/lerobot/smolvla_base), our pretrained 450M model, with the lerobot training framework:
```python
python lerobot/scripts/train.py \
--policy.path=lerobot/smolvla_base \
--dataset.repo_id=lerobot/svla_so100_stacking \
--batch_size=64 \
--steps=200000
```
<p align="center">
<img src="https://cdn-uploads.huggingface.co/production/uploads/640e21ef3c82bd463ee5a76d/S-3vvVCulChREwHDkquoc.gif" alt="Comparison of SmolVLA across task variations." width="500"/>
<br/>
<em>Figure 1: Comparison of SmolVLA across task variations. From left to right: (1) asynchronous pick-place cube counting, (2) synchronous pick-place cube counting, (3) pick-place cube counting under perturbations, and (4) generalization on pick-and-place of the lego block with real-world SO101.</em>
</p>
### Train from scratch
If you'd like to build from the architecture (pretrained VLM + action expert) rather than a pretrained checkpoint:
```python
python lerobot/scripts/train.py \
--policy.type=smolvla \
--dataset.repo_id=lerobot/svla_so100_stacking \
--batch_size=64 \
--steps=200000
```
You can also load `SmolVLAPolicy` directly:
```python
from lerobot.common.policies.smolvla.modeling_smolvla import SmolVLAPolicy
policy = SmolVLAPolicy.from_pretrained("lerobot/smolvla_base")
```
## Evaluate the pretrained policy and run it in real-time
If you want to record the evaluation process and safe the videos on the hub, login to your HF account by running:
```python
huggingface-cli login --token ${HUGGINGFACE_TOKEN} --add-to-git-credential
```
Store your Hugging Face repository name in a variable to run these commands:
```python
HF_USER=$(huggingface-cli whoami | head -n 1)
echo $HF_USER
```
Now, indicate the path to the policy, which is `lerobot/smolvla_base` in this case, and run:
```python
python lerobot/scripts/control_robot.py \
--robot.type=so100 \
--control.type=record \
--control.fps=30 \
--control.single_task="Grasp a lego block and put it in the bin." \
--control.repo_id=${HF_USER}/eval_svla_base_test \
--control.tags='["tutorial"]' \
--control.warmup_time_s=5 \
--control.episode_time_s=30 \
--control.reset_time_s=30 \
--control.num_episodes=10 \
--control.push_to_hub=true \
--control.policy.path=lerobot/smolvla_base
```
Depending on your evaluation setup, you can configure the duration and the number of episodes to record for your evaluation suite.

125
realman.md Normal file
View File

@@ -0,0 +1,125 @@
# Install
Create a virtual environment with Python 3.10 and activate it, e.g. with [`miniconda`](https://docs.anaconda.com/free/miniconda/index.html):
```bash
conda create -y -n lerobot python=3.10
conda activate lerobot
```
Install 🤗 LeRobot:
```bash
pip install -e . -i https://pypi.tuna.tsinghua.edu.cn/simple
# pip uninstall numpy
# pip install numpy==1.26.0
# pip install pynput
```
/!\ For Linux only, ffmpeg and opencv requires conda install for now. Run this exact sequence of commands:
```bash
conda install ffmpeg=7.1.1 -c conda-forge
# pip uninstall opencv-python
# conda install "opencv>=4.10.0"
```
Install Realman SDK:
```bash
pip install Robotic_Arm==1.0.4.1
pip install pygame
```
# piper集成lerobot
见lerobot_piper_tutorial/1. 🤗 LeRobot新增机械臂的一般流程.pdf
# Teleoperate
```bash
cd piper_scripts/
bash can_activate.sh can0 1000000
cd ..
python lerobot/scripts/control_robot.py \
--robot.type=piper \
--robot.inference_time=false \
--control.type=teleoperate
```
# Record
Set dataset root path
```bash
HF_USER=$PWD/data
echo $HF_USER
```
```bash
python lerobot/scripts/control_robot.py \
--robot.type=piper \
--robot.inference_time=false \
--control.type=record \
--control.fps=30 \
--control.single_task="move" \
--control.repo_id=${HF_USER}/test \
--control.num_episodes=2 \
--control.warmup_time_s=2 \
--control.episode_time_s=10 \
--control.reset_time_s=10 \
--control.play_sounds=true \
--control.push_to_hub=false
```
Press right arrow -> at any time during episode recording to early stop and go to resetting. Same during resetting, to early stop and to go to the next episode recording.
Press left arrow <- at any time during episode recording or resetting to early stop, cancel the current episode, and re-record it.
Press escape ESC at any time during episode recording to end the session early and go straight to video encoding and dataset uploading.
# visualize
```bash
python lerobot/scripts/visualize_dataset.py \
--repo-id ${HF_USER}/test \
--episode-index 0
```
# Replay
```bash
python lerobot/scripts/control_robot.py \
--robot.type=piper \
--robot.inference_time=false \
--control.type=replay \
--control.fps=30 \
--control.repo_id=${HF_USER}/test \
--control.episode=0
```
# Caution
1. In lerobots/common/datasets/video_utils, the vcodec is set to **libopenh264**, please find your vcodec by **ffmpeg -codecs**
# Train
具体的训练流程见lerobot_piper_tutorial/2. 🤗 AutoDL训练.pdf
```bash
python lerobot/scripts/train.py \
--dataset.repo_id=${HF_USER}/jack \
--policy.type=act \
--output_dir=outputs/train/act_jack \
--job_name=act_jack \
--device=cuda \
--wandb.enable=true
```
# Inference
还是使用control_robot.py中的record loop配置 **--robot.inference_time=true** 可以将手柄移出。
```bash
python lerobot/scripts/control_robot.py \
--robot.type=piper \
--robot.inference_time=true \
--control.type=record \
--control.fps=30 \
--control.single_task="move" \
--control.repo_id=$USER/eval_act_jack \
--control.num_episodes=1 \
--control.warmup_time_s=2 \
--control.episode_time_s=30 \
--control.reset_time_s=10 \
--control.push_to_hub=false \
--control.policy.path=outputs/train/act_koch_pick_place_lego/checkpoints/latest/pretrained_model
```

View File

@@ -0,0 +1,31 @@
from Robotic_Arm.rm_robot_interface import *
armleft = RoboticArm(rm_thread_mode_e.RM_TRIPLE_MODE_E)
armright = RoboticArm()
lefthandle = armleft.rm_create_robot_arm("169.254.128.18", 8080)
print("机械臂ID", lefthandle.id)
righthandle = armright.rm_create_robot_arm("169.254.128.19", 8080)
print("机械臂ID", righthandle.id)
# software_info = armleft.rm_get_arm_software_info()
# if software_info[0] == 0:
# print("\n================== Arm Software Information ==================")
# print("Arm Model: ", software_info[1]['product_version'])
# print("Algorithm Library Version: ", software_info[1]['algorithm_info']['version'])
# print("Control Layer Software Version: ", software_info[1]['ctrl_info']['version'])
# print("Dynamics Version: ", software_info[1]['dynamic_info']['model_version'])
# print("Planning Layer Software Version: ", software_info[1]['plan_info']['version'])
# print("==============================================================\n")
# else:
# print("\nFailed to get arm software information, Error code: ", software_info[0], "\n")
print("Left: ", armleft.rm_get_current_arm_state())
print("Left: ", armleft.rm_get_arm_all_state())
armleft.rm_movej_p()
# print("Right: ", armright.rm_get_current_arm_state())
# 断开所有连接,销毁线程
RoboticArm.rm_destory()

352
realman_src/realman_xbox.py Normal file
View File

@@ -0,0 +1,352 @@
#!/usr/bin/env python3
# -*-coding:utf8-*-
from typing import Optional
import time
from Robotic_Arm.rm_robot_interface import *
import pygame
import threading
from typing import Dict
def enable_fun(arm: RoboticArm):
'''
使能机械臂并检测使能状态,尝试5s,如果使能超时则退出程序
'''
enable_flag = False
# 设置超时时间(秒)
timeout = 5
# 记录进入循环前的时间
start_time = time.time()
elapsed_time_flag = False
while not enable_flag:
elapsed_time = time.time() - start_time
print("--------------------")
# 获取机械臂状态
ret = arm.rm_get_current_arm_state()
if ret[0] == 0: # 成功获取状态
arm_state = ret[1]
enable_flag = True
print("使能状态:", enable_flag)
print("--------------------")
# 检查是否超过超时时间
if elapsed_time > timeout:
print("超时....")
elapsed_time_flag = True
enable_flag = True
break
time.sleep(1)
if elapsed_time_flag:
print("程序自动使能超时,退出程序")
exit(0)
class EndPoseController:
def __init__(self, init_joint, init_pose):
# 初始化pygame和手柄
pygame.init()
pygame.joystick.init()
# 检查是否有连接的手柄
if pygame.joystick.get_count() == 0:
raise Exception("未检测到手柄")
# 初始化手柄
self.joystick = pygame.joystick.Joystick(0)
self.joystick.init()
# 摇杆死区
self.deadzone = 0.15
# 精细控制模式
self.fine_control_mode = False
# 初始化末端姿态 [X, Y, Z, RX, RY, RZ] XYZ meter RX RY RZ rad
self.init_joint = init_joint
self.init_pose = init_pose
self.joint = self.init_joint
self.pose = self.init_pose
self.pose_speeds = [0.0] * 6
# 末端位姿限制
self.pose_limits = [
(-0.850, 0.850), # X (m)
(-0.850, 0.850), # Y (m)
(0.850, 0.850), # Z (m) - 设置最小高度防止碰撞
(-3.14, 3.14), # RX (rad)
(-3.14, 3.14), # RY (rad)
(-3.14, 3.14) # RZ (rad)
]
# 控制参数
self.linear_step = 0.0015 # 线性移动步长(m)
self.angular_step = 0.001 # 角度步长(rad) - 从度转换为弧度
# 夹爪状态和速度
self.gripper_open = False
self.gripper_speed = 10
# 启动更新线程
self.running = True
self.thread = threading.Thread(target=self.update_controller)
self.thread.start()
print("机械臂末端位姿控制器已启动")
def _apply_nonlinear_mapping(self, value):
"""应用非线性映射以提高控制精度"""
# 保持符号,但对数值应用平方映射以提高精度
sign = 1 if value >= 0 else -1
return sign * (abs(value) ** 2)
def _normalize_angle(self, angle):
"""将角度归一化到[-π, π]范围内"""
import math
while angle > math.pi:
angle -= 2 * math.pi
while angle < -math.pi:
angle += 2 * math.pi
return angle
def update_controller(self):
while self.running:
try:
pygame.event.pump()
except Exception as e:
print(f"控制器错误: {e}")
self.stop()
continue
# 检查精细控制模式切换 (使用L3按钮)
if self.joystick.get_button(10): # L3按钮
self.fine_control_mode = not self.fine_control_mode
print(f"切换到{'精细' if self.fine_control_mode else '普通'}控制模式")
time.sleep(0.3) # 防止多次触发
# 检查重置按钮 (7号按钮通常是Start按钮)
if self.joystick.get_button(7): # Start按钮
print("重置机械臂到初始位置...")
# 重置位姿
self.joint = self.init_joint
self.pose = self.init_pose
self.pose_speeds = [0.0] * 6
self.gripper_open = False
self.gripper_speed = 10
print("机械臂已重置到初始位置")
time.sleep(0.3) # 防止多次触发
# 更新末端位姿
self.update_end_pose()
# 夹爪控制(圈/叉)
circle = self.joystick.get_button(1) # 夹爪开
cross = self.joystick.get_button(0) # 夹爪关
self.gripper_speed = 10 if circle else (10 if cross else 0)
self.gripper_open = True if circle else False
# 更新夹爪
# self.gripper += self.gripper_speed
# self.gripper = max(0.0, min(0.1, self.gripper))
time.sleep(0.02)
def update_end_pose(self):
print("1", self.pose, "griper", self.gripper_open)
"""更新末端位姿控制"""
# 根据控制模式调整步长
current_linear_step = self.linear_step * (0.1 if self.fine_control_mode else 1.0)
current_angular_step = self.angular_step * (0.1 if self.fine_control_mode else 1.0)
# print(f"步长设置 - 线性: {current_linear_step}, 角度: {current_angular_step}")
print(f"精细控制模式: {self.fine_control_mode}")
# 方向键控制XY
hat = self.joystick.get_hat(0)
hat_up = hat[1] == 1 # Y+
hat_down = hat[1] == -1 # Y-
hat_left = hat[0] == -1 # X-
hat_right = hat[0] == 1 # X+
# print(f"方向键状态: up={hat_up}, down={hat_down}, left={hat_left}, right={hat_right}")
# 右摇杆控制Z
right_y_raw = -self.joystick.get_axis(4)
# print(f"右摇杆原始值(axis 4): {self.joystick.get_axis(4)}")
# print(f"右摇杆处理值: {right_y_raw}")
# 左摇杆控制RZ
left_y_raw = -self.joystick.get_axis(1)
# print(f"左摇杆原始值(axis 1): {self.joystick.get_axis(1)}")
# print(f"左摇杆处理值: {left_y_raw}")
# 应用死区
right_y = 0.0 if abs(right_y_raw) < self.deadzone else right_y_raw
left_y = 0.0 if abs(left_y_raw) < self.deadzone else left_y_raw
# print(f"死区处理后 - 右摇杆: {right_y}, 左摇杆: {left_y}")
# 计算各轴速度
self.pose_speeds[0] = current_linear_step if hat_up else (-current_linear_step if hat_down else 0.0) # X
self.pose_speeds[1] = current_linear_step if hat_left else (-current_linear_step if hat_right else 0.0) # Y
# 设置Z速度右摇杆Y轴控制
z_mapping = self._apply_nonlinear_mapping(right_y)
# print(f"Z轴非线性映射: {right_y} -> {z_mapping}")
self.pose_speeds[2] = z_mapping * current_linear_step # Z
# L1/R1控制RX旋转
LB = self.joystick.get_button(4) # RX-
RB = self.joystick.get_button(5) # RX+
self.pose_speeds[3] = (-current_angular_step if LB else (current_angular_step if RB else 0.0))
# △/□控制RY旋转
triangle = self.joystick.get_button(2) # RY+
square = self.joystick.get_button(3) # RY-
self.pose_speeds[4] = (current_angular_step if triangle else (-current_angular_step if square else 0.0))
# 左摇杆Y轴控制RZ旋转
rz_mapping = self._apply_nonlinear_mapping(left_y)
# print(f"RZ轴非线性映射: {left_y} -> {rz_mapping}")
self.pose_speeds[5] = rz_mapping * current_angular_step * 2 # RZ
# print(f"计算出的速度: {self.pose_speeds}")
# 更新末端位姿
old_pose = self.pose.copy()
for i in range(6):
self.pose[i] += self.pose_speeds[i]
# print(f"位姿更新: {old_pose} -> {self.pose}")
# 位置限制
# pose_before_limit = self.pose.copy()
# for i in range(3):
# min_val, max_val = self.pose_limits[i]
# self.pose[i] = max(min_val, min(max_val, self.pose[i]))
# if pose_before_limit != self.pose:
# print(f"位置限制生效: {pose_before_limit} -> {self.pose}")
# 角度归一化处理
pose_before_normalize = self.pose.copy()
for i in range(3, 6):
self.pose[i] = self._normalize_angle(self.pose[i])
# if pose_before_normalize != self.pose:
# print(f"角度归一化生效: {pose_before_normalize} -> {self.pose}")
# print("2", self.pose)
# print("=" * 50)
def update_state(self, end_pose, joint_state):
"""更新状态信息(从机械臂获取当前状态)"""
# 这里可以选择是否要同步机械臂的实际位置到控制器
# 如果需要严格同步,可以取消下面的注释
# self.pose = end_pose.copy()
pass
def get_action(self) -> Dict:
"""获取当前控制命令"""
return {
'X': self.pose[0],
'Y': self.pose[1],
'Z': self.pose[2],
'RX': self.pose[3],
'RY': self.pose[4],
'RZ': self.pose[5],
'gripper_speed': self.gripper_speed,
'gripper_open': self.gripper_open
}
def stop(self):
"""停止控制器"""
self.running = False
if self.thread.is_alive():
self.thread.join()
pygame.quit()
print("控制器已退出")
def reset(self):
"""重置到初始状态"""
self.joint = self.init_joint
self.pose = self.init_pose
self.pose_speeds = [0.0] * 6
self.gripper_open = False
self.gripper_speed = 10
print("已重置到初始状态")
if __name__ == "__main__":
# 初始化睿尔曼机械臂
arm = RoboticArm(rm_thread_mode_e.RM_TRIPLE_MODE_E)
init_joint = [-90, 90, 90, 90, -90, -90, 90]
init_pose = [-0.030, 0.255, 0.161, 3.142, 0, -1.57]
# 创建机械臂连接
handle = arm.rm_create_robot_arm("192.168.3.18", 8080)
print(f"机械臂连接ID: {handle.id}")
# 使能机械臂
enable_fun(arm=arm)
teleop = EndPoseController(init_joint, init_pose)
try:
while True:
# 获取当前控制命令
action = teleop.get_action()
# 构建目标位姿列表 [X, Y, Z, RX, RY, RZ]
target_pose = [
action['X'], # X (m)
action['Y'], # Y (m)
action['Z'], # Z (m)
action['RX'], # RX (rad)
action['RY'], # RY (rad)
action['RZ'] # RZ (rad)
]
# 使用笛卡尔空间直线运动控制末端位姿
# 参数: 目标位姿, 速度比例(20%), 交融半径(0), 连接标志(0), 阻塞模式(0-非阻塞)
result = arm.rm_movej_p(target_pose, 50, 0, 0, 1)
if result != 0:
print(f"运动控制错误,错误码: {result}")
if action['gripper_open']:
# arm.rm_set_gripper_release(action['gripper_speed'], block=True)
arm.rm_set_gripper_position(1000, True, 1)
else:
# arm.rm_set_gripper_pick(action['gripper_speed'], force=50, block=True)
arm.rm_set_gripper_position(1, True, 1)
# 获取当前机械臂状态
ret = arm.rm_get_current_arm_state()
if ret[0] == 0: # 成功获取状态
current_pose = ret[1].get('pose', target_pose)
current_joint = ret[1].get('joint', [0]*7)
teleop.update_state(current_pose, current_joint)
print("控制模式: 末端控制")
print(f"目标位姿: {target_pose}")
print(f"当前位姿: {current_pose}")
print(f"关节位置: {current_joint}")
else:
print(f"获取机械臂状态失败,错误码: {ret[0]}")
time.sleep(0.1)
except KeyboardInterrupt:
print("程序被用户中断")
finally:
# 清理资源
teleop.stop()
arm.rm_delete_robot_arm()
print("程序退出完成")

View File

@@ -0,0 +1,23 @@
from Robotic_Arm.rm_robot_interface import *
robot = RoboticArm(rm_thread_mode_e.RM_TRIPLE_MODE_E)
handle = robot.rm_create_robot_arm("192.168.3.18", 8080)
print("机械臂ID", handle.id)
software_info = robot.rm_get_arm_software_info()
if software_info[0] == 0:
print("\n================== Arm Software Information ==================")
print("Arm Model: ", software_info[1]['product_version'])
print("Algorithm Library Version: ", software_info[1]['algorithm_info']['version'])
print("Control Layer Software Version: ", software_info[1]['ctrl_info']['version'])
print("Dynamics Version: ", software_info[1]['dynamic_info']['model_version'])
print("Planning Layer Software Version: ", software_info[1]['plan_info']['version'])
print("==============================================================\n")
else:
print("\nFailed to get arm software information, Error code: ", software_info[0], "\n")
print("Left: ", robot.rm_get_current_arm_state())
print("Left: ", robot.rm_get_arm_all_state())
# 断开所有连接,销毁线程
RoboticArm.rm_destory()

View File

@@ -0,0 +1,21 @@
from Robotic_Arm.rm_robot_interface import *
armleft = RoboticArm(rm_thread_mode_e.RM_TRIPLE_MODE_E)
lefthandle = armleft.rm_create_robot_arm("192.168.3.18", 8080)
print("机械臂ID", lefthandle.id)
print("Left: ", armleft.rm_get_current_arm_state())
print("Left: ", armleft.rm_get_arm_all_state())
# armleft.rm_movej([-90, 90, 90, -90, -90, 90], 50, 0, 0, 1)
# armleft.rm_movej([-90, 90, 90, -90, -90, 90], 50, 0, 0, 1)
# armleft.rm_movej_p([-0.185, 0.315, 0.080, -1.500, -0.800, -0.000], 50, 0, 0, 1)s
# armleft.rm_movel([-0.185, 0.315, 0.080, -1.500, -0.800, -0.000], 50, 0, 0, 1)
armleft.rm_set_gripper_position(1000, True, 2)
import time
time.sleep(3)
armleft.rm_set_gripper_position(1, True, 2)
# 断开所有连接,销毁线程
RoboticArm.rm_destory()