Tolan/backend/performance.py

"""
性能监控模块
"""
import time
from typing import List, Dict
from statistics import mean, median
from loguru import logger


class PerformanceMonitor:
    """性能监控器"""
    
    def __init__(self):
        self.latencies: List[float] = []
        self.audio_sent_total = 0
        self.audio_received_total = 0
        self.session_count = 0
        self.error_count = 0
        self.start_time = time.time()
    
    def record_latency(self, latency_ms: float):
        """记录延迟"""
        self.latencies.append(latency_ms)
        
        # 只保留最近1000次
        if len(self.latencies) > 1000:
            self.latencies.pop(0)
        
        # 警告高延迟
        if latency_ms > 500:
            logger.warning(f"高延迟检测: {latency_ms:.2f}ms")
    
    def record_audio_sent(self, bytes_count: int):
        """记录发送的音频"""
        self.audio_sent_total += bytes_count
    
    def record_audio_received(self, bytes_count: int):
        """记录接收的音频"""
        self.audio_received_total += bytes_count
    
    def record_session(self):
        """记录会话"""
        self.session_count += 1
    
    def record_error(self):
        """记录错误"""
        self.error_count += 1
    
    def get_stats(self) -> Dict:
        """获取统计信息"""
        stats = {
            'uptime_seconds': int(time.time() - self.start_time),
            'total_sessions': self.session_count,
            'total_errors': self.error_count,
            'audio_sent_mb': round(self.audio_sent_total / 1024 / 1024, 2),
            'audio_received_mb': round(self.audio_received_total / 1024 / 1024, 2),
        }
        
        if self.latencies:
            stats.update({
                'avg_latency_ms': round(mean(self.latencies), 2),
                'median_latency_ms': round(median(self.latencies), 2),
                'min_latency_ms': round(min(self.latencies), 2),
                'max_latency_ms': round(max(self.latencies), 2),
                'p95_latency_ms': round(self._percentile(self.latencies, 0.95), 2),
                'p99_latency_ms': round(self._percentile(self.latencies, 0.99), 2),
            })
        
        return stats
    
    def _percentile(self, data: List[float], percentile: float) -> float:
        """计算百分位数"""
        if not data:
            return 0
        sorted_data = sorted(data)
        index = int(len(sorted_data) * percentile)
        return sorted_data[min(index, len(sorted_data) - 1)]
    
    def reset(self):
        """重置统计"""
        self.latencies.clear()
        self.audio_sent_total = 0
        self.audio_received_total = 0
        self.session_count = 0
        self.error_count = 0
        self.start_time = time.time()
        logger.info("性能监控已重置")


# 全局性能监控实例
performance_monitor = PerformanceMonitor()