Add French docstrings and README

- Docstrings for all modules, classes and methods
- README.md with installation and usage instructions
- Update CLAUDE.md with dns.py

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-01-26 18:25:20 +01:00
parent 9f3c052e2a
commit 2c40f9c12c
8 changed files with 341 additions and 18 deletions

View File

@@ -24,7 +24,7 @@ python lan_checker.py
- `lan_checker.py` - Main script: config loading, MQTT client, check scheduler - `lan_checker.py` - Main script: config loading, MQTT client, check scheduler
- `checkers/` - Modular check implementations - `checkers/` - Modular check implementations
- `base.py` - `BaseChecker` abstract class and `CheckResult` dataclass - `base.py` - `BaseChecker` abstract class and `CheckResult` dataclass
- `ping.py`, `http.py`, `snmp.py` - Concrete checker implementations - `ping.py`, `http.py`, `dns.py`, `snmp.py` - Concrete checker implementations
- `config.yaml.example` - Configuration template (copy to `config.yaml`) - `config.yaml.example` - Configuration template (copy to `config.yaml`)
## Adding a New Checker ## Adding a New Checker

47
README.md Normal file
View File

@@ -0,0 +1,47 @@
# LAN Checker
Surveillance réseau avec reporting MQTT pour Home Assistant.
Vérifie périodiquement l'état de services et équipements réseau (ping, HTTP, DNS, SNMP), puis publie les résultats via MQTT. Les entités sont créées automatiquement dans Home Assistant grâce au protocole MQTT Discovery.
## Installation
```bash
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```
## Configuration
```bash
cp config.yaml.example config.yaml
# Éditer config.yaml avec vos équipements
```
## Usage
```bash
python lan_checker.py
python lan_checker.py -c /chemin/vers/config.yaml
```
## Types de checks
| Type | Description |
|------|-------------|
| `ping` | Vérifie la disponibilité via ICMP |
| `http` | Vérifie un service web (code HTTP) |
| `dns` | Vérifie un serveur DNS (ping + requête) |
| `snmp` | Vérifie un équipement via SNMP (+ température optionnelle) |
## Entités Home Assistant
Pour chaque équipement configuré :
- **binary_sensor** : état online/offline
- **sensor** : latence (ms)
- **sensor** : température (SNMP uniquement, si `temperature_oid` configuré)
## Licence
MIT

View File

@@ -1,3 +1,10 @@
"""
Module de base pour les checkers.
Définit la classe abstraite BaseChecker et le dataclass CheckResult
utilisés par tous les checkers.
"""
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any from typing import Any
@@ -5,6 +12,15 @@ from typing import Any
@dataclass @dataclass
class CheckResult: class CheckResult:
"""
Résultat d'une vérification.
Attributes:
success: True si la vérification a réussi, False sinon.
message: Message décrivant le résultat.
response_time: Temps de réponse en millisecondes (None si échec).
details: Informations supplémentaires (optionnel).
"""
success: bool success: bool
message: str message: str
response_time: float | None = None response_time: float | None = None
@@ -12,10 +28,34 @@ class CheckResult:
class BaseChecker(ABC): class BaseChecker(ABC):
"""
Classe abstraite de base pour tous les checkers.
Chaque checker doit hériter de cette classe et implémenter
la méthode check().
Attributes:
name: Nom du checker (pour l'affichage).
config: Configuration du checker (depuis le fichier YAML).
"""
def __init__(self, name: str, config: dict): def __init__(self, name: str, config: dict):
"""
Initialise le checker.
Args:
name: Nom du checker.
config: Dictionnaire de configuration.
"""
self.name = name self.name = name
self.config = config self.config = config
@abstractmethod @abstractmethod
def check(self) -> CheckResult: def check(self) -> CheckResult:
"""
Exécute la vérification.
Returns:
CheckResult contenant le résultat de la vérification.
"""
pass pass

View File

