#!/usr/bin/env python3 """ LAN Checker - Network health monitoring with MQTT reporting for Home Assistant. """ 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: def __init__(self, config_path: str): 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: 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): 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): 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): logger.warning(f"Disconnected from MQTT broker: {reason_code}") def _publish_discovery(self): """Publish MQTT Discovery messages for Home Assistant.""" for check in self.config["checks"]: device_id = check["id"] device_name = check["name"] # Binary sensor for online/offline status 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 for response time 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 for temperature (SNMP only, if temperature_oid configured) 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): """Initialize check instances from configuration.""" 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): """Execute a single check and publish results.""" 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: # Extract temperature for SNMP checks 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): """Main loop.""" 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): """Stop the checker gracefully.""" logger.info("Stopping LAN Checker...") self.running = False if self.mqtt_client: self.mqtt_client.loop_stop() self.mqtt_client.disconnect() def main(): 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): 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()