v1.0.1 - corrige le demarrage et ajoute le build Windows

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.
This commit is contained in:
2026-09-11 08:12:33 +02:00
parent b47d67299c
commit 4adde135d7
16 changed files with 2967 additions and 0 deletions
+82
View File
@@ -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
+195
View File
@@ -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
+13
View File
@@ -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
+46
View File
@@ -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,
)
+11
View File
@@ -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"
+1
View File
@@ -0,0 +1 @@
# Configuration package
+13
View File
@@ -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"]
}
View File
+429
View File
@@ -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
+1
View File
@@ -0,0 +1 @@
# Module GUI
+1452
View File
File diff suppressed because it is too large Load Diff
+41
View File
@@ -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)
+1
View File
@@ -0,0 +1 @@
# Module de monitoring
+654
View File
@@ -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
+23
View File
@@ -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'
Executable
+5
View File
@@ -0,0 +1,5 @@
#!/bin/bash
# Script de lancement pour StabilityTester
cd "$(dirname "$0")"
python3 main.py