392 lines
15 KiB
Python
392 lines
15 KiB
Python
import os
|
||
import subprocess
|
||
import json
|
||
import sys
|
||
import platform
|
||
import re
|
||
import logging
|
||
import importlib
|
||
import warnings
|
||
import time
|
||
|
||
# 获取当前脚本所在目录的父目录（项目根目录）
|
||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||
project_root = os.path.dirname(script_dir)
|
||
|
||
# 配置日志 - 使用相对路径
|
||
log_file = os.path.join(project_root, 'environment_verify.log')
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s - %(levelname)s - %(message)s',
|
||
handlers=[
|
||
logging.StreamHandler(),
|
||
logging.FileHandler(log_file)
|
||
]
|
||
)
|
||
|
||
def run_command(command, description=None, show_output=False):
|
||
"""运行命令并打印状态"""
|
||
if description:
|
||
logging.info(f"=== {description} ===")
|
||
logging.info(f"执行: {command}")
|
||
|
||
try:
|
||
# 明确指定使用bash执行命令
|
||
result = subprocess.run(['/bin/bash', '-c', command],
|
||
check=True,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE)
|
||
output = result.stdout.decode('utf-8')
|
||
logging.info(f"命令执行成功: {command}")
|
||
if show_output:
|
||
logging.info(f"输出:\n{output}")
|
||
return True, output
|
||
except subprocess.CalledProcessError as e:
|
||
error = e.stderr.decode('utf-8')
|
||
logging.error(f"命令执行失败: {command}")
|
||
logging.error(f"错误信息: {error}")
|
||
return False, error
|
||
|
||
class SapienRenderTester:
|
||
def __init__(self):
|
||
warnings.simplefilter(action='ignore', category=FutureWarning)
|
||
warnings.simplefilter(action='ignore', category=UserWarning)
|
||
|
||
# 添加tools路径
|
||
sys.path.append(os.path.join(project_root, 'tools'))
|
||
|
||
def test_render_pipeline(self):
|
||
"""测试Sapien渲染管线是否正常工作"""
|
||
try:
|
||
# 导入必要的包
|
||
import numpy as np
|
||
import torch
|
||
import sapien.core as sapien
|
||
from sapien.utils.viewer import Viewer
|
||
import gymnasium as gym
|
||
import transforms3d as t3d
|
||
from collections import OrderedDict
|
||
|
||
# 设置日志级别
|
||
import toppra as ta
|
||
ta.setup_logging("CRITICAL") # 隐藏日志
|
||
|
||
# 初始化引擎
|
||
engine = sapien.Engine()
|
||
|
||
# 声明sapien渲染器
|
||
from sapien.render import set_global_config
|
||
set_global_config(max_num_materials=50000, max_num_textures=50000)
|
||
renderer = sapien.SapienRenderer()
|
||
|
||
# 将渲染器提供给sapien模拟器
|
||
engine.set_renderer(renderer)
|
||
|
||
# 设置光线追踪
|
||
sapien.render.set_camera_shader_dir("rt")
|
||
sapien.render.set_ray_tracing_samples_per_pixel(32)
|
||
sapien.render.set_ray_tracing_path_depth(8)
|
||
sapien.render.set_ray_tracing_denoiser("oidn")
|
||
|
||
# 声明sapien场景
|
||
scene_config = sapien.SceneConfig()
|
||
scene = engine.create_scene(scene_config)
|
||
|
||
# 如果以上步骤都没有异常，则渲染管线工作正常
|
||
return True, "Sapien渲染管线初始化成功"
|
||
except Exception as e:
|
||
return False, f"Sapien渲染管线初始化失败: {str(e)}"
|
||
|
||
def verify_robotwin_environment():
|
||
"""验证RoboTwin环境是否正常安装和配置"""
|
||
verification_results = []
|
||
all_passed = True
|
||
|
||
# 获取conda路径
|
||
success, conda_path_output = run_command("conda info --base", "获取conda基础路径")
|
||
if success:
|
||
conda_path = conda_path_output.strip()
|
||
logging.info(f"Conda 基础路径: {conda_path}")
|
||
else:
|
||
logging.error("无法获取conda路径，环境验证失败")
|
||
verification_results.append({
|
||
"component": "Conda",
|
||
"status": "失败",
|
||
"message": "无法获取conda路径"
|
||
})
|
||
all_passed = False
|
||
return all_passed, verification_results
|
||
|
||
# 检查RoboTwin环境是否存在
|
||
check_env_cmd = "conda env list | grep -w RoboTwin || echo ''"
|
||
success, env_check_output = run_command(check_env_cmd, "检查RoboTwin环境是否存在")
|
||
|
||
if not success or not env_check_output.strip() or "RoboTwin" not in env_check_output:
|
||
logging.error("RoboTwin环境不存在")
|
||
verification_results.append({
|
||
"component": "RoboTwin环境",
|
||
"status": "失败",
|
||
"message": "RoboTwin conda环境不存在"
|
||
})
|
||
all_passed = False
|
||
return all_passed, verification_results
|
||
|
||
# 验证RoboTwin环境
|
||
activate_command = f"source {conda_path}/etc/profile.d/conda.sh && conda activate RoboTwin && "
|
||
|
||
# 1. 检查环境是否可激活
|
||
check_activate_cmd = activate_command + "conda info --envs"
|
||
success, activate_output = run_command(check_activate_cmd, "激活RoboTwin环境")
|
||
if not success:
|
||
logging.error("无法激活RoboTwin环境")
|
||
verification_results.append({
|
||
"component": "RoboTwin环境激活",
|
||
"status": "失败",
|
||
"message": "无法激活RoboTwin环境"
|
||
})
|
||
all_passed = False
|
||
else:
|
||
verification_results.append({
|
||
"component": "RoboTwin环境激活",
|
||
"status": "成功",
|
||
"message": "可以成功激活RoboTwin环境"
|
||
})
|
||
|
||
# 2. 检查Python版本
|
||
check_py_cmd = activate_command + "python --version"
|
||
success, py_version_output = run_command(check_py_cmd, "检查Python版本")
|
||
if success:
|
||
# 提取版本号
|
||
match = re.search(r"Python (\d+\.\d+\.\d+)", py_version_output)
|
||
if match:
|
||
py_version = match.group(1)
|
||
if py_version.startswith("3.8.") or py_version.startswith("3.10."):
|
||
verification_results.append({
|
||
"component": "Python版本",
|
||
"status": "成功",
|
||
"message": f"Python版本 {py_version} 符合要求"
|
||
})
|
||
else:
|
||
verification_results.append({
|
||
"component": "Python版本",
|
||
"status": "失败",
|
||
"message": f"Python版本 {py_version} 不符合要求 (需要3.8或3.10)"
|
||
})
|
||
all_passed = False
|
||
else:
|
||
verification_results.append({
|
||
"component": "Python版本",
|
||
"status": "失败",
|
||
"message": "无法确定Python版本"
|
||
})
|
||
all_passed = False
|
||
else:
|
||
verification_results.append({
|
||
"component": "Python版本",
|
||
"status": "失败",
|
||
"message": "无法获取Python版本"
|
||
})
|
||
all_passed = False
|
||
|
||
# 3. 检查核心依赖包
|
||
core_packages = [
|
||
"torch", "torchvision", "sapien", "scipy",
|
||
"mplib", "gymnasium", "trimesh", "open3d",
|
||
"imageio", "pydantic", "zarr", "huggingface_hub",
|
||
"pytorch3d" # 将pytorch3d添加到核心依赖包列表中
|
||
]
|
||
|
||
# 使用importlib.util.find_spec快速检查包是否安装，而不实际导入它们
|
||
import importlib.util
|
||
|
||
logging.info("=== 检查核心依赖包 ===")
|
||
start_total = time.time()
|
||
|
||
for package in core_packages:
|
||
try:
|
||
# 仅检查包是否存在，不导入
|
||
spec = importlib.util.find_spec(package)
|
||
if spec is not None:
|
||
verification_results.append({
|
||
"component": f"依赖包 {package}",
|
||
"status": "成功",
|
||
"message": f"已安装 {package}"
|
||
})
|
||
else:
|
||
verification_results.append({
|
||
"component": f"依赖包 {package}",
|
||
"status": "失败",
|
||
"message": f"未安装 {package}"
|
||
})
|
||
all_passed = False
|
||
|
||
except Exception as e:
|
||
verification_results.append({
|
||
"component": f"依赖包 {package}",
|
||
"status": "失败",
|
||
"message": f"检查 {package} 时发生错误: {str(e)}"
|
||
})
|
||
all_passed = False
|
||
|
||
elapsed_total = time.time() - start_total
|
||
logging.info(f"=== 核心依赖包检查完成 (总耗时: {elapsed_total:.2f}秒) ===")
|
||
|
||
# 5. 检查Vulkan
|
||
check_vulkan_cmd = "DISPLAY='' vulkaninfo --summary 2>/dev/null || echo 'Vulkan not installed'"
|
||
success, vulkan_output = run_command(check_vulkan_cmd, "检查Vulkan")
|
||
|
||
if success and "Vulkan not installed" not in vulkan_output:
|
||
if "Vulkan Instance Version" in vulkan_output:
|
||
# 提取Vulkan版本
|
||
version_match = re.search(r"Vulkan Instance Version: (\d+\.\d+\.\d+)", vulkan_output)
|
||
version = version_match.group(1) if version_match else "未知"
|
||
|
||
verification_results.append({
|
||
"component": "Vulkan",
|
||
"status": "成功",
|
||
"message": f"Vulkan 已安装 (版本 {version})"
|
||
})
|
||
else:
|
||
verification_results.append({
|
||
"component": "Vulkan",
|
||
"status": "成功",
|
||
"message": "Vulkan 已安装"
|
||
})
|
||
else:
|
||
verification_results.append({
|
||
"component": "Vulkan",
|
||
"status": "失败",
|
||
"message": "Vulkan 未安装或安装不正确"
|
||
})
|
||
all_passed = False
|
||
|
||
# 6. 检查资源文件
|
||
resource_directories = ["aloha_maniskill_sim", "models"]
|
||
for directory in resource_directories:
|
||
check_dir_cmd = f"ls -la {os.path.join(project_root, directory)} 2>/dev/null || echo 'not found'"
|
||
success, dir_output = run_command(check_dir_cmd, f"检查{directory}资源目录")
|
||
|
||
if success and "not found" not in dir_output:
|
||
verification_results.append({
|
||
"component": f"资源目录 {directory}",
|
||
"status": "成功",
|
||
"message": f"{directory} 目录存在"
|
||
})
|
||
else:
|
||
verification_results.append({
|
||
"component": f"资源目录 {directory}",
|
||
"status": "失败",
|
||
"message": f"{directory} 目录不存在"
|
||
})
|
||
all_passed = False
|
||
|
||
# 7. 检查mplib修改
|
||
logging.info("=== 检查mplib修改 ===")
|
||
|
||
try:
|
||
# 直接使用Python获取mplib包路径
|
||
import importlib.util
|
||
mplib_spec = importlib.util.find_spec("mplib")
|
||
|
||
if mplib_spec and mplib_spec.origin:
|
||
# 从模块文件路径获取包路径
|
||
mplib_path = os.path.dirname(os.path.dirname(mplib_spec.origin))
|
||
planner_path = os.path.join(mplib_path, "mplib", "planner.py")
|
||
|
||
if os.path.exists(planner_path):
|
||
# 直接读取文件内容
|
||
with open(planner_path, 'r') as f:
|
||
file_content = f.read()
|
||
|
||
# 检查修改1: convex=True 是否已注释
|
||
convex_exists = "convex=True," in file_content
|
||
convex_commented = "# convex=True," in file_content
|
||
|
||
if not convex_exists or convex_commented:
|
||
verification_results.append({
|
||
"component": "mplib修改 - convex=True",
|
||
"status": "成功",
|
||
"message": "convex=True 参数已经被注释或移除"
|
||
})
|
||
else:
|
||
verification_results.append({
|
||
"component": "mplib修改 - convex=True",
|
||
"status": "失败",
|
||
"message": "convex=True 参数未被注释或移除"
|
||
})
|
||
all_passed = False
|
||
|
||
# 检查修改2: 'or collide' 是否已移除
|
||
collide_exists = "or collide or not within_joint_limit" in file_content
|
||
|
||
if not collide_exists:
|
||
verification_results.append({
|
||
"component": "mplib修改 - or collide",
|
||
"status": "成功",
|
||
"message": "'or collide' 条件已被移除"
|
||
})
|
||
else:
|
||
verification_results.append({
|
||
"component": "mplib修改 - or collide",
|
||
"status": "失败",
|
||
"message": "'or collide' 条件未被移除"
|
||
})
|
||
all_passed = False
|
||
else:
|
||
verification_results.append({
|
||
"component": "mplib修改",
|
||
"status": "失败",
|
||
"message": f"mplib planner.py 文件不存在: {planner_path}"
|
||
})
|
||
all_passed = False
|
||
else:
|
||
verification_results.append({
|
||
"component": "mplib修改",
|
||
"status": "失败",
|
||
"message": "无法找到mplib安装位置"
|
||
})
|
||
all_passed = False
|
||
except Exception as e:
|
||
verification_results.append({
|
||
"component": "mplib修改",
|
||
"status": "失败",
|
||
"message": f"检查mplib修改时发生错误: {str(e)}"
|
||
})
|
||
all_passed = False
|
||
|
||
|
||
|
||
# 生成最终报告
|
||
logging.info("\n=== RoboTwin环境验证总结 ===")
|
||
success_count = sum(1 for result in verification_results if result["status"] == "成功")
|
||
total_items = len(verification_results)
|
||
logging.info(f"总检查项: {total_items}, 成功: {success_count}, 失败: {total_items - success_count}")
|
||
|
||
for result in verification_results:
|
||
status_symbol = "✓" if result["status"] == "成功" else "✗"
|
||
logging.info(f"{status_symbol} {result['component']}: {result['message']}")
|
||
|
||
return all_passed, verification_results
|
||
|
||
if __name__ == "__main__":
|
||
# 确保当前工作目录是项目根目录
|
||
os.chdir(project_root)
|
||
|
||
# 运行环境验证
|
||
is_healthy, check_results = verify_robotwin_environment()
|
||
|
||
# 准备输出结果
|
||
output_result = {
|
||
"env_check_result": "healthy" if is_healthy else "unhealthy",
|
||
"check_result_message": check_results
|
||
}
|
||
|
||
# 输出到终端
|
||
print(json.dumps(output_result, indent=2, ensure_ascii=False))
|
||
|
||
# 保存结果到文件
|
||
with open(os.path.join(project_root, 'environment_verify_results.json'), 'w', encoding='utf-8') as f:
|
||
json.dump(output_result, f, indent=2, ensure_ascii=False)
|
||
|
||
# 退出状态码
|
||
sys.exit(0 if is_healthy else 1) |