@@ -1,3 +1,10 @@
"""
Checker DNS.
Vérifie la disponibilité d'un serveur DNS et sa capacité à résoudre
des requêtes.
"""
import time import time
import dns.resolver import dns.resolver
@@ -7,13 +14,38 @@ from .ping import PingChecker
class DnsChecker(BaseChecker): class DnsChecker(BaseChecker):
"""
Vérifie la disponibilité d'un serveur DNS.
Effectue d'abord un ping pour vérifier que le serveur est joignable,
puis exécute une requête DNS configurable.
Configuration YAML:
host: Adresse IP du serveur DNS (obligatoire).
query: Nom de domaine à résoudre (obligatoire).
record_type: Type d'enregistrement DNS (défaut: A).
Valeurs possibles: A, AAAA, MX, TXT, CNAME, etc.
timeout: Délai d'attente en secondes (défaut: 5).
"""
def check(self) -> CheckResult: def check(self) -> CheckResult:
"""
Vérifie le serveur DNS.
Étapes:
1. Ping du serveur DNS pour vérifier sa disponibilité.
2. Requête DNS du type configuré.
Returns:
CheckResult avec success=True si le serveur répond
et la requête DNS aboutit.
"""
host = self.config["host"] host = self.config["host"]
query_name = self.config["query"] query_name = self.config["query"]
query_type = self.config.get("record_type", "A") query_type = self.config.get("record_type", "A")
timeout = self.config.get("timeout", 5) timeout = self.config.get("timeout", 5)
# First check if host is reachable via ping # Vérifie d'abord que le serveur est joignable
ping_checker = PingChecker(self.name, {"host": host, "timeout": timeout}) ping_checker = PingChecker(self.name, {"host": host, "timeout": timeout})
ping_result = ping_checker.check() ping_result = ping_checker.check()
@@ -24,7 +56,7 @@ class DnsChecker(BaseChecker):
response_time=None response_time=None
) )
# Now perform DNS query # Exécute la requête DNS
resolver = dns.resolver.Resolver() resolver = dns.resolver.Resolver()
resolver.nameservers = [host] resolver.nameservers = [host]
resolver.timeout = timeout resolver.timeout = timeout

View File

@@ -1,3 +1,9 @@
"""
Checker HTTP.
Vérifie la disponibilité d'un service web via requête HTTP.
"""
import time import time
import requests import requests
@@ -6,7 +12,26 @@ from .base import BaseChecker, CheckResult
class HttpChecker(BaseChecker): class HttpChecker(BaseChecker):
"""
Vérifie la disponibilité d'un service web.
Configuration YAML:
url: URL à vérifier (obligatoire).
method: Méthode HTTP (défaut: GET).
expected_status: Code HTTP attendu (défaut: 200).
timeout: Délai d'attente en secondes (défaut: 10).
verify_ssl: Vérifier le certificat SSL (défaut: true).
headers: En-têtes HTTP additionnels (optionnel).
"""
def check(self) -> CheckResult: def check(self) -> CheckResult:
"""
Exécute une requête HTTP vers l'URL configurée.
Returns:
CheckResult avec success=True si le code HTTP correspond
à expected_status.
"""
url = self.config["url"] url = self.config["url"]
method = self.config.get("method", "GET").upper() method = self.config.get("method", "GET").upper()
timeout = self.config.get("timeout", 10) timeout = self.config.get("timeout", 10)

View File

@@ -1,3 +1,9 @@
"""
Checker Ping.
Vérifie la disponibilité d'un hôte via ICMP ping.
"""
import subprocess import subprocess
import time import time
import platform import platform
@@ -6,12 +12,30 @@ from .base import BaseChecker, CheckResult
class PingChecker(BaseChecker): class PingChecker(BaseChecker):
"""
Vérifie la disponibilité d'un hôte via ping ICMP.
Configuration YAML:
host: Adresse IP ou nom d'hôte à vérifier (obligatoire).
count: Nombre de paquets à envoyer (défaut: 1).
timeout: Délai d'attente en secondes (défaut: 5).
"""
def check(self) -> CheckResult: def check(self) -> CheckResult:
"""
Exécute un ping vers l'hôte configuré.
Adapte automatiquement la commande ping selon le système
d'exploitation (Windows ou Linux/macOS).
Returns:
CheckResult avec success=True si l'hôte répond.
"""
host = self.config["host"] host = self.config["host"]
count = self.config.get("count", 1) count = self.config.get("count", 1)
timeout = self.config.get("timeout", 5) timeout = self.config.get("timeout", 5)
# Adapt ping command for OS # Adapte la commande ping selon l'OS
if platform.system().lower() == "windows": if platform.system().lower() == "windows":
cmd = ["ping", "-n", str(count), "-w", str(timeout * 1000), host] cmd = ["ping", "-n", str(count), "-w", str(timeout * 1000), host]
else: else:

