Trixy3/trixy_core/ml/performance_monitor.py

"""
Performance monitoring and optimization for Trixy ML Integration.

This module provides comprehensive performance monitoring, metrics collection,
resource optimization, and performance analytics for all ML components.
"""

import asyncio
import logging
import time
import threading
from collections import defaultdict, deque
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple, Callable, Union
import json
import psutil
import numpy as np
from concurrent.futures import ThreadPoolExecutor


class MetricType(Enum):
    """Types of performance metrics."""
    LATENCY = "latency"
    THROUGHPUT = "throughput"
    CPU_USAGE = "cpu_usage"
    MEMORY_USAGE = "memory_usage"
    AUDIO_QUALITY = "audio_quality"
    MODEL_ACCURACY = "model_accuracy"
    ERROR_RATE = "error_rate"
    BUFFER_UTILIZATION = "buffer_utilization"
    NETWORK_LATENCY = "network_latency"
    PROCESSING_TIME = "processing_time"


class AlertLevel(Enum):
    """Alert severity levels."""
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"
    EMERGENCY = "emergency"


@dataclass
class PerformanceMetric:
    """Individual performance metric data point."""
    timestamp: datetime
    component: str
    metric_type: MetricType
    value: Union[float, int]
    unit: str
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class PerformanceThreshold:
    """Performance threshold configuration."""
    metric_type: MetricType
    warning_threshold: float
    critical_threshold: float
    unit: str
    check_interval: float = 1.0  # seconds
    min_samples: int = 5


@dataclass
class OptimizationRule:
    """Optimization rule for automatic performance tuning."""
    name: str
    condition: Callable[[Dict[str, Any]], bool]
    action: Callable[[Dict[str, Any]], None]
    cooldown: float = 60.0  # seconds
    description: str = ""


@dataclass
class PerformanceReport:
    """Comprehensive performance report."""
    timestamp: datetime
    duration: timedelta
    component_stats: Dict[str, Dict[str, Any]]
    system_stats: Dict[str, Any]
    alerts: List[Dict[str, Any]]
    recommendations: List[str]


class MetricsCollector:
    """Collects and aggregates performance metrics."""
    
    def __init__(self, max_history: int = 10000):
        self.max_history = max_history
        self.metrics: Dict[str, deque] = defaultdict(lambda: deque(maxlen=max_history))
        self._lock = threading.RLock()
        self.logger = logging.getLogger(__name__)
    
    def record_metric(self, metric: PerformanceMetric) -> None:
        """Record a performance metric."""
        with self._lock:
            key = f"{metric.component}:{metric.metric_type.value}"
            self.metrics[key].append(metric)
    
    def get_metrics(self, component: str, metric_type: MetricType, 
                   since: Optional[datetime] = None, 
                   limit: Optional[int] = None) -> List[PerformanceMetric]:
        """Get metrics for a component and type."""
        with self._lock:
            key = f"{component}:{metric_type.value}"
            metrics = list(self.metrics[key])
            
            if since:
                metrics = [m for m in metrics if m.timestamp >= since]
            
            if limit:
                metrics = metrics[-limit:]
            
            return metrics
    
    def get_aggregated_stats(self, component: str, metric_type: MetricType,
                           window: timedelta) -> Dict[str, float]:
        """Get aggregated statistics for a time window."""
        since = datetime.now() - window
        metrics = self.get_metrics(component, metric_type, since=since)
        
        if not metrics:
            return {}
        
        values = [m.value for m in metrics]
        return {
            'count': len(values),
            'mean': np.mean(values),
            'median': np.median(values),
            'std': np.std(values),
            'min': np.min(values),
            'max': np.max(values),
            'p95': np.percentile(values, 95),
            'p99': np.percentile(values, 99)
        }
    
    def clear_old_metrics(self, older_than: timedelta) -> int:
        """Clear metrics older than specified time."""
        cutoff = datetime.now() - older_than
        removed_count = 0
        
        with self._lock:
            for key in self.metrics:
                original_length = len(self.metrics[key])
                self.metrics[key] = deque(
                    (m for m in self.metrics[key] if m.timestamp >= cutoff),
                    maxlen=self.max_history
                )
                removed_count += original_length - len(self.metrics[key])
        
        return removed_count


