""" 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}")