smart_boost.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
PaimonCore Boost - Adaptive Hybrid Core Scheduler
Inspired by Genshin Impact's Idle Boost Phenomenon

作者: Geoffrey Wang
版本: 0.1.0
许可证: Apache License 2.0
仓库: https://github.com/GeoffreyWang1117/PaimonCore-Boost

"Just like how Paimon helps the Traveler navigate Teyvat,
PaimonCore helps your CPU navigate hybrid core scheduling!"
"""

import os
import sys
import time
import json
import csv
import threading
import argparse
import psutil
import numpy as np
from datetime import datetime
from colorama import init, Fore, Style
from tqdm import tqdm

# 初始化colorama用于彩色输出
init(autoreset=True)

class SmartBoostTool:
    def __init__(self):
        self.is_running = False
        self.monitoring_data = []
        self.p_core_activator_thread = None
        self.monitor_thread = None
        
        # 配置参数
        self.config = {
            "p_core_load_intensity": 0.3,  # P核激活负载强度(0-1)
            "monitor_interval": 1.0,       # 监控间隔(秒)
            "record_duration": 300,        # 默认录制时长(秒)
            "cpu_threshold": 80,           # CPU使用率告警阈值
            "temp_threshold": 75           # 温度告警阈值(摄氏度)
        }
        
        print(f"{Fore.CYAN}🚀 SmartBoostTool v0.1.0 已启动")
        print(f"{Fore.GREEN}✅ 检测到CPU: {psutil.cpu_count()} 核心")
        
    def get_cpu_info(self):
        """获取CPU基础信息"""
        cpu_info = {
            "physical_cores": psutil.cpu_count(logical=False),
            "logical_cores": psutil.cpu_count(logical=True),
            "cpu_freq": psutil.cpu_freq(),
            "cpu_percent": psutil.cpu_percent(interval=1),
            "load_avg": os.getloadavg() if hasattr(os, 'getloadavg') else None
        }
        return cpu_info
    
    def activate_p_cores(self):
        """P核心激活器 - 通过计算密集型任务保持P核活跃"""
        print(f"{Fore.YELLOW}🔥 正在激活P核心...")
        
        def cpu_intensive_task():
            """CPU密集型计算任务"""
            while self.is_running:
                # 执行一些数学计算来激活P核
                for _ in range(1000):
                    if not self.is_running:
                        break
                    # 矩阵运算 - 偏向于P核心处理
                    matrix = np.random.rand(50, 50)
                    result = np.linalg.inv(matrix @ matrix.T + np.eye(50))
                    
                # 短暂休息，避免100%占用
                time.sleep(0.1)
        
        # 创建多个线程来激活不同的P核心
        threads = []
        for i in range(min(4, psutil.cpu_count(logical=False))):  # 最多4个线程
            thread = threading.Thread(target=cpu_intensive_task, daemon=True)
            thread.start()
            threads.append(thread)
            
        print(f"{Fore.GREEN}✅ P核心激活器已启动 ({len(threads)} 个激活线程)")
        return threads
    
    def monitor_performance(self):
        """性能监控器"""
        print(f"{Fore.BLUE}📊 开始性能监控...")
        
        while self.is_running:
            try:
                # 获取系统性能数据
                cpu_percent = psutil.cpu_percent(interval=None, percpu=True)
                cpu_freq = psutil.cpu_freq(percpu=True) if psutil.cpu_freq(percpu=True) else [psutil.cpu_freq()]
                memory = psutil.virtual_memory()
                
                # 尝试获取CPU温度（需要特定的传感器支持）
                temps = []
                try:
                    if hasattr(psutil, "sensors_temperatures"):
                        sensors = psutil.sensors_temperatures()
                        if sensors:
                            for name, entries in sensors.items():
                                temps.extend([entry.current for entry in entries if entry.current])
                except:
                    temps = [0]  # 如果无法获取温度，使用占位符
                
                # 记录性能数据
                performance_data = {
                    "timestamp": datetime.now().isoformat(),
                    "cpu_percent_per_core": cpu_percent,
                    "cpu_percent_total": sum(cpu_percent) / len(cpu_percent),
                    "cpu_freq": [freq.current if hasattr(freq, 'current') else 0 for freq in cpu_freq],
                    "memory_percent": memory.percent,
                    "memory_used_gb": memory.used / (1024**3),
                    "temperatures": temps,
                    "avg_temp": sum(temps) / len(temps) if temps else 0
                }
                
                self.monitoring_data.append(performance_data)
                
                # 实时显示关键指标
                avg_cpu = performance_data["cpu_percent_total"]
                avg_temp = performance_data["avg_temp"]
                
                status_color = Fore.GREEN
                if avg_cpu > self.config["cpu_threshold"] or avg_temp > self.config["temp_threshold"]:
                    status_color = Fore.RED
                elif avg_cpu > self.config["cpu_threshold"] * 0.7:
                    status_color = Fore.YELLOW
                
                print(f"\r{status_color}CPU: {avg_cpu:.1f}% | 内存: {performance_data['memory_percent']:.1f}% | 温度: {avg_temp:.1f}°C", end="")
                
            except Exception as e:
                print(f"{Fore.RED}❌ 监控错误: {e}")
            
            time.sleep(self.config["monitor_interval"])
    
    def run_baseline_test(self, test_name="default", duration=60):
        """运行基准测试"""
        print(f"{Fore.CYAN}🧪 开始基准测试: {test_name}")
        print(f"⏱️  测试时长: {duration}秒")
        
        # 清空之前的监控数据
        self.monitoring_data = []
        
        # 开始监控
        self.is_running = True
        self.monitor_thread = threading.Thread(target=self.monitor_performance, daemon=True)
        self.monitor_thread.start()
        
        # 进度条
        with tqdm(total=duration, desc="基准测试进行中", unit="s") as pbar:
            for i in range(duration):
                time.sleep(1)
                pbar.update(1)
        
        # 停止监控
        self.is_running = False
        if self.monitor_thread:
            self.monitor_thread.join(timeout=2)
        
        # 保存测试结果
        self.save_baseline_results(test_name)
        
        print(f"\n{Fore.GREEN}✅ 基准测试完成")
    
    def save_baseline_results(self, test_name):
        """保存基准测试结果"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"baseline_{test_name}_{timestamp}.json"
        filepath = os.path.join("results", filename)
        
        # 确保results目录存在
        os.makedirs("results", exist_ok=True)
        
        # 计算统计数据
        if self.monitoring_data:
            cpu_values = [data["cpu_percent_total"] for data in self.monitoring_data]
            temp_values = [data["avg_temp"] for data in self.monitoring_data if data["avg_temp"] > 0]
            
            stats = {
                "test_info": {
                    "name": test_name,
                    "timestamp": timestamp,
                    "duration": len(self.monitoring_data),
                    "system_info": self.get_cpu_info()
                },
                "statistics": {
                    "cpu_avg": np.mean(cpu_values),
                    "cpu_max": np.max(cpu_values),
                    "cpu_min": np.min(cpu_values),
                    "cpu_std": np.std(cpu_values),
                    "temp_avg": np.mean(temp_values) if temp_values else 0,
                    "temp_max": np.max(temp_values) if temp_values else 0,
                },
                "raw_data": self.monitoring_data
            }
            
            # 保存到JSON文件
            with open(filepath, 'w', encoding='utf-8') as f:
                json.dump(stats, f, indent=2, ensure_ascii=False)
            
            # 同时保存CSV格式
            csv_filepath = filepath.replace('.json', '.csv')
            with open(csv_filepath, 'w', newline='', encoding='utf-8') as f:
                if self.monitoring_data:
                    fieldnames = self.monitoring_data[0].keys()
                    writer = csv.DictWriter(f, fieldnames=fieldnames)
                    writer.writeheader()
                    writer.writerows(self.monitoring_data)
            
            print(f"{Fore.GREEN}📁 结果已保存到: {filepath}")
            print(f"{Fore.GREEN}📊 CSV数据已保存到: {csv_filepath}")
            
            # 显示测试总结
            self.print_test_summary(stats)
    
    def print_test_summary(self, stats):
        """显示测试总结"""
        print(f"\n{Fore.CYAN}📋 测试总结:")
        print(f"   测试名称: {stats['test_info']['name']}")
        print(f"   持续时间: {stats['test_info']['duration']}秒")
        print(f"   平均CPU使用率: {stats['statistics']['cpu_avg']:.2f}%")
        print(f"   最高CPU使用率: {stats['statistics']['cpu_max']:.2f}%")
        print(f"   平均温度: {stats['statistics']['temp_avg']:.2f}°C")
        print(f"   最高温度: {stats['statistics']['temp_max']:.2f}°C")
    
    def start_p_core_activation(self):
        """启动P核心激活模式"""
        print(f"{Fore.GREEN}🎯 启动P核心激活模式")
        
        self.is_running = True
        
        # 启动P核激活线程
        self.p_core_activator_thread = self.activate_p_cores()
        
        # 启动监控线程
        self.monitor_thread = threading.Thread(target=self.monitor_performance, daemon=True)
        self.monitor_thread.start()
        
        try:
            print(f"{Fore.YELLOW}💡 P核心激活器正在运行... (按 Ctrl+C 停止)")
            while self.is_running:
                time.sleep(1)
        except KeyboardInterrupt:
            print(f"\n{Fore.CYAN}🛑 正在停止P核心激活器...")
            self.is_running = False
            
        print(f"{Fore.GREEN}✅ P核心激活器已停止")

