Antoine Van Elstraete
b0d571ae26
- 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>
357 lines
12 KiB
Python
357 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
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 json
|
|
import logging
|
|
import signal
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
import paho.mqtt.client as mqtt
|
|
|
|
from checkers import CHECKERS
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s - %(levelname)s - %(message)s"
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
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):
|
|
"""
|
|
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.mqtt_client: mqtt.Client | None = None
|
|
self.running = False
|
|
self.checks = []
|
|
|
|
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)
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Configuration file not found: {config_path}")
|
|
|
|
with open(path, "r") as f:
|
|
return yaml.safe_load(f)
|
|
|
|
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"]
|
|
self.mqtt_client = mqtt.Client(
|
|
mqtt.CallbackAPIVersion.VERSION2,
|
|
client_id=mqtt_config.get("client_id", "lan_checker")
|
|
)
|
|
|
|
if mqtt_config.get("username"):
|
|
self.mqtt_client.username_pw_set(
|
|
mqtt_config["username"],
|
|
mqtt_config.get("password", "")
|
|
)
|
|
|
|
self.mqtt_client.on_connect = self._on_mqtt_connect
|
|
self.mqtt_client.on_disconnect = self._on_mqtt_disconnect
|
|
|
|
self.mqtt_client.connect(
|
|
mqtt_config["host"],
|
|
mqtt_config.get("port", 1883),
|
|
keepalive=60
|
|
)
|
|
self.mqtt_client.loop_start()
|
|
|
|
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:
|
|
logger.info("Connected to MQTT broker")
|
|
self._publish_discovery()
|
|
else:
|
|
logger.error(f"MQTT connection failed: {reason_code}")
|
|
|
|
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}")
|
|
|
|
def _publish_discovery(self):
|
|
"""
|
|
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"]:
|
|
device_id = check["id"]
|
|
device_name = check["name"]
|
|
|
|
# Binary sensor pour l'état online/offline
|
|
status_config = {
|
|
"name": f"{device_name} Status",
|
|
"unique_id": f"lan_checker_{device_id}_status",
|
|
"state_topic": f"lan_checker/{device_id}/state",
|
|
"value_template": "{{ value_json.state }}",
|
|
"payload_on": "online",
|
|
"payload_off": "offline",
|
|
"device_class": "connectivity",
|
|
"device": {
|
|
"identifiers": [f"lan_checker_{device_id}"],
|
|
"name": device_name,
|
|
"manufacturer": "LAN Checker",
|
|
},
|
|
}
|
|
|
|
self.mqtt_client.publish(
|
|
f"homeassistant/binary_sensor/lan_checker_{device_id}/config",
|
|
json.dumps(status_config),
|
|
retain=True
|
|
)
|
|
|
|
# Sensor pour le temps de réponse
|
|
latency_config = {
|
|
"name": f"{device_name} Latency",
|
|
"unique_id": f"lan_checker_{device_id}_latency",
|
|
"state_topic": f"lan_checker/{device_id}/state",
|
|
"value_template": "{{ value_json.response_time | default('unavailable') }}",
|
|
"unit_of_measurement": "ms",
|
|
"device_class": "duration",
|
|
"state_class": "measurement",
|
|
"device": {
|
|
"identifiers": [f"lan_checker_{device_id}"],
|
|
"name": device_name,
|
|
"manufacturer": "LAN Checker",
|
|
},
|
|
}
|
|
|
|
self.mqtt_client.publish(
|
|
f"homeassistant/sensor/lan_checker_{device_id}_latency/config",
|
|
json.dumps(latency_config),
|
|
retain=True
|
|
)
|
|
|
|
# Sensor pour la température (SNMP uniquement)
|
|
if check.get("type") == "snmp" and check.get("temperature_oid"):
|
|
temp_config = {
|
|
"name": f"{device_name} Temperature",
|
|
"unique_id": f"lan_checker_{device_id}_temperature",
|
|
"state_topic": f"lan_checker/{device_id}/state",
|
|
"value_template": "{{ value_json.temperature | default('unavailable') }}",
|
|
"unit_of_measurement": "°C",
|
|
"device_class": "temperature",
|
|
"state_class": "measurement",
|
|
"device": {
|
|
"identifiers": [f"lan_checker_{device_id}"],
|
|
"name": device_name,
|
|
"manufacturer": "LAN Checker",
|
|
},
|
|
}
|
|
|
|
self.mqtt_client.publish(
|
|
f"homeassistant/sensor/lan_checker_{device_id}_temperature/config",
|
|
json.dumps(temp_config),
|
|
retain=True
|
|
)
|
|
|
|
logger.info(f"Published discovery for: {device_name}")
|
|
|
|
def _setup_checks(self):
|
|
"""
|
|
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"]:
|
|
check_type = check_config["type"]
|
|
if check_type not in CHECKERS:
|
|
logger.error(f"Unknown check type: {check_type}")
|
|
continue
|
|
|
|
checker_class = CHECKERS[check_type]
|
|
checker = checker_class(check_config["name"], check_config)
|
|
self.checks.append({
|
|
"id": check_config["id"],
|
|
"name": check_config["name"],
|
|
"checker": checker,
|
|
"interval": check_config.get("interval", self.config.get("default_interval", 60)),
|
|
"last_check": 0,
|
|
})
|
|
|
|
def _run_check(self, check: dict):
|
|
"""
|
|
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()
|
|
|
|
state = "online" if result.success else "offline"
|
|
payload = {
|
|
"state": state,
|
|
"message": result.message,
|
|
"response_time": round(result.response_time, 2) if result.response_time else None,
|
|
"last_check": time.strftime("%Y-%m-%dT%H:%M:%S"),
|
|
}
|
|
|
|
if result.details:
|
|
# Extrait la température des détails pour la mettre à la racine
|
|
if "temperature" in result.details:
|
|
payload["temperature"] = result.details.pop("temperature")
|
|
if result.details:
|
|
payload["details"] = result.details
|
|
|
|
topic = f"lan_checker/{check['id']}/state"
|
|
self.mqtt_client.publish(topic, json.dumps(payload), retain=True)
|
|
|
|
log_level = logging.INFO if result.success else logging.WARNING
|
|
logger.log(log_level, f"{check['name']}: {state} - {result.message}")
|
|
|
|
def run(self):
|
|
"""
|
|
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_checks()
|
|
|
|
self.running = True
|
|
logger.info(f"Starting LAN Checker with {len(self.checks)} checks")
|
|
|
|
while self.running:
|
|
current_time = time.time()
|
|
|
|
for check in self.checks:
|
|
if current_time - check["last_check"] >= check["interval"]:
|
|
self._run_check(check)
|
|
check["last_check"] = current_time
|
|
|
|
time.sleep(1)
|
|
|
|
def stop(self):
|
|
"""
|
|
Arrête proprement le gestionnaire.
|
|
|
|
Stoppe la boucle principale et déconnecte le client MQTT.
|
|
"""
|
|
logger.info("Stopping LAN Checker...")
|
|
self.running = False
|
|
if self.mqtt_client:
|
|
self.mqtt_client.loop_stop()
|
|
self.mqtt_client.disconnect()
|
|
|
|
|
|
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.add_argument(
|
|
"-c", "--config",
|
|
default="config.yaml",
|
|
help="Path to configuration file (default: config.yaml)"
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
checker = LanChecker(args.config)
|
|
|
|
def signal_handler(sig, frame):
|
|
"""Gestionnaire de signaux pour arrêt propre."""
|
|
checker.stop()
|
|
sys.exit(0)
|
|
|
|
signal.signal(signal.SIGINT, signal_handler)
|
|
signal.signal(signal.SIGTERM, signal_handler)
|
|
|
|
try:
|
|
checker.run()
|
|
except Exception as e:
|
|
logger.error(f"Fatal error: {e}")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|