class SystemMonitor:
    """Monitors system-level performance metrics."""
    
    def __init__(self, process_name: str = "trixy"):
        self.process_name = process_name
        self.process = None
        self._find_process()
        self.logger = logging.getLogger(__name__)
    
    def _find_process(self) -> None:
        """Find the Trixy process."""
        try:
            for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
                if self.process_name.lower() in proc.info['name'].lower():
                    self.process = psutil.Process(proc.info['pid'])
                    break
        except Exception as e:
            self.logger.warning(f"Could not find process {self.process_name}: {e}")
    
    def get_cpu_usage(self) -> float:
        """Get current CPU usage percentage."""
        try:
            if self.process and self.process.is_running():
                return self.process.cpu_percent()
            return psutil.cpu_percent()
        except Exception:
            return 0.0
    
    def get_memory_usage(self) -> Dict[str, float]:
        """Get memory usage statistics."""
        try:
            if self.process and self.process.is_running():
                memory_info = self.process.memory_info()
                return {
                    'rss': memory_info.rss / 1024 / 1024,  # MB
                    'vms': memory_info.vms / 1024 / 1024,  # MB
                    'percent': self.process.memory_percent()
                }
            
            memory = psutil.virtual_memory()
            return {
                'used': memory.used / 1024 / 1024 / 1024,  # GB
                'available': memory.available / 1024 / 1024 / 1024,  # GB
                'percent': memory.percent
            }
        except Exception:
            return {'rss': 0.0, 'vms': 0.0, 'percent': 0.0}
    
    def get_disk_usage(self) -> Dict[str, float]:
        """Get disk usage statistics."""
        try:
            disk = psutil.disk_usage('/')
            return {
                'used': disk.used / 1024 / 1024 / 1024,  # GB
                'free': disk.free / 1024 / 1024 / 1024,  # GB
                'percent': (disk.used / disk.total) * 100
            }
        except Exception:
            return {'used': 0.0, 'free': 0.0, 'percent': 0.0}
    
    def get_network_stats(self) -> Dict[str, float]:
        """Get network I/O statistics."""
        try:
            net_io = psutil.net_io_counters()
            return {
                'bytes_sent': net_io.bytes_sent / 1024 / 1024,  # MB
                'bytes_recv': net_io.bytes_recv / 1024 / 1024,  # MB
                'packets_sent': net_io.packets_sent,
                'packets_recv': net_io.packets_recv
            }
        except Exception:
            return {'bytes_sent': 0.0, 'bytes_recv': 0.0, 'packets_sent': 0, 'packets_recv': 0}


class PerformanceOptimizer:
    """Automatic performance optimization engine."""
    
    def __init__(self, metrics_collector: MetricsCollector):
        self.metrics_collector = metrics_collector
        self.optimization_rules: List[OptimizationRule] = []
        self.last_optimization: Dict[str, datetime] = {}
        self.logger = logging.getLogger(__name__)
        self._setup_default_rules()
    
    def _setup_default_rules(self) -> None:
        """Setup default optimization rules."""
        
        # High CPU usage optimization
        def high_cpu_condition(context: Dict[str, Any]) -> bool:
            cpu_usage = context.get('cpu_usage', 0)
            return cpu_usage > 80.0
        
        def reduce_cpu_action(context: Dict[str, Any]) -> None:
            self.logger.info("High CPU detected, reducing audio processing frequency")
            # Implementation would adjust audio processing parameters
        
        self.optimization_rules.append(OptimizationRule(
            name="high_cpu_optimization",
            condition=high_cpu_condition,
            action=reduce_cpu_action,
            cooldown=30.0,
            description="Reduce processing load when CPU usage is high"
        ))
        
        # High memory usage optimization
        def high_memory_condition(context: Dict[str, Any]) -> bool:
            memory_percent = context.get('memory_percent', 0)
            return memory_percent > 85.0
        
        def reduce_memory_action(context: Dict[str, Any]) -> None:
            self.logger.info("High memory detected, clearing buffers and caches")
            # Implementation would clear audio buffers and model caches
        
        self.optimization_rules.append(OptimizationRule(
            name="high_memory_optimization",
            condition=high_memory_condition,
            action=reduce_memory_action,
            cooldown=60.0,
            description="Clear caches when memory usage is high"
        ))
        
        # High latency optimization
        def high_latency_condition(context: Dict[str, Any]) -> bool:
            avg_latency = context.get('avg_latency', 0)
            return avg_latency > 500.0  # 500ms
        
        def reduce_latency_action(context: Dict[str, Any]) -> None:
            self.logger.info("High latency detected, switching to faster models")
            # Implementation would switch to faster but less accurate models
        
        self.optimization_rules.append(OptimizationRule(
            name="high_latency_optimization",
            condition=high_latency_condition,
            action=reduce_latency_action,
            cooldown=120.0,
            description="Use faster models when latency is high"
        ))
    
    def add_rule(self, rule: OptimizationRule) -> None:
        """Add a custom optimization rule."""
        self.optimization_rules.append(rule)
    
    def run_optimization(self, context: Dict[str, Any]) -> List[str]:
        """Run optimization rules and return applied optimizations."""
        applied_optimizations = []
        current_time = datetime.now()
        
        for rule in self.optimization_rules:
            # Check if rule is in cooldown
            last_run = self.last_optimization.get(rule.name)
            if last_run and (current_time - last_run).total_seconds() < rule.cooldown:
                continue
            
            # Check if rule condition is met
            try:
                if rule.condition(context):
                    rule.action(context)
                    self.last_optimization[rule.name] = current_time
                    applied_optimizations.append(rule.name)
                    self.logger.info(f"Applied optimization: {rule.description}")
            except Exception as e:
                self.logger.error(f"Error applying optimization rule {rule.name}: {e}")
        
        return applied_optimizations


