home_control/lib/_arrival_detection.py

"""iPhone arrival detection backed by a Zyxel router's active-client list."""

from __future__ import annotations

import logging
import time
from datetime import datetime
from pathlib import Path

from lib._zyxel import ZyxelError, ZyxelRouter, normalize_mac
from notify import send_notification

LOG = logging.getLogger("home_control.arrival_detection")


class ArrivalDetector:
    """Debounce connection samples and emit absent -> present transitions."""

    def __init__(self, absent_after: int, present_after: int) -> None:
        if absent_after < 1 or present_after < 1:
            raise ValueError("absent_after and present_after must be at least 1")
        self.absent_after = absent_after
        self.present_after = present_after
        self.state: bool | None = None
        self.present_samples = 0
        self.absent_samples = 0

    def sample(self, present: bool) -> bool:
        self.present_samples = self.present_samples + 1 if present else 0
        self.absent_samples = self.absent_samples + 1 if not present else 0
        if self.state is None:
            if self.present_samples >= self.present_after:
                self.state = True
            elif self.absent_samples >= self.absent_after:
                self.state = False
            return False
        if self.state and self.absent_samples >= self.absent_after:
            self.state = False
        elif not self.state and self.present_samples >= self.present_after:
            self.state = True
            return True
        return False


def _build_router(config: dict) -> ZyxelRouter:
    router = config["zyxel"]
    return ZyxelRouter(
        router["host"], router["username"], router["password"],
        timeout=float(router.get("timeout", 10)),
        verify_tls=bool(router.get("verify_tls", True)),
        hosts_oid=str(router.get("hosts_oid", "lanhosts")),
    )


def _notify(config: dict, device: dict) -> None:
    LOG.info("ARRIVAL: %s", device["name"])
    notification = config.get("notification", {})
    message = str(notification.get("message", "{name} arrived home")).format(
        name=device["name"], mac=device["mac"]
    )
    event_log = notification.get("event_log")
    if event_log:
        timestamp = datetime.now().astimezone().isoformat(timespec="seconds")
        with Path(event_log).open("a", encoding="utf-8") as stream:
            stream.write(f"{timestamp} {message}\n")
    if not notification.get("enabled", True):
        return
    send_notification(
        message,
        topic_url=notification.get("topic_url") or None,
        timeout=float(notification.get("timeout", 10)),
    )


def run(config: dict, stop_requested, *, once: bool = False) -> int:
    """Run arrival detection until stop_requested returns true."""
    device = dict(config["device"])
    device["mac"] = normalize_mac(device["mac"])
    polling = config.get("polling", {})
    detector = ArrivalDetector(
        int(polling.get("absent_after", 3)), int(polling.get("present_after", 2))
    )
    interval = float(polling.get("interval_seconds", 10))
    retry = float(polling.get("error_retry_seconds", 30))
    router = _build_router(config)
    try:
        while not stop_requested():
            try:
                present = router.is_connected(device["mac"])
                LOG.debug("%s is %s", device["name"], "connected" if present else "absent")
                if detector.sample(present):
                    _notify(config, device)
                if once:
                    print("connected" if present else "absent")
                    return 0
                time.sleep(interval)
            except (ZyxelError, OSError, RuntimeError) as error:
                LOG.error("poll failed: %s", error)
                router.close()
                if once:
                    return 1
                time.sleep(retry)
    finally:
        router.close()
    return 0