""" Module de monitoring - Surveillance continue et gestion des alertes """ import threading import time import json import os from datetime import datetime, timedelta from typing import Callable, Optional, List, Dict, Any from dataclasses import dataclass, field from enum import Enum import subprocess import sys # Import conditionnel pour winsound (Windows uniquement) try: import winsound WINSOUND_AVAILABLE = True except ImportError: WINSOUND_AVAILABLE = False # Pour la compatibilité multiplateforme try: import plyer from plyer import notification PLYER_AVAILABLE = True except ImportError: PLYER_AVAILABLE = False from core.network_tester import NetworkTester, TestResult class AlertLevel(Enum): """Niveaux d'alerte""" INFO = "info" WARNING = "warning" CRITICAL = "critical" @dataclass class Alert: """Alerte générée par le système de monitoring""" timestamp: datetime level: AlertLevel message: str test_type: str = "" value: Optional[float] = None threshold: Optional[float] = None def to_dict(self) -> Dict[str, Any]: return { 'timestamp': self.timestamp.isoformat(), 'level': self.level.value, 'message': self.message, 'test_type': self.test_type, 'value': self.value, 'threshold': self.threshold } @dataclass class MonitorConfig: """Configuration du monitoring""" interval: int = 60 # Intervalle entre les tests (secondes) ping_threshold: float = 100.0 # Seuil de latence pour alerte (ms) download_threshold: float = 5.0 # Seuil de téléchargement (Mbps) upload_threshold: float = 2.0 # Seuil d'upload (Mbps) packet_loss_threshold: float = 1.0 # Seuil de perte de paquets (%) jitter_threshold: float = 50.0 # Seuil de jitter (ms) enable_sound: bool = True # Activer les alertes sonores enable_notifications: bool = True # Activer les notifications enable_logging: bool = True # Activer la journalisation max_history: int = 100 # Nombre maximal d'historique à conserver test_types: List[str] = field(default_factory=lambda: ["ping", "connection", "dns"]) def to_dict(self) -> Dict[str, Any]: return { 'interval': self.interval, 'ping_threshold': self.ping_threshold, 'download_threshold': self.download_threshold, 'upload_threshold': self.upload_threshold, 'packet_loss_threshold': self.packet_loss_threshold, 'jitter_threshold': self.jitter_threshold, 'enable_sound': self.enable_sound, 'enable_notifications': self.enable_notifications, 'enable_logging': self.enable_logging, 'max_history': self.max_history, 'test_types': self.test_types } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'MonitorConfig': return cls( interval=data.get('interval', 60), ping_threshold=data.get('ping_threshold', 100.0), download_threshold=data.get('download_threshold', 5.0), upload_threshold=data.get('upload_threshold', 2.0), packet_loss_threshold=data.get('packet_loss_threshold', 1.0), jitter_threshold=data.get('jitter_threshold', 50.0), enable_sound=data.get('enable_sound', True), enable_notifications=data.get('enable_notifications', True), enable_logging=data.get('enable_logging', True), max_history=data.get('max_history', 100), test_types=data.get('test_types', ["ping", "connection", "dns"]) ) class AlertHandler: """Gestionnaire des alertes""" def __init__(self, config: MonitorConfig): self.config = config self.alert_callbacks: List[Callable[[Alert], None]] = [] self.alert_history: List[Alert] = [] def register_callback(self, callback: Callable[[Alert], None]): """Enregistre un callback pour les alertes""" self.alert_callbacks.append(callback) def trigger_alert(self, alert: Alert): """Déclenche une alerte""" self.alert_history.append(alert) # Limiter l'historique if len(self.alert_history) > self.config.max_history: self.alert_history = self.alert_history[-self.config.max_history:] # Appeler tous les callbacks for callback in self.alert_callbacks: try: callback(alert) except Exception as e: print(f"Erreur dans le callback d'alerte: {e}") # Notification système if self.config.enable_notifications: self._show_notification(alert) # Son d'alerte if self.config.enable_sound: self._play_sound(alert) def _show_notification(self, alert: Alert): """Affiche une notification système""" if PLYER_AVAILABLE: try: notification.notify( title="StabilityTester - Alerte", message=alert.message, app_name="StabilityTester", timeout=10 ) except Exception as e: print(f"Impossible d'afficher la notification: {e}") else: # Fallback pour les systèmes sans plyer print(f"[ALERTE {alert.level.value.upper()}] {alert.message}") def _play_sound(self, alert: Alert): """Joue un son d'alerte""" try: if sys.platform == 'win32' and WINSOUND_AVAILABLE: # Windows if alert.level == AlertLevel.CRITICAL: winsound.Beep(1000, 500) # Son aigu pour critique elif alert.level == AlertLevel.WARNING: winsound.Beep(800, 300) # Son moyen pour warning else: winsound.Beep(600, 200) # Son bas pour info elif sys.platform == 'darwin': # macOS subprocess.run(['afplay', '/System/Library/Sounds/Ping.aiff']) else: # Linux et autres subprocess.run(['paplay', '/usr/share/sounds/freedesktop/stereo/complete.oga'], stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL) except Exception as e: print(f"Impossible de jouer le son: {e}") def get_alert_history(self, limit: int = None) -> List[Alert]: """Récupère l'historique des alertes""" if limit: return self.alert_history[-limit:] return self.alert_history.copy() def clear_history(self): """Efface l'historique des alertes""" self.alert_history.clear() class StabilityMonitor: """ Moniteur de stabilité réseau - Effectue des tests réguliers et génère des alertes """ def __init__(self, config: MonitorConfig = None): """ Initialisation du moniteur Args: config: Configuration du monitoring """ self.config = config or MonitorConfig() self.tester = NetworkTester(timeout=10) self.alert_handler = AlertHandler(self.config) self.running = False self.monitor_thread: Optional[threading.Thread] = None self.test_history: List[Dict[str, TestResult]] = [] self.status_callbacks: List[Callable[[Dict[str, Any]], None]] = [] # État actuel self.current_status = { 'connected': False, 'last_test': None, 'uptime': timedelta(0), 'downtime': timedelta(0), 'current_streak': timedelta(0), 'connection_loss_count': 0 } # Historique de connectivité self.connection_history: List[bool] = [] def register_status_callback(self, callback: Callable[[Dict[str, Any]], None]): """Enregistre un callback pour les mises à jour de statut""" self.status_callbacks.append(callback) def register_alert_callback(self, callback: Callable[[Alert], None]): """Enregistre un callback pour les alertes""" self.alert_handler.register_callback(callback) def _notify_status_update(self): """Notifie les callbacks de mise à jour de statut""" for callback in self.status_callbacks: try: callback(self.current_status) except Exception as e: print(f"Erreur dans le callback de statut: {e}") def start(self): """Démarre le monitoring""" if self.running: return self.running = True self.monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True) self.monitor_thread.start() print("Monitoring démarré...") def stop(self): """Arrête le monitoring""" self.running = False if self.monitor_thread: self.monitor_thread.join(timeout=5) print("Monitoring arrêté.") def _monitor_loop(self): """Boucle principale de monitoring""" start_time = datetime.now() while self.running: iteration_start = datetime.now() # Effectuer les tests results = self._run_tests() # Analyser les résultats et générer des alertes self._analyze_results(results) # Mettre à jour le statut self._update_status(results) # Enregistrer l'historique self.test_history.append(results) if len(self.test_history) > self.config.max_history: self.test_history = self.test_history[-self.config.max_history:] # Calculer le temps d'attente elapsed = (datetime.now() - iteration_start).total_seconds() sleep_time = max(0, self.config.interval - elapsed) # Attendre avant le prochain test for _ in range(int(sleep_time * 10)): # Boucle avec vérification toutes les 0.1s if not self.running: break time.sleep(0.1) def _run_tests(self) -> Dict[str, TestResult]: """Effectue les tests configurés""" results = {} try: # Toujours tester la connectivité de base if "connection" in self.config.test_types: results['connection'] = self.tester.connection_test() if "dns" in self.config.test_types: results['dns'] = self.tester.dns_test() if "ping" in self.config.test_types: # Test de ping vers plusieurs serveurs ping_results = self.tester.ping_all_servers() for i, result in enumerate(ping_results): results[f'ping_{i}'] = result if "download" in self.config.test_types: results['download'] = self.tester.download_speed_test(duration=3) if "upload" in self.config.test_types: results['upload'] = self.tester.upload_speed_test(data_size=256 * 1024) if "full_speed" in self.config.test_types: results['full_speed'] = self.tester.full_speed_test() except Exception as e: print(f"Erreur lors des tests: {e}") # Créer un résultat d'erreur results['error'] = TestResult( timestamp=datetime.now(), test_type="error", success=False, error_message=str(e) ) return results def _analyze_results(self, results: Dict[str, TestResult]): """Analyse les résultats et génère des alertes si nécessaire""" for test_name, result in results.items(): if not result.success: # Alerte pour échec de test alert = Alert( timestamp=datetime.now(), level=AlertLevel.CRITICAL, message=f"Test {result.test_type} échoué: {result.error_message}", test_type=result.test_type, value=None, threshold=None ) self.alert_handler.trigger_alert(alert) else: # Vérifier les seuils self._check_thresholds(test_name, result) def _check_thresholds(self, test_name: str, result: TestResult): """Vérifie si les valeurs dépassent les seuils""" if result.latency_ms is not None: if result.latency_ms > self.config.ping_threshold: alert = Alert( timestamp=datetime.now(), level=AlertLevel.WARNING, message=f"Latence élevée: {result.latency_ms:.1f}ms (seuil: {self.config.ping_threshold}ms)", test_type=result.test_type, value=result.latency_ms, threshold=self.config.ping_threshold ) self.alert_handler.trigger_alert(alert) if result.jitter is not None: if result.jitter > self.config.jitter_threshold: alert = Alert( timestamp=datetime.now(), level=AlertLevel.WARNING, message=f"Jitter élevé: {result.jitter:.1f}ms (seuil: {self.config.jitter_threshold}ms)", test_type=result.test_type, value=result.jitter, threshold=self.config.jitter_threshold ) self.alert_handler.trigger_alert(alert) if result.download_speed is not None: if result.download_speed < self.config.download_threshold: alert = Alert( timestamp=datetime.now(), level=AlertLevel.WARNING, message=f"Vitesse de téléchargement faible: {result.download_speed:.2f}Mbps (seuil: {self.config.download_threshold}Mbps)", test_type=result.test_type, value=result.download_speed, threshold=self.config.download_threshold ) self.alert_handler.trigger_alert(alert) if result.upload_speed is not None: if result.upload_speed < self.config.upload_threshold: alert = Alert( timestamp=datetime.now(), level=AlertLevel.WARNING, message=f"Vitesse d'upload faible: {result.upload_speed:.2f}Mbps (seuil: {self.config.upload_threshold}Mbps)", test_type=result.test_type, value=result.upload_speed, threshold=self.config.upload_threshold ) self.alert_handler.trigger_alert(alert) def _update_status(self, results: Dict[str, TestResult]): """Met à jour l'état du moniteur""" now = datetime.now() # Vérifier si la connexion est active connected = True for result in results.values(): if result.test_type in ["connection", "ping"] and not result.success: connected = False break # Mettre à jour l'historique de connectivité self.connection_history.append(connected) # Mettre à jour les compteurs if len(self.connection_history) > 1: prev_connected = self.connection_history[-2] if len(self.connection_history) >= 2 else True if connected and not prev_connected: # Connexion rétablie self.current_status['connection_loss_count'] += 1 self.current_status['current_streak'] = timedelta(0) elif not connected and prev_connected: # Connexion perdue self.current_status['current_streak'] = timedelta(0) # Calculer le temps de disponibilité uptime = sum(1 for c in self.connection_history if c) * self.config.interval downtime = sum(1 for c in self.connection_history if not c) * self.config.interval self.current_status['connected'] = connected self.current_status['last_test'] = now self.current_status['uptime'] = timedelta(seconds=uptime) self.current_status['downtime'] = timedelta(seconds=downtime) self.current_status['current_streak'] = timedelta(seconds=self.config.interval) # Notifier les callbacks self._notify_status_update() def run_single_test(self) -> Dict[str, TestResult]: """Effectue un test unique (pour l'interface utilisateur)""" return self._run_tests() def get_current_status(self) -> Dict[str, Any]: """Récupère l'état actuel""" return self.current_status.copy() def get_test_history(self, limit: int = None) -> List[Dict[str, TestResult]]: """Récupère l'historique des tests""" if limit: return self.test_history[-limit:] return self.test_history.copy() def get_statistics(self) -> Dict[str, Any]: """Calcule les statistiques globales""" if not self.test_history: return { 'total_tests': 0, 'success_rate': 0, 'avg_latency': 0, 'avg_download': 0, 'avg_upload': 0, 'connection_loss_count': self.current_status['connection_loss_count'], 'uptime_percentage': 100 } total_tests = 0 success_count = 0 total_latency = 0 latency_count = 0 total_download = 0 download_count = 0 total_upload = 0 upload_count = 0 for results in self.test_history: for result in results.values(): total_tests += 1 if result.success: success_count += 1 if result.latency_ms is not None: total_latency += result.latency_ms latency_count += 1 if result.download_speed is not None: total_download += result.download_speed download_count += 1 if result.upload_speed is not None: total_upload += result.upload_speed upload_count += 1 return { 'total_tests': total_tests, 'success_rate': (success_count / total_tests * 100) if total_tests > 0 else 0, 'avg_latency': (total_latency / latency_count) if latency_count > 0 else 0, 'avg_download': (total_download / download_count) if download_count > 0 else 0, 'avg_upload': (total_upload / upload_count) if upload_count > 0 else 0, 'connection_loss_count': self.current_status['connection_loss_count'], 'uptime_percentage': (self.current_status['uptime'].total_seconds() / (self.current_status['uptime'].total_seconds() + self.current_status['downtime'].total_seconds() + 0.001) * 100) if (self.current_status['uptime'].total_seconds() + self.current_status['downtime'].total_seconds()) > 0 else 100 } def get_stability_score(self) -> Dict[str, Any]: """ Calcule un score de stabilité globale basé sur plusieurs critères Returns: Dictionnaire avec le score et les détails """ stats = self.get_statistics() # Score de base à 100 score = 100.0 # Critère 1 : Taux de succès des tests (poids: 30%) success_rate = stats.get('success_rate', 100) score -= (100 - success_rate) * 0.3 # Critère 2 : Temps de disponibilité (poids: 25%) uptime_pct = stats.get('uptime_percentage', 100) score -= (100 - uptime_pct) * 0.25 # Critère 3 : Latence moyenne (poids: 20%) avg_latency = stats.get('avg_latency', 0) # Score latence: 100 si < 50ms, 0 si > 200ms latency_score = max(0, 100 - (avg_latency - 50) * 1.25) if avg_latency > 50 else 100 latency_score = min(100, latency_score) score -= (100 - latency_score) * 0.20 # Critère 4 : Vitesse de téléchargement (poids: 15%) avg_download = stats.get('avg_download', 0) # Score download: 100 si > 10Mbps, 0 si < 1Mbps download_score = min(100, avg_download * 10) score -= (100 - download_score) * 0.15 # Critère 5 : Vitesse d'upload (poids: 10%) avg_upload = stats.get('avg_upload', 0) # Score upload: 100 si > 5Mbps, 0 si < 1Mbps upload_score = min(100, avg_upload * 25) score -= (100 - upload_score) * 0.10 # Arrondir le score score = round(score, 1) # Déterminer le niveau de stabilité if score >= 80: stability_level = "EXCELLENT" color = "green" elif score >= 60: stability_level = "BON" color = "orange" elif score >= 40: stability_level = "MOYEN" color = "orange" else: stability_level = "FAIBLE" color = "red" return { 'score': score, 'stability_level': stability_level, 'color': color, 'success_rate': success_rate, 'uptime_percentage': uptime_pct, 'avg_latency': avg_latency, 'avg_download': avg_download, 'avg_upload': avg_upload, 'connection_losses': stats.get('connection_loss_count', 0), 'criteria': { 'success_rate': {'value': success_rate, 'weight': 30, 'score': success_rate}, 'uptime': {'value': uptime_pct, 'weight': 25, 'score': uptime_pct}, 'latency': {'value': avg_latency, 'weight': 20, 'score': latency_score}, 'download': {'value': avg_download, 'weight': 15, 'score': download_score}, 'upload': {'value': avg_upload, 'weight': 10, 'score': upload_score} } } def save_state(self, filepath: str = "monitor_state.json"): """Sauvegarde l'état actuel dans un fichier""" try: state = { 'config': self.config.to_dict(), 'current_status': { k: str(v) if isinstance(v, timedelta) else v for k, v in self.current_status.items() }, 'test_history': [ {name: result.to_dict() for name, result in results.items()} for results in self.test_history ], 'connection_history': self.connection_history, 'alert_history': [alert.to_dict() for alert in self.alert_handler.get_alert_history()] } with open(filepath, 'w') as f: json.dump(state, f, indent=2) print(f"État sauvegardé dans {filepath}") return True except Exception as e: print(f"Erreur lors de la sauvegarde: {e}") return False def load_state(self, filepath: str = "monitor_state.json") -> bool: """Charge l'état depuis un fichier""" try: with open(filepath, 'r') as f: state = json.load(f) self.config = MonitorConfig.from_dict(state['config']) self.alert_handler = AlertHandler(self.config) # Reconstruire l'historique for alert_data in state.get('alert_history', []): alert = Alert( timestamp=datetime.fromisoformat(alert_data['timestamp']), level=AlertLevel(alert_data['level']), message=alert_data['message'], test_type=alert_data['test_type'], value=alert_data.get('value'), threshold=alert_data.get('threshold') ) self.alert_handler.alert_history.append(alert) # Reconstruire l'historique des tests for test_data in state.get('test_history', []): results = {} for name, result_data in test_data.items(): result = TestResult( timestamp=datetime.fromisoformat(result_data['timestamp']), test_type=result_data['test_type'], success=result_data['success'], latency_ms=result_data.get('latency_ms'), download_speed=result_data.get('download_speed'), upload_speed=result_data.get('upload_speed'), packet_loss=result_data.get('packet_loss'), jitter=result_data.get('jitter'), error_message=result_data.get('error_message'), server=result_data.get('server') ) results[name] = result self.test_history.append(results) self.connection_history = state.get('connection_history', []) print(f"État chargé depuis {filepath}") return True except FileNotFoundError: print(f"Fichier {filepath} introuvable, utilisation des valeurs par défaut") return False except Exception as e: print(f"Erreur lors du chargement: {e}") return False