system_env.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
PaimonCore Boost - System Environment Manager
动态检测和配置系统环境，消除硬编码路径依赖

作者: Geoffrey Wang
项目: PaimonCore Boost - Adaptive Hybrid Core Scheduler
许可证: Apache License 2.0
"""

import os
import sys
import json
import platform
import subprocess
import shutil
from pathlib import Path
from typing import Dict, List, Optional, Tuple
import logging

class SystemEnvironment:
    """系统环境检测和配置管理类"""
    
    def __init__(self, config_file: str = "paimon_config.json"):
        self.config_file = Path(config_file)
        self.project_root = Path(__file__).parent
        self.config = self._load_config()
        self.logger = self._setup_logging()
        
    def _load_config(self) -> Dict:
        """加载配置文件"""
        try:
            if self.config_file.exists():
                with open(self.config_file, 'r', encoding='utf-8') as f:
                    return json.load(f)
            else:
                self.logger.warning(f"配置文件 {self.config_file} 不存在，使用默认配置")
                return self._get_default_config()
        except Exception as e:
            self.logger.error(f"加载配置文件失败: {e}")
            return self._get_default_config()
    
    def _get_default_config(self) -> Dict:
        """获取默认配置"""
        return {
            "compilers": {
                "msvc": {"name": "MSVC", "version_check_cmd": "cl"},
                "clang": {"name": "Clang", "version_check_cmd": "clang++ --version"},
                "gcc": {"name": "GCC", "version_check_cmd": "g++ --version"}
            },
            "paths": {
                "cpp_benchmark_dir": "cpp_benchmark",
                "results_dir": "cpp_compile_results",
                "temp_dir": "temp"
            },
            "performance": {
                "default_intensity": 0.7,
                "max_temperature": 85
            }
        }
    
    def _setup_logging(self) -> logging.Logger:
        """设置日志记录"""
        logger = logging.getLogger('PaimonCore')
        if not logger.handlers:
            handler = logging.StreamHandler()
            formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
            handler.setFormatter(formatter)
            logger.addHandler(handler)
            logger.setLevel(logging.INFO)
        return logger
    
    def detect_system_info(self) -> Dict:
        """检测系统信息"""
        try:
            import psutil
            
            system_info = {
                "platform": platform.platform(),
                "system": platform.system(),
                "version": platform.version(),
                "architecture": platform.architecture()[0],
                "processor": platform.processor(),
                "cpu_count": psutil.cpu_count(),
                "physical_cores": psutil.cpu_count(logical=False),
                "total_memory": psutil.virtual_memory().total,
                "python_version": platform.python_version(),
                "is_admin": self._check_admin_privileges()
            }
            
            self.logger.info(f"系统检测完成: {system_info['system']} {system_info['version']}")
            return system_info
            
        except ImportError:
            self.logger.error("psutil模块未安装，请运行: pip install psutil")
            return {}
        except Exception as e:
            self.logger.error(f"系统信息检测失败: {e}")
            return {}
    
    def _check_admin_privileges(self) -> bool:
        """检查管理员权限"""
        try:
            if platform.system() == "Windows":
                import ctypes
                return ctypes.windll.shell32.IsUserAnAdmin() != 0
            else:
                return os.geteuid() == 0
        except:
            return False
    
    def detect_compilers(self) -> Dict[str, Dict]:
        """自动检测可用的编译器"""
        detected_compilers = {}
        
        # 检测 MSVC
        msvc_info = self._detect_msvc()
        if msvc_info:
            detected_compilers['msvc'] = msvc_info
        
        # 检测 Clang
        clang_info = self._detect_clang()
        if clang_info:
            detected_compilers['clang'] = clang_info
        
        # 检测 GCC
        gcc_info = self._detect_gcc()
        if gcc_info:
            detected_compilers['gcc'] = gcc_info
        
        self.logger.info(f"检测到 {len(detected_compilers)} 个编译器: {list(detected_compilers.keys())}")
        return detected_compilers
    
    def _detect_msvc(self) -> Optional[Dict]:
        """检测MSVC编译器"""
        try:
            # 尝试常见的Visual Studio路径
            common_paths = [
                "C:\\Program Files\\Microsoft Visual Studio\\2022\\Community\\VC\\Auxiliary\\Build\\vcvarsall.bat",
                "C:\\Program Files\\Microsoft Visual Studio\\2022\\Professional\\VC\\Auxiliary\\Build\\vcvarsall.bat",
                "C:\\Program Files\\Microsoft Visual Studio\\2022\\Enterprise\\VC\\Auxiliary\\Build\\vcvarsall.bat",
                "C:\\Program Files (x86)\\Microsoft Visual Studio\\2019\\Community\\VC\\Auxiliary\\Build\\vcvarsall.bat",
                "C:\\Program Files (x86)\\Microsoft Visual Studio\\2019\\Professional\\VC\\Auxiliary\\Build\\vcvarsall.bat"
            ]
            
            for path in common_paths:
                if os.path.exists(path):
                    # 测试编译器版本
                    try:
                        cmd = f'"{path}" x64 && cl'
                        result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=10)
                        if "Microsoft (R) C/C++ Optimizing Compiler" in result.stderr:
                            version_line = [line for line in result.stderr.split('\n') 
                                          if "Microsoft (R) C/C++ Optimizing Compiler" in line]
                            version = version_line[0].split()[-2] if version_line else "Unknown"
                            
                            return {
                                "name": "MSVC",
                                "version": version,
                                "vcvarsall_path": path,
                                "compile_flags": "/EHsc /O2 /std:c++17",
                                "output_flag": "/Fe:",
                                "available": True
                            }
                    except (subprocess.TimeoutExpired, subprocess.CalledProcessError):
                        continue
            
            return None
            
        except Exception as e:
            self.logger.warning(f"MSVC检测失败: {e}")
            return None
    
    def _detect_clang(self) -> Optional[Dict]:
        """检测Clang编译器"""
        try:
            result = subprocess.run(['clang++', '--version'], 
                                  capture_output=True, text=True, timeout=5)
            if result.returncode == 0:
                version_line = result.stdout.split('\n')[0]
                version = version_line.split()[2] if len(version_line.split()) > 2 else "Unknown"
                
                return {
                    "name": "Clang",
                    "version": version,
                    "executable": "clang++",
                    "compile_flags": "-std=c++17 -O2 -Wall",
                    "output_flag": "-o",
                    "available": True
                }
        except (subprocess.TimeoutExpired, subprocess.CalledProcessError, FileNotFoundError):
            pass
        
        return None
    
    def _detect_gcc(self) -> Optional[Dict]:
        """检测GCC编译器"""
        try:
            result = subprocess.run(['g++', '--version'], 
                                  capture_output=True, text=True, timeout=5)
            if result.returncode == 0:
                version_line = result.stdout.split('\n')[0]
                version = version_line.split()[2] if len(version_line.split()) > 2 else "Unknown"
                
                return {
                    "name": "GCC",
                    "version": version,
                    "executable": "g++",
                    "compile_flags": "-std=c++17 -O2 -Wall",
                    "output_flag": "-o",
                    "available": True
                }
        except (subprocess.TimeoutExpired, subprocess.CalledProcessError, FileNotFoundError):
            pass
        
        return None
    
    def get_project_paths(self) -> Dict[str, Path]:
        """获取项目相关路径"""
        paths = {}
        base_paths = self.config.get("paths", {})
        
        for key, relative_path in base_paths.items():
            full_path = self.project_root / relative_path
            full_path.mkdir(parents=True, exist_ok=True)
            paths[key] = full_path
        
        return paths
    
    def validate_environment(self) -> Tuple[bool, List[str]]:
        """验证运行环境"""
        issues = []
        
        # 检查操作系统
        if platform.system() != "Windows":
            issues.append("PaimonCore Boost目前仅支持Windows系统")
        
        # 检查Python版本
        python_version = tuple(map(int, platform.python_version().split('.')))
        if python_version < (3, 8):
            issues.append("需要Python 3.8或更高版本")
        
        # 检查必要的Python模块
        required_modules = ['psutil', 'numpy', 'matplotlib']
        for module in required_modules:
            try:
                __import__(module)
            except ImportError:
                issues.append(f"缺少必要模块: {module}")
        
        # 检查编译器
        compilers = self.detect_compilers()
        if not compilers:
            issues.append("未检测到任何C++编译器 (MSVC/Clang/GCC)")
        
        # 检查CPU核心数
        try:
            import psutil
            if psutil.cpu_count() < self.config.get("system", {}).get("min_cores", 4):
                issues.append("CPU核心数量不足，建议至少4核心")
        except ImportError:
            pass
        
        is_valid = len(issues) == 0
        return is_valid, issues
    
    def generate_system_report(self) -> str:
        """生成系统环境报告"""
        system_info = self.detect_system_info()
        compilers = self.detect_compilers()
        is_valid, issues = self.validate_environment()
        
        report = [
            "🎮 PaimonCore Boost 系统环境报告",
            "=" * 50,
            "",
            "📊 系统信息:",
            f"  操作系统: {system_info.get('platform', 'Unknown')}",
            f"  处理器: {system_info.get('processor', 'Unknown')}",
            f"  CPU核心: {system_info.get('physical_cores', 'Unknown')} 物理核心, {system_info.get('cpu_count', 'Unknown')} 逻辑核心",
            f"  内存: {system_info.get('total_memory', 0) // (1024**3)} GB",
            f"  Python版本: {system_info.get('python_version', 'Unknown')}",
            f"  管理员权限: {'是' if system_info.get('is_admin', False) else '否'}",
            "",
            "🔧 编译器检测:",
        ]
        
        if compilers:
            for name, info in compilers.items():
                report.append(f"  ✅ {info['name']} {info['version']}")
        else:
            report.append("  ❌ 未检测到可用编译器")
        
        report.extend([
            "",
            "🎯 环境验证:",
        ])
        
        if is_valid:
            report.append("  ✅ 环境验证通过，可以正常运行PaimonCore Boost")
        else:
            report.append("  ❌ 环境验证失败，发现以下问题:")
            for issue in issues:
                report.append(f"     • {issue}")
        
        return "\n".join(report)

def main():
    """主函数 - 用于测试环境检测"""
    env = SystemEnvironment()
    print(env.generate_system_report())

if __name__ == "__main__":
    main()