diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..7244347
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,82 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# PyInstaller
+*.manifest
+*.spec
+# ...sauf la spec de build du projet, qui fait partie des sources.
+!StabilityTester.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+
+# Translations
+*.mo
+*.pot
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# IDE
+.idea/
+.vscode/
+*.swp
+*.swo
+*~
+
+# OS
+.DS_Store
+Thumbs.db
+
+# Logs
+*.log
+monitor_state.json
+
+# Data exports
+*.json
+# ...sauf la configuration par defaut, qui fait partie des sources.
+!config/default_config.json
diff --git a/Makefile b/Makefile
new file mode 100644
index 0000000..0fc639e
--- /dev/null
+++ b/Makefile
@@ -0,0 +1,195 @@
+# Makefile for StabilityTester
+# Version: 1.0.1
+# Interface: PySide6 (Qt for Python)
+
+.PHONY: help install run test clean all check-python venv prerequis clean-venv info build-windows
+
+# Variables
+VENV_DIR := .venv
+PYTHON := python3
+PIP := pip
+REQUIREMENTS := requirements.txt
+
+# Build Windows (croise, via Wine)
+WINE_PREFIX := $(HOME)/.wine-stabilitytester
+WIN_PYTHON_VERSION := 3.13.15
+WIN_PYTHON := C:\\Python313\\python.exe
+# PySide6 est epingle pour le build Windows : a partir de 6.10, Qt6Core.dll
+# depend de icuuc.dll (ICU systeme de Windows 10+) que Wine n'implemente pas,
+# ce qui rend l'analyse PyInstaller impossible. 6.9.3 s'en passe.
+WIN_PYSIDE6 := PySide6==6.9.3
+
+# Detect OS
+UNAME := $(shell uname -s)
+
+# Default target
+help:
+ @echo "🌐 StabilityTester - Makefile"
+ @echo "============================"
+ @echo ""
+ @echo "Targets:"
+ @echo " prerequis - 📋 Install system prerequisites"
+ @echo " venv - 🐍 Create Python virtual environment"
+ @echo " install - 📦 Install Python dependencies"
+ @echo " run - 🚀 Run the application"
+ @echo " test - 🧪 Run basic tests"
+ @echo " build-windows - 🪟 Build the Windows .exe (via Wine)"
+ @echo " clean - 🧹 Clean build artifacts"
+ @echo " clean-venv - 🧹 Remove virtual environment"
+ @echo " all - 🔄 Install prerequisites, venv, dependencies and run"
+ @echo " info - 📊 Show environment info"
+ @echo " check-py - 🐍 Check Python version"
+ @echo ""
+
+# Check Python version
+check-python:
+ @$(PYTHON) --version
+
+# Install system prerequisites
+prerequis:
+ @echo "📋 Installing system prerequisites..."
+ @echo "Detected OS: $(UNAME)"
+ifeq ($(UNAME),Linux)
+ @echo "🐧 Linux detected - Installing required packages..."
+ @if [ -f /etc/os-release ]; then \
+ . /etc/os-release && \
+ if [ "$$ID" = "ubuntu" ] || [ "$$ID" = "debian" ] || [ "$$ID_LIKE" = "debian" ]; then \
+ sudo apt-get update -qq && \
+ sudo apt-get install -y -qq python3 python3-pip python3-venv libgl1 libxcb-xinerama0 libxkbcommon-x11-0 libxcb-cursor0 || echo "⚠️ Some packages may have failed to install"; \
+ fi; \
+ fi
+ @echo "✅ System prerequisites installed!"
+else ifeq ($(UNAME),Darwin)
+ @echo "🍎 macOS detected"
+ @echo "✅ macOS prerequisites are typically pre-installed!"
+ @echo "If you have issues, install: brew install pythonqt"
+else
+ @echo "🪟 Windows detected"
+ @echo "✅ Windows prerequisites are typically pre-installed!"
+ @echo "Make sure Python is in your PATH"
+endif
+
+# Create virtual environment
+venv:
+ @echo "🐍 Creating Python virtual environment..."
+ @if [ ! -d "$(VENV_DIR)" ]; then \
+ $(PYTHON) -m venv $(VENV_DIR) && \
+ echo "✅ Virtual environment created at $(VENV_DIR)"; \
+ else \
+ echo "ℹ️ Virtual environment already exists at $(VENV_DIR)"; \
+ fi
+ @echo ""
+ @echo "To activate the virtual environment, run:"
+ @echo " source $(VENV_DIR)/bin/activate"
+
+# Install dependencies
+install:
+ @echo "📦 Installing Python dependencies..."
+ @if [ -d "$(VENV_DIR)" ]; then \
+ echo "Using virtual environment at $(VENV_DIR)"; \
+ $(VENV_DIR)/bin/pip install -r $(REQUIREMENTS) && \
+ echo "✅ Dependencies installed in virtual environment!"; \
+ else \
+ $(PIP) install -r $(REQUIREMENTS) && \
+ echo "✅ Dependencies installed globally!"; \
+ fi
+
+# Run the application
+run:
+ @echo "🚀 Starting StabilityTester with PySide6..."
+ @if [ -d "$(VENV_DIR)" ]; then \
+ $(VENV_DIR)/bin/python main.py; \
+ else \
+ $(PYTHON) main.py; \
+ fi
+
+# Run basic tests
+test:
+ @echo "🧪 Running basic tests..."
+ @if [ -d "$(VENV_DIR)" ]; then \
+ $(VENV_DIR)/bin/python -c "from core.network_tester import NetworkTester; t = NetworkTester(); r = t.connection_test(); print(f'✅ Connection test: {r.success}')" && \
+ $(VENV_DIR)/bin/python -c "from monitor.monitor import StabilityMonitor; m = StabilityMonitor(); print(f'✅ StabilityMonitor created successfully')" && \
+ $(VENV_DIR)/bin/python -c "from gui.app import StabilityTesterWindow; print(f'✅ PySide6 GUI loaded successfully')"; \
+ else \
+ $(PYTHON) -c "from core.network_tester import NetworkTester; t = NetworkTester(); r = t.connection_test(); print(f'✅ Connection test: {r.success}')" && \
+ $(PYTHON) -c "from monitor.monitor import StabilityMonitor; m = StabilityMonitor(); print(f'✅ StabilityMonitor created successfully')" && \
+ $(PYTHON) -c "from gui.app import StabilityTesterWindow; print(f'✅ PySide6 GUI loaded successfully')"; \
+ fi
+ @echo "🎉 All basic tests passed!"
+
+# Build de l'executable Windows depuis Linux.
+# PyInstaller ne sait pas cross-compiler : on execute un Python Windows sous
+# Wine. Le prefixe est provisionne une seule fois puis reutilise.
+build-windows:
+ @echo "🪟 Building Windows executable via Wine..."
+ @command -v wine >/dev/null 2>&1 || { echo "❌ wine is required (apt install wine winetricks)"; exit 1; }
+ @if [ ! -f "$(WINE_PREFIX)/drive_c/Python313/python.exe" ]; then \
+ echo "📥 Provisioning Wine prefix (one-time, downloads ~1 GB)..."; \
+ WINEPREFIX=$(WINE_PREFIX) WINEARCH=win64 wineboot --init >/dev/null 2>&1; \
+ echo "📥 Installing native UCRT (Wine's builtin lacks the C99 complex math NumPy needs)..."; \
+ WINEPREFIX=$(WINE_PREFIX) winetricks -q -f vcrun2022 ucrtbase2019 >/dev/null 2>&1; \
+ curl -sL -o /tmp/python-win.exe \
+ https://www.python.org/ftp/python/$(WIN_PYTHON_VERSION)/python-$(WIN_PYTHON_VERSION)-amd64.exe && \
+ WINEPREFIX=$(WINE_PREFIX) wine /tmp/python-win.exe /quiet InstallAllUsers=1 \
+ PrependPath=1 Include_test=0 TargetDir='C:\\Python313' && \
+ rm -f /tmp/python-win.exe; \
+ else \
+ echo "ℹ️ Wine prefix already provisioned at $(WINE_PREFIX)"; \
+ fi
+ @echo "📦 Installing Windows dependencies..."
+ @WINEPREFIX=$(WINE_PREFIX) WINEDEBUG=-all wine $(WIN_PYTHON) -m pip install --no-input \
+ --disable-pip-version-check -q -r $(REQUIREMENTS) "$(WIN_PYSIDE6)" pyinstaller
+ @echo "🔨 Running PyInstaller..."
+ @WINEPREFIX=$(WINE_PREFIX) wineserver -k 2>/dev/null || true
+ @rm -f dist/StabilityTester.exe
+ @WINEPREFIX=$(WINE_PREFIX) WINEDEBUG=-all wine $(WIN_PYTHON) -m PyInstaller \
+ --noconfirm --clean --distpath dist --workpath build StabilityTester.spec
+ @WINEPREFIX=$(WINE_PREFIX) wineserver -k 2>/dev/null || true
+ @echo "✅ Windows executable: dist/StabilityTester.exe"
+ @ls -lh dist/StabilityTester.exe
+
+# Clean build artifacts
+clean:
+ @echo "🧹 Cleaning build artifacts..."
+ find . -type d -name "__pycache__" -exec rm -rf {} + 2>/dev/null || true
+ find . -name "*.pyc" -delete 2>/dev/null || true
+ find . -name "*.pyo" -delete 2>/dev/null || true
+ @echo "✅ Cleanup complete!"
+
+# Remove virtual environment
+clean-venv:
+ @echo "🧹 Removing virtual environment..."
+ @if [ -d "$(VENV_DIR)" ]; then \
+ rm -rf $(VENV_DIR) && \
+ echo "✅ Virtual environment removed!"; \
+ else \
+ echo "ℹ️ No virtual environment found"; \
+ fi
+
+# Install everything and run
+all: prerequis venv install run
+
+# Show environment info
+info:
+ @echo "📊 StabilityTester - Environment Info"
+ @echo "================================="
+ @echo ""
+ @echo "Python version:"
+ @$(PYTHON) --version
+ @echo ""
+ @echo "Pip version:"
+ @$(PIP) --version 2>/dev/null || echo "pip not found"
+ @echo ""
+ @echo "Virtual environment:"
+ @if [ -d "$(VENV_DIR)" ]; then \
+ ls -la $(VENV_DIR)/bin/python* 2>/dev/null | head -1 || echo "$(VENV_DIR) exists"; \
+ else \
+ echo "No virtual environment"; \
+ fi
+ @echo ""
+ @echo "Installed packages:"
+ @if [ -d "$(VENV_DIR)" ]; then \
+ $(VENV_DIR)/bin/pip list 2>/dev/null | grep -E "(PySide6|requests|matplotlib|speedtest|ping3)" || echo "No packages in venv"; \
+ else \
+ $(PIP) list 2>/dev/null | grep -E "(PySide6|requests|matplotlib|speedtest|ping3)" || echo "No packages found"; \
+ fi
diff --git a/README.md b/README.md
index d6c34a6..ec163f5 100644
--- a/README.md
+++ b/README.md
@@ -68,6 +68,19 @@ make install
pip install -r requirements.txt
```
+### Executable Windows
+Un `.exe` autonome est joint a chaque release. Pour le reconstruire depuis
+Linux (PyInstaller ne sait pas cross-compiler : un Python Windows est execute
+sous Wine) :
+
+```bash
+sudo apt install wine winetricks
+make build-windows # -> dist/StabilityTester.exe
+```
+
+Le premier lancement provisionne le prefixe Wine (~1 Go de telechargements),
+les suivants le reutilisent.
+
---
## 🎮 Utilisation
diff --git a/StabilityTester.spec b/StabilityTester.spec
new file mode 100644
index 0000000..61f95c1
--- /dev/null
+++ b/StabilityTester.spec
@@ -0,0 +1,46 @@
+# -*- mode: python ; coding: utf-8 -*-
+"""
+Definition du build Windows de StabilityTester (PyInstaller).
+
+Construit depuis Linux via Wine : voir la cible `build-windows` du Makefile.
+"""
+
+a = Analysis(
+ ['main.py'],
+ pathex=['.'],
+ binaries=[],
+ datas=[],
+ # plyer charge sa facade de notification dynamiquement : PyInstaller ne
+ # peut pas la detecter par analyse statique.
+ hiddenimports=['plyer.platforms.win.notification'],
+ hookspath=[],
+ hooksconfig={},
+ runtime_hooks=[],
+ # tkinter et les autres bindings Qt seraient embarques par matplotlib
+ # alors que l'application n'utilise que PySide6.
+ excludes=['tkinter', 'PyQt5', 'PyQt6', 'PySide2'],
+ noarchive=False,
+)
+
+pyz = PYZ(a.pure)
+
+exe = EXE(
+ pyz,
+ a.scripts,
+ a.binaries,
+ a.datas,
+ [],
+ name='StabilityTester',
+ debug=False,
+ bootloader_ignore_signals=False,
+ strip=False,
+ upx=False,
+ runtime_tmpdir=None,
+ # Application graphique : pas de console Windows a l'ouverture.
+ console=False,
+ disable_windowed_traceback=False,
+ argv_emulation=False,
+ target_arch=None,
+ codesign_identity=None,
+ entitlements_file=None,
+)
diff --git a/__init__.py b/__init__.py
new file mode 100644
index 0000000..4243622
--- /dev/null
+++ b/__init__.py
@@ -0,0 +1,11 @@
+# StabilityTester - Application de surveillance de connexion internet
+# Version: 1.0.1
+# Auteur: StabilityTester
+
+"""
+Application complète pour tester, surveiller et alerter sur la stabilité
+d'une connexion internet avec interface graphique et fonctionnement en tâche de fond.
+"""
+
+__version__ = "1.0.1"
+__author__ = "StabilityTester"
diff --git a/config/__init__.py b/config/__init__.py
new file mode 100644
index 0000000..ab0e2ff
--- /dev/null
+++ b/config/__init__.py
@@ -0,0 +1 @@
+# Configuration package
diff --git a/config/default_config.json b/config/default_config.json
new file mode 100644
index 0000000..d9ca558
--- /dev/null
+++ b/config/default_config.json
@@ -0,0 +1,13 @@
+{
+ "interval": 60,
+ "ping_threshold": 100.0,
+ "download_threshold": 5.0,
+ "upload_threshold": 2.0,
+ "packet_loss_threshold": 1.0,
+ "jitter_threshold": 50.0,
+ "enable_sound": true,
+ "enable_notifications": true,
+ "enable_logging": true,
+ "max_history": 100,
+ "test_types": ["ping", "connection", "dns"]
+}
diff --git a/core/__init__.py b/core/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/core/network_tester.py b/core/network_tester.py
new file mode 100644
index 0000000..3c1700e
--- /dev/null
+++ b/core/network_tester.py
@@ -0,0 +1,429 @@
+"""
+Module de test réseau - Effectue divers tests de connectivité et de performance
+"""
+
+import subprocess
+import time
+import requests
+import socket
+import json
+import threading
+from dataclasses import dataclass
+from typing import Optional, Dict, Any
+from datetime import datetime
+
+# Import optionnel de speedtest
+try:
+ import speedtest
+ SPEEDTEST_AVAILABLE = True
+except ImportError:
+ SPEEDTEST_AVAILABLE = False
+ speedtest = None
+
+# Import optionnel de ping3
+try:
+ import ping3
+ PING3_AVAILABLE = True
+except ImportError:
+ PING3_AVAILABLE = False
+ ping3 = None
+
+
+@dataclass
+class TestResult:
+ """Résultat d'un test réseau"""
+ timestamp: datetime
+ test_type: str
+ success: bool
+ latency_ms: Optional[float] = None
+ download_speed: Optional[float] = None # en Mbps
+ upload_speed: Optional[float] = None # en Mbps
+ packet_loss: Optional[float] = None # en %
+ jitter: Optional[float] = None # en ms
+ error_message: Optional[str] = None
+ server: Optional[str] = None
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {
+ 'timestamp': self.timestamp.isoformat(),
+ 'test_type': self.test_type,
+ 'success': self.success,
+ 'latency_ms': self.latency_ms,
+ 'download_speed': self.download_speed,
+ 'upload_speed': self.upload_speed,
+ 'packet_loss': self.packet_loss,
+ 'jitter': self.jitter,
+ 'error_message': self.error_message,
+ 'server': self.server
+ }
+
+
+class NetworkTester:
+ """
+ Classe pour effectuer divers tests de connectivité et de performance réseau
+ """
+
+ # Serveurs de test par défaut
+ PING_SERVERS = [
+ "8.8.8.8", # Google DNS
+ "1.1.1.1", # Cloudflare DNS
+ "9.9.9.9", # Quad9 DNS
+ "8.8.4.4", # Google DNS secondaire
+ "208.67.222.222" # OpenDNS
+ ]
+
+ SPEEDTEST_SERVERS = [
+ {"name": "Paris", "id": 12345}, # ID à adapter selon votre région
+ {"name": "Lyon", "id": 67890},
+ ]
+
+ # URL pour tester le téléchargement
+ DOWNLOAD_TEST_URL = "https://speedtest.ftp.otenet.gr/files/test1Mb.db"
+ UPLOAD_TEST_URL = "https://httpbin.org/post"
+
+ def __init__(self, timeout: int = 5):
+ """
+ Initialisation du testeur réseau
+
+ Args:
+ timeout: Timeout en secondes pour les tests
+ """
+ self.timeout = timeout
+ self.last_test_time = None
+
+ def ping_test(self, server: str = None, count: int = 4) -> TestResult:
+ """
+ Effectue un test de ping vers un serveur
+
+ Args:
+ server: Adresse IP ou hostname à pinger (par défaut, utilise le premier serveur)
+ count: Nombre de paquets à envoyer
+
+ Returns:
+ TestResult: Résultat du test
+ """
+ if server is None:
+ server = self.PING_SERVERS[0]
+
+ if not PING3_AVAILABLE:
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="ping",
+ success=False,
+ error_message="ping3 non installé. Installez avec: pip install ping3"
+ )
+
+ try:
+ # Utilisation de ping3 pour plus de fiabilité
+ start_time = time.time()
+
+ # Effectuer le ping
+ latency_ms = ping3.ping(server, unit='ms', timeout=self.timeout)
+
+ if latency_ms is None or latency_ms is False:
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="ping",
+ success=False,
+ error_message=f"Échec du ping vers {server}",
+ server=server
+ )
+
+ # Calcul du jitter (variation de latence)
+ latencies = []
+ for _ in range(count):
+ lat = ping3.ping(server, unit='ms', timeout=self.timeout)
+ if lat is not None and lat is not False:
+ latencies.append(lat)
+
+ jitter = max(latencies) - min(latencies) if latencies else 0
+
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="ping",
+ success=True,
+ latency_ms=latency_ms,
+ jitter=jitter,
+ server=server
+ )
+
+ except Exception as e:
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="ping",
+ success=False,
+ error_message=f"Erreur lors du ping: {str(e)}",
+ server=server
+ )
+
+ def ping_all_servers(self) -> list:
+ """
+ Ping tous les serveurs configurés
+
+ Returns:
+ Liste de TestResult pour chaque serveur
+ """
+ results = []
+ for server in self.PING_SERVERS:
+ result = self.ping_test(server, count=2)
+ results.append(result)
+ time.sleep(0.5) # Petite pause entre les tests
+ return results
+
+ def connection_test(self, url: str = "https://www.google.com") -> TestResult:
+ """
+ Teste la connectivité HTTP vers une URL
+
+ Args:
+ url: URL à tester
+
+ Returns:
+ TestResult: Résultat du test
+ """
+ try:
+ start_time = time.time()
+ response = requests.get(url, timeout=self.timeout)
+ latency = (time.time() - start_time) * 1000 # en ms
+
+ if response.status_code < 400:
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="http_connection",
+ success=True,
+ latency_ms=latency,
+ server=url
+ )
+ else:
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="http_connection",
+ success=False,
+ error_message=f"Code HTTP: {response.status_code}",
+ server=url
+ )
+
+ except requests.exceptions.RequestException as e:
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="http_connection",
+ success=False,
+ error_message=f"Erreur de connexion: {str(e)}",
+ server=url
+ )
+
+ def dns_test(self, domain: str = "google.com") -> TestResult:
+ """
+ Teste la résolution DNS
+
+ Args:
+ domain: Domaine à résoudre
+
+ Returns:
+ TestResult: Résultat du test
+ """
+ try:
+ start_time = time.time()
+ # Résolution DNS
+ ip_addresses = socket.getaddrinfo(domain, None)
+ latency = (time.time() - start_time) * 1000
+
+ if ip_addresses:
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="dns_resolution",
+ success=True,
+ latency_ms=latency,
+ server=domain
+ )
+ else:
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="dns_resolution",
+ success=False,
+ error_message="Aucune adresse IP trouvée",
+ server=domain
+ )
+
+ except socket.gaierror as e:
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="dns_resolution",
+ success=False,
+ error_message=f"Erreur DNS: {str(e)}",
+ server=domain
+ )
+
+ def download_speed_test(self, url: str = None, duration: int = 5) -> TestResult:
+ """
+ Teste la vitesse de téléchargement
+
+ Args:
+ url: URL du fichier à télécharger
+ duration: Durée du test en secondes
+
+ Returns:
+ TestResult: Résultat du test
+ """
+ if url is None:
+ url = self.DOWNLOAD_TEST_URL
+
+ try:
+ start_time = time.time()
+ response = requests.get(url, stream=True, timeout=self.timeout)
+
+ # Lire les données par blocs pour mesurer la vitesse
+ bytes_received = 0
+ chunk_size = 1024 * 1024 # 1 Mo
+
+ while time.time() - start_time < duration:
+ chunk = response.raw.read(chunk_size)
+ if not chunk:
+ break
+ bytes_received += len(chunk)
+
+ elapsed = time.time() - start_time
+ speed_mbps = (bytes_received * 8) / (1024 * 1024) / elapsed if elapsed > 0 else 0
+
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="download_speed",
+ success=True,
+ download_speed=round(speed_mbps, 2),
+ server=url
+ )
+
+ except Exception as e:
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="download_speed",
+ success=False,
+ error_message=f"Erreur de téléchargement: {str(e)}",
+ server=url
+ )
+
+ def upload_speed_test(self, url: str = None, data_size: int = 1024 * 1024) -> TestResult:
+ """
+ Teste la vitesse d'upload
+
+ Args:
+ url: URL pour le test d'upload
+ data_size: Taille des données à uploader en octets
+
+ Returns:
+ TestResult: Résultat du test
+ """
+ if url is None:
+ url = self.UPLOAD_TEST_URL
+
+ try:
+ # Créer des données aléatoires
+ data = b'0' * data_size
+
+ start_time = time.time()
+ response = requests.post(url, data=data, timeout=self.timeout * 2)
+ elapsed = time.time() - start_time
+
+ speed_mbps = (data_size * 8) / (1024 * 1024) / elapsed if elapsed > 0 else 0
+
+ if response.status_code < 400:
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="upload_speed",
+ success=True,
+ upload_speed=round(speed_mbps, 2),
+ server=url
+ )
+ else:
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="upload_speed",
+ success=False,
+ error_message=f"Code HTTP: {response.status_code}",
+ server=url
+ )
+
+ except Exception as e:
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="upload_speed",
+ success=False,
+ error_message=f"Erreur d'upload: {str(e)}",
+ server=url
+ )
+
+ def full_speed_test(self) -> TestResult:
+ """
+ Effectue un test complet de vitesse (download + upload) avec speedtest-cli
+
+ Returns:
+ TestResult: Résultat du test
+ """
+ if not SPEEDTEST_AVAILABLE:
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="full_speed_test",
+ success=False,
+ error_message="speedtest-cli non installé. Installez avec: pip install speedtest-cli"
+ )
+
+ try:
+ st = speedtest.Speedtest()
+
+ # Sélection du meilleur serveur
+ st.get_best_server()
+
+ # Test de téléchargement
+ st.download()
+ download_speed = st.results.dict()["download"] / 1024 / 1024 # en Mbps
+
+ # Test d'upload
+ st.upload()
+ upload_speed = st.results.dict()["upload"] / 1024 / 1024 # en Mbps
+
+ # Latence (ping)
+ latency = st.results.dict()["ping"]
+
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="full_speed_test",
+ success=True,
+ latency_ms=latency,
+ download_speed=round(download_speed, 2),
+ upload_speed=round(upload_speed, 2),
+ server=st.results.dict()["server"]["name"]
+ )
+
+ except Exception as e:
+ return TestResult(
+ timestamp=datetime.now(),
+ test_type="full_speed_test",
+ success=False,
+ error_message=f"Erreur lors du speedtest: {str(e)}"
+ )
+
+ def comprehensive_test(self) -> Dict[str, TestResult]:
+ """
+ Effectue une série complète de tests réseau
+
+ Returns:
+ Dictionnaire avec les résultats de tous les tests
+ """
+ results = {}
+
+ # Test de connectivité
+ results['connection'] = self.connection_test()
+
+ # Test DNS
+ results['dns'] = self.dns_test()
+
+ # Tests de ping
+ results['ping_google'] = self.ping_test("8.8.8.8")
+ results['ping_cloudflare'] = self.ping_test("1.1.1.1")
+
+ # Test de vitesse de téléchargement (rapide)
+ results['download'] = self.download_speed_test(duration=3)
+
+ # Test de vitesse d'upload (rapide)
+ results['upload'] = self.upload_speed_test(data_size=512 * 1024)
+
+ return results
diff --git a/gui/__init__.py b/gui/__init__.py
new file mode 100644
index 0000000..89f9476
--- /dev/null
+++ b/gui/__init__.py
@@ -0,0 +1 @@
+# Module GUI
diff --git a/gui/app.py b/gui/app.py
new file mode 100644
index 0000000..c8aff57
--- /dev/null
+++ b/gui/app.py
@@ -0,0 +1,1452 @@
+"""
+StabilityTester - Interface Graphique avec PySide6
+Application moderne pour surveiller la stabilite de connexion internet
+"""
+
+import sys
+import os
+from datetime import datetime, timedelta
+import threading
+import time
+import json
+
+# ============================================================================
+# CONFIGURATION MATPLOTLIB - Doit être fait AVANT tout autre import de matplotlib
+# ============================================================================
+# Forcer matplotlib à utiliser PySide6 comme backend Qt
+# C'est CRUCIAL pour éviter les erreurs : "Failed to import PyQt5, PySide2"
+
+# Méthode 1: Définir la variable d'environnement (la plus fiable)
+os.environ.setdefault('MPLBACKEND', 'QtAgg')
+
+# Méthode 2: Importer matplotlib et définir le backend
+import matplotlib
+
+# Essayer les backends compatibles avec PySide6 (Qt6)
+# Note: QtAgg est compatible avec Qt5 et Qt6
+backends_to_try = ["QtAgg"]
+for backend in backends_to_try:
+ try:
+ matplotlib.use(backend)
+ break
+ except Exception as e:
+ print(f"Warning: Cannot use matplotlib backend {backend}: {e}")
+ continue
+else:
+ # Si QtAgg échoue, essayer Qt5Agg
+ try:
+ matplotlib.use("Qt5Agg")
+ print("Using Qt5Agg backend (Qt5)")
+ except Exception:
+ # Fallback sans Qt (pour les graphiques non-interactifs)
+ matplotlib.use("Agg")
+ print("Using Agg backend (non-interactive)")
+
+# Maintenant importer les modules matplotlib de base
+import matplotlib.pyplot as plt
+from matplotlib.figure import Figure
+
+# ============================================================================
+# IMPORT PYSIDE6 - Doit être fait APRES la configuration de matplotlib
+# ============================================================================
+try:
+ from PySide6.QtWidgets import (
+ QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
+ QLabel, QPushButton, QTabWidget, QFrame, QGroupBox,
+ QRadioButton, QCheckBox, QButtonGroup, QSpinBox, QDoubleSpinBox,
+ QTextEdit, QTreeWidget, QTreeWidgetItem, QScrollArea, QFileDialog,
+ QMessageBox, QProgressBar, QSizePolicy, QSplitter, QStyleFactory
+ )
+ from PySide6.QtCore import Qt, QTimer
+ from PySide6.QtGui import (
+ QFont, QColor, QPalette, QPainter, QLinearGradient, QBrush, QPen, QIcon
+ )
+
+ # ========================================================================
+ # IMPORT MATPLOTLIB QT BACKEND - Doit être fait APRES PySide6
+ # ========================================================================
+ # Avec PySide6 (Qt6) le backend est backend_qtagg ; backend_qt5agg sert de repli.
+ try:
+ from matplotlib.backends.backend_qtagg import FigureCanvasQTAgg
+ except ImportError:
+ try:
+ from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg
+ except ImportError:
+ # Fallback pour les systèmes sans Qt
+ print("Warning: No Qt backend available for matplotlib, using Agg")
+ from matplotlib.backends.backend_agg import FigureCanvasAgg as FigureCanvasQTAgg
+
+ PYQt_AVAILABLE = True
+
+except ImportError as e:
+ print(f"Erreur: PySide6 non disponible: {e}")
+ print("Installez PySide6 avec: pip install PySide6")
+ print("Si le problème persiste, essayez: pip install --upgrade PySide6 matplotlib")
+ PYQt_AVAILABLE = False
+ sys.exit(1)
+
+# Import de la logique metier
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from core.network_tester import NetworkTester, TestResult
+from monitor.monitor import StabilityMonitor, MonitorConfig, Alert, AlertLevel, AlertHandler
+
+
+class StabilityScoreBar(QWidget):
+ """Barre de progression personnalisée pour le score de stabilité"""
+
+ def __init__(self, parent=None):
+ super().__init__(parent)
+ self.score = 0
+ self.level = "Inconnu"
+ self._setup_ui()
+
+ def _setup_ui(self):
+ layout = QHBoxLayout(self)
+ layout.setContentsMargins(0, 0, 0, 0)
+ layout.setSpacing(10)
+
+ # Label du score
+ self.score_label = QLabel("Score: 0/100")
+ self.score_label.setFont(QFont("Segoe UI", 10, QFont.Weight.Bold))
+ layout.addWidget(self.score_label)
+
+ # Barre de progression
+ self.progress_bar = QProgressBar()
+ self.progress_bar.setRange(0, 100)
+ self.progress_bar.setValue(0)
+ self.progress_bar.setTextVisible(False)
+ self.progress_bar.setFixedHeight(20)
+
+ # Style personnalisé
+ self.progress_bar.setStyleSheet("""
+ QProgressBar {
+ border: 1px solid #ccc;
+ border-radius: 5px;
+ background-color: #f0f0f0;
+ text-align: center;
+ }
+ QProgressBar::chunk {
+ border-radius: 4px;
+ margin: 0.5px;
+ }
+ """)
+ layout.addWidget(self.progress_bar, 1)
+
+ # Label du niveau
+ self.level_label = QLabel("Niveau: Inconnu")
+ self.level_label.setFont(QFont("Segoe UI", 10, QFont.Weight.Bold))
+ layout.addWidget(self.level_label)
+
+ self.setLayout(layout)
+
+ def set_score(self, score, level="Inconnu", color="gray"):
+ """Met à jour le score et le niveau"""
+ self.score = score
+ self.level = level
+ self.score_label.setText(f"Score: {score:.1f}/100")
+ self.level_label.setText(f"Niveau: {level}")
+ self.progress_bar.setValue(int(score))
+
+ # Définir la couleur de la barre
+ colors = {
+ "green": "#5cb85c",
+ "orange": "#f0ad4e",
+ "red": "#d9534f",
+ "gray": "#999999"
+ }
+ color_value = colors.get(color, "#999999")
+
+ # Appliquer le style avec la couleur
+ style = f"""
+ QProgressBar::chunk {{
+ background-color: {color_value};
+ }}
+ """
+ self.progress_bar.setStyleSheet("""
+ QProgressBar {
+ border: 1px solid #ccc;
+ border-radius: 5px;
+ background-color: #f0f0f0;
+ text-align: center;
+ }
+ """ + style)
+
+
+class DashboardTab(QWidget):
+ """Onglet Tableau de bord"""
+
+ def __init__(self, monitor, parent=None):
+ super().__init__(parent)
+ self.monitor = monitor
+ self._setup_ui()
+
+ def _setup_ui(self):
+ layout = QVBoxLayout()
+ layout.setContentsMargins(15, 15, 15, 15)
+ layout.setSpacing(15)
+
+ # Frame des indicateurs principaux
+ indicators_frame = QFrame()
+ indicators_layout = QHBoxLayout()
+ indicators_layout.setSpacing(20)
+
+ # Statut de connexion
+ self.connection_status_label = QLabel("Statut: Inconnu")
+ self.connection_status_label.setFont(QFont("Segoe UI", 14, QFont.Weight.Bold))
+ indicators_layout.addWidget(self.connection_status_label)
+
+ # Temps de disponibilité
+ self.uptime_label = QLabel("Disponibilité: 0s")
+ self.uptime_label.setFont(QFont("Segoe UI", 12))
+ indicators_layout.addWidget(self.uptime_label)
+
+ # Temps d'indisponibilité
+ self.downtime_label = QLabel("Indisponibilité: 0s")
+ self.downtime_label.setFont(QFont("Segoe UI", 12))
+ indicators_layout.addWidget(self.downtime_label)
+
+ # Pourcentage de disponibilité
+ self.uptime_pct_label = QLabel("Dispo: 100%")
+ self.uptime_pct_label.setFont(QFont("Segoe UI", 12))
+ indicators_layout.addWidget(self.uptime_pct_label)
+
+ indicators_frame.setLayout(indicators_layout)
+ layout.addWidget(indicators_frame)
+
+ # Barre de score de stabilité
+ self.score_bar = StabilityScoreBar()
+ layout.addWidget(self.score_bar)
+
+ # Séparateur
+ layout.addWidget(QFrame()) # Espace vide
+
+ # Métriques principales
+ metrics_frame = QFrame()
+ metrics_layout = QHBoxLayout()
+ metrics_layout.setSpacing(10)
+
+ # Latence
+ latency_group = QGroupBox("Latence")
+ latency_vbox = QVBoxLayout()
+ self.latency_label = QLabel("-- ms")
+ self.latency_label.setFont(QFont("Segoe UI", 14, QFont.Weight.Bold))
+ self.latency_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
+ latency_vbox.addWidget(self.latency_label)
+ latency_group.setLayout(latency_vbox)
+ metrics_layout.addWidget(latency_group, 1)
+
+ # Download
+ download_group = QGroupBox("Download")
+ download_vbox = QVBoxLayout()
+ self.download_label = QLabel("-- Mbps")
+ self.download_label.setFont(QFont("Segoe UI", 14, QFont.Weight.Bold))
+ self.download_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
+ download_vbox.addWidget(self.download_label)
+ download_group.setLayout(download_vbox)
+ metrics_layout.addWidget(download_group, 1)
+
+ # Upload
+ upload_group = QGroupBox("Upload")
+ upload_vbox = QVBoxLayout()
+ self.upload_label = QLabel("-- Mbps")
+ self.upload_label.setFont(QFont("Segoe UI", 14, QFont.Weight.Bold))
+ self.upload_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
+ upload_vbox.addWidget(self.upload_label)
+ upload_group.setLayout(upload_vbox)
+ metrics_layout.addWidget(upload_group, 1)
+
+ # Jitter
+ jitter_group = QGroupBox("Jitter")
+ jitter_vbox = QVBoxLayout()
+ self.jitter_label = QLabel("-- ms")
+ self.jitter_label.setFont(QFont("Segoe UI", 14, QFont.Weight.Bold))
+ self.jitter_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
+ jitter_vbox.addWidget(self.jitter_label)
+ jitter_group.setLayout(jitter_vbox)
+ metrics_layout.addWidget(jitter_group, 1)
+
+ metrics_frame.setLayout(metrics_layout)
+ layout.addWidget(metrics_frame)
+
+ self.setLayout(layout)
+
+ def update_status(self, status):
+ """Met à jour les indicateurs de statut"""
+ if status.get('connected', False):
+ self.connection_status_label.setText("Statut: Connecté")
+ else:
+ self.connection_status_label.setText("Statut: Déconnecté")
+
+ self.uptime_label.setText(f"Disponibilité: {self._format_timedelta(status.get('uptime', timedelta(0)))}")
+ self.downtime_label.setText(f"Indisponibilité: {self._format_timedelta(status.get('downtime', timedelta(0)))}")
+
+ total = status['uptime'].total_seconds() + status['downtime'].total_seconds()
+ uptime_pct = (status['uptime'].total_seconds() / total * 100) if total > 0 else 100
+
+ if uptime_pct > 95:
+ color = "#5cb85c"
+ elif uptime_pct > 80:
+ color = "#f0ad4e"
+ else:
+ color = "#d9534f"
+
+ self.uptime_pct_label.setText(f"Dispo: {uptime_pct:.1f}%")
+
+ def update_metrics(self, result):
+ """Met à jour les métriques"""
+ if result.latency_ms is not None:
+ self.latency_label.setText(f"{result.latency_ms:.1f} ms")
+ if result.jitter is not None:
+ self.jitter_label.setText(f"{result.jitter:.1f} ms")
+ if result.download_speed is not None:
+ self.download_label.setText(f"{result.download_speed:.2f} Mbps")
+ if result.upload_speed is not None:
+ self.upload_label.setText(f"{result.upload_speed:.2f} Mbps")
+
+ def update_score(self, score_data):
+ """Met à jour la barre de score"""
+ score = score_data.get('score', 0)
+ level = score_data.get('stability_level', 'Inconnu')
+ color = score_data.get('color', 'gray')
+ self.score_bar.set_score(score, level, color)
+
+ def _format_timedelta(self, td):
+ ts = int(td.total_seconds())
+ if ts < 60:
+ return f"{ts}s"
+ elif ts < 3600:
+ return f"{ts//60}m {ts%60}s"
+ else:
+ return f"{ts//3600}h {(ts%3600)//60}m"
+
+
+class TestsTab(QWidget):
+ """Onglet Tests"""
+
+ def __init__(self, monitor, tester, parent=None):
+ super().__init__(parent)
+ self.monitor = monitor
+ self.tester = tester
+ self._setup_ui()
+
+ def _setup_ui(self):
+ layout = QVBoxLayout()
+ layout.setContentsMargins(15, 15, 15, 15)
+ layout.setSpacing(15)
+
+ # Frame de contrôle
+ control_frame = QFrame()
+ control_layout = QHBoxLayout()
+
+ # Sélection du type de test
+ test_label = QLabel("Type de test:")
+ control_layout.addWidget(test_label)
+
+ self.test_type_group = QButtonGroup()
+ test_types = [
+ ("Complet", "comprehensive"),
+ ("Ping", "ping"),
+ ("Connexion", "connection"),
+ ("DNS", "dns"),
+ ("Download", "download"),
+ ("Upload", "upload"),
+ ("Speedtest", "full_speed"),
+ ("Ping tous", "ping_all")
+ ]
+
+ for label, value in test_types:
+ rb = QRadioButton(label)
+ rb.setProperty("test_value", value)
+ self.test_type_group.addButton(rb)
+ control_layout.addWidget(rb)
+
+ # Bouton Exécuter
+ self.run_button = QPushButton("Exécuter")
+ self.run_button.setStyleSheet("""
+ QPushButton {
+ background-color: #007bff;
+ color: white;
+ border: none;
+ padding: 8px 16px;
+ border-radius: 4px;
+ }
+ QPushButton:hover {
+ background-color: #0056b3;
+ }
+ """)
+ self.run_button.clicked.connect(self._run_selected_test)
+ control_layout.addWidget(self.run_button)
+
+ control_frame.setLayout(control_layout)
+ layout.addWidget(control_frame)
+
+ # Zone de résultats
+ self.result_text = QTextEdit()
+ self.result_text.setReadOnly(True)
+ self.result_text.setFont(QFont("Courier New", 10))
+ layout.addWidget(self.result_text)
+
+ self.setLayout(layout)
+
+ def _run_selected_test(self):
+ """Exécute le test sélectionné"""
+ test_type = self.test_type_group.checkedButton()
+ if not test_type:
+ return
+
+ test_value = test_type.property("test_value")
+ self.result_text.clear()
+ self.result_text.append("Démarrage du test...")
+
+ def run():
+ try:
+ if test_value == "comprehensive":
+ results = self.monitor.run_single_test()
+ elif test_value == "ping":
+ results = {'ping': self.tester.ping_test()}
+ elif test_value == "connection":
+ results = {'connection': self.tester.connection_test()}
+ elif test_value == "dns":
+ results = {'dns': self.tester.dns_test()}
+ elif test_value == "download":
+ results = {'download': self.tester.download_speed_test(duration=3)}
+ elif test_value == "upload":
+ results = {'upload': self.tester.upload_speed_test(data_size=256*1024)}
+ elif test_value == "full_speed":
+ results = {'full_speed': self.tester.full_speed_test()}
+ elif test_value == "ping_all":
+ results = {f'ping_{i}': r for i, r in enumerate(self.tester.ping_all_servers())}
+ else:
+ results = {}
+
+ for test_name, result in results.items():
+ self.result_text.append(f"\n{'-'*40}")
+ self.result_text.append(f"Test: {result.test_type}")
+ self.result_text.append(f" Statut: {'OK' if result.success else 'ERROR'}")
+ self.result_text.append(f" Heure: {result.timestamp.strftime('%H:%M:%S')}")
+
+ if result.latency_ms is not None:
+ self.result_text.append(f" Latence: {result.latency_ms:.2f} ms")
+ if result.jitter is not None:
+ self.result_text.append(f" Jitter: {result.jitter:.2f} ms")
+ if result.download_speed is not None:
+ self.result_text.append(f" Download: {result.download_speed:.2f} Mbps")
+ if result.upload_speed is not None:
+ self.result_text.append(f" Upload: {result.upload_speed:.2f} Mbps")
+ if result.error_message:
+ self.result_text.append(f" Erreur: {result.error_message}")
+
+ self.result_text.append(f"\n{'='*40}")
+ self.result_text.append("Test terminé.")
+
+ except Exception as e:
+ self.result_text.append(f"Erreur: {str(e)}")
+
+ thread = threading.Thread(target=run, daemon=True)
+ thread.start()
+
+
+class HistoryTab(QWidget):
+ """Onglet Historique"""
+
+ def __init__(self, monitor, parent=None):
+ super().__init__(parent)
+ self.monitor = monitor
+ self._setup_ui()
+
+ def _setup_ui(self):
+ layout = QVBoxLayout()
+ layout.setContentsMargins(15, 15, 15, 15)
+ layout.setSpacing(10)
+
+ # Frame de contrôle
+ control_frame = QFrame()
+ control_layout = QHBoxLayout()
+
+ # Période
+ control_layout.addWidget(QLabel("Période:"))
+
+ self.period_group = QButtonGroup()
+ periods = [
+ ("Tout", "all"),
+ ("1h", "hour"),
+ ("24h", "day"),
+ ("Semaine", "week")
+ ]
+
+ for label, value in periods:
+ rb = QRadioButton(label)
+ rb.setProperty("period_value", value)
+ self.period_group.addButton(rb)
+ control_layout.addWidget(rb)
+
+ # Bouton Rafraîchir
+ refresh_button = QPushButton("Rafraîchir")
+ refresh_button.clicked.connect(self._refresh)
+ control_layout.addWidget(refresh_button)
+
+ control_frame.setLayout(control_layout)
+ layout.addWidget(control_frame)
+
+ # Arbre de l'historique
+ self.tree = QTreeWidget()
+ self.tree.setColumnCount(6)
+ self.tree.setHeaderLabels(["Heure", "Type", "Statut", "Latence", "Download", "Upload"])
+ self.tree.setColumnWidth(0, 150)
+ self.tree.setColumnWidth(1, 120)
+ self.tree.setColumnWidth(2, 80)
+ self.tree.setColumnWidth(3, 100)
+ self.tree.setColumnWidth(4, 100)
+ self.tree.setColumnWidth(5, 100)
+
+ layout.addWidget(self.tree)
+ self.setLayout(layout)
+
+ # Timer pour rafraîchissement automatique
+ self.refresh_timer = QTimer()
+ self.refresh_timer.timeout.connect(self._refresh)
+ self.refresh_timer.start(5000) # 5 secondes
+
+ def _refresh(self):
+ """Rafraîchit l'historique"""
+ self.tree.clear()
+
+ now = datetime.now()
+ period = self.period_group.checkedButton()
+ if period:
+ period_value = period.property("period_value")
+ else:
+ period_value = "all"
+
+ for results in self.monitor.get_test_history():
+ for result in results.values():
+ if period_value == "hour" and (now - result.timestamp).total_seconds() > 3600:
+ continue
+ elif period_value == "day" and (now - result.timestamp).total_seconds() > 86400:
+ continue
+ elif period_value == "week" and (now - result.timestamp).total_seconds() > 604800:
+ continue
+
+ item = QTreeWidgetItem()
+ item.setText(0, result.timestamp.strftime('%H:%M:%S'))
+ item.setText(1, result.test_type)
+ item.setText(2, 'OK' if result.success else 'ERROR')
+
+ # Couleur selon le statut
+ if result.success:
+ item.setTextColor(2, QColor("#5cb85c"))
+ else:
+ item.setTextColor(2, QColor("#d9534f"))
+
+ # Valeurs
+ if result.latency_ms is not None:
+ item.setText(3, f"{result.latency_ms:.0f}ms")
+ if result.download_speed is not None:
+ item.setText(4, f"{result.download_speed:.1f}Mbps")
+ if result.upload_speed is not None:
+ item.setText(5, f"{result.upload_speed:.1f}Mbps")
+
+ self.tree.addTopLevelItem(item)
+
+
+class AlertsTab(QWidget):
+ """Onglet Alertes"""
+
+ def __init__(self, monitor, parent=None):
+ super().__init__(parent)
+ self.monitor = monitor
+ self._setup_ui()
+
+ def _setup_ui(self):
+ layout = QVBoxLayout()
+ layout.setContentsMargins(15, 15, 15, 15)
+ layout.setSpacing(10)
+
+ # Frame de contrôle
+ control_frame = QFrame()
+ control_layout = QHBoxLayout()
+
+ control_layout.addWidget(QLabel("Niveau:"))
+
+ self.level_group = QButtonGroup()
+ levels = [
+ ("Toutes", "all"),
+ ("Critique", "critical"),
+ ("Warning", "warning"),
+ ("Info", "info")
+ ]
+
+ for label, value in levels:
+ rb = QRadioButton(label)
+ rb.setProperty("level_value", value)
+ self.level_group.addButton(rb)
+ control_layout.addWidget(rb)
+
+ # Boutons
+ clear_button = QPushButton("Effacer")
+ clear_button.clicked.connect(self._clear_alerts)
+ control_layout.addWidget(clear_button)
+
+ refresh_button = QPushButton("Rafraîchir")
+ refresh_button.clicked.connect(self._refresh)
+ control_layout.addWidget(refresh_button)
+
+ control_frame.setLayout(control_layout)
+ layout.addWidget(control_frame)
+
+ # Arbre des alertes
+ self.tree = QTreeWidget()
+ self.tree.setColumnCount(5)
+ self.tree.setHeaderLabels(["Heure", "Niveau", "Message", "Type", "Valeur"])
+ self.tree.setColumnWidth(0, 150)
+ self.tree.setColumnWidth(1, 80)
+ self.tree.setColumnWidth(2, 300)
+ self.tree.setColumnWidth(3, 120)
+ self.tree.setColumnWidth(4, 100)
+
+ layout.addWidget(self.tree)
+ self.setLayout(layout)
+
+ # Timer pour rafraîchissement automatique
+ self.refresh_timer = QTimer()
+ self.refresh_timer.timeout.connect(self._refresh)
+ self.refresh_timer.start(5000)
+
+ def _refresh(self):
+ """Rafraîchit les alertes"""
+ self.tree.clear()
+
+ level = self.level_group.checkedButton()
+ level_value = level.property("level_value") if level else "all"
+
+ for alert in self.monitor.alert_handler.get_alert_history():
+ if level_value != "all" and alert.level.value != level_value:
+ continue
+
+ item = QTreeWidgetItem()
+ item.setText(0, alert.timestamp.strftime('%H:%M:%S'))
+ item.setText(1, alert.level.value.upper())
+ item.setText(2, alert.message[:50])
+ item.setText(3, alert.test_type)
+ item.setText(4, f"{alert.value:.2f}" if alert.value else "--")
+
+ # Couleur selon le niveau
+ if alert.level == AlertLevel.CRITICAL:
+ item.setTextColor(1, QColor("#dc3545"))
+ elif alert.level == AlertLevel.WARNING:
+ item.setTextColor(1, QColor("#ffc107"))
+ else:
+ item.setTextColor(1, QColor("#17a2b8"))
+
+ self.tree.addTopLevelItem(item)
+
+ def _clear_alerts(self):
+ """Efface l'historique des alertes"""
+ reply = QMessageBox.question(
+ self, "Confirmer",
+ "Voulez-vous vraiment effacer l'historique des alertes?",
+ QMessageBox.Yes | QMessageBox.No
+ )
+ if reply == QMessageBox.Yes:
+ self.monitor.alert_handler.clear_history()
+ self._refresh()
+
+
+class SettingsTab(QWidget):
+ """Onglet Paramètres"""
+
+ def __init__(self, monitor, parent=None):
+ super().__init__(parent)
+ self.monitor = monitor
+ self._setup_ui()
+
+ def _setup_ui(self):
+ layout = QVBoxLayout()
+ layout.setContentsMargins(15, 15, 15, 15)
+ layout.setSpacing(15)
+
+ # Paramètres généraux
+ general_group = QGroupBox("Paramètres généraux")
+ general_layout = QVBoxLayout()
+
+ # Intervalle
+ interval_layout = QHBoxLayout()
+ interval_layout.addWidget(QLabel("Intervalle (secondes):"))
+ self.interval_spin = QSpinBox()
+ self.interval_spin.setRange(10, 3600)
+ self.interval_spin.setValue(self.monitor.config.interval)
+ interval_layout.addWidget(self.interval_spin)
+ general_layout.addLayout(interval_layout)
+
+ # Historique max
+ history_layout = QHBoxLayout()
+ history_layout.addWidget(QLabel("Historique maximal:"))
+ self.max_history_spin = QSpinBox()
+ self.max_history_spin.setRange(10, 1000)
+ self.max_history_spin.setValue(self.monitor.config.max_history)
+ history_layout.addWidget(self.max_history_spin)
+ general_layout.addLayout(history_layout)
+
+ general_group.setLayout(general_layout)
+ layout.addWidget(general_group)
+
+ # Seuils
+ thresholds_group = QGroupBox("Seuils d'alerte")
+ thresholds_layout = QVBoxLayout()
+
+ # Latence
+ ping_layout = QHBoxLayout()
+ ping_layout.addWidget(QLabel("Latence (ms):"))
+ self.ping_threshold_spin = QDoubleSpinBox()
+ self.ping_threshold_spin.setRange(1, 1000)
+ self.ping_threshold_spin.setValue(self.monitor.config.ping_threshold)
+ ping_layout.addWidget(self.ping_threshold_spin)
+ thresholds_layout.addLayout(ping_layout)
+
+ # Jitter
+ jitter_layout = QHBoxLayout()
+ jitter_layout.addWidget(QLabel("Jitter (ms):"))
+ self.jitter_threshold_spin = QDoubleSpinBox()
+ self.jitter_threshold_spin.setRange(1, 500)
+ self.jitter_threshold_spin.setValue(self.monitor.config.jitter_threshold)
+ jitter_layout.addWidget(self.jitter_threshold_spin)
+ thresholds_layout.addLayout(jitter_layout)
+
+ # Download
+ download_layout = QHBoxLayout()
+ download_layout.addWidget(QLabel("Download (Mbps):"))
+ self.download_threshold_spin = QDoubleSpinBox()
+ self.download_threshold_spin.setRange(0.1, 100)
+ self.download_threshold_spin.setValue(self.monitor.config.download_threshold)
+ download_layout.addWidget(self.download_threshold_spin)
+ thresholds_layout.addLayout(download_layout)
+
+ # Upload
+ upload_layout = QHBoxLayout()
+ upload_layout.addWidget(QLabel("Upload (Mbps):"))
+ self.upload_threshold_spin = QDoubleSpinBox()
+ self.upload_threshold_spin.setRange(0.1, 100)
+ self.upload_threshold_spin.setValue(self.monitor.config.upload_threshold)
+ upload_layout.addWidget(self.upload_threshold_spin)
+ thresholds_layout.addLayout(upload_layout)
+
+ thresholds_group.setLayout(thresholds_layout)
+ layout.addWidget(thresholds_group)
+
+ # Notifications
+ notification_group = QGroupBox("Notifications")
+ notification_layout = QVBoxLayout()
+
+ self.enable_sound_check = QCheckBox("Alertes sonores")
+ self.enable_sound_check.setChecked(self.monitor.config.enable_sound)
+ notification_layout.addWidget(self.enable_sound_check)
+
+ self.enable_notifications_check = QCheckBox("Notifications système")
+ self.enable_notifications_check.setChecked(self.monitor.config.enable_notifications)
+ notification_layout.addWidget(self.enable_notifications_check)
+
+ notification_group.setLayout(notification_layout)
+ layout.addWidget(notification_group)
+
+ # Types de tests
+ tests_group = QGroupBox("Types de tests à exécuter")
+ tests_layout = QVBoxLayout()
+
+ self.test_checks = {}
+ test_types = [
+ ("Ping", "ping"),
+ ("Connexion HTTP", "connection"),
+ ("Résolution DNS", "dns"),
+ ("Test Download", "download"),
+ ("Test Upload", "upload"),
+ ("Speedtest complet", "full_speed")
+ ]
+
+ for label, key in test_types:
+ cb = QCheckBox(label)
+ cb.setChecked(key in self.monitor.config.test_types)
+ self.test_checks[key] = cb
+ tests_layout.addWidget(cb)
+
+ tests_group.setLayout(tests_layout)
+ layout.addWidget(tests_group)
+
+ # Boutons
+ button_layout = QHBoxLayout()
+ apply_button = QPushButton("Appliquer")
+ apply_button.clicked.connect(self._apply_settings)
+ button_layout.addWidget(apply_button)
+
+ reset_button = QPushButton("Réinitialiser")
+ reset_button.clicked.connect(self._reset_settings)
+ button_layout.addWidget(reset_button)
+
+ layout.addLayout(button_layout)
+ self.setLayout(layout)
+
+ def _apply_settings(self):
+ """Applique les paramètres"""
+ try:
+ test_types = [
+ key for key, cb in self.test_checks.items()
+ if cb.isChecked()
+ ]
+
+ new_config = MonitorConfig(
+ interval=self.interval_spin.value(),
+ ping_threshold=self.ping_threshold_spin.value(),
+ download_threshold=self.download_threshold_spin.value(),
+ upload_threshold=self.upload_threshold_spin.value(),
+ jitter_threshold=self.jitter_threshold_spin.value(),
+ enable_sound=self.enable_sound_check.isChecked(),
+ enable_notifications=self.enable_notifications_check.isChecked(),
+ max_history=self.max_history_spin.value(),
+ test_types=test_types
+ )
+
+ self.monitor.config = new_config
+ self.monitor.alert_handler = AlertHandler(new_config)
+ QMessageBox.information(self, "Succès", "Paramètres appliqués!")
+
+ except Exception as e:
+ QMessageBox.critical(self, "Erreur", f"Erreur: {str(e)}")
+
+ def _reset_settings(self):
+ """Réinitialise les paramètres"""
+ reply = QMessageBox.question(
+ self, "Confirmer",
+ "Voulez-vous vraiment réinitialiser les paramètres?",
+ QMessageBox.Yes | QMessageBox.No
+ )
+ if reply == QMessageBox.Yes:
+ self.interval_spin.setValue(60)
+ self.max_history_spin.setValue(100)
+ self.ping_threshold_spin.setValue(100.0)
+ self.jitter_threshold_spin.setValue(50.0)
+ self.download_threshold_spin.setValue(5.0)
+ self.upload_threshold_spin.setValue(2.0)
+ self.enable_sound_check.setChecked(True)
+ self.enable_notifications_check.setChecked(True)
+ for cb in self.test_checks.values():
+ cb.setChecked(True)
+
+
+class StatisticsTab(QWidget):
+ """Onglet Statistiques avec graphiques"""
+
+ def __init__(self, monitor, parent=None):
+ super().__init__(parent)
+ self.monitor = monitor
+ self._setup_ui()
+
+ def _setup_ui(self):
+ layout = QVBoxLayout()
+ layout.setContentsMargins(15, 15, 15, 15)
+ layout.setSpacing(15)
+
+ # Statistiques globales
+ stats_frame = QFrame()
+ stats_layout = QHBoxLayout()
+ stats_layout.setSpacing(20)
+
+ self.total_tests_label = QLabel("Tests: 0")
+ self.total_tests_label.setFont(QFont("Segoe UI", 12))
+ stats_layout.addWidget(self.total_tests_label)
+
+ self.success_rate_label = QLabel("Succès: 0%")
+ self.success_rate_label.setFont(QFont("Segoe UI", 12))
+ stats_layout.addWidget(self.success_rate_label)
+
+ self.connection_loss_label = QLabel("Pertes: 0")
+ self.connection_loss_label.setFont(QFont("Segoe UI", 12))
+ stats_layout.addWidget(self.connection_loss_label)
+
+ self.avg_latency_label = QLabel("Latence avg: 0ms")
+ self.avg_latency_label.setFont(QFont("Segoe UI", 12))
+ stats_layout.addWidget(self.avg_latency_label)
+
+ stats_frame.setLayout(stats_layout)
+ layout.addWidget(stats_frame)
+
+ # Onglets des graphiques
+ self.chart_tabs = QTabWidget()
+ layout.addWidget(self.chart_tabs)
+
+ # Graphique de latence
+ self._create_latency_chart()
+
+ # Graphique de vitesse
+ self._create_speed_chart()
+
+ # Graphique de connectivité
+ self._create_connectivity_chart()
+
+ self.setLayout(layout)
+
+ # Timer pour mise à jour
+ self.update_timer = QTimer()
+ self.update_timer.timeout.connect(self._update_statistics)
+ self.update_timer.start(5000)
+
+ def _create_latency_chart(self):
+ """Crée le graphique de latence"""
+ chart_widget = QWidget()
+ chart_layout = QVBoxLayout()
+
+ self.latency_fig = Figure(figsize=(8, 4), dpi=100)
+ self.latency_ax = self.latency_fig.add_subplot(111)
+ self.latency_ax.set_title("Latence (ms)")
+ self.latency_ax.set_ylim(0, 500)
+ self.latency_ax.grid(True, alpha=0.3)
+
+ self.latency_canvas = FigureCanvasQTAgg(self.latency_fig)
+ chart_layout.addWidget(self.latency_canvas)
+
+ chart_widget.setLayout(chart_layout)
+ self.chart_tabs.addTab(chart_widget, "Latence")
+
+ self.latency_times = []
+ self.latency_values = []
+
+ def _create_speed_chart(self):
+ """Crée le graphique de vitesse"""
+ chart_widget = QWidget()
+ chart_layout = QVBoxLayout()
+
+ self.speed_fig = Figure(figsize=(8, 4), dpi=100)
+ self.speed_ax = self.speed_fig.add_subplot(111)
+ self.speed_ax.set_title("Vitesses (Mbps)")
+ self.speed_ax.set_ylim(0, 100)
+ self.speed_ax.grid(True, alpha=0.3)
+
+ self.speed_canvas = FigureCanvasQTAgg(self.speed_fig)
+ chart_layout.addWidget(self.speed_canvas)
+
+ chart_widget.setLayout(chart_layout)
+ self.chart_tabs.addTab(chart_widget, "Vitesse")
+
+ self.speed_times = []
+ self.download_values = []
+ self.upload_values = []
+
+ def _create_connectivity_chart(self):
+ """Crée le graphique de connectivité"""
+ chart_widget = QWidget()
+ chart_layout = QVBoxLayout()
+
+ self.connectivity_fig = Figure(figsize=(8, 4), dpi=100)
+ self.connectivity_ax = self.connectivity_fig.add_subplot(111)
+ self.connectivity_ax.set_title("Connectivité")
+ self.connectivity_ax.set_ylim(-0.5, 1.5)
+ self.connectivity_ax.set_yticks([0, 1])
+ self.connectivity_ax.set_yticklabels(['Déconnecté', 'Connecté'])
+ self.connectivity_ax.grid(True, alpha=0.3)
+
+ self.connectivity_canvas = FigureCanvasQTAgg(self.connectivity_fig)
+ chart_layout.addWidget(self.connectivity_canvas)
+
+ chart_widget.setLayout(chart_layout)
+ self.chart_tabs.addTab(chart_widget, "Connectivité")
+
+ self.connectivity_times = []
+ self.connectivity_values = []
+
+ def _update_statistics(self):
+ """Met à jour les statistiques"""
+ try:
+ stats = self.monitor.get_statistics()
+ self.total_tests_label.setText(f"Tests: {stats['total_tests']}")
+ self.success_rate_label.setText(f"Succès: {stats['success_rate']:.1f}%")
+ self.connection_loss_label.setText(f"Pertes: {stats['connection_loss_count']}")
+ self.avg_latency_label.setText(f"Latence avg: {stats['avg_latency']:.1f}ms")
+
+ # Mettre à jour les graphiques
+ self._update_latency_chart()
+ self._update_speed_chart()
+ self._update_connectivity_chart()
+
+ except Exception as e:
+ print(f"Erreur dans _update_statistics: {e}")
+
+ def _update_latency_chart(self):
+ """Met à jour le graphique de latence"""
+ try:
+ self.latency_times.clear()
+ self.latency_values.clear()
+
+ for i, results in enumerate(self.monitor.get_test_history(limit=20)):
+ for result in results.values():
+ if result.latency_ms is not None:
+ self.latency_times.append(i)
+ self.latency_values.append(result.latency_ms)
+ break
+
+ if self.latency_times:
+ self.latency_ax.clear()
+ self.latency_ax.plot(
+ self.latency_times,
+ self.latency_values,
+ 'b-',
+ linewidth=2
+ )
+ self.latency_ax.fill_between(
+ self.latency_times,
+ self.latency_values,
+ alpha=0.3
+ )
+ self.latency_ax.axhline(
+ y=self.monitor.config.ping_threshold,
+ color='r',
+ linestyle='--'
+ )
+ self.latency_ax.set_title("Latence (ms)")
+ self.latency_ax.set_ylim(0, max(max(self.latency_values) * 1.2, 100))
+ self.latency_ax.grid(True, alpha=0.3)
+ self.latency_canvas.draw()
+ except Exception as e:
+ print(f"Erreur dans _update_latency_chart: {e}")
+
+ def _update_speed_chart(self):
+ """Met à jour le graphique de vitesse"""
+ try:
+ self.speed_times.clear()
+ self.download_values.clear()
+ self.upload_values.clear()
+
+ for i, results in enumerate(self.monitor.get_test_history(limit=20)):
+ for result in results.values():
+ if result.download_speed is not None:
+ self.speed_times.append(i)
+ self.download_values.append(result.download_speed)
+ if result.upload_speed is not None:
+ if len(self.upload_values) <= i:
+ self.upload_values.append(result.upload_speed)
+ else:
+ self.upload_values[i] = result.upload_speed
+
+ if self.speed_times:
+ self.speed_ax.clear()
+ if self.download_values:
+ self.speed_ax.plot(
+ self.speed_times,
+ self.download_values,
+ 'g-',
+ label='Download',
+ linewidth=2
+ )
+ if self.upload_values:
+ self.speed_ax.plot(
+ self.speed_times,
+ self.upload_values,
+ 'r-',
+ label='Upload',
+ linewidth=2
+ )
+
+ self.speed_ax.axhline(
+ y=self.monitor.config.download_threshold,
+ color='g',
+ linestyle='--',
+ alpha=0.5
+ )
+ self.speed_ax.axhline(
+ y=self.monitor.config.upload_threshold,
+ color='r',
+ linestyle='--',
+ alpha=0.5
+ )
+
+ self.speed_ax.set_title("Vitesses (Mbps)")
+ max_speed = max(
+ max(self.download_values or [0]) * 1.2,
+ max(self.upload_values or [0]) * 1.2,
+ 10
+ )
+ self.speed_ax.set_ylim(0, max_speed)
+ self.speed_ax.legend()
+ self.speed_ax.grid(True, alpha=0.3)
+ self.speed_canvas.draw()
+ except Exception as e:
+ print(f"Erreur dans _update_speed_chart: {e}")
+
+ def _update_connectivity_chart(self):
+ """Met à jour le graphique de connectivité"""
+ try:
+ ch = (self.monitor.connection_history[-20:]
+ if len(self.monitor.connection_history) > 20
+ else self.monitor.connection_history)
+
+ self.connectivity_times = list(range(len(ch)))
+ self.connectivity_values = [1 if c else 0 for c in ch]
+
+ if self.connectivity_times:
+ self.connectivity_ax.clear()
+ self.connectivity_ax.step(
+ self.connectivity_times,
+ self.connectivity_values,
+ where='post',
+ color='blue',
+ linewidth=2
+ )
+
+ for i in range(len(self.connectivity_times) - 1):
+ color = 'red' if self.connectivity_values[i] == 0 else 'green'
+ self.connectivity_ax.axvspan(
+ self.connectivity_times[i],
+ self.connectivity_times[i + 1],
+ facecolor=color,
+ alpha=0.3
+ )
+
+ self.connectivity_ax.set_title("Connectivité")
+ self.connectivity_ax.set_ylim(-0.5, 1.5)
+ self.connectivity_ax.set_yticks([0, 1])
+ self.connectivity_ax.set_yticklabels(['Déconnecté', 'Connecté'])
+ self.connectivity_ax.grid(True, alpha=0.3)
+ self.connectivity_canvas.draw()
+ except Exception as e:
+ print(f"Erreur dans _update_connectivity_chart: {e}")
+
+
+class StabilityTesterWindow(QMainWindow):
+ """Fenêtre principale de l'application"""
+
+ def __init__(self):
+ super().__init__()
+ self.setWindowTitle("StabilityTester")
+ self.setGeometry(100, 100, 1000, 700)
+ self.setMinimumSize(800, 600)
+
+ # Initialisation
+ self.monitor = StabilityMonitor()
+ self.tester = NetworkTester()
+ self.running = False
+
+ # Configuration du style
+ self._configure_style()
+
+ # Création de l'interface
+ self._create_widgets()
+
+ # Connexion des signaux
+ self.monitor.register_status_callback(self._update_status)
+ self.monitor.register_alert_callback(self._show_alert)
+
+ # Timer pour mise à jour
+ self.update_timer = QTimer()
+ self.update_timer.timeout.connect(self._update_ui)
+ self.update_timer.start(1000) # 1 seconde
+
+ # Gestion de la fermeture
+ self.setWindowIconText("StabilityTester")
+
+ def _configure_style(self):
+ """Configure le style de l'application"""
+ # Utiliser le style Fusion pour un look moderne
+ QApplication.setStyle(QStyleFactory.create("Fusion"))
+
+ # Palette moderne
+ palette = QPalette()
+ palette.setColor(QPalette.Window, QColor(248, 249, 250))
+ palette.setColor(QPalette.WindowText, QColor(51, 51, 51))
+ palette.setColor(QPalette.Base, QColor(255, 255, 255))
+ palette.setColor(QPalette.AlternateBase, QColor(240, 240, 240))
+ palette.setColor(QPalette.ToolTipBase, QColor(255, 255, 255))
+ palette.setColor(QPalette.ToolTipText, QColor(0, 0, 0))
+ palette.setColor(QPalette.Text, QColor(51, 51, 51))
+ palette.setColor(QPalette.Button, QColor(240, 240, 240))
+ palette.setColor(QPalette.ButtonText, QColor(51, 51, 51))
+ palette.setColor(QPalette.BrightText, QColor(255, 255, 255))
+ palette.setColor(QPalette.Link, QColor(0, 100, 180))
+ palette.setColor(QPalette.Highlight, QColor(0, 120, 200))
+ palette.setColor(QPalette.HighlightedText, QColor(255, 255, 255))
+
+ QApplication.setPalette(palette)
+
+ def _create_widgets(self):
+ """Crée tous les widgets"""
+ # Widget central
+ central_widget = QWidget()
+ central_layout = QVBoxLayout()
+ central_layout.setContentsMargins(0, 0, 0, 0)
+ central_layout.setSpacing(0)
+
+ # Barre d'outils
+ self._create_toolbar(central_layout)
+
+ # Onglets
+ self.tab_widget = QTabWidget()
+ central_layout.addWidget(self.tab_widget)
+
+ # Créer les onglets
+ self.dashboard_tab = DashboardTab(self.monitor)
+ self.tab_widget.addTab(self.dashboard_tab, "📋 Tableau de bord")
+
+ self.tests_tab = TestsTab(self.monitor, self.tester)
+ self.tab_widget.addTab(self.tests_tab, "🧪 Tests")
+
+ self.history_tab = HistoryTab(self.monitor)
+ self.tab_widget.addTab(self.history_tab, "📜 Historique")
+
+ self.alerts_tab = AlertsTab(self.monitor)
+ self.tab_widget.addTab(self.alerts_tab, "⚠️ Alertes")
+
+ self.settings_tab = SettingsTab(self.monitor)
+ self.tab_widget.addTab(self.settings_tab, "⚙️ Paramètres")
+
+ self.statistics_tab = StatisticsTab(self.monitor)
+ self.tab_widget.addTab(self.statistics_tab, "📈 Statistiques")
+
+ central_widget.setLayout(central_layout)
+ self.setCentralWidget(central_widget)
+
+ # Mettre à jour l'UI
+ self._update_ui()
+
+ def _create_toolbar(self, layout):
+ """Crée la barre d'outils"""
+ toolbar = QWidget()
+ toolbar_layout = QHBoxLayout()
+ toolbar_layout.setContentsMargins(15, 10, 15, 10)
+ toolbar_layout.setSpacing(10)
+
+ # Bouton Démarrer
+ self.start_button = QPushButton("🟢 Démarrer")
+ self.start_button.setFixedHeight(35)
+ self.start_button.clicked.connect(self._start_monitoring)
+ toolbar_layout.addWidget(self.start_button)
+
+ # Bouton Arrêter
+ self.stop_button = QPushButton("🔴 Arrêter")
+ self.stop_button.setFixedHeight(35)
+ self.stop_button.setEnabled(False)
+ self.stop_button.clicked.connect(self._stop_monitoring)
+ toolbar_layout.addWidget(self.stop_button)
+
+ # Bouton Test unique
+ test_button = QPushButton("🧪 Test unique")
+ test_button.setFixedHeight(35)
+ test_button.clicked.connect(self._run_single_test)
+ toolbar_layout.addWidget(test_button)
+
+ # Séparateur
+ toolbar_layout.addStretch(1)
+
+ # Boutons Exporter, Sauvegarder, Charger
+ export_button = QPushButton("📤 Exporter")
+ export_button.setFixedHeight(35)
+ export_button.clicked.connect(self._export_data)
+ toolbar_layout.addWidget(export_button)
+
+ save_button = QPushButton("💾 Sauvegarder")
+ save_button.setFixedHeight(35)
+ save_button.clicked.connect(self._save_state)
+ toolbar_layout.addWidget(save_button)
+
+ load_button = QPushButton("📂 Charger")
+ load_button.setFixedHeight(35)
+ load_button.clicked.connect(self._load_state)
+ toolbar_layout.addWidget(load_button)
+
+ # Séparateur
+ toolbar_layout.addStretch(1)
+
+ # Bouton Minimiser
+ minimize_button = QPushButton("📉 Minimiser")
+ minimize_button.setFixedHeight(35)
+ minimize_button.clicked.connect(self.showMinimized)
+ toolbar_layout.addWidget(minimize_button)
+
+ # Bouton Quitter
+ quit_button = QPushButton("❌ Quitter")
+ quit_button.setFixedHeight(35)
+ quit_button.clicked.connect(self.close)
+ toolbar_layout.addWidget(quit_button)
+
+ # Indicateurs d'état
+ toolbar_layout.addStretch(1)
+
+ self.connection_indicator = QLabel("● Déconnecté")
+ self.connection_indicator.setStyleSheet("color: #d9534f; font-weight: bold;")
+ toolbar_layout.addWidget(self.connection_indicator)
+
+ self.monitor_indicator = QLabel("● Surveillance: Inactive")
+ self.monitor_indicator.setStyleSheet("color: #999; font-weight: bold;")
+ toolbar_layout.addWidget(self.monitor_indicator)
+
+ toolbar.setLayout(toolbar_layout)
+ layout.addWidget(toolbar)
+
+ def _start_monitoring(self):
+ """Démarre la surveillance"""
+ if not self.running:
+ self.running = True
+ self.start_button.setEnabled(False)
+ self.stop_button.setEnabled(True)
+ self.monitor_indicator.setText("● Surveillance: Active")
+ self.monitor_indicator.setStyleSheet("color: #5cb85c; font-weight: bold;")
+ self.monitor.start()
+
+ def _stop_monitoring(self):
+ """Arrête la surveillance"""
+ if self.running:
+ self.running = False
+ self.start_button.setEnabled(True)
+ self.stop_button.setEnabled(False)
+ self.monitor_indicator.setText("● Surveillance: Inactive")
+ self.monitor_indicator.setStyleSheet("color: #999; font-weight: bold;")
+ self.monitor.stop()
+
+ def _run_single_test(self):
+ """Effectue un test unique"""
+ def run():
+ try:
+ results = self.monitor.run_single_test()
+ for test_name, result in results.items():
+ self.dashboard_tab.update_metrics(result)
+ if result.test_type in ["connection", "ping"]:
+ if result.success:
+ self.connection_indicator.setText("● Connecté")
+ self.connection_indicator.setStyleSheet("color: #5cb85c; font-weight: bold;")
+ else:
+ self.connection_indicator.setText("● Déconnecté")
+ self.connection_indicator.setStyleSheet("color: #d9534f; font-weight: bold;")
+ except Exception as e:
+ print(f"Erreur lors du test: {e}")
+
+ thread = threading.Thread(target=run, daemon=True)
+ thread.start()
+
+ def _update_status(self, status):
+ """Met à jour l'état"""
+ self.dashboard_tab.update_status(status)
+ if status.get('connected', False):
+ self.connection_indicator.setText("● Connecté")
+ self.connection_indicator.setStyleSheet("color: #5cb85c; font-weight: bold;")
+ else:
+ self.connection_indicator.setText("● Déconnecté")
+ self.connection_indicator.setStyleSheet("color: #d9534f; font-weight: bold;")
+
+ def _show_alert(self, alert):
+ """Affiche une alerte"""
+ print(f"[ALERTE {alert.level.value.upper()}] {alert.message}")
+ # Clignoter l'onglet Alertes
+ if self.tab_widget.currentIndex() != 3:
+ self._blink_tab(3)
+
+ def _blink_tab(self, tab_index):
+ """Fait clignoter un onglet"""
+ original_text = self.tab_widget.tabText(tab_index)
+ for _ in range(3):
+ self.tab_widget.setTabText(tab_index, f"⚠️ {original_text}")
+ QApplication.processEvents()
+ time.sleep(0.3)
+ self.tab_widget.setTabText(tab_index, original_text)
+ QApplication.processEvents()
+ time.sleep(0.3)
+
+ def _update_ui(self):
+ """Met à jour l'interface"""
+ try:
+ # Mettre à jour le score de stabilité
+ score_data = self.monitor.get_stability_score()
+ self.dashboard_tab.update_score(score_data)
+
+ # Mettre à jour les statistiques
+ self.statistics_tab._update_statistics()
+
+ # Mettre à jour l'historique
+ self.history_tab._refresh()
+
+ # Mettre à jour les alertes
+ self.alerts_tab._refresh()
+
+ except Exception as e:
+ print(f"Erreur dans _update_ui: {e}")
+
+ def _export_data(self):
+ """Exporte les données"""
+ filepath, _ = QFileDialog.getSaveFileName(
+ self, "Exporter les données",
+ "", "JSON Files (*.json);;All Files (*)"
+ )
+ if filepath:
+ try:
+ data = {
+ 'config': self.monitor.config.to_dict(),
+ 'status': self.monitor.get_current_status(),
+ 'stats': self.monitor.get_statistics(),
+ 'history': [
+ {n: r.to_dict() for n, r in res.items()}
+ for res in self.monitor.get_test_history()
+ ],
+ 'alerts': [
+ a.to_dict() for a in self.monitor.alert_handler.get_alert_history()
+ ],
+ 'exported_at': datetime.now().isoformat()
+ }
+ with open(filepath, 'w') as f:
+ json.dump(data, f, indent=2)
+ QMessageBox.information(self, "Succès", f"Données exportées vers {filepath}")
+ except Exception as e:
+ QMessageBox.critical(self, "Erreur", f"Erreur lors de l'export: {str(e)}")
+
+ def _save_state(self):
+ """Sauvegarde l'état"""
+ try:
+ self.monitor.save_state()
+ QMessageBox.information(self, "Succès", "État sauvegardé!")
+ except Exception as e:
+ QMessageBox.critical(self, "Erreur", f"Erreur lors de la sauvegarde: {str(e)}")
+
+ def _load_state(self):
+ """Charge l'état"""
+ try:
+ if self.monitor.load_state():
+ QMessageBox.information(self, "Succès", "État chargé!")
+ self._update_ui()
+
+ # Mettre à jour les paramètres de l'interface
+ self.settings_tab.interval_spin.setValue(self.monitor.config.interval)
+ self.settings_tab.max_history_spin.setValue(self.monitor.config.max_history)
+ self.settings_tab.ping_threshold_spin.setValue(self.monitor.config.ping_threshold)
+ self.settings_tab.jitter_threshold_spin.setValue(self.monitor.config.jitter_threshold)
+ self.settings_tab.download_threshold_spin.setValue(self.monitor.config.download_threshold)
+ self.settings_tab.upload_threshold_spin.setValue(self.monitor.config.upload_threshold)
+ self.settings_tab.enable_sound_check.setChecked(self.monitor.config.enable_sound)
+ self.settings_tab.enable_notifications_check.setChecked(self.monitor.config.enable_notifications)
+ for key, cb in self.settings_tab.test_checks.items():
+ cb.setChecked(key in self.monitor.config.test_types)
+ else:
+ QMessageBox.information(self, "Info", "Aucun état trouvé, valeurs par défaut")
+ except Exception as e:
+ QMessageBox.critical(self, "Erreur", f"Erreur lors du chargement: {str(e)}")
+
+ def closeEvent(self, event):
+ """Gère la fermeture de la fenêtre"""
+ reply = QMessageBox.question(
+ self, "Quitter",
+ "Voulez-vous vraiment quitter StabilityTester?",
+ QMessageBox.Yes | QMessageBox.No
+ )
+ if reply == QMessageBox.Yes:
+ self._stop_monitoring()
+ event.accept()
+ else:
+ event.ignore()
+
+
+def run_application():
+ """Lance l'application PySide6"""
+ app = QApplication(sys.argv)
+ app.setApplicationName("StabilityTester")
+ app.setApplicationVersion("1.0.1")
+
+ window = StabilityTesterWindow()
+ window.show()
+
+ sys.exit(app.exec())
+
+
+if __name__ == "__main__":
+ run_application()
diff --git a/main.py b/main.py
new file mode 100644
index 0000000..4f4d699
--- /dev/null
+++ b/main.py
@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+"""
+StabilityTester - Application de surveillance de connexion internet
+Lancez ce fichier pour demarrer l'application GUI avec PySide6
+"""
+
+import sys
+import os
+
+# Sous Windows la sortie standard utilise cp1252 : les emojis des messages
+# ci-dessous la font echouer avec UnicodeEncodeError des qu'elle est
+# redirigee. En mode fenetre (executable PyInstaller) stdout vaut None.
+for _stream in (sys.stdout, sys.stderr):
+ if _stream is not None:
+ _stream.reconfigure(encoding="utf-8", errors="replace")
+
+# Ajouter le repertoire parent au path pour les imports
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+# Forcer le backend matplotlib avant d'importer l'interface
+# Cela évite les erreurs "Failed to import PyQt5, PySide2"
+import matplotlib
+matplotlib.use("QtAgg")
+
+# Essayer d'importer l'interface PySide6
+try:
+ from gui.app import run_application
+ print("✅ Utilisation de PySide6 pour l'interface graphique")
+ run_application()
+except ImportError as e:
+ print(f"❌ Erreur: Impossible de charger l'interface PySide6: {e}")
+ print("")
+ print("Pour résoudre ce problème:")
+ print("1. Installez les dépendances: pip install -r requirements.txt")
+ print("2. Assurez-vous que PySide6 est installé: pip install PySide6")
+ print("3. Essayez: pip install --upgrade PySide6 matplotlib")
+ print("")
+ print("Si vous utilisez un environnement virtuel, activez-le d'abord:")
+ print(" source .venv/bin/activate # Linux/macOS")
+ print(" .\\.venv\\Scripts\\activate # Windows")
+ sys.exit(1)
diff --git a/monitor/__init__.py b/monitor/__init__.py
new file mode 100644
index 0000000..46f9857
--- /dev/null
+++ b/monitor/__init__.py
@@ -0,0 +1 @@
+# Module de monitoring
diff --git a/monitor/monitor.py b/monitor/monitor.py
new file mode 100644
index 0000000..3dbf4f3
--- /dev/null
+++ b/monitor/monitor.py
@@ -0,0 +1,654 @@
+"""
+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
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..997dd58
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,23 @@
+# Dependencies for StabilityTester
+# Install with: pip install -r requirements.txt
+
+# Network testing
+requests>=2.25.0
+speedtest-cli>=2.1.3
+ping3>=4.0.3
+
+# GUI - PySide6 (Qt for Python)
+PySide6>=6.4.0
+
+# Matplotlib for graphs
+matplotlib>=3.4.0
+Pillow>=8.0.0
+
+# Optional for better notifications
+plyer>=2.0.0
+
+# Optional for system tray icon
+pystray>=0.19.0
+
+# For Windows sound support
+pywin32>=300.0; sys_platform == 'win32'
diff --git a/run.sh b/run.sh
new file mode 100755
index 0000000..a1cae78
--- /dev/null
+++ b/run.sh
@@ -0,0 +1,5 @@
+#!/bin/bash
+# Script de lancement pour StabilityTester
+
+cd "$(dirname "$0")"
+python3 main.py