lan_checker/lan_checker.py

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

            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:
            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()