cyy/RoboGauge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'master' of https://github.com/wty-yy/RoboGauge

Browse Source
This commit is contained in:
wty-yy
2026-01-06 01:22:22 +08:00
parent e9754538a5 ab986953b1
commit 22aedec6fb
13 changed files with 472 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
UPDATE.md
View File

@@ -1,6 +1,7 @@
# UPDATE
## 20260106
## 20260103
### v1.0.4
1. 加入save_additional_output参数, 启动会自动记录每个step的latent和weights
Fix bug: 修复高频评估会将之前的评估冲掉问题
## 20260101
### v1.0.3

1
robogauge/tasks/custom/go2/go2_flat_task.py
View File

@@ -7,7 +7,6 @@
@Blog : https://wty-yy.github.io/
@Desc : Go2 Flat Task Configuration
'''
from robogauge.tasks.robots import Go2Config, Go2MoEConfig
from robogauge.tasks.gauge import FlatGaugeConfig
from robogauge.tasks.simulator.mujoco_config import MujocoConfig

1
robogauge/tasks/custom/go2/go2_obstacle_task.py
View File

@@ -7,7 +7,6 @@
@Blog : https://wty-yy.github.io/
@Desc : Go2 Obstacle Task Configuration
'''
from robogauge.tasks.robots import Go2Config, Go2MoEConfig
from robogauge.tasks.gauge import ObstacleGaugeConfig
from robogauge.tasks.simulator.mujoco_config import MujocoConfig

1
robogauge/tasks/custom/go2/go2_slope_task.py
View File

@@ -7,7 +7,6 @@
@Blog : https://wty-yy.github.io/
@Desc : Go2 Slope Task Configuration
'''
from robogauge.tasks.robots import Go2Config, Go2MoEConfig
from robogauge.tasks.gauge import SlopeForwardGaugeConfig, SlopeBackwardGaugeConfig
from robogauge.tasks.simulator.mujoco_config import MujocoConfig

11
robogauge/tasks/custom/go2/go2_wave_task.py
View File

@@ -1,5 +1,12 @@
from robogauge.tasks.robots import Go2Config, Go2MoEConfig
# -*- coding: utf-8 -*-
'''
@File : go2_wave_task.py
@Time : 2026/01/02 23:16:45
@Author : wty-yy
@Version : 1.0
@Blog : https://wty-yy.github.io/
@Desc : Go2 Wave Task Configuration
'''
from robogauge.tasks.gauge import WaveGaugeConfig
from robogauge.tasks.simulator.mujoco_config import MujocoConfig

1
robogauge/tasks/robots/base_robot_config.py
View File

@@ -37,6 +37,7 @@ class RobotConfig(Config):
0.1, 1.0, -1.5, -0.1, 1.0, -1.5]
mj2model_dof_indices = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]
save_additional_output = False
class scales:
lin_vel = 2.0

3
robogauge/tasks/robots/go2/go2_config.py
View File

@@ -7,7 +7,7 @@
@Blog : https://wty-yy.github.io/
@Desc : Go2 Robot Configuration
'''
from typing_extensions import Literal
from typing import Literal
from robogauge.tasks.robots import RobotConfig
class Go2Config(RobotConfig):
@@ -37,6 +37,7 @@ class Go2Config(RobotConfig):
0.1, 1.0, -1.5, -0.1, 1.0, -1.5]
mj2model_dof_indices = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]
save_additional_output = False
class scales(RobotConfig.control.scales):
lin_vel = 2.0

30
robogauge/tasks/robots/go2/go2_moe.py
View File