class AlertManager:
    """Manages performance alerts and notifications."""
    
    def __init__(self, metrics_collector: MetricsCollector):
        self.metrics_collector = metrics_collector
        self.thresholds: Dict[str, PerformanceThreshold] = {}
        self.active_alerts: Dict[str, Dict[str, Any]] = {}
        self.alert_history: List[Dict[str, Any]] = []
        self.logger = logging.getLogger(__name__)
        self._setup_default_thresholds()
    
    def _setup_default_thresholds(self) -> None:
        """Setup default performance thresholds."""
        
        self.thresholds['cpu_usage'] = PerformanceThreshold(
            metric_type=MetricType.CPU_USAGE,
            warning_threshold=70.0,
            critical_threshold=90.0,
            unit="percent"
        )
        
        self.thresholds['memory_usage'] = PerformanceThreshold(
            metric_type=MetricType.MEMORY_USAGE,
            warning_threshold=80.0,
            critical_threshold=95.0,
            unit="percent"
        )
        
        self.thresholds['latency'] = PerformanceThreshold(
            metric_type=MetricType.LATENCY,
            warning_threshold=300.0,
            critical_threshold=1000.0,
            unit="milliseconds"
        )
        
        self.thresholds['error_rate'] = PerformanceThreshold(
            metric_type=MetricType.ERROR_RATE,
            warning_threshold=5.0,
            critical_threshold=15.0,
            unit="percent"
        )
    
    def add_threshold(self, name: str, threshold: PerformanceThreshold) -> None:
        """Add a custom performance threshold."""
        self.thresholds[name] = threshold
    
    def check_thresholds(self, component: str) -> List[Dict[str, Any]]:
        """Check all thresholds for a component."""
        alerts = []
        window = timedelta(minutes=5)
        
        for name, threshold in self.thresholds.items():
            stats = self.metrics_collector.get_aggregated_stats(
                component, threshold.metric_type, window
            )
            
            if not stats or stats['count'] < threshold.min_samples:
                continue
            
            current_value = stats['mean']
            alert_level = None
            
            if current_value >= threshold.critical_threshold:
                alert_level = AlertLevel.CRITICAL
            elif current_value >= threshold.warning_threshold:
                alert_level = AlertLevel.WARNING
            
            if alert_level:
                alert_key = f"{component}:{name}"
                alert = {
                    'key': alert_key,
                    'component': component,
                    'metric_type': threshold.metric_type.value,
                    'level': alert_level.value,
                    'current_value': current_value,
                    'threshold': threshold.critical_threshold if alert_level == AlertLevel.CRITICAL else threshold.warning_threshold,
                    'unit': threshold.unit,
                    'timestamp': datetime.now(),
                    'message': f"{component} {threshold.metric_type.value} is {current_value:.2f} {threshold.unit}"
                }
                
                alerts.append(alert)
                self.active_alerts[alert_key] = alert
                self.alert_history.append(alert)
                
                self.logger.warning(f"Performance alert: {alert['message']}")
        
        return alerts
    
    def clear_alert(self, alert_key: str) -> bool:
        """Clear an active alert."""
        if alert_key in self.active_alerts:
            del self.active_alerts[alert_key]
            return True
        return False
    
    def get_active_alerts(self, component: Optional[str] = None) -> List[Dict[str, Any]]:
        """Get currently active alerts."""
        alerts = list(self.active_alerts.values())
        if component:
            alerts = [a for a in alerts if a['component'] == component]
        return alerts