View File

@@ -1,3 +1,10 @@
"""
Checker SNMP.
Vérifie la disponibilité d'un équipement réseau via SNMP et peut
récupérer sa température.
"""
import asyncio import asyncio
import time import time
@@ -11,10 +18,47 @@ from .base import BaseChecker, CheckResult
class SnmpChecker(BaseChecker): class SnmpChecker(BaseChecker):
"""
Vérifie la disponibilité d'un équipement via SNMP.
Utilise SNMPv2c pour interroger un OID et optionnellement
récupérer la température de l'équipement.
Configuration YAML:
host: Adresse IP de l'équipement (obligatoire).
port: Port SNMP (défaut: 161).
community: Communauté SNMP (défaut: public).
oid: OID à interroger (défaut: sysDescr).
temperature_oid: OID de la température (optionnel).
Note: Utiliser l'OID complet (feuille, pas branche).
Exemples:
- Mikrotik: 1.3.6.1.4.1.14988.1.1.3.100.1.3.52.0
- Synology: 1.3.6.1.4.1.6574.1.2.0
timeout: Délai d'attente en secondes (défaut: 5).
"""
def check(self) -> CheckResult: def check(self) -> CheckResult:
"""
Exécute la vérification SNMP.
Wrapper synchrone autour de _async_check() pour compatibilité
avec l'interface BaseChecker.
Returns:
CheckResult avec success=True si l'équipement répond.
"""
return asyncio.run(self._async_check()) return asyncio.run(self._async_check())
async def _async_check(self) -> CheckResult: async def _async_check(self) -> CheckResult:
"""
Exécute la requête SNMP de manière asynchrone.
Interroge l'OID principal et optionnellement l'OID de température
en une seule requête SNMP GET.
Returns:
CheckResult contenant le résultat et la température si configurée.
"""
host = self.config["host"] host = self.config["host"]
port = self.config.get("port", 161) port = self.config.get("port", 161)
community = self.config.get("community", "public") community = self.config.get("community", "public")
@@ -25,7 +69,7 @@ class SnmpChecker(BaseChecker):
start = time.time() start = time.time()
try: try:
with Slim() as slim: with Slim() as slim:
# Build list of OIDs to query # Construit la liste des OIDs à interroger
oids = [ObjectType(ObjectIdentity(oid))] oids = [ObjectType(ObjectIdentity(oid))]
if temperature_oid: if temperature_oid:
oids.append(ObjectType(ObjectIdentity(temperature_oid))) oids.append(ObjectType(ObjectIdentity(temperature_oid)))
@@ -54,15 +98,15 @@ class SnmpChecker(BaseChecker):
response_time=None response_time=None
) )
else: else:
# Only include main OID in details, not temperature # Inclut uniquement l'OID principal dans les détails
details = {str(var_binds[0][0]): str(var_binds[0][1])} details = {str(var_binds[0][0]): str(var_binds[0][1])}
# Extract temperature if configured (second OID in response) # Extrait la température si configurée (deuxième OID)
if temperature_oid and len(var_binds) >= 2: if temperature_oid and len(var_binds) >= 2:
try: try:
details["temperature"] = int(var_binds[1][1]) details["temperature"] = int(var_binds[1][1])
except (ValueError, TypeError): except (ValueError, TypeError):
pass # Ignore if not a valid integer pass # Ignore si la valeur n'est pas un entier
return CheckResult( return CheckResult(
success=True, success=True,

View File

@@ -1,6 +1,17 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
LAN Checker - Network health monitoring with MQTT reporting for Home Assistant. LAN Checker - Surveillance réseau avec reporting MQTT pour Home Assistant.
Ce script vérifie périodiquement l'état de services et équipements réseau,
puis publie les résultats via MQTT en utilisant le protocole MQTT Discovery
de Home Assistant pour créer automatiquement les entités.
Usage:
python lan_checker.py [-c CONFIG]
Exemples:
python lan_checker.py
python lan_checker.py -c /etc/lan_checker/config.yaml
""" """
import argparse import argparse
@@ -24,13 +35,47 @@ logger = logging.getLogger(__name__)
class LanChecker: class LanChecker:
"""
Gestionnaire principal de surveillance réseau.
Charge la configuration, établit la connexion MQTT, exécute les
vérifications périodiques et publie les résultats.
Attributes:
config: Configuration chargée depuis le fichier YAML.
mqtt_client: Client MQTT pour la publication des résultats.
running: Flag d'exécution de la boucle principale.
checks: Liste des vérifications configurées.
"""
def __init__(self, config_path: str): def __init__(self, config_path: str):
"""
Initialise le gestionnaire.
Args:
config_path: Chemin vers le fichier de configuration YAML.
Raises:
FileNotFoundError: Si le fichier de configuration n'existe pas.
"""
self.config = self._load_config(config_path) self.config = self._load_config(config_path)
self.mqtt_client: mqtt.Client | None = None self.mqtt_client: mqtt.Client | None = None
self.running = False self.running = False
self.checks = [] self.checks = []
def _load_config(self, config_path: str) -> dict: def _load_config(self, config_path: str) -> dict:
"""
Charge la configuration depuis un fichier YAML.
Args:
config_path: Chemin vers le fichier de configuration.
Returns:
Dictionnaire contenant la configuration.
Raises:
FileNotFoundError: Si le fichier n'existe pas.
"""
path = Path(config_path) path = Path(config_path)
if not path.exists(): if not path.exists():
raise FileNotFoundError(f"Configuration file not found: {config_path}") raise FileNotFoundError(f"Configuration file not found: {config_path}")
@@ -39,6 +84,12 @@ class LanChecker:
return yaml.safe_load(f) return yaml.safe_load(f)
def _setup_mqtt(self): def _setup_mqtt(self):
"""
Configure et connecte le client MQTT.
Utilise les paramètres de la section 'mqtt' de la configuration.
Configure l'authentification si un nom d'utilisateur est fourni.
"""
mqtt_config = self.config["mqtt"] mqtt_config = self.config["mqtt"]
self.mqtt_client = mqtt.Client( self.mqtt_client = mqtt.Client(
mqtt.CallbackAPIVersion.VERSION2, mqtt.CallbackAPIVersion.VERSION2,
@@ -62,6 +113,19 @@ class LanChecker:
self.mqtt_client.loop_start() self.mqtt_client.loop_start()
def _on_mqtt_connect(self, client, userdata, flags, reason_code, properties): def _on_mqtt_connect(self, client, userdata, flags, reason_code, properties):
"""
Callback appelé lors de la connexion au broker MQTT.
Publie les messages de découverte Home Assistant si la connexion
est réussie.
Args:
client: Instance du client MQTT.
userdata: Données utilisateur (non utilisé).
flags: Flags de connexion.
reason_code: Code de résultat de la connexion.
properties: Propriétés MQTT v5 (non utilisé).
"""
if reason_code == 0: if reason_code == 0:
logger.info("Connected to MQTT broker") logger.info("Connected to MQTT broker")
self._publish_discovery() self._publish_discovery()
@@ -69,15 +133,35 @@ class LanChecker:
logger.error(f"MQTT connection failed: {reason_code}") logger.error(f"MQTT connection failed: {reason_code}")
def _on_mqtt_disconnect(self, client, userdata, flags, reason_code, properties): def _on_mqtt_disconnect(self, client, userdata, flags, reason_code, properties):
"""
Callback appelé lors de la déconnexion du broker MQTT.
Args:
client: Instance du client MQTT.
userdata: Données utilisateur (non utilisé).
flags: Flags de déconnexion.
reason_code: Code de raison de la déconnexion.
properties: Propriétés MQTT v5 (non utilisé).
"""
logger.warning(f"Disconnected from MQTT broker: {reason_code}") logger.warning(f"Disconnected from MQTT broker: {reason_code}")
def _publish_discovery(self): def _publish_discovery(self):
"""Publish MQTT Discovery messages for Home Assistant.""" """
Publie les messages MQTT Discovery pour Home Assistant.
Pour chaque check configuré, publie:
- Un binary_sensor pour l'état online/offline
- Un sensor pour la latence (temps de réponse)
- Un sensor pour la température (SNMP uniquement, si configuré)
Les entités sont automatiquement créées dans Home Assistant
grâce au protocole MQTT Discovery.
"""
for check in self.config["checks"]: for check in self.config["checks"]:
device_id = check["id"] device_id = check["id"]
device_name = check["name"] device_name = check["name"]
# Binary sensor for online/offline status # Binary sensor pour l'état online/offline
status_config = { status_config = {
"name": f"{device_name} Status", "name": f"{device_name} Status",
"unique_id": f"lan_checker_{device_id}_status", "unique_id": f"lan_checker_{device_id}_status",
@@ -99,7 +183,7 @@ class LanChecker:
retain=True retain=True
) )
# Sensor for response time # Sensor pour le temps de réponse
latency_config = { latency_config = {
"name": f"{device_name} Latency", "name": f"{device_name} Latency",
"unique_id": f"lan_checker_{device_id}_latency", "unique_id": f"lan_checker_{device_id}_latency",
@@ -121,7 +205,7 @@ class LanChecker:
retain=True retain=True
) )
# Sensor for temperature (SNMP only, if temperature_oid configured) # Sensor pour la température (SNMP uniquement)
if check.get("type") == "snmp" and check.get("temperature_oid"): if check.get("type") == "snmp" and check.get("temperature_oid"):
temp_config = { temp_config = {
"name": f"{device_name} Temperature", "name": f"{device_name} Temperature",
@@ -147,7 +231,12 @@ class LanChecker:
logger.info(f"Published discovery for: {device_name}") logger.info(f"Published discovery for: {device_name}")
def _setup_checks(self): def _setup_checks(self):
"""Initialize check instances from configuration.""" """
Initialise les instances de checkers depuis la configuration.
Parcourt la liste des checks dans la configuration et crée
une instance du checker approprié pour chacun.
"""
for check_config in self.config["checks"]: for check_config in self.config["checks"]:
check_type = check_config["type"] check_type = check_config["type"]
if check_type not in CHECKERS: if check_type not in CHECKERS:
@@ -165,7 +254,13 @@ class LanChecker:
}) })
def _run_check(self, check: dict): def _run_check(self, check: dict):
"""Execute a single check and publish results.""" """
Exécute une vérification et publie le résultat via MQTT.
Args:
check: Dictionnaire contenant l'id, le nom, le checker
et l'intervalle de vérification.
"""
result = check["checker"].check() result = check["checker"].check()
state = "online" if result.success else "offline" state = "online" if result.success else "offline"
@@ -177,7 +272,7 @@ class LanChecker:
} }
if result.details: if result.details:
# Extract temperature for SNMP checks # Extrait la température des détails pour la mettre à la racine
if "temperature" in result.details: if "temperature" in result.details:
payload["temperature"] = result.details.pop("temperature") payload["temperature"] = result.details.pop("temperature")
if result.details: if result.details:
@@ -190,7 +285,12 @@ class LanChecker:
logger.log(log_level, f"{check['name']}: {state} - {result.message}") logger.log(log_level, f"{check['name']}: {state} - {result.message}")
def run(self): def run(self):
"""Main loop.""" """
Lance la boucle principale de surveillance.
Configure MQTT, initialise les checkers, puis exécute les
vérifications en boucle selon leurs intervalles respectifs.
"""
self._setup_mqtt() self._setup_mqtt()
self._setup_checks() self._setup_checks()
@@ -208,7 +308,11 @@ class LanChecker:
time.sleep(1) time.sleep(1)
def stop(self): def stop(self):
"""Stop the checker gracefully.""" """
Arrête proprement le gestionnaire.
Stoppe la boucle principale et déconnecte le client MQTT.
"""
logger.info("Stopping LAN Checker...") logger.info("Stopping LAN Checker...")
self.running = False self.running = False
if self.mqtt_client: if self.mqtt_client:
@@ -217,6 +321,12 @@ class LanChecker:
def main(): def main():
"""
Point d'entrée principal.
Parse les arguments de ligne de commande, configure les gestionnaires
de signaux et lance le checker.
"""
parser = argparse.ArgumentParser(description="LAN Checker - Network health monitoring") parser = argparse.ArgumentParser(description="LAN Checker - Network health monitoring")
parser.add_argument( parser.add_argument(
"-c", "--config", "-c", "--config",
@@ -228,6 +338,7 @@ def main():
checker = LanChecker(args.config) checker = LanChecker(args.config)
def signal_handler(sig, frame): def signal_handler(sig, frame):
"""Gestionnaire de signaux pour arrêt propre."""
checker.stop() checker.stop()
sys.exit(0) sys.exit(0)