@@ -9,20 +9,40 @@
'''
import torch
import numpy as np
from collections import defaultdict
from robogauge.tasks.robots.go2.go2 import Go2
from robogauge.utils.logger import logger
class Go2MoE(Go2):
def __init__(self, cfg):
super().__init__(cfg)
self.save_info = defaultdict(list)
self.save_count = 0
def get_action(self, obs: np.ndarray):
obs_tensor = torch.tensor(obs, dtype=torch.float32).unsqueeze(0).to(self.device)
action, results = self.model(obs_tensor)
if isinstance(results, tuple):
if isinstance(results, tuple) and len(results) == 2:
weights, latent = results
latent = latent.detach().cpu().numpy().squeeze(0)
else:
weights = results
latent = latent.detach().cpu().numpy().squeeze(0) if latent is not None else None
weights = weights.detach().cpu().numpy().squeeze(0) if weights is not None else None
if self.cfg.control.save_additional_output:
self.save_info['latent'].append(latent)
self.save_info['weights'].append(weights)
action = action.detach().cpu().numpy().squeeze(0)[self.model2mj_idx]
weights = weights.detach().cpu().numpy().squeeze(0)
self.last_action = action
target_dof_pos = action * self.action_scale + self.default_dof_pos
return target_dof_pos, self.p_gains, self.d_gains, self.control_type
def reset(self):
super().reset()
save_path = logger.log_dir / f"moe_info_{self.save_count}.npz"
if self.cfg.control.save_additional_output:
np.savez_compressed(save_path,
weights=np.array(self.save_info['weights']),
latent=np.array(self.save_info['latent'])
)
logger.info(f"Saved MoE info to {save_path}")
self.save_count += 1
self.save_info = defaultdict(list)

1
robogauge/tasks/robots/go2/go2_moe_config.py
View File

@@ -14,6 +14,7 @@ class Go2MoEConfig(Go2Config):
class control(Go2Config.control):
model_path = "{ROBOGAUGE_ROOT_DIR}/resources/models/go2/go2_moe_cts_124k.pt"
save_additional_output = False
class Go2MoETerrainConfig(Go2MoEConfig):
""" Go2 MoE Robot Configuration for Terrain Tasks (wave, stairs up/down, slope, obstacles) """

140
robogauge/utils/visualize/plot_latent_pca.py Normal file
View File

@@ -0,0 +1,140 @@
import numpy as np
import matplotlib.pyplot as plt
from sklearn.decomposition import PCA
import os
import glob
# ================= 配置区域 =================
data_root = "/root/Coding/RoboGauge/logs_latent"
# 地形列表 (我们需要遍历所有地形来收集同一个指令的数据)
terrains = ['flat', 'wave', 'slope_fd', 'slope_bd', 'stairs_fd', 'stairs_bd', 'obstacle']
# 指令 ID 与含义的映射 (根据你的描述)
command_map = {
0: "Pos X",
1: "Neg X",
2: "Pos Y",
3: "Neg Y",
4: "Pos Yaw",
5: "Neg Yaw"
}
# 颜色映射 (为6种指令分配不同颜色)
cmd_colors = {
0: '#d62728', # 红
1: '#1f77b4', # 蓝
2: '#2ca02c', # 绿
3: '#ff7f0e', # 橙
4: '#9467bd', # 紫
5: '#8c564b' # 棕
}
# 每种指令最大保留样本数
# 因为我们要把7个地形的数据合起来，数据量会很大，必须采样
# 建议 1000 - 2000，太少看不出分布，太多会糊成一团
MAX_SAMPLES_PER_CMD = 3000
# ================= 功能函数 =================
def load_combined_command_data(root_path, terrain_list, cmd_id):
"""
遍历所有地形文件夹，寻找指定 cmd_id 的 npz 文件，并将它们全部合并
"""
cmd_latents = []
filename = f"moe_info_{cmd_id}.npz"
for t_name in terrain_list:
# 搜索路径: /root/.../go2_moe_{terrain}_latent/*/moe_info_{id}.npz
search_pattern = os.path.join(root_path, f"go2_moe_{t_name}_latent", "*", filename)
files = glob.glob(search_pattern)
for f in files:
try:
data = np.load(f)
if 'latent' in data:
cmd_latents.append(data['latent'])
except:
pass
if not cmd_latents:
return None
# 合并该指令下所有地形的数据
combined = np.concatenate(cmd_latents, axis=0)
return combined
# ================= 主程序 =================
print(f"Start processing. Grouping by COMMAND (0-5)...")
# 1. 数据收集与预处理
all_data = [] # 存放 latent 向量
all_labels = [] # 存放对应的指令 ID (0, 1, 2...)
for cmd_id, cmd_name in command_map.items():
print(f" - Loading data for Command {cmd_id}: {cmd_name} ...", end=" ")
# 获取该指令在所有地形下的数据汇总
raw_data = load_combined_command_data(data_root, terrains, cmd_id)
if raw_data is not None:
# 随机下采样 (防止数据量过大)
n_total = len(raw_data)
if n_total > MAX_SAMPLES_PER_CMD:
indices = np.random.choice(n_total, MAX_SAMPLES_PER_CMD, replace=False)
data_sample = raw_data[indices]
else:
data_sample = raw_data
all_data.append(data_sample)
# 记录标签：有多少个数据，就存多少个 label
all_labels.extend([cmd_id] * len(data_sample))
print(f"Got {len(data_sample)} samples (from {n_total})")
else:
print("No data found!")
if not all_data:
print("Error: No data loaded.")
exit()
# 将列表转换为大矩阵
X = np.concatenate(all_data, axis=0)
y = np.array(all_labels)
# 2. PCA 降维
print(f"Running PCA on total {X.shape[0]} samples...")
pca = PCA(n_components=2)
X_2d = pca.fit_transform(X)
# 3. 可视化绘制
plt.figure(figsize=(10, 8), dpi=120)
# 遍历 0-5 进行绘制
for cmd_id in command_map.keys():
# 提取属于当前指令的 2D 点
indices = (y == cmd_id)
points = X_2d[indices]
if len(points) > 0:
plt.scatter(
points[:, 0],
points[:, 1],
c=cmd_colors[cmd_id],
label=command_map[cmd_id],
alpha=0.6, # 透明度
s=15 # 点大小
)
plt.title("PCA of Latent Space grouped by Control Command (All Terrains Mixed)")
plt.xlabel("PC 1")
plt.ylabel("PC 2")
plt.legend(title="Control Commands", markerscale=1.5)
plt.grid(True, linestyle='--', alpha=0.4)
save_path = 'latent_pca_by_command_mixed.png'
plt.savefig(save_path)
print(f"Done! Visualization saved to {save_path}")
plt.show()

198
robogauge/utils/visualize/plot_latent_tsne.py Normal file
View File

@@ -0,0 +1,198 @@
import numpy as np
import matplotlib.pyplot as plt
from sklearn.manifold import TSNE
import os
import glob
import pandas as pd
from sklearn.metrics import silhouette_score, silhouette_samples, calinski_harabasz_score
# ================= 配置区域 =================
data_root = "/root/Coding/RoboGauge/logs_latent/moe"
terrains = ['flat', 'wave', 'slope_fd', 'slope_bd', 'stairs_fd', 'stairs_bd', 'obstacle']
command_map = {
0: "Pos X (Forward)",
1: "Neg X (Backward)",
2: "Pos Y (Left)",
3: "Neg Y (Right)",
4: "Pos Yaw (Turn Left)",
5: "Neg Yaw (Turn Right)"
}
cmd_colors = {
0: '#d62728', 1: '#1f77b4', 2: '#2ca02c',
3: '#ff7f0e', 4: '#9467bd', 5: '#8c564b'
}
# t-SNE 计算较慢，且点太多会严重重叠
# 建议每种指令采样 800 - 1000 个点即可，足以看清分布
MAX_SAMPLES_PER_CMD = 2000
# t-SNE 参数
# perplexity: 困惑度，通常在 5-50 之间。越大越关注全局结构，越小越关注局部邻居。
TSNE_PERPLEXITY = 30
TSNE_ITER = 1000
# ================= 功能函数 =================
# key = 'weights' # 'latent'
key = 'latent'
def load_combined_command_data(root_path, terrain_list, cmd_id):
"""读取指定指令在所有地形下的数据并合并"""
cmd_latents = []
filename = f"moe_info_{cmd_id}.npz"
for t_name in terrain_list:
search_pattern = os.path.join(root_path, f"go2_moe_{t_name}_latent", "*", filename)
files = glob.glob(search_pattern)
for f in files:
try:
data = np.load(f)
if key in data:
cmd_latents.append(data[key])
except: pass
if not cmd_latents: return None
return np.concatenate(cmd_latents, axis=0)
# ================= 主程序 =================
print(f"Start processing t-SNE visualization...")
all_data = []
all_labels = []
# 1. 数据收集
for cmd_id, cmd_name in command_map.items():
raw_data = load_combined_command_data(data_root, terrains, cmd_id)
if raw_data is not None:
# 随机下采样
n_total = len(raw_data)
if n_total > MAX_SAMPLES_PER_CMD:
indices = np.random.choice(n_total, MAX_SAMPLES_PER_CMD, replace=False)
data_sample = raw_data[indices]
else:
data_sample = raw_data
all_data.append(data_sample)
all_labels.extend([cmd_id] * len(data_sample))
print(f" - Cmd {cmd_id}: {len(data_sample)} samples")
if not all_data:
print("Error: No data loaded.")
exit()
X = np.concatenate(all_data, axis=0)
y = np.array(all_labels)
print(f"Running t-SNE on {X.shape} matrix...")
print(f" (Perplexity={TSNE_PERPLEXITY}, Iterations={TSNE_ITER})")
print(" This may take a moment...")
# 2. t-SNE 降维
# init='pca' 通常比 'random' 更稳定，能更好地保留全局结构
tsne = TSNE(
n_components=2,
perplexity=TSNE_PERPLEXITY,
n_iter=TSNE_ITER,
init='pca',
learning_rate='auto',
random_state=42,
verbose=1,
)
X_embedded = tsne.fit_transform(X)
# 3. 可视化
plt.figure(figsize=(12, 10), dpi=120)
for cmd_id in command_map.keys():
indices = (y == cmd_id)
points = X_embedded[indices]
if len(points) > 0:
plt.scatter(
points[:, 0],
points[:, 1],
c=cmd_colors[cmd_id],
label=command_map[cmd_id],
alpha=0.6,
s=20 #稍微大一点的点
)
plt.title(f"t-SNE Visualization of {key.capitalize()} Space by Command\n(Perplexity={TSNE_PERPLEXITY}, Mixed Terrains)")
# t-SNE 的坐标轴没有物理意义，所以隐藏刻度通常更好看
plt.xticks([])
plt.yticks([])
plt.legend(title="Control Commands", markerscale=2, bbox_to_anchor=(1.05, 1), loc='upper left')
plt.tight_layout()
save_path = os.path.join(data_root, f'{key}_tsne_by_command.png')
plt.savefig(save_path)
print(f"Done! Visualization saved to {save_path}")
plt.show()
def evaluate_latent_space(X, y, command_map):
"""
X: (N, D) 原始高维 Latent 向量 (例如 32维)
y: (N,) 对应的指令标签 (0-5)
command_map: label到名称的映射字典
"""
print(f"--- Evaluating Latent Space (Samples: {X.shape[0]}, Dim: {X.shape[1]}) ---")
# 1. 全局指标 (Global Metrics)
# 轮廓系数 (Silhouette): 越接近 1 越好
global_sil = silhouette_score(X, y)
# Calinski-Harabasz Index (CHI): 数值越大越好 (表示类间离散度高，类内离散度低)
global_chi = calinski_harabasz_score(X, y)
print(f"Global Silhouette Score: {global_sil:.4f} (范围 -1 到 1, 越大越好)")
print(f"Global Calinski-Harabasz: {global_chi:.1f} (数值越大越好)")
print("-" * 30)
# 2. 逐类指标 (Per-Class Metrics)
# 计算每个样本的轮廓系数
sample_silhouette_values = silhouette_samples(X, y)
class_scores = []
for label in sorted(command_map.keys()):
# 提取当前类的所有样本的轮廓系数
ith_class_silhouette_values = sample_silhouette_values[y == label]
# 计算该类的平均分
avg_score = np.mean(ith_class_silhouette_values)
# 计算该类的紧密度 (Intra-class distance) - 可选
# 这里直接用轮廓系数代表聚集程度
class_scores.append({
"Command ID": label,
"Command Name": command_map[label],
"Silhouette Score": avg_score,
"Sample Count": len(ith_class_silhouette_values)
})
# 转为 DataFrame 展示
df = pd.DataFrame(class_scores)
print(df.to_string(index=False))
print("-" * 30 + "\n")
return df, global_sil
# 1. 评估原始高维空间 (Intrinsic Quality)
print(">>> Evaluating Original High-Dim Latent Space Quality...")
df_original, score_original = evaluate_latent_space(X, y, command_map)
print(f"Original Space Score: {score_original:.4f}")
# 2. 评估 t-SNE 降维后的 2D 空间 (Visual Cluster Quality)
print("\n>>> Evaluating t-SNE Embedded Space Quality (2D)...")
# 直接传入 X_embedded 即可，evaluate_latent_space 函数对维度不敏感
df_tsne, score_tsne = evaluate_latent_space(X_embedded, y, command_map)
# 保存 t-SNE 的评估结果
save_path_tsne = os.path.join(data_root, f'{key}_tsne_2d_evaluation.csv')
df_tsne.to_csv(save_path_tsne, index=False)
print(f"t-SNE 2D Evaluation results saved to {save_path_tsne}")
print(f"t-SNE Space Score: {score_tsne:.4f}")

12
run.bash → scripts/run.bash
View File

@@ -7,7 +7,7 @@ python robogauge/scripts/run.py \
--experiment-name debug \
--stress-benchmark \
--stress-terrain-names flat slope_fd slope_bd stairs_fd stairs_bd wave obstacle \
--num-processes 62 \
--num-processes 35 \
--seeds 0 1 2 \
--search-seeds 0 1 2 3 4 \
--frictions 0.5 0.75 1.0 1.25 1.5 1.75 2.0 2.25 2.5 \
@@ -20,7 +20,7 @@ python robogauge/scripts/run.py \
--experiment-name debug \
--stress-benchmark \
--stress-terrain-names flat slope_fd slope_bd stairs_fd stairs_bd wave obstacle \
--num-processes 62 \
--num-processes 35 \
--seeds 0 1 2 \
--search-seeds 0 1 2 3 4 \
--frictions 0.5 0.75 1.0 1.25 1.5 1.75 2.0 2.25 2.5 \
@@ -33,7 +33,7 @@ python robogauge/scripts/run.py \
--experiment-name debug \
--stress-benchmark \
--stress-terrain-names flat slope_fd slope_bd stairs_fd stairs_bd wave obstacle \
--num-processes 62 \
--num-processes 35 \
--seeds 0 1 2 \
--search-seeds 0 1 2 3 4 \
--frictions 0.5 0.75 1.0 1.25 1.5 1.75 2.0 2.25 2.5 \
@@ -45,7 +45,7 @@ python robogauge/scripts/run.py \
--experiment-name debug \
--stress-benchmark \
--stress-terrain-names flat slope_fd slope_bd stairs_fd stairs_bd wave obstacle \
--num-processes 62 \
--num-processes 35 \
--seeds 0 1 2 \
--search-seeds 0 1 2 3 4 \
--frictions 0.5 0.75 1.0 1.25 1.5 1.75 2.0 2.25 2.5 \
@@ -58,7 +58,7 @@ python robogauge/scripts/run.py \
--experiment-name debug \
--stress-benchmark \
--stress-terrain-names flat slope_fd slope_bd stairs_fd stairs_bd wave obstacle \
--num-processes 62 \
--num-processes 35 \
--seeds 0 1 2 \
--search-seeds 0 1 2 3 4 \
--frictions 0.5 0.75 1.0 1.25 1.5 1.75 2.0 2.25 2.5 \
@@ -71,7 +71,7 @@ python robogauge/scripts/run.py \
--experiment-name debug \
--stress-benchmark \
--stress-terrain-names flat slope_fd slope_bd stairs_fd stairs_bd wave obstacle \
--num-processes 62 \
--num-processes 35 \
--seeds 0 1 2 \
--search-seeds 0 1 2 3 4 \
--frictions 0.5 0.75 1.0 1.25 1.5 1.75 2.0 2.25 2.5 \

88
scripts/run_save_latent.bash Executable file
View File

@@ -0,0 +1,88 @@
#!/bin/bash
# Change robogauge/tasks/robots/go2/go2_moe_config.py `save_additional_output = True`
# This
source /root/Programs/miniforge3/bin/activate robot
# python robogauge/scripts/run.py \
# --task-name go2_moe \
# --model-path /root/Coding/RoboGauge/mytest/go2_moe_cts_111k_61.09%.pt \
# --experiment-name debug \
# --stress-benchmark \
# --stress-terrain-names flat slope_fd slope_bd stairs_fd stairs_bd wave obstacle \
# --num-processes 35 \
# --seeds 0 1 2 \
# --search-seeds 0 1 2 3 4 \
# --frictions 0.5 0.75 1.0 1.25 1.5 1.75 2.0 2.25 2.5 \
# --compress-logs \
# --headless
python robogauge/scripts/run.py \
--task-name go2_moe.flat \
--model-path /root/Coding/RoboGauge/mytest/go2_moe_cts_111k_61.09%.pt \
--experiment-name latent \
--seed 0 \
--friction 1.5 \
--goals max_velocity \
--headless
python robogauge/scripts/run.py \
--task-name go2_moe.obstacle \
--level 9 \
--model-path /root/Coding/RoboGauge/mytest/go2_moe_cts_111k_61.09%.pt \
--experiment-name latent \
--seed 0 \
--friction 1.5 \
--goals max_velocity \
--headless
python robogauge/scripts/run.py \
--task-name go2_moe.slope_bd \
--level 6 \
--model-path /root/Coding/RoboGauge/mytest/go2_moe_cts_111k_61.09%.pt \
--experiment-name latent \
--seed 0 \
--friction 1.5 \
--goals max_velocity \
--headless
python robogauge/scripts/run.py \
--task-name go2_moe.slope_fd \
--level 6 \
--model-path /root/Coding/RoboGauge/mytest/go2_moe_cts_111k_61.09%.pt \
--experiment-name latent \
--seed 0 \
--friction 1.5 \
--goals max_velocity \
--headless
python robogauge/scripts/run.py \
--task-name go2_moe.stairs_bd \
--level 7 \
--model-path /root/Coding/RoboGauge/mytest/go2_moe_cts_111k_61.09%.pt \
--experiment-name latent \
--seed 0 \
--friction 1.5 \
--goals max_velocity \
--headless
python robogauge/scripts/run.py \
--task-name go2_moe.stairs_fd \
--level 9 \
--model-path /root/Coding/RoboGauge/mytest/go2_moe_cts_111k_61.09%.pt \
--experiment-name latent \
--seed 0 \
--friction 1.5 \
--goals max_velocity \
--headless
python robogauge/scripts/run.py \
--task-name go2_moe.wave \
--level 6 \
--model-path /root/Coding/RoboGauge/mytest/go2_moe_cts_111k_61.09%.pt \
--experiment-name latent \
--seed 0 \
--friction 1.5 \
--goals max_velocity \
--headless