tuxgyver
4adde135d7
L'application ne demarrait pas du tout : gui/app.py importait pyqtSignal, qui est le nom PyQt du signal, absent de PySide6. Trois autres defauts bloquaient juste derriere, une fois celui-ci leve : - MPLBACKEND etait fixe a 'module://backend_interagg', le backend interne de PyCharm, indisponible hors de cet IDE ; - gui/app.py et monitor/monitor.py utilisaient des imports relatifs remontant au-dela du paquet racine charge par main.py ; - core/ n'avait pas de __init__.py. Corrige aussi get_statistics(), dont le retour anticipe "historique vide" omettait connection_loss_count et uptime_percentage : le KeyError qui en resultait cassait silencieusement tout l'onglet Statistiques a chaque rafraichissement. Les deux chemins renvoient desormais les memes cles. Ajoute le build Windows (StabilityTester.spec + make build-windows), qui produit un .exe autonome depuis Linux via Wine. PySide6 y est epingle a 6.9.3 : a partir de 6.10, Qt6Core.dll depend de l'ICU systeme de Windows que Wine n'implemente pas. La sortie standard est reconfiguree en UTF-8 dans main.py, les emojis des messages faisant echouer cp1252 sous Windows. make test etait casse par la meme confusion de paquet et passe a nouveau. Makefile : libgl1-mesa-glx, retire des depots Debian/Ubuntu recents, remplace par libgl1 et les dependances Qt6 manquantes.
655 lines
26 KiB
Python
655 lines
26 KiB
Python
"""
|
|
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
|