"""iPhone arrival detection backed by a Zyxel router's active-client list.""" from __future__ import annotations import logging from datetime import datetime from pathlib import Path from threading import Event, Thread from typing import Callable from lib._zyxel import ZyxelError, ZyxelRouter, normalize_mac 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 transition(self, present: bool) -> str | None: """Return ``arrived`` or ``departed`` after a confirmed state change.""" 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 None if self.state and self.absent_samples >= self.absent_after: self.state = False return "departed" elif not self.state and self.present_samples >= self.present_after: self.state = True return "arrived" return None def sample(self, present: bool) -> bool: """Backward-compatible arrival-only result.""" return self.transition(present) == "arrived" 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")), debug=bool(router.get("debug", False)), ) def _record_event(config: dict, event: dict, mac: str = "") -> None: """Record a confirmed presence event before handing it to the controller.""" LOG.info("%s: %s", event["event"].upper(), event["id"]) event_log = config.get("event_log") if event_log: timestamp = datetime.now().astimezone().isoformat(timespec="seconds") mac_field = f" mac={mac}" if mac else "" with Path(event_log).open("a", encoding="utf-8") as stream: stream.write( f"{timestamp} {event['event'].upper()} id={event['id']}{mac_field}\n" ) def _record_arrival(config: dict, device: dict) -> None: """Compatibility helper for callers that record an arrival directly.""" _record_event( config, {"sender": "_arrival_detection", "event": "arrived", "id": device["id"], "text": ""}, device["mac"], ) def _configured_devices(config: dict) -> list[dict]: devices = config.get("devices") if not isinstance(devices, list) or not devices: raise ValueError("arrival_detection.devices must contain at least one device") if len(devices) > 3: raise ValueError("arrival_detection.devices supports at most 3 devices") normalized = [] identifiers = set() for index, device in enumerate(devices, start=1): if not isinstance(device, dict) or not device.get("id") or not device.get("mac"): raise ValueError(f"arrival_detection.devices entry {index} requires id and mac") item = dict(device) item["mac"] = normalize_mac(str(item["mac"])) item["id"] = str(item["id"]).strip().lower() if not item["id"]: raise ValueError(f"arrival_detection.devices entry {index} requires a non-empty id") if item["id"] in identifiers: raise ValueError(f"duplicate arrival_detection device id: {item['id']}") identifiers.add(item["id"]) normalized.append(item) return normalized class ArrivalMonitor: """Background Zyxel monitor that emits generic controller events.""" def __init__(self, config: dict, on_event: Callable[[dict], None]) -> None: self.config = config self.on_event = on_event self.devices = _configured_devices(config) polling = config.get("polling", {}) self.detectors = { device["mac"]: ArrivalDetector( int(polling.get("absent_after", 3)), int(polling.get("present_after", 2)), ) for device in self.devices } self.interval = float(polling.get("interval_seconds", 10)) self.retry = float(polling.get("error_retry_seconds", 30)) self.router = _build_router(config) self._stop = Event() self._thread: Thread | None = None self._house_occupied: bool | None = None def _connection_states(self) -> dict[str, bool]: return self.router.connection_states([device["mac"] for device in self.devices]) def check_once(self) -> list[str]: states = self._connection_states() return [ f"{device['id']}={'connected' if states[device['mac']] else 'absent'}" for device in self.devices ] def _process_states(self, states: dict[str, bool]) -> None: arrivals: list[dict] = [] for device in self.devices: present = states[device["mac"]] LOG.debug( "%s is %s", device["id"], "connected" if present else "absent", ) transition = self.detectors[device["mac"]].transition(present) if transition: event = { "sender": "_arrival_detection", "event": transition, "id": device["id"], "text": "", } _record_event(self.config, event, device["mac"]) self.on_event(event) if transition == "arrived": arrivals.append(device) known_states = [detector.state for detector in self.detectors.values()] if all(state is not None for state in known_states): occupied = any(known_states) if self._house_occupied is True and not occupied: event = { "sender": "_arrival_detection", "event": "empty", "id": "house", "text": "", } _record_event(self.config, event) self.on_event(event) elif self._house_occupied is False and occupied: first = arrivals[0] event = { "sender": "_arrival_detection", "event": "first_arrival", "id": first["id"], "text": "", } _record_event(self.config, event, first["mac"]) self.on_event(event) self._house_occupied = occupied def _run(self) -> None: try: while not self._stop.is_set(): try: states = self._connection_states() self._process_states(states) self._stop.wait(self.interval) except (ZyxelError, OSError, RuntimeError) as error: LOG.error("poll failed: %s", error) self.router.close() self._stop.wait(self.retry) finally: self.router.close() def start(self) -> None: if self._thread and self._thread.is_alive(): raise RuntimeError("arrival monitor is already running") self._thread = Thread(target=self._run, name="zyxel-arrival-monitor") self._thread.start() LOG.info("Zyxel arrival monitor started") def stop(self) -> None: self._stop.set() def join(self, timeout: float | None = None) -> None: if self._thread: self._thread.join(timeout) @property def is_alive(self) -> bool: return bool(self._thread and self._thread.is_alive()) def close(self) -> None: """Release resources after a one-shot check.""" self.router.close() def create(config: dict, on_event: Callable[[dict], None]) -> ArrivalMonitor: """Generic controller plugin factory.""" return ArrivalMonitor(config, on_event)