def main():
    parser = argparse.ArgumentParser(description="SmartBoostTool - Windows P核激活优化工具")
    parser.add_argument("--mode", choices=["activate", "monitor", "baseline"], 
                       default="activate", help="运行模式")
    parser.add_argument("--test-name", default="default", 
                       help="基准测试名称")
    parser.add_argument("--duration", type=int, default=60, 
                       help="基准测试持续时间(秒)")
    
    args = parser.parse_args()
    
    # 检查管理员权限
    try:
        import ctypes
        is_admin = ctypes.windll.shell32.IsUserAnAdmin()
        if not is_admin:
            print(f"{Fore.YELLOW}⚠️  建议以管理员权限运行以获得最佳效果")
    except:
        pass
    
    tool = SmartBoostTool()
    
    if args.mode == "activate":
        tool.start_p_core_activation()
    elif args.mode == "baseline":
        tool.run_baseline_test(args.test_name, args.duration)
    elif args.mode == "monitor":
        tool.is_running = True
        tool.monitor_thread = threading.Thread(target=tool.monitor_performance, daemon=True)
        tool.monitor_thread.start()
        
        try:
            print(f"{Fore.YELLOW}📊 监控模式运行中... (按 Ctrl+C 停止)")
            while tool.is_running:
                time.sleep(1)
        except KeyboardInterrupt:
            tool.is_running = False
        
        print(f"\n{Fore.GREEN}✅ 监控已停止")

if __name__ == "__main__":
    main()