# Description: Analytics and monitoring system for EU-Utility # Privacy-focused usage tracking and performance monitoring """ EU-Utility Analytics System Privacy-focused analytics with opt-in tracking: - Feature usage statistics - Performance monitoring (FPS, memory, CPU) - Error reporting - Health checks All data is stored locally by default. """ from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QCheckBox, QProgressBar, QTableWidget, QTableWidgetItem, QTabWidget, QGroupBox ) from PyQt6.QtCore import QTimer, pyqtSignal, Qt from plugins.base_plugin import BasePlugin from datetime import datetime, timedelta import json import os import psutil import time class AnalyticsPlugin(BasePlugin): """ Analytics and monitoring dashboard for EU-Utility. Tracks: - Feature usage (opt-in) - System performance - Error occurrences - Health status """ name = "Analytics" version = "1.0.0" author = "LemonNexus" description = "Usage analytics and performance monitoring" icon = "bar-chart" def initialize(self): """Initialize analytics system.""" # Load settings self.enabled = self.load_data("enabled", False) self.track_performance = self.load_data("track_performance", True) self.track_usage = self.load_data("track_usage", False) # Data storage self.usage_data = self.load_data("usage", {}) self.performance_data = self.load_data("performance", []) self.error_data = self.load_data("errors", []) # Performance tracking self.start_time = time.time() self.session_events = [] # Setup timers if self.track_performance: self._setup_performance_monitoring() def _setup_performance_monitoring(self): """Setup periodic performance monitoring.""" self.performance_timer = QTimer() self.performance_timer.timeout.connect(self._record_performance) self.performance_timer.start(30000) # Every 30 seconds # Initial record self._record_performance() def _record_performance(self): """Record current system performance.""" try: # Get system stats cpu_percent = psutil.cpu_percent(interval=1) memory = psutil.virtual_memory() # Get process info process = psutil.Process() process_memory = process.memory_info().rss / 1024 / 1024 # MB record = { 'timestamp': datetime.now().isoformat(), 'cpu_percent': cpu_percent, 'memory_percent': memory.percent, 'memory_used_mb': process_memory, 'uptime_seconds': time.time() - self.start_time } self.performance_data.append(record) # Keep only last 1000 records if len(self.performance_data) > 1000: self.performance_data = self.performance_data[-1000:] self.save_data("performance", self.performance_data) except Exception as e: self.log_error(f"Performance recording failed: {e}") def record_event(self, event_type, details=None): """Record a usage event (if tracking enabled).""" if not self.track_usage: return event = { 'type': event_type, 'timestamp': datetime.now().isoformat(), 'details': details or {} } self.session_events.append(event) # Update usage stats if event_type not in self.usage_data: self.usage_data[event_type] = {'count': 0, 'last_used': None} self.usage_data[event_type]['count'] += 1 self.usage_data[event_type]['last_used'] = datetime.now().isoformat() self.save_data("usage", self.usage_data) def record_error(self, error_message, context=None): """Record an error occurrence.""" error = { 'message': str(error_message), 'timestamp': datetime.now().isoformat(), 'context': context or {}, 'session_uptime': time.time() - self.start_time } self.error_data.append(error) # Keep only last 100 errors if len(self.error_data) > 100: self.error_data = self.error_data[-100:] self.save_data("errors", self.error_data) def get_ui(self): """Create analytics dashboard UI.""" widget = QWidget() layout = QVBoxLayout(widget) layout.setSpacing(15) # Title title = QLabel("Analytics & Monitoring") title.setStyleSheet("font-size: 20px; font-weight: bold; color: #4a9eff;") layout.addWidget(title) # Tabs tabs = QTabWidget() # Overview tab tabs.addTab(self._create_overview_tab(), "Overview") # Performance tab tabs.addTab(self._create_performance_tab(), "Performance") # Usage tab tabs.addTab(self._create_usage_tab(), "Usage") # Errors tab tabs.addTab(self._create_errors_tab(), "Errors") # Settings tab tabs.addTab(self._create_settings_tab(), "Settings") layout.addWidget(tabs) # Refresh button refresh_btn = QPushButton("Refresh Data") refresh_btn.clicked.connect(self._refresh_all) layout.addWidget(refresh_btn) return widget def _create_overview_tab(self): """Create overview dashboard.""" tab = QWidget() layout = QVBoxLayout(tab) # System health health_group = QGroupBox("System Health") health_layout = QVBoxLayout(health_group) self.health_status = QLabel("Checking...") self.health_status.setStyleSheet("font-size: 16px; font-weight: bold;") health_layout.addWidget(self.health_status) # Health metrics self.cpu_label = QLabel("CPU: --") self.memory_label = QLabel("Memory: --") self.uptime_label = QLabel("Uptime: --") health_layout.addWidget(self.cpu_label) health_layout.addWidget(self.memory_label) health_layout.addWidget(self.uptime_label) layout.addWidget(health_group) # Session stats stats_group = QGroupBox("Session Statistics") stats_layout = QVBoxLayout(stats_group) self.events_label = QLabel(f"Events recorded: {len(self.session_events)}") self.errors_label = QLabel(f"Errors: {len(self.error_data)}") self.plugins_label = QLabel(f"Active plugins: --") stats_layout.addWidget(self.events_label) stats_layout.addWidget(self.errors_label) stats_layout.addWidget(self.plugins_label) layout.addWidget(stats_group) layout.addStretch() # Update immediately self._update_overview() return tab def _create_performance_tab(self): """Create performance monitoring tab.""" tab = QWidget() layout = QVBoxLayout(tab) # Current stats current_group = QGroupBox("Current Performance") current_layout = QVBoxLayout(current_group) self.perf_cpu = QLabel("CPU Usage: --") self.perf_memory = QLabel("Memory Usage: --") self.perf_process = QLabel("Process Memory: --") current_layout.addWidget(self.perf_cpu) current_layout.addWidget(self.perf_memory) current_layout.addWidget(self.perf_process) layout.addWidget(current_group) # Historical data history_group = QGroupBox("Performance History (Last Hour)") history_layout = QVBoxLayout(history_group) self.perf_table = QTableWidget() self.perf_table.setColumnCount(4) self.perf_table.setHorizontalHeaderLabels(["Time", "CPU %", "Memory %", "Process MB"]) self.perf_table.horizontalHeader().setStretchLastSection(True) history_layout.addWidget(self.perf_table) layout.addWidget(history_group) layout.addStretch() self._update_performance_tab() return tab def _create_usage_tab(self): """Create usage statistics tab.""" tab = QWidget() layout = QVBoxLayout(tab) # Usage table self.usage_table = QTableWidget() self.usage_table.setColumnCount(3) self.usage_table.setHorizontalHeaderLabels(["Feature", "Usage Count", "Last Used"]) self.usage_table.horizontalHeader().setStretchLastSection(True) layout.addWidget(self.usage_table) # Update data self._update_usage_tab() return tab def _create_errors_tab(self): """Create error log tab.""" tab = QWidget() layout = QVBoxLayout(tab) # Error table self.error_table = QTableWidget() self.error_table.setColumnCount(3) self.error_table.setHorizontalHeaderLabels(["Time", "Error", "Context"]) self.error_table.horizontalHeader().setStretchLastSection(True) layout.addWidget(self.error_table) # Clear button clear_btn = QPushButton("Clear Error Log") clear_btn.clicked.connect(self._clear_errors) layout.addWidget(clear_btn) self._update_errors_tab() return tab def _create_settings_tab(self): """Create analytics settings tab.""" tab = QWidget() layout = QVBoxLayout(tab) # Privacy notice privacy = QLabel( "🔒 Privacy Notice:\n" "All analytics data is stored locally on your machine. " "No data is sent to external servers unless you explicitly configure it." ) privacy.setStyleSheet("background-color: #2a3a40; padding: 10px; border-radius: 4px;") privacy.setWordWrap(True) layout.addWidget(privacy) # Enable analytics self.enable_checkbox = QCheckBox("Enable Analytics") self.enable_checkbox.setChecked(self.enabled) self.enable_checkbox.toggled.connect(self._on_enable_changed) layout.addWidget(self.enable_checkbox) # Performance tracking self.perf_checkbox = QCheckBox("Track System Performance") self.perf_checkbox.setChecked(self.track_performance) self.perf_checkbox.toggled.connect(self._on_perf_changed) layout.addWidget(self.perf_checkbox) # Usage tracking self.usage_checkbox = QCheckBox("Track Feature Usage (Opt-in)") self.usage_checkbox.setChecked(self.track_usage) self.usage_checkbox.toggled.connect(self._on_usage_changed) layout.addWidget(self.usage_checkbox) # Data management layout.addWidget(QLabel("Data Management:")) export_btn = QPushButton("Export Analytics Data") export_btn.clicked.connect(self._export_data) layout.addWidget(export_btn) clear_all_btn = QPushButton("Clear All Analytics Data") clear_all_btn.setStyleSheet("color: #f44336;") clear_all_btn.clicked.connect(self._clear_all_data) layout.addWidget(clear_all_btn) layout.addStretch() return tab def _update_overview(self): """Update overview tab.""" try: # Get current stats cpu = psutil.cpu_percent(interval=0.5) memory = psutil.virtual_memory() # Health status if cpu < 50 and memory.percent < 80: status = "✓ Healthy" color = "#4caf50" elif cpu < 80 and memory.percent < 90: status = "⚠ Warning" color = "#ff9800" else: status = "✗ Critical" color = "#f44336" self.health_status.setText(status) self.health_status.setStyleSheet(f"font-size: 16px; font-weight: bold; color: {color};") self.cpu_label.setText(f"CPU: {cpu:.1f}%") self.memory_label.setText(f"Memory: {memory.percent:.1f}%") # Uptime uptime = time.time() - self.start_time hours = int(uptime // 3600) minutes = int((uptime % 3600) // 60) self.uptime_label.setText(f"Uptime: {hours}h {minutes}m") # Stats self.events_label.setText(f"Events recorded: {len(self.session_events)}") self.errors_label.setText(f"Total errors: {len(self.error_data)}") except Exception as e: self.log_error(f"Overview update failed: {e}") def _update_performance_tab(self): """Update performance tab.""" try: # Current stats cpu = psutil.cpu_percent(interval=0.5) memory = psutil.virtual_memory() process = psutil.Process() process_mem = process.memory_info().rss / 1024 / 1024 self.perf_cpu.setText(f"CPU Usage: {cpu:.1f}%") self.perf_memory.setText(f"Memory Usage: {memory.percent:.1f}%") self.perf_process.setText(f"Process Memory: {process_mem:.1f} MB") # Historical data (last 20 records) recent_data = self.performance_data[-20:] self.perf_table.setRowCount(len(recent_data)) for i, record in enumerate(reversed(recent_data)): time_str = record['timestamp'][11:19] # HH:MM:SS self.perf_table.setItem(i, 0, QTableWidgetItem(time_str)) self.perf_table.setItem(i, 1, QTableWidgetItem(f"{record['cpu_percent']:.1f}%")) self.perf_table.setItem(i, 2, QTableWidgetItem(f"{record['memory_percent']:.1f}%")) self.perf_table.setItem(i, 3, QTableWidgetItem(f"{record['memory_used_mb']:.1f}")) except Exception as e: self.log_error(f"Performance tab update failed: {e}") def _update_usage_tab(self): """Update usage tab.""" self.usage_table.setRowCount(len(self.usage_data)) for i, (feature, data) in enumerate(sorted(self.usage_data.items())): self.usage_table.setItem(i, 0, QTableWidgetItem(feature)) self.usage_table.setItem(i, 1, QTableWidgetItem(str(data['count']))) last_used = data.get('last_used', 'Never') if last_used and last_used != 'Never': last_used = last_used[:16].replace('T', ' ') # Format datetime self.usage_table.setItem(i, 2, QTableWidgetItem(last_used)) def _update_errors_tab(self): """Update errors tab.""" self.error_table.setRowCount(len(self.error_data)) for i, error in enumerate(reversed(self.error_data[-50:])): # Last 50 errors time_str = error['timestamp'][11:19] self.error_table.setItem(i, 0, QTableWidgetItem(time_str)) self.error_table.setItem(i, 1, QTableWidgetItem(error['message'][:50])) self.error_table.setItem(i, 2, QTableWidgetItem(str(error.get('context', ''))[:50])) def _refresh_all(self): """Refresh all tabs.""" self._update_overview() self._update_performance_tab() self._update_usage_tab() self._update_errors_tab() def _on_enable_changed(self, checked): """Handle analytics enable toggle.""" self.enabled = checked self.save_data("enabled", checked) if checked and self.track_performance: self._setup_performance_monitoring() elif not checked and hasattr(self, 'performance_timer'): self.performance_timer.stop() def _on_perf_changed(self, checked): """Handle performance tracking toggle.""" self.track_performance = checked self.save_data("track_performance", checked) if checked and self.enabled: self._setup_performance_monitoring() elif hasattr(self, 'performance_timer'): self.performance_timer.stop() def _on_usage_changed(self, checked): """Handle usage tracking toggle.""" self.track_usage = checked self.save_data("track_usage", checked) def _export_data(self): """Export analytics data.""" data = { 'exported_at': datetime.now().isoformat(), 'usage': self.usage_data, 'performance_samples': len(self.performance_data), 'errors': len(self.error_data) } # Save to file export_path = os.path.expanduser('~/.eu-utility/analytics_export.json') os.makedirs(os.path.dirname(export_path), exist_ok=True) with open(export_path, 'w') as f: json.dump(data, f, indent=2) self.notify_info("Export Complete", f"Data exported to:\n{export_path}") def _clear_all_data(self): """Clear all analytics data.""" self.usage_data = {} self.performance_data = [] self.error_data = [] self.session_events = [] self.save_data("usage", {}) self.save_data("performance", []) self.save_data("errors", []) self._refresh_all() self.notify_info("Data Cleared", "All analytics data has been cleared.") def _clear_errors(self): """Clear error log.""" self.error_data = [] self.save_data("errors", []) self._update_errors_tab() def on_show(self): """Update when tab shown.""" self._refresh_all() def shutdown(self): """Cleanup on shutdown.""" if hasattr(self, 'performance_timer'): self.performance_timer.stop() # Record final stats if self.enabled: self.save_data("final_session", { 'session_duration': time.time() - self.start_time, 'events_recorded': len(self.session_events), 'timestamp': datetime.now().isoformat() })