diff --git a/README.md b/README.md index a740650..254f1ba 100644 --- a/README.md +++ b/README.md @@ -15,13 +15,39 @@ Check any MAC address directly through the Zyxel client with: python lib/_zyxel.py --mac AA:BB:CC:DD:EE:FF ``` -Omit `--mac` to check `arrival_detection.device.mac` from `service.yaml`. +Omit `--mac` to check the first entry in `arrival_detection.devices` from +`service.yaml`. Arrival detection supports one to three devices and tracks each +device independently. + +Set `arrival_detection.zyxel.debug` to `true` to log every Zyxel HTTP/DAL call +and MAC presence result. Under systemd, view these messages with +`journalctl -u home-control.service -f`. An arrival is logged after the phone was confirmed absent and then present. It -also calls `send_notification()` from `notify.py`; its message, ntfy topic, and -timeout are configured under `arrival_detection.notification`. Set +also calls `send_notification()` from `lib/_notify.py`; its message, ntfy topic, +and timeout are configured under the top-level `notify` section. Set `arrival_detection.enabled` to `false` to disable this function. +Plugins emit dictionaries containing `sender`, `event`, `id`, and `text`. The +controller routes each sender to one or more actions through `controller.plugins`. +For example, `_arrival_detection` emits an `arrived` event and can be routed to +both actions through `on_event: [_notify, _cloud_logger]`. A single action name +such as `on_event: _notify` remains supported. + +Arrival detection emits `arrived` and `departed` for individual device IDs. It +also emits `empty` with ID `house` after the last present device departs. Initial +startup state never generates these transition events. + +Plugin modules, factories, configuration sections, and event actions are all +declared under `controller`. The controller contains no plugin-specific imports +or startup logic. A plugin module exposes a configured factory (normally +`create`), and an action module exposes a configured handler (normally +`on_event`). + +The `_notify` action can filter by event or ID under `notify.filter`. Empty +`accept` lists allow all values; populated lists act as allow-lists. Values in +`ignore` are always rejected, even when also accepted. + On the iPhone, open **Settings > Wi-Fi**, tap the info button beside the home network, and copy **Wi-Fi Address**. If Private Wi-Fi Address is enabled, that per-network address is the correct one to configure. @@ -50,4 +76,4 @@ sudo journalctl -u home-control.service -f Confirmed arrival notifications are also appended to `/log/detect-arrivals.txt`. The path is configurable as -`arrival_detection.notification.event_log`. +`arrival_detection.event_log`. diff --git a/cloud/config.yaml b/cloud/config.yaml new file mode 100644 index 0000000..e21e281 --- /dev/null +++ b/cloud/config.yaml @@ -0,0 +1,4 @@ +HOME_CONTROL_LOG: flask_home_control_log.log +ip_address: 127.0.0.1 +port: 5000 +url_prefix: /home_logger diff --git a/cloud/flask_home_control_log.log b/cloud/flask_home_control_log.log new file mode 100644 index 0000000..5ca9113 --- /dev/null +++ b/cloud/flask_home_control_log.log @@ -0,0 +1,2 @@ +{"timestamp":"2026-08-23T06:09:35.779477+00:00","remote_addr":"127.0.0.1","event":{"sender":"_arrival_detection","event":"departed","id":"ignace","text":""}} +{"timestamp":"2026-08-23T06:10:47.665139+00:00","remote_addr":"127.0.0.1","event":{"sender":"_arrival_detection","event":"arrived","id":"ignace","text":""}} diff --git a/cloud/flask_home_control_log.py b/cloud/flask_home_control_log.py new file mode 100644 index 0000000..f6145d2 --- /dev/null +++ b/cloud/flask_home_control_log.py @@ -0,0 +1,70 @@ +import json +import os +from datetime import datetime, timezone +from pathlib import Path + +import yaml +from flask import Flask, Response, jsonify, request + + +CONFIG_FILE = Path(__file__).with_name("config.yaml") +with CONFIG_FILE.open(encoding="utf-8") as stream: + CONFIG = yaml.safe_load(stream) or {} + +LOG_FILE = Path(CONFIG["HOME_CONTROL_LOG"]) +if not LOG_FILE.is_absolute(): + LOG_FILE = CONFIG_FILE.parent / LOG_FILE +IP_ADDRESS = str(CONFIG.get("ip_address", "127.0.0.1")) +PORT = int(CONFIG.get("port", 5000)) +URL_PREFIX = "/" + str(CONFIG.get("url_prefix", "")).strip("/") +if URL_PREFIX == "/": + URL_PREFIX = "" + +app = Flask(__name__) + + +def _tail(path: Path, line_count: int = 50) -> str: + try: + with path.open("rb") as logfile: + logfile.seek(0, os.SEEK_END) + end = logfile.tell() + data = b"" + + while end > 0 and data.count(b"\n") <= line_count: + size = min(4096, end) + end -= size + logfile.seek(end) + data = logfile.read(size) + data + + return b"\n".join(data.splitlines()[-line_count:]).decode("utf-8", "replace") + except FileNotFoundError: + return "" + + +@app.route(f"{URL_PREFIX}/event", methods=["GET", "POST"]) +def add_event(): + data = request.get_json(silent=True) + if data is None: + data = request.values.to_dict(flat=False) + if not data: + data = request.get_data(as_text=True) + + entry = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "remote_addr": request.remote_addr, + "event": data, + } + LOG_FILE.parent.mkdir(parents=True, exist_ok=True) + with LOG_FILE.open("a", encoding="utf-8") as logfile: + logfile.write(json.dumps(entry, ensure_ascii=False, separators=(",", ":")) + "\n") + + return jsonify(ok=True), 201 + + +@app.get(URL_PREFIX or "/", strict_slashes=False) +def show_log(): + return Response(_tail(LOG_FILE), mimetype="text/plain") + + +if __name__ == "__main__": + app.run(host=IP_ADDRESS, port=PORT) diff --git a/lib/_arrival_detection.py b/lib/_arrival_detection.py index 2255e0c..0e21e6d 100644 --- a/lib/_arrival_detection.py +++ b/lib/_arrival_detection.py @@ -3,12 +3,12 @@ from __future__ import annotations import logging -import time 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 -from notify import send_notification LOG = logging.getLogger("home_control.arrival_detection") @@ -25,7 +25,8 @@ class ArrivalDetector: self.present_samples = 0 self.absent_samples = 0 - def sample(self, present: bool) -> bool: + 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: @@ -33,13 +34,18 @@ class ArrivalDetector: self.state = True elif self.absent_samples >= self.absent_after: self.state = False - return 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 True - return False + 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: @@ -49,57 +55,157 @@ def _build_router(config: dict) -> ZyxelRouter: 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 _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") +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} {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)), + 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 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 +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: + 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) + + 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) + 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) diff --git a/lib/_cloud_logger.py b/lib/_cloud_logger.py new file mode 100644 index 0000000..25cd68c --- /dev/null +++ b/lib/_cloud_logger.py @@ -0,0 +1,80 @@ +"""Send controller events to the cloud event logger.""" + +import logging +from datetime import datetime, timezone + +import requests + + +LOG = logging.getLogger("home_control.cloud_logger") + + +def _filter_values(config: dict, mode: str, field: str) -> set[str]: + value = config.get("filter", {}).get(mode, {}).get(field, []) + if not isinstance(value, list) or any(not isinstance(item, str) for item in value): + raise ValueError( + f"cloud_logger.filter.{mode}.{field} must be a list of strings" + ) + return set(value) + + +def accepts_event(config: dict, message: dict) -> bool: + """Return whether an event passes configured event and ID filters.""" + accepted_events = _filter_values(config, "accept", "events") + accepted_ids = _filter_values(config, "accept", "ids") + ignored_events = _filter_values(config, "ignore", "events") + ignored_ids = _filter_values(config, "ignore", "ids") + + if message["event"] in ignored_events or message["id"] in ignored_ids: + return False + if accepted_events and message["event"] not in accepted_events: + return False + if accepted_ids and message["id"] not in accepted_ids: + return False + return True + + +def send_event(message: dict, event_url: str, timeout: float = 10) -> None: + """POST an event dictionary as JSON to the configured logger endpoint.""" + if not isinstance(message, dict): + raise TypeError("message must be a dictionary") + if not event_url: + raise ValueError("event_url is required") + + response = requests.post(event_url, json=message, timeout=timeout) + response.raise_for_status() + + +def on_event(config: dict, message: dict) -> None: + """Controller action that forwards event data to the cloud logger.""" + if not config.get("enabled", True): + return + + try: + if not accepts_event(config, message): + LOG.debug( + "cloud event ignored by filter: event=%s id=%s", + message["event"], + message["id"], + ) + return + + timestamp = datetime.now(timezone.utc).isoformat() + values = {**message, "datetime": timestamp} + messages = config.get("messages", {}) + template = messages.get( + message["event"], + config.get("message", "{datetime} {sender}: {event} ({id})"), + ) + payload = { + **message, + "datetime": timestamp, + "text": message.get("text") or str(template).format(**values), + } + send_event( + payload, + event_url=str(config.get("event_url", "")), + timeout=float(config.get("timeout", 10)), + ) + except (TypeError, ValueError, requests.RequestException) as error: + LOG.error("cloud event logging failed: %s", error) diff --git a/lib/_notify.py b/lib/_notify.py new file mode 100644 index 0000000..91fc584 --- /dev/null +++ b/lib/_notify.py @@ -0,0 +1,115 @@ +#!/usr/bin/env python3 +"""Send text notifications to a configured ntfy topic.""" + +import argparse +import logging +import os +import sys +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + + +LOG = logging.getLogger("home_control.notify") + +def send_notification(message, topic_url=None, timeout=10): + """Send *message* to ntfy and return the server response body.""" + if not isinstance(message, str): + raise TypeError("message must be a string") + if not message: + raise ValueError("message must not be empty") + url = topic_url or os.environ.get("NTFY_TOPIC_URL") + if not url: + raise ValueError("topic_url is required (or set NTFY_TOPIC_URL)") + request = Request( + url, + data=message.encode("utf-8"), + headers={"Content-Type": "text/plain; charset=utf-8"}, + method="POST", + ) + with urlopen(request, timeout=timeout) as response: + return response.read().decode("utf-8") + + +def _filter_values(config: dict, mode: str, field: str) -> set[str]: + value = config.get("filter", {}).get(mode, {}).get(field, []) + if not isinstance(value, list) or any(not isinstance(item, str) for item in value): + raise ValueError(f"notify.filter.{mode}.{field} must be a list of strings") + return set(value) + + +def accepts_event(config: dict, message: dict) -> bool: + """Return whether an event passes configured event and ID filters.""" + accepted_events = _filter_values(config, "accept", "events") + accepted_ids = _filter_values(config, "accept", "ids") + ignored_events = _filter_values(config, "ignore", "events") + ignored_ids = _filter_values(config, "ignore", "ids") + + if message["event"] in ignored_events or message["id"] in ignored_ids: + return False + if accepted_events and message["event"] not in accepted_events: + return False + if accepted_ids and message["id"] not in accepted_ids: + return False + return True + + +def on_event(config: dict, message: dict) -> None: + """Generic controller action for event dictionaries.""" + if not config.get("enabled", True): + return + if not accepts_event(config, message): + LOG.debug( + "notification ignored by filter: event=%s id=%s", + message["event"], + message["id"], + ) + return + messages = config.get("messages", {}) + template = messages.get(message["event"], config.get("message", "{sender}: {event} ({id})")) + text = message["text"] or str(template).format(**message) + try: + send_notification( + text, + topic_url=config.get("topic_url") or None, + timeout=float(config.get("timeout", 10)), + ) + except (OSError, RuntimeError, ValueError) as error: + LOG.error( + "notification failed for %s/%s/%s: %s", + message["sender"], + message["event"], + message["id"], + error, + ) + + +def parse_arguments(): + parser = argparse.ArgumentParser(description="Send a text message to an ntfy topic.") + parser.add_argument("message", nargs="+", help="notification text") + parser.add_argument("--topic-url", help="ntfy topic URL (or use NTFY_TOPIC_URL)") + parser.add_argument("--timeout", type=float, default=10, help="request timeout in seconds") + return parser.parse_args() + + +def main(): + args = parse_arguments() + try: + print( + send_notification( + " ".join(args.message), + topic_url=args.topic_url, + timeout=args.timeout, + ) + ) + return 0 + except HTTPError as error: + print(f"ntfy returned HTTP {error.code}: {error.reason}", file=sys.stderr) + except URLError as error: + print(f"Could not reach ntfy: {error.reason}", file=sys.stderr) + except (TypeError, ValueError) as error: + print(f"Invalid notification: {error}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/lib/_zyxel.py b/lib/_zyxel.py index fc350fa..69c0183 100644 --- a/lib/_zyxel.py +++ b/lib/_zyxel.py @@ -5,6 +5,7 @@ from __future__ import annotations import argparse import base64 import json +import logging import os import sys from dataclasses import dataclass @@ -19,6 +20,9 @@ from Crypto.PublicKey import RSA from Crypto.Util.Padding import pad, unpad +LOG = logging.getLogger("home_control.zyxel") + + class ZyxelError(RuntimeError): """The router rejected a request or returned an unexpected response.""" @@ -50,6 +54,7 @@ class ZyxelRouter: timeout: float = 10, verify_tls: bool = True, hosts_oid: str = "lanhosts", + debug: bool = False, ) -> None: if "://" not in host: host = f"http://{host}" @@ -62,9 +67,11 @@ class ZyxelRouter: self.timeout = timeout self.verify_tls = verify_tls self.hosts_oid = hosts_oid + self.debug = debug self.session = requests.Session() self._aes_key: bytes | None = None self._session_key: str | None = None + self._has_logged_connection = False def __enter__(self) -> "ZyxelRouter": self.login() @@ -79,6 +86,8 @@ class ZyxelRouter: try: response = self.session.request(method, f"{self.url}{path}", **kwargs) response.raise_for_status() + if self.debug: + LOG.info("Zyxel HTTP call: %s %s -> %s", method, path, response.status_code) return response except requests.RequestException as error: raise ZyxelError(f"Zyxel request failed: {error}") from error @@ -120,6 +129,12 @@ class ZyxelRouter: if result.get("result") != "ZCFG_SUCCESS" or not result.get("sessionkey"): raise ZyxelError(f"Zyxel login failed: {result.get('result', 'unknown error')}") self._session_key = str(result["sessionkey"]) + self._log_first_connection() + + def _log_first_connection(self) -> None: + if not self._has_logged_connection: + LOG.info("Connected successfully to Zyxel router at %s", self.url) + self._has_logged_connection = True def _decrypt(self, envelope: dict[str, Any]) -> dict[str, Any]: if "content" not in envelope or "iv" not in envelope: @@ -143,7 +158,10 @@ class ZyxelRouter: response = self._request( "GET", "/cgi-bin/DAL", params={"oid": oid, "sessionkey": self._session_key} ) - return self._decrypt(self._json(response)) + result = self._decrypt(self._json(response)) + if self.debug: + LOG.info("Zyxel DAL call: oid=%s -> %s", oid, result.get("result", "no result")) + return result def get_lan_hosts(self) -> list[LanHost]: result = self.dal_get(self.hosts_oid) @@ -184,9 +202,23 @@ class ZyxelRouter: )) return hosts + def connection_states(self, macs: list[str]) -> dict[str, bool]: + """Check several MAC addresses using one router query.""" + wanted = [normalize_mac(mac) for mac in macs] + active_macs = {host.mac for host in self.get_lan_hosts() if host.active} + states = {mac: mac in active_macs for mac in wanted} + if self.debug: + for mac, connected in states.items(): + LOG.info( + "Zyxel presence result: mac=%s -> %s", + mac, + "connected" if connected else "absent", + ) + return states + def is_connected(self, mac: str) -> bool: wanted = normalize_mac(mac) - return any(host.mac == wanted and host.active for host in self.get_lan_hosts()) + return self.connection_states([wanted])[wanted] def close(self) -> None: if self._session_key: @@ -204,7 +236,7 @@ def _load_cli_config(path: Path) -> tuple[dict[str, Any], str]: with path.open(encoding="utf-8") as stream: root = yaml.safe_load(stream) arrival = root["arrival_detection"] - return arrival["zyxel"], str(arrival["device"]["mac"]) + return arrival["zyxel"], str(arrival["devices"][0]["mac"]) except (OSError, TypeError, KeyError, yaml.YAMLError) as error: raise ZyxelError(f"could not load configuration from {path}: {error}") from error @@ -221,7 +253,7 @@ def main() -> int: ) parser.add_argument( "--mac", - help="MAC address to check (default: arrival_detection.device.mac)", + help="MAC address to check (default: first arrival_detection.devices entry)", ) args = parser.parse_args() @@ -235,6 +267,7 @@ def main() -> int: timeout=float(config.get("timeout", 10)), verify_tls=bool(config.get("verify_tls", True)), hosts_oid=str(config.get("hosts_oid", "lanhosts")), + debug=bool(config.get("debug", False)), ) try: connected = router.is_connected(mac) diff --git a/log/detect-arrivals.txt b/log/detect-arrivals.txt new file mode 100644 index 0000000..c7746e3 --- /dev/null +++ b/log/detect-arrivals.txt @@ -0,0 +1,3 @@ +2026-08-22T22:27:23+02:00 ARRIVAL name=Ignace mac=ce:68:53:a8:fc:07 +2026-08-23T08:09:35+02:00 DEPARTED id=ignace mac=ce:68:53:a8:fc:07 +2026-08-23T08:10:47+02:00 ARRIVED id=ignace mac=ce:68:53:a8:fc:07 diff --git a/notify.py b/notify.py deleted file mode 100644 index 8c18172..0000000 --- a/notify.py +++ /dev/null @@ -1,95 +0,0 @@ -#!/usr/bin/env python3 - -"""Send a text notification to the Home Alert ntfy topic.""" - -import argparse -import os -import sys -from urllib.error import HTTPError, URLError -from urllib.request import Request, urlopen - - -DEFAULT_TOPIC_URL = ( - "https://ntfy.sh/VK20_Home_Alert_______4671938674123049873459" -) - - -def send_notification( - message, - topic_url=None, - timeout=10, -): - """Send *message* to ntfy and return the server response body.""" - - if not isinstance(message, str): - raise TypeError("message must be a string") - - if not message: - raise ValueError("message must not be empty") - - url = topic_url or os.environ.get( - "NTFY_TOPIC_URL", - DEFAULT_TOPIC_URL, - ) - request = Request( - url, - data=message.encode("utf-8"), - headers={"Content-Type": "text/plain; charset=utf-8"}, - method="POST", - ) - - with urlopen(request, timeout=timeout) as response: - return response.read().decode("utf-8") - - -def parse_arguments(): - """Parse command-line arguments.""" - - parser = argparse.ArgumentParser( - description="Send a text message to the Home Alert ntfy topic.", - ) - parser.add_argument( - "message", - nargs="+", - help="notification text", - ) - parser.add_argument( - "--topic-url", - help="override the ntfy topic URL", - ) - parser.add_argument( - "--timeout", - type=float, - default=10, - help="request timeout in seconds (default: 10)", - ) - return parser.parse_args() - - -def main(): - args = parse_arguments() - - try: - response = send_notification( - " ".join(args.message), - topic_url=args.topic_url, - timeout=args.timeout, - ) - print(response) - return 0 - - except HTTPError as exc: - print( - f"ntfy returned HTTP {exc.code}: {exc.reason}", - file=sys.stderr, - ) - except URLError as exc: - print(f"Could not reach ntfy: {exc.reason}", file=sys.stderr) - except (TypeError, ValueError) as exc: - print(f"Invalid notification: {exc}", file=sys.stderr) - - return 1 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/service.py b/service.py index 1ddeb09..e77ef89 100644 --- a/service.py +++ b/service.py @@ -1,16 +1,21 @@ #!/usr/bin/env python3 -"""Entry point and orchestrator for enabled home-control functions.""" +"""Configuration-driven event controller for home-control plugins.""" from __future__ import annotations import argparse +import importlib import logging import signal from pathlib import Path +from threading import Event +from typing import Any, Callable import yaml -from lib import _arrival_detection + +LOG = logging.getLogger("home_control") +EVENT_FIELDS = ("sender", "event", "id", "text") def load_config(path: Path) -> dict: @@ -21,33 +26,151 @@ def load_config(path: Path) -> dict: return config -def run(config: dict, once: bool = False) -> int: - stopped = False +def _configured_callable(spec: dict, key: str, default: str) -> Callable: + module_name = spec.get("module") + if not isinstance(module_name, str) or not module_name.startswith("lib."): + raise ValueError("component module must be inside the lib package") + attribute = spec.get(key, default) + if not isinstance(attribute, str): + raise ValueError(f"component {key} must be a string") + try: + function = getattr(importlib.import_module(module_name), attribute) + except (ImportError, AttributeError) as error: + raise ValueError(f"could not load {module_name}.{attribute}: {error}") from error + if not callable(function): + raise ValueError(f"configured component {module_name}.{attribute} is not callable") + return function - def stop(*_: object) -> None: - nonlocal stopped - stopped = True - signal.signal(signal.SIGTERM, stop) - signal.signal(signal.SIGINT, stop) +class Controller: + """Load, supervise, and route plugins without plugin-specific code.""" - arrival_config = config.get("arrival_detection") - if not isinstance(arrival_config, dict): - raise ValueError("arrival_detection configuration is required") - if not arrival_config.get("enabled", True): - logging.getLogger("home_control").info("arrival detection is disabled") + def __init__(self, config: dict) -> None: + self.config = config + controller = config.get("controller", {}) + self.plugin_specs = controller.get("plugins", {}) + self.action_specs = controller.get("actions", {}) + if not isinstance(self.plugin_specs, dict): + raise ValueError("controller.plugins must be a mapping") + if not isinstance(self.action_specs, dict): + raise ValueError("controller.actions must be a mapping") + self.stopped = Event() + self.plugins: list[Any] = [] + + @staticmethod + def _validate_event(message: dict) -> None: + if not isinstance(message, dict): + raise ValueError("plugin event must be a dictionary") + missing = [field for field in EVENT_FIELDS if field not in message] + if missing: + raise ValueError(f"plugin event is missing fields: {', '.join(missing)}") + if any(not isinstance(message[field], str) for field in EVENT_FIELDS): + raise ValueError("plugin event fields must all be strings") + + def _component_config(self, spec: dict) -> dict: + section = spec.get("config") + if not isinstance(section, str): + raise ValueError("component config must name a configuration section") + component_config = self.config.get(section) + if not isinstance(component_config, dict): + raise ValueError(f"configuration section {section!r} is required") + return component_config + + def handle_event(self, message: dict) -> None: + """Route a validated plugin event to its configured action.""" + self._validate_event(message) + plugin_spec = self.plugin_specs.get(message["sender"]) + if not isinstance(plugin_spec, dict): + LOG.warning("no controller route for plugin %s", message["sender"]) + return + configured_actions = plugin_spec.get("on_event") + action_names = ( + [configured_actions] + if isinstance(configured_actions, str) + else configured_actions + ) + if not isinstance(action_names, list) or any( + not isinstance(name, str) for name in action_names + ): + raise ValueError( + f"on_event for plugin {message['sender']!r} must be a string or list of strings" + ) + + for action_name in action_names: + action_spec = self.action_specs.get(action_name) + if not isinstance(action_spec, dict): + raise ValueError( + f"unknown on_event action {action_name!r} " + f"for plugin {message['sender']!r}" + ) + handler = _configured_callable(action_spec, "handler", "on_event") + handler(self._component_config(action_spec), message) + + def _load_plugins(self) -> None: + for name, spec in self.plugin_specs.items(): + if not isinstance(spec, dict): + raise ValueError(f"controller plugin {name!r} must be a mapping") + plugin_config = self._component_config(spec) + if not plugin_config.get("enabled", True): + LOG.info("plugin %s is disabled", name) + continue + factory = _configured_callable(spec, "factory", "create") + plugin = factory(plugin_config, self.handle_event) + self.plugins.append(plugin) + + def run(self) -> int: + self._load_plugins() + for plugin in self.plugins: + plugin.start() + try: + while not self.stopped.wait(1): + if any(not plugin.is_alive for plugin in self.plugins): + LOG.error("a controller plugin stopped unexpectedly") + return 1 + finally: + for plugin in self.plugins: + plugin.stop() + for plugin in self.plugins: + plugin.join(timeout=20) return 0 - return _arrival_detection.run(arrival_config, lambda: stopped, once=once) + + def stop(self, *_: object) -> None: + self.stopped.set() + + def check_once(self) -> int: + self._load_plugins() + result = 0 + for plugin in self.plugins: + try: + check_once = getattr(plugin, "check_once", None) + if callable(check_once): + output = check_once() + if output: + print("\n".join(output) if isinstance(output, list) else output) + except (OSError, RuntimeError) as error: + LOG.error("plugin check failed: %s", error) + result = 1 + finally: + close = getattr(plugin, "close", None) + if callable(close): + close() + return result + + +def run(config: dict, once: bool = False) -> int: + controller = Controller(config) + signal.signal(signal.SIGTERM, controller.stop) + signal.signal(signal.SIGINT, controller.stop) + return controller.check_once() if once else controller.run() def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--config", type=Path, default=Path(__file__).with_name("service.yaml")) - parser.add_argument("--once", action="store_true", help="check once and print connected/absent") + parser.add_argument("--once", action="store_true", help="check plugins once") args = parser.parse_args() config = load_config(args.config) - arrival = config.get("arrival_detection", {}) - logging_config = arrival.get("logging", {}) if isinstance(arrival, dict) else {} + logging_config = config.get("controller", {}).get("logging", {}) logging.basicConfig( level=getattr(logging, str(logging_config.get("level", "INFO")).upper()), format="%(asctime)s %(levelname)s %(message)s", diff --git a/service.yaml b/service.yaml index d4994d9..70a6280 100644 --- a/service.yaml +++ b/service.yaml @@ -1,3 +1,23 @@ +controller: + plugins: + _arrival_detection: + module: lib._arrival_detection + factory: create + config: arrival_detection + on_event: [_notify, _cloud_logger] + actions: + _notify: + module: lib._notify + handler: on_event + config: notify + _cloud_logger: + module: lib._cloud_logger + handler: on_event + config: cloud_logger + + logging: + level: INFO + arrival_detection: enabled: true @@ -8,27 +28,56 @@ arrival_detection: timeout: 10 verify_tls: true hosts_oid: lanhosts - - device: - name: Ignace - # Use the iPhone's Wi-Fi Address shown for your home SSID. If Private Wi-Fi - # Address is enabled, this is the per-network private address, not hardware MAC. - mac: "ce:68:53:a8:fc:07" - + # Log every router/API call and its result without logging credentials. + debug: true + + devices: + - id: ignace + mac: "ce:68:53:a8:fc:07" + - id: janine + mac: "dc:10:57:9b:85:0e" + - id: Fatima + mac: "02:f8:26:33:2f:d9" + polling: interval_seconds: 5 present_after: 2 absent_after: 4 error_retry_seconds: 30 - - notification: - enabled: true - # Available placeholders: {name}, {mac}. - message: "{name} arrived home" - event_log: /log/detect-arrivals.txt - # Leave blank to use the default ntfy topic configured in notify.py. - topic_url: "" - timeout: 10 - - logging: - level: INFO + + event_log: log/detect-arrivals.txt + +notify: + enabled: true + filter: + # Empty accept lists allow all values. Ignore lists always take precedence. + accept: + events: [empty] + ids: [] + ignore: + events: [arrived, departed] + ids: [] + # Available placeholders: {sender}, {event}, {id}, {text}. + message: "{sender}: {event} ({id})" + messages: + arrived: "{id} arrived home" + departed: "{id} left home" + empty: "The house is empty" + # May also be supplied through the NTFY_TOPIC_URL environment variable. + topic_url: "https://ntfy.sh/VK20_Home_Alert_______4671938674123049873459" + timeout: 10 + +cloud_logger: + enabled: true + event_url: http://127.0.0.1:5000/home_logger/event + timeout: 10 + filter: + # Empty accept lists allow all values. Ignore lists always take precedence. + accept: + events: [empty] + ids: [] + ignore: + events: [arrived, departed] + ids: [] + # Available placeholders: {sender}, {event}, {id}, {text}. + message: "{datetime} {sender}: {event} ({id})" \ No newline at end of file diff --git a/tests/test_service.py b/tests/test_service.py index 89df4ac..efb983e 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -4,7 +4,14 @@ from tempfile import TemporaryDirectory from unittest.mock import patch from lib._zyxel import ZyxelRouter, normalize_mac -from lib._arrival_detection import ArrivalDetector, _notify +from lib._arrival_detection import ( + ArrivalDetector, + ArrivalMonitor, + _configured_devices, + _record_arrival, +) +from lib._notify import accepts_event +from service import Controller class ArrivalDetectorTests(unittest.TestCase): @@ -26,6 +33,93 @@ class ArrivalDetectorTests(unittest.TestCase): self.assertFalse(detector.sample(False)) self.assertFalse(detector.sample(True)) + def test_accepts_up_to_three_devices(self): + devices = _configured_devices( + { + "devices": [ + {"id": "one", "mac": "00:00:00:00:00:01"}, + {"id": "two", "mac": "00:00:00:00:00:02"}, + {"id": "three", "mac": "00:00:00:00:00:03"}, + ] + } + ) + self.assertEqual(len(devices), 3) + + def test_rejects_more_than_three_devices(self): + config = { + "devices": [ + {"id": str(index), "mac": f"00:00:00:00:00:0{index}"} + for index in range(1, 5) + ] + } + with self.assertRaisesRegex(ValueError, "at most 3"): + _configured_devices(config) + + @patch("lib._arrival_detection._build_router") + def test_monitor_emits_generic_arrived_event(self, _build_router): + events = [] + monitor = ArrivalMonitor( + { + "devices": [ + {"id": "ignace", "mac": "00:00:00:00:00:01"} + ], + "polling": {"absent_after": 1, "present_after": 1}, + }, + events.append, + ) + mac = "00:00:00:00:00:01" + monitor._process_states({mac: False}) + monitor._process_states({mac: True}) + self.assertEqual( + events, + [ + { + "sender": "_arrival_detection", + "event": "arrived", + "id": "ignace", + "text": "", + } + ], + ) + + @patch("lib._arrival_detection._build_router") + def test_monitor_emits_departure_and_empty_after_last_device_leaves(self, _build_router): + events = [] + monitor = ArrivalMonitor( + { + "devices": [ + {"id": "ignace", "mac": "00:00:00:00:00:01"}, + {"id": "janine", "mac": "00:00:00:00:00:02"}, + ], + "polling": {"absent_after": 1, "present_after": 1}, + }, + events.append, + ) + first = "00:00:00:00:00:01" + second = "00:00:00:00:00:02" + monitor._process_states({first: True, second: True}) + self.assertEqual(events, []) + monitor._process_states({first: False, second: True}) + self.assertEqual([event["event"] for event in events], ["departed"]) + monitor._process_states({first: False, second: False}) + self.assertEqual( + [(event["event"], event["id"]) for event in events], + [("departed", "ignace"), ("departed", "janine"), ("empty", "house")], + ) + + @patch("lib._arrival_detection._build_router") + def test_initially_empty_house_emits_no_events(self, _build_router): + events = [] + monitor = ArrivalMonitor( + { + "devices": [{"id": "ignace", "mac": "00:00:00:00:00:01"}], + "polling": {"absent_after": 1, "present_after": 1}, + }, + events.append, + ) + monitor._process_states({"00:00:00:00:00:01": False}) + self.assertEqual(events, []) + class MacTests(unittest.TestCase): def test_normalizes_common_formats(self): @@ -57,38 +151,175 @@ class MacTests(unittest.TestCase): self.assertEqual(hosts[0].mac, "2c:db:07:50:2b:5c") self.assertTrue(router.is_connected("2c-db-07-50-2b-5c")) + def test_logs_only_the_first_successful_zyxel_connection(self): + router = ZyxelRouter("192.0.2.1", "user", "password") + with self.assertLogs("home_control.zyxel", level="INFO") as logs: + router._log_first_connection() + router._log_first_connection() + self.assertEqual(len(logs.output), 1) + self.assertIn("Connected successfully", logs.output[0]) + + def test_debug_logs_presence_result(self): + router = ZyxelRouter("192.0.2.1", "user", "password", debug=True) + router.get_lan_hosts = lambda: [] + with self.assertLogs("home_control.zyxel", level="INFO") as logs: + self.assertFalse(router.is_connected("2c:db:07:50:2b:5c")) + self.assertIn("2c:db:07:50:2b:5c -> absent", logs.output[0]) + class NotificationTests(unittest.TestCase): - @patch("lib._arrival_detection.send_notification") - def test_uses_notify_module(self, send_notification): + @patch("lib._cloud_logger.send_event") + @patch("lib._notify.send_notification") + def test_controller_routes_event_to_multiple_actions( + self, send_notification, send_event + ): config = { - "notification": { + "controller": { + "plugins": { + "source": { + "module": "lib.example", + "config": "source", + "on_event": ["_notify", "_cloud_logger"], + } + }, + "actions": { + "_notify": { + "module": "lib._notify", + "config": "notify", + }, + "_cloud_logger": { + "module": "lib._cloud_logger", + "config": "cloud_logger", + }, + }, + }, + "source": {}, + "notify": {"topic_url": "https://ntfy.example/home"}, + "cloud_logger": {"event_url": "https://logger.example/event"}, + } + message = { + "sender": "source", + "event": "arrived", + "id": "ignace", + "text": "hello", + } + + Controller(config).handle_event(message) + + send_notification.assert_called_once() + send_event.assert_called_once() + payload = send_event.call_args.args[0] + self.assertEqual(payload["sender"], "source") + self.assertEqual(payload["event"], "arrived") + self.assertEqual(payload["id"], "ignace") + self.assertEqual(payload["text"], "hello") + self.assertIn("datetime", payload) + self.assertEqual( + send_event.call_args.kwargs, + {"event_url": "https://logger.example/event", "timeout": 10.0}, + ) + + @patch("lib._notify.send_notification") + def test_controller_routes_generic_event_to_notify(self, send_notification): + config = { + "controller": { + "plugins": { + "_arrival_detection": { + "module": "lib._arrival_detection", + "config": "arrival_detection", + "on_event": "_notify", + } + }, + "actions": { + "_notify": { + "module": "lib._notify", + "handler": "on_event", + "config": "notify", + } + }, + }, + "arrival_detection": {}, + "notify": { "enabled": True, - "message": "Welcome home, {name} ({mac})", + "message": "Welcome home, {id}", "topic_url": "https://ntfy.example/home", "timeout": 4, - } + }, } - _notify(config, {"name": "Ignace", "mac": "aa:bb:cc:dd:ee:ff"}) + Controller(config).handle_event( + { + "sender": "_arrival_detection", + "event": "arrived", + "id": "ignace", + "text": "", + } + ) send_notification.assert_called_once_with( - "Welcome home, Ignace (aa:bb:cc:dd:ee:ff)", + "Welcome home, ignace", topic_url="https://ntfy.example/home", timeout=4.0, ) - @patch("lib._arrival_detection.send_notification") - def test_appends_notification_to_event_log(self, send_notification): + def test_controller_has_no_plugin_specific_setup(self): + config = { + "controller": { + "plugins": { + "example": { + "module": "lib.example", + "factory": "build", + "config": "example_settings", + "on_event": "example_action", + } + }, + "actions": {}, + }, + "example_settings": {}, + } + controller = Controller(config) + self.assertEqual(controller.plugin_specs["example"]["factory"], "build") + + def test_notify_accept_filter(self): + config = { + "filter": { + "accept": {"events": ["arrived"], "ids": ["ignace"]}, + "ignore": {"events": [], "ids": []}, + } + } + self.assertTrue( + accepts_event(config, {"event": "arrived", "id": "ignace"}) + ) + self.assertFalse( + accepts_event(config, {"event": "departed", "id": "ignace"}) + ) + self.assertFalse( + accepts_event(config, {"event": "arrived", "id": "janine"}) + ) + + def test_notify_ignore_filter_takes_precedence(self): + config = { + "filter": { + "accept": {"events": ["departed"], "ids": ["ignace"]}, + "ignore": {"events": ["departed"], "ids": []}, + } + } + self.assertFalse( + accepts_event(config, {"event": "departed", "id": "ignace"}) + ) + + def test_monitor_appends_arrival_to_event_log(self): with TemporaryDirectory() as directory: event_log = Path(directory) / "arrivals.txt" config = { - "notification": { - "enabled": True, - "message": "{name} arrived home", - "event_log": str(event_log), - } + "event_log": str(event_log), } - _notify(config, {"name": "Ignace", "mac": "aa:bb:cc:dd:ee:ff"}) - self.assertTrue(event_log.read_text().endswith(" Ignace arrived home\n")) + _record_arrival( + config, {"id": "ignace", "mac": "aa:bb:cc:dd:ee:ff"} + ) + self.assertTrue( + event_log.read_text().endswith( + " ARRIVED id=ignace mac=aa:bb:cc:dd:ee:ff\n" + ) + ) if __name__ == "__main__":