#!/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()