class PerformanceMonitor:
    """Main performance monitoring system for Trixy ML Integration."""
    
    def __init__(self, config: Optional[Dict[str, Any]] = None):
        self.config = config or {}
        self.logger = logging.getLogger(__name__)
        
        # Initialize components
        self.metrics_collector = MetricsCollector(
            max_history=self.config.get('max_metrics_history', 10000)
        )
        self.system_monitor = SystemMonitor()
        self.optimizer = PerformanceOptimizer(self.metrics_collector)
        self.alert_manager = AlertManager(self.metrics_collector)
        
        # Monitoring state
        self.monitoring_active = False
        self.monitoring_interval = self.config.get('monitoring_interval', 1.0)
        self.monitoring_task: Optional[asyncio.Task] = None
        self.executor = ThreadPoolExecutor(max_workers=2)
        
        # Performance context
        self.start_time = datetime.now()
        self.component_timers: Dict[str, float] = {}
        
        self.logger.info("Performance monitor initialized")
    
    async def start_monitoring(self) -> None:
        """Start continuous performance monitoring."""
        if self.monitoring_active:
            return
        
        self.monitoring_active = True
        self.monitoring_task = asyncio.create_task(self._monitoring_loop())
        self.logger.info("Performance monitoring started")
    
    async def stop_monitoring(self) -> None:
        """Stop performance monitoring."""
        self.monitoring_active = False
        if self.monitoring_task:
            self.monitoring_task.cancel()
            try:
                await self.monitoring_task
            except asyncio.CancelledError:
                pass
        
        self.executor.shutdown(wait=True)
        self.logger.info("Performance monitoring stopped")
    
    async def _monitoring_loop(self) -> None:
        """Main monitoring loop."""
        while self.monitoring_active:
            try:
                await self._collect_system_metrics()
                await self._run_optimization()
                await self._check_alerts()
                await asyncio.sleep(self.monitoring_interval)
            except asyncio.CancelledError:
                break
            except Exception as e:
                self.logger.error(f"Error in monitoring loop: {e}")
                await asyncio.sleep(self.monitoring_interval)
    
    async def _collect_system_metrics(self) -> None:
        """Collect system-level metrics."""
        try:
            timestamp = datetime.now()
            
            # CPU usage
            cpu_usage = await asyncio.get_event_loop().run_in_executor(
                self.executor, self.system_monitor.get_cpu_usage
            )
            self.record_metric("system", MetricType.CPU_USAGE, cpu_usage, "percent")
            
            # Memory usage
            memory_stats = await asyncio.get_event_loop().run_in_executor(
                self.executor, self.system_monitor.get_memory_usage
            )
            self.record_metric("system", MetricType.MEMORY_USAGE, 
                             memory_stats.get('percent', 0), "percent")
            
        except Exception as e:
            self.logger.error(f"Error collecting system metrics: {e}")
    
    async def _run_optimization(self) -> None:
        """Run performance optimization."""
        try:
            # Build optimization context
            context = {
                'cpu_usage': await asyncio.get_event_loop().run_in_executor(
                    self.executor, self.system_monitor.get_cpu_usage
                ),
                'memory_percent': self.system_monitor.get_memory_usage().get('percent', 0),
            }
            
            # Add latency metrics
            latency_stats = self.metrics_collector.get_aggregated_stats(
                "ml_manager", MetricType.LATENCY, timedelta(minutes=5)
            )
            context['avg_latency'] = latency_stats.get('mean', 0) if latency_stats else 0
            
            applied_optimizations = await asyncio.get_event_loop().run_in_executor(
                self.executor, self.optimizer.run_optimization, context
            )
            
        except Exception as e:
            self.logger.error(f"Error running optimization: {e}")
    
    async def _check_alerts(self) -> None:
        """Check performance alerts."""
        try:
            components = ['system', 'ml_manager', 'audio_pipeline', 'wakeword', 'voice_recognition']
            for component in components:
                await asyncio.get_event_loop().run_in_executor(
                    self.executor, self.alert_manager.check_thresholds, component
                )
        except Exception as e:
            self.logger.error(f"Error checking alerts: {e}")
    
    def record_metric(self, component: str, metric_type: MetricType, 
                     value: Union[float, int], unit: str,
                     metadata: Optional[Dict[str, Any]] = None) -> None:
        """Record a performance metric."""
        metric = PerformanceMetric(
            timestamp=datetime.now(),
            component=component,
            metric_type=metric_type,
            value=value,
            unit=unit,
            metadata=metadata or {}
        )
        self.metrics_collector.record_metric(metric)
    
    def start_timer(self, component: str, operation: str) -> str:
        """Start a timer for an operation."""
        timer_key = f"{component}:{operation}"
        self.component_timers[timer_key] = time.time()
        return timer_key
    
    def end_timer(self, timer_key: str) -> float:
        """End a timer and record the duration."""
        if timer_key not in self.component_timers:
            return 0.0
        
        duration = (time.time() - self.component_timers[timer_key]) * 1000  # milliseconds
        del self.component_timers[timer_key]
        
        # Extract component from timer key
        component = timer_key.split(':')[0]
        self.record_metric(component, MetricType.PROCESSING_TIME, duration, "milliseconds")
        
        return duration
    
    def get_performance_report(self, window: timedelta = timedelta(hours=1)) -> PerformanceReport:
        """Generate a comprehensive performance report."""
        timestamp = datetime.now()
        components = ['system', 'ml_manager', 'audio_pipeline', 'wakeword', 'voice_recognition']
        
        component_stats = {}
        for component in components:
            component_stats[component] = {}
            for metric_type in MetricType:
                stats = self.metrics_collector.get_aggregated_stats(
                    component, metric_type, window
                )
                if stats:
                    component_stats[component][metric_type.value] = stats
        
        # System stats
        system_stats = {
            'uptime': (timestamp - self.start_time).total_seconds(),
            'cpu_usage': self.system_monitor.get_cpu_usage(),
            'memory_usage': self.system_monitor.get_memory_usage(),
            'disk_usage': self.system_monitor.get_disk_usage(),
            'network_stats': self.system_monitor.get_network_stats()
        }
        
        # Alerts
        alerts = self.alert_manager.get_active_alerts()
        
        # Recommendations
        recommendations = self._generate_recommendations(component_stats, system_stats)
        
        return PerformanceReport(
            timestamp=timestamp,
            duration=window,
            component_stats=component_stats,
            system_stats=system_stats,
            alerts=alerts,
            recommendations=recommendations
        )
    
    def _generate_recommendations(self, component_stats: Dict[str, Any], 
                                system_stats: Dict[str, Any]) -> List[str]:
        """Generate performance optimization recommendations."""
        recommendations = []
        
        # CPU recommendations
        cpu_usage = system_stats.get('cpu_usage', 0)
        if cpu_usage > 80:
            recommendations.append("Consider reducing audio processing frequency or using lower quality models")
        
        # Memory recommendations
        memory_percent = system_stats.get('memory_usage', {}).get('percent', 0)
        if memory_percent > 85:
            recommendations.append("Consider clearing audio buffers more frequently or reducing buffer sizes")
        
        # Latency recommendations
        ml_stats = component_stats.get('ml_manager', {})
        latency_stats = ml_stats.get('latency', {})
        if latency_stats and latency_stats.get('mean', 0) > 500:
            recommendations.append("Consider switching to faster models or optimizing audio preprocessing")
        
        # Error rate recommendations
        error_stats = ml_stats.get('error_rate', {})
        if error_stats and error_stats.get('mean', 0) > 10:
            recommendations.append("High error rate detected - check model configurations and input quality")
        
        return recommendations
    
    def export_metrics(self, file_path: str, window: timedelta = timedelta(hours=24)) -> None:
        """Export metrics to a JSON file."""
        try:
            report = self.get_performance_report(window)
            
            # Convert report to JSON-serializable format
            export_data = {
                'timestamp': report.timestamp.isoformat(),
                'duration_hours': report.duration.total_seconds() / 3600,
                'component_stats': report.component_stats,
                'system_stats': report.system_stats,
                'alerts': [
                    {**alert, 'timestamp': alert['timestamp'].isoformat()}
                    for alert in report.alerts
                ],
                'recommendations': report.recommendations
            }
            
            with open(file_path, 'w') as f:
                json.dump(export_data, f, indent=2, default=str)
            
            self.logger.info(f"Metrics exported to {file_path}")
            
        except Exception as e:
            self.logger.error(f"Error exporting metrics: {e}")