diff --git a/.gitignore b/.gitignore index 7f61e484..c8ffdc7c 100644 --- a/.gitignore +++ b/.gitignore @@ -53,4 +53,7 @@ superpowers/ # coverage (legacy locations) coverage.lcov lcov.info -htmlcov/ \ No newline at end of file +htmlcov/ + +# Local session runner logs (scripts/run_session.py) +logs/ diff --git a/Makefile b/Makefile index 28328130..29fd58aa 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,69 @@ -.PHONY: setup test lint sync verify-dist +.PHONY: help setup test coverage lint sync verify-dist run-file run-live simulate-file simulate-live live-forget logs status stop call + +# Describe every target. `setup` stays the default goal (see .DEFAULT_GOAL). +help: + @echo "Usage: make [OPTION=value ...]" + @echo "" + @echo "Development" + @echo " setup Install the package in editable mode and the pre-commit hooks" + @echo " test Run the full test suite (coverage needs the whole suite)" + @echo " coverage Run the tests under coverage and write an lcov report" + @echo " lint Run every linter through pre-commit" + @echo " sync Generate the sync pyhiveapi package from the async source" + @echo " verify-dist Build the wheel and check it imports in a clean venv" + @echo "" + @echo "Local session runner (read-only unless started with WRITE=1)" + @echo "Each run-*/simulate-* target starts a background process that keeps" + @echo "polling until 'make stop', and writes to logs/.log (gitignored)." + @echo " run-file Session against the bundled fixture (no account, no network)" + @echo " run-live Session against a real Hive account. Asks for email and" + @echo " password (hidden, not stored) and, on the first run, the SMS" + @echo " code, then moves to the background" + @echo " simulate-file As run-file, but strips a state key from every poll after" + @echo " the first, to imitate Hive returning incomplete state" + @echo " simulate-live As run-live, with the same stripping applied to real data" + @echo " logs Follow the runner logs (Ctrl-C stops following, not the runners)" + @echo " status List the runners that are running" + @echo " stop Stop the runners" + @echo " call Run an admin command inside a running session, e.g." + @echo " CMD=\"mode Hallway\". With no CMD it lists the commands" + @echo " live-forget Delete the device credentials run-live saved in the macOS Keychain" + @echo "" + @echo "Options" + @echo " INTERVAL= Seconds between polls (default 120; simulate-file 5, simulate-live 30)" + @echo " POLLS= Stop after n polls (default: run until stopped)" + @echo " DROP= State key the simulate-* targets strip (default mode)" + @echo " DEBUG=1 Turn on the library's debug logging" + @echo " WRITE=1 Start a session whose admin commands may change devices" + @echo " CMD=\"...\" The admin command for 'make call'" + @echo " NAME= Limit logs/stop/call to one of: file, live, simulate-file, simulate-live" + @echo "" + @echo "Admin commands (make call CMD=\"\")" + @echo "DEVICE is a name or the 8-character ID shown by 'devices'." + @echo " help List the commands and whether writes are allowed" + @echo " devices List the devices the other commands can address" + @echo " poll Poll Hive now" + @echo " mode DEVICE The mode the library reports (heating, hot water)" + @echo " state DEVICE The state as last received, without its schedule" + @echo " drop KEY... | off Start or stop stripping state keys from polls" + @echo " debug on | off Library debug logging" + @echo " interval SECONDS Seconds between polls, from the next wait" + @echo "Only in a session started with WRITE=1 (these change real devices):" + @echo " set-mode DEVICE MODE Set a heating or hot water mode" + @echo " set-temp DEVICE TEMP Set a heating target temperature" + @echo " boost DEVICE MINS [TEMP] Start a boost (heating needs TEMP)" + @echo " boost-off DEVICE Cancel a boost" + @echo " turn-on DEVICE Turn a light or plug on" + @echo " turn-off DEVICE Turn a light or plug off" + @echo "" + @echo "Examples" + @echo " make simulate-file INTERVAL=2 POLLS=8" + @echo " make logs NAME=simulate-file" + @echo " make stop NAME=live" + @echo " make call CMD=\"drop mode\" # start stripping mode from polls" + @echo " make run-live WRITE=1 # then: make call CMD=\"set-mode Hallway OFF\"" + +.DEFAULT_GOAL := setup setup: pip install -e ".[dev]" @@ -22,3 +87,57 @@ verify-dist: rm -rf dist python -m build python scripts/verify_dist.py dist + +# --- Local session runner (scripts/run_session.py) -------------------------- +# Each run-*/simulate-* target starts a background process that polls until +# `make stop`, and writes to logs/ (gitignored). They are read-only unless +# started with WRITE=1, which lets `make call` commands change devices. +# Options: INTERVAL=, DROP=, +# DEBUG=1 (library debug logging), POLLS= (stop after n polls). +RUN_SESSION = python3 scripts/run_session.py +RUN_FLAGS = $(if $(POLLS),--polls $(POLLS)) $(if $(DEBUG),--debug) \ + $(if $(WRITE),--allow-writes) +RUN_OPTS = $(RUN_FLAGS) $(if $(INTERVAL),--interval $(INTERVAL)) + +# Session against the bundled fixture (no account, no network). +run-file: + $(RUN_SESSION) --file $(RUN_OPTS) + +# Session against a real Hive account. Prompts for the password (hidden, never +# stored) and, on the first run, the SMS code, then moves to the background. +run-live: + $(RUN_SESSION) --live $(RUN_OPTS) + +# Simulate Hive returning incomplete state: DROP (default `mode`) is stripped +# from every poll after the first. +simulate-file: + $(RUN_SESSION) --file --drop $(or $(DROP),mode) --interval $(or $(INTERVAL),5) \ + $(RUN_FLAGS) + +simulate-live: + $(RUN_SESSION) --live --drop $(or $(DROP),mode) --interval $(or $(INTERVAL),30) \ + $(RUN_FLAGS) + +# Follow the logs of every runner (Ctrl-C stops following, not the runners). +# NAME= follows just one. +logs: + @mkdir -p logs && touch logs/$(or $(NAME),file).log + tail -n 40 -F logs/$(or $(NAME),*).log + +# List the running background runners. +status: + @$(RUN_SESSION) --status + +# Stop every runner, or just NAME=. +stop: + @$(RUN_SESSION) --stop $(if $(NAME),--name $(NAME)) + +# Run an admin command inside a running session: make call CMD="mode Hallway". +# `make call` alone lists the commands. NAME= picks the session when +# more than one is running. +call: + @$(RUN_SESSION) $(if $(NAME),--name $(NAME)) --call $(or $(CMD),help) + +# Remove the device credentials saved in the macOS Keychain by run-live. +live-forget: + $(RUN_SESSION) --forget diff --git a/scripts/run_session.py b/scripts/run_session.py new file mode 100644 index 00000000..f2384993 --- /dev/null +++ b/scripts/run_session.py @@ -0,0 +1,693 @@ +"""Run a Hive session in the background for local testing. + +Starts a session against the bundled fixture (``--file``) or a real Hive +account (``--live``) as a background process that polls until it is stopped +(``--stop``), logging the shape of each product's state to ``logs/``. + +A running session takes admin commands (``--call``) over a local socket in +``logs/``. It is read-only unless started with ``--allow-writes``: without +that flag every command that would change a device is refused. + +Live mode asks for the password and SMS code at hidden/interactive prompts. +The password is never stored. After the first SMS login the device +credentials are saved in the macOS Keychain so later runs skip the SMS. + +Live mode logs in in the foreground, then hands the session to the background +process over a pipe, so nothing sensitive is put on a command line. + +The summary shows product types, shortened IDs, the mode and state key names +only. Library log lines are printed as the library emits them and can include +device names. +""" + +from __future__ import annotations + +import argparse +import asyncio +import getpass +import inspect +import json +import logging +import logging.handlers +import os +import shutil +import signal +import subprocess +import sys +from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field +from pathlib import Path + +from apyhiveapi import SMS_REQUIRED, Auth, Hive +from apyhiveapi.helper.hive_exceptions import HiveInvalidDeviceAuthentication + +FILE_USERNAME = "use@file.com" +KEYCHAIN_SERVICE = "pyhive-dev" +KEYCHAIN_ACCOUNT = "live-session" +SHORT_ID_LENGTH = 8 +LOG_DIR = Path(__file__).resolve().parent.parent / "logs" +LOG_MAX_BYTES = 5_000_000 +LOG_BACKUPS = 2 + +CALL_TIMEOUT = 60 +MODE_TYPES = ("climate", "water_heater") +SWITCHABLE_TYPES = ("light", "switch") + +LOG = logging.getLogger("run_session") + + +@dataclass +class Settings: + """What a running session can have changed by an admin command.""" + + interval: float + allow_writes: bool = False + drop_keys: list[str] = field(default_factory=list) + drop_on: bool = False + + +def _keychain_available() -> bool: + return sys.platform == "darwin" and shutil.which("security") is not None + + +def load_stored() -> dict: + """Return the stored username and device credentials, or an empty dict.""" + if not _keychain_available(): + return {} + result = subprocess.run( + [ + "security", + "find-generic-password", + "-s", + KEYCHAIN_SERVICE, + "-a", + KEYCHAIN_ACCOUNT, + "-w", + ], + capture_output=True, + text=True, + check=False, + ) + if result.returncode != 0: + return {} + try: + return json.loads(bytes.fromhex(result.stdout.strip())) + except ValueError: + return {} + + +def save_stored(username: str, device_data: list) -> bool: + """Save the username and device credentials in the Keychain.""" + if not _keychain_available(): + return False + value = json.dumps({"username": username, "device_data": device_data}) + # Sent on stdin (hex-encoded, so no quoting) to keep it off the process list. + command = ( + f"add-generic-password -U -s {KEYCHAIN_SERVICE} -a {KEYCHAIN_ACCOUNT} " + f"-w {value.encode().hex()}\n" + ) + result = subprocess.run( + ["security", "-i"], + input=command, + capture_output=True, + text=True, + check=False, + ) + return result.returncode == 0 + + +def forget_stored() -> None: + """Delete the stored username and device credentials.""" + if not _keychain_available(): + print("No Keychain on this platform - nothing is stored.") + return + result = subprocess.run( + [ + "security", + "delete-generic-password", + "-s", + KEYCHAIN_SERVICE, + "-a", + KEYCHAIN_ACCOUNT, + ], + capture_output=True, + check=False, + ) + print( + "Stored credentials removed." if result.returncode == 0 else "Nothing stored." + ) + + +def _tokens(result: dict) -> dict: + auth_result = result["AuthenticationResult"] + return { + "token": auth_result["IdToken"], + "refreshToken": auth_result["RefreshToken"], + "accessToken": auth_result["AccessToken"], + } + + +async def live_config() -> dict: + """Log in to a real Hive account and return the start_session config.""" + stored = load_stored() + username = stored.get("username") or input("Hive email: ").strip() + password = getpass.getpass("Hive password (hidden, not stored): ") + device_data = stored.get("device_data") + + result: dict = {} + if device_data: + print("Logging in with the stored device credentials (no SMS)...") + auth = Auth(username, password, *device_data) + try: + # The password login comes first; Hive answers it with a device + # challenge, which device_login() then completes. + result = await auth.login() + if result.get("ChallengeName") == Auth.DEVICE_VERIFIER_CHALLENGE: + result = await auth.device_login() + except HiveInvalidDeviceAuthentication: + result = {} + if "AuthenticationResult" not in result: + print("Hive rejected the stored device credentials - signing in again.") + forget_stored() + device_data = None + + if not device_data: + print("Logging in...") + auth = Auth(username, password) + result = await auth.login() + if result.get("ChallengeName") == SMS_REQUIRED: + code = input("SMS code: ").strip() + result = await auth.sms_2fa(code, result) + if "AuthenticationResult" not in result: + sys.exit(f"Login did not complete: {result.get('ChallengeName')}") + if auth.device_key: + # Cognito ties the new tokens to this device: without registering + # it the refresh token is rejected and the next login needs an SMS. + await auth.device_registration() + device_data = list((await auth.get_device_data())[:3]) + if all(device_data) and save_stored(username, device_data): + print("Device credentials saved in the macOS Keychain.") + else: + print("Device credentials not saved - the next run will need an SMS.") + + config = {"tokens": _tokens(result), "username": username, "password": password} + if all(device_data): + config["device_data"] = device_data + return config + + +def install_drop(hive: Hive, settings: Settings) -> None: + """Strip ``settings.drop_keys`` from product state while ``drop_on`` is set. + + Simulates the incomplete responses Hive occasionally returns. + """ + + def _strip(response): + if settings.drop_on and isinstance(response, dict): + for product in (response.get("parsed") or {}).get("products", []): + state = product.get("state") + if isinstance(state, dict): + for key in settings.drop_keys: + state.pop(key, None) + return response + + # Fixture sessions read through open_file, live sessions through get_all. + open_file = hive.open_file + get_all = hive.api.get_all + + async def _get_all(): + return _strip(await get_all()) + + hive.open_file = lambda file: _strip(open_file(file)) + hive.api.get_all = _get_all + + +def log_summary(hive: Hive, label: str) -> None: + """Log one line per product: type, short ID, mode and state key names.""" + LOG.info("--- %s: %d products ---", label, len(hive.data.products)) + rows = sorted( + hive.data.products.items(), key=lambda item: (str(item[1].get("type")), item[0]) + ) + for product_id, product in rows: + state = product.get("state") if isinstance(product.get("state"), dict) else {} + LOG.info( + "%-20s %-9s mode=%-9s keys=%s", + product.get("type"), + product_id[-SHORT_ID_LENGTH:], + state.get("mode"), + ", ".join(sorted(state)), + ) + + +class Admin: + """The commands a running session accepts over its control socket. + + A fixed list rather than "call anything": the session object also holds + the account's tokens and password, which no command hands out. + """ + + def __init__(self, hive: Hive, settings: Settings) -> None: + """Bind the commands to a started session.""" + self.hive = hive + self.settings = settings + # name: (handler, changes a device, arguments, description) + self.commands: dict[str, tuple[Callable[..., Awaitable], bool, str, str]] = { + "help": (self.help, False, "", "list the commands"), + "devices": (self.devices, False, "", "list the controllable devices"), + "poll": (self.poll, False, "", "poll Hive now"), + "mode": (self.mode, False, "DEVICE", "mode the library reports"), + "state": (self.state, False, "DEVICE", "state as last received"), + "drop": (self.drop, False, "KEY... | off", "strip state keys from polls"), + "debug": (self.debug, False, "on | off", "library debug logging"), + "interval": (self.interval, False, "SECONDS", "seconds between polls"), + "set-mode": (self.set_mode, True, "DEVICE MODE", "set heating/hot water"), + "set-temp": (self.set_temp, True, "DEVICE TEMP", "set target temperature"), + "boost": (self.boost, True, "DEVICE MINS [TEMP]", "start a boost"), + "boost-off": (self.boost_off, True, "DEVICE", "cancel a boost"), + "turn-on": (self.turn_on, True, "DEVICE", "turn a light or plug on"), + "turn-off": (self.turn_off, True, "DEVICE", "turn a light or plug off"), + } + + def _find(self, ref: str, types: tuple[str, ...]): + """Return the one device of *types* matching a name or ID suffix.""" + matches = [ + device + for ha_type in types + for device in self.hive.device_list.get(ha_type, []) + if device.ha_name.lower() == ref.lower() or device.hive_id.endswith(ref) + ] + if len(matches) != 1: + found = "no" if not matches else f"{len(matches)}" + raise ValueError( + f"{found} {'/'.join(types)} device matches {ref!r} - see `devices`" + ) + return matches[0] + + def _module(self, device): + return { + "climate": self.hive.heating, + "water_heater": self.hive.hotwater, + "light": self.hive.light, + "switch": self.hive.switch, + }[device.ha_type] + + async def run(self, name: str, args: list[str]): + """Run one command and return its JSON-serialisable result.""" + if name not in self.commands: + raise ValueError(f"unknown command {name!r} - try `help`") + handler, writes, _, _ = self.commands[name] + if writes and not self.settings.allow_writes: + raise PermissionError( + f"{name} changes a device and this session is read-only. " + "Restart it with WRITE=1 to allow that." + ) + try: + inspect.signature(handler).bind(*args) + except TypeError as error: + raise ValueError(self._usage(name)) from error + return await handler(*args) + + def _usage(self, name: str) -> str: + return f"usage: {name} {self.commands[name][2]}" + + async def help(self): + """List the commands, marking the ones this session refuses.""" + return { + "writes_allowed": self.settings.allow_writes, + "commands": [ + f"{name} {usage}".strip().ljust(28) + + description + + (" [needs WRITE=1]" if writes else "") + for name, (_, writes, usage, description) in self.commands.items() + ], + } + + async def devices(self): + """List the devices the other commands can address.""" + return [ + f"{device.ha_type:<13}{device.hive_id[-SHORT_ID_LENGTH:]} {device.ha_name}" + for ha_type in MODE_TYPES + SWITCHABLE_TYPES + for device in self.hive.device_list.get(ha_type, []) + ] + + async def poll(self): + """Poll now; False means it failed or one was already running.""" + ok = await self.hive.force_update() + log_summary(self.hive, f"admin poll{'' if ok else ' (FAILED)'}") + return ok + + async def mode(self, ref: str): + """Return what get_mode reports for a heating or hot water device.""" + device = self._find(ref, MODE_TYPES) + return await self._module(device).get_mode(device) + + async def state(self, ref: str): + """Return a device's state block, without its schedule.""" + device = self._find(ref, MODE_TYPES + SWITCHABLE_TYPES) + state = self.hive.data.products.get(device.hive_id, {}).get("state") or {} + return {key: value for key, value in state.items() if key != "schedule"} + + async def drop(self, *keys: str): + """Strip *keys* from every later poll, or stop with ``off``.""" + if not keys: + raise ValueError(self._usage("drop")) + self.settings.drop_on = keys != ("off",) + if self.settings.drop_on: + self.settings.drop_keys[:] = keys + return {"dropping": self.settings.drop_keys if self.settings.drop_on else []} + + async def debug(self, switch: str): + """Turn the library's debug logging on or off.""" + if switch not in ("on", "off"): + raise ValueError(self._usage("debug")) + level = logging.DEBUG if switch == "on" else logging.WARNING + logging.getLogger("apyhiveapi").setLevel(level) + return {"debug": switch} + + async def interval(self, seconds: str): + """Change the polling interval, from the next wait onwards.""" + value = float(seconds) + if value < 1: + raise ValueError("interval must be at least 1 second") + self.settings.interval = value + return {"interval": value, "applies": "after the current wait"} + + async def set_mode(self, ref: str, new_mode: str): + """Set a heating or hot water mode.""" + device = self._find(ref, MODE_TYPES) + return await self._module(device).set_mode(device, new_mode.upper()) + + async def set_temp(self, ref: str, temp: str): + """Set a heating target temperature.""" + device = self._find(ref, ("climate",)) + return await self.hive.heating.set_target_temperature(device, float(temp)) + + async def boost(self, ref: str, mins: str, temp: str | None = None): + """Start a heating (needs TEMP) or hot water boost.""" + device = self._find(ref, MODE_TYPES) + if device.ha_type == "water_heater": + return await self.hive.hotwater.set_boost_on(device, int(mins)) + if temp is None: + raise ValueError( + "a heating boost needs a temperature: boost DEVICE MINS TEMP" + ) + return await self.hive.heating.set_boost_on(device, mins, float(temp)) + + async def boost_off(self, ref: str): + """Cancel a boost.""" + device = self._find(ref, MODE_TYPES) + return await self._module(device).set_boost_off(device) + + async def turn_on(self, ref: str): + """Turn a light or plug on.""" + device = self._find(ref, SWITCHABLE_TYPES) + if device.ha_type == "light": + return await self.hive.light.turn_on(device, None, None, None) + return await self.hive.switch.turn_on(device) + + async def turn_off(self, ref: str): + """Turn a light or plug off.""" + device = self._find(ref, SWITCHABLE_TYPES) + return await self._module(device).turn_off(device) + + async def serve(self, reader, writer) -> None: + """Answer one request: a JSON line in, a JSON line out.""" + try: + request = json.loads(await reader.readline()) + name, args = request["command"], request.get("args", []) + LOG.info("admin: %s %s", name, " ".join(args)) + try: + reply = {"ok": True, "result": await self.run(name, args)} + except (ValueError, PermissionError) as error: + reply = {"ok": False, "error": str(error)} + except Exception as error: # pylint: disable=broad-exception-caught + LOG.exception("admin: %s failed", name) + reply = {"ok": False, "error": f"{type(error).__name__}: {error}"} + LOG.info("admin: %s -> %s", name, json.dumps(reply, default=str)) + writer.write(json.dumps(reply, default=str).encode() + b"\n") + await writer.drain() + finally: + writer.close() + + +async def run(args: argparse.Namespace, config: dict) -> None: + """Start the session, then poll and log a summary until stopped.""" + hive = Hive(username=config["username"], password=config.get("password", "")) + settings = Settings( + interval=args.interval, + allow_writes=args.allow_writes, + drop_keys=list(args.drop), + ) + async with hive: + install_drop(hive, settings) + device_list = await hive.start_session(config) + entities = ", ".join(f"{k}={len(v)}" for k, v in sorted(device_list.items())) + LOG.info("Session started. Entities: %s", entities) + log_summary(hive, "poll 1") + + if args.drop: + settings.drop_on = True + LOG.info("Dropping state keys from later polls: %s", ", ".join(args.drop)) + if settings.allow_writes: + LOG.warning("Admin commands that change devices are ENABLED.") + socket_path = _socket_file(args.worker) + socket_path.unlink(missing_ok=True) + server = await asyncio.start_unix_server( + Admin(hive, settings).serve, path=str(socket_path) + ) + socket_path.chmod(0o600) + async with server: + poll = 1 + while args.polls == 0 or poll < args.polls: + poll += 1 + await asyncio.sleep(settings.interval) + ok = await hive.force_update() + log_summary(hive, f"poll {poll}{'' if ok else ' (FAILED)'}") + + +def _pid_file(name: str) -> Path: + return LOG_DIR / f"{name}.pid" + + +def _socket_file(name: str) -> Path: + return LOG_DIR / f"{name}.sock" + + +def _running_pid(name: str) -> int | None: + """Return the PID of the named runner if it is still our process.""" + try: + pid = int(_pid_file(name).read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + result = subprocess.run( + ["ps", "-p", str(pid), "-o", "command="], + capture_output=True, + text=True, + check=False, + ) + if Path(__file__).name in result.stdout: + return pid + _pid_file(name).unlink(missing_ok=True) + return None + + +def _names() -> list[str]: + return sorted(path.stem for path in LOG_DIR.glob("*.pid")) + + +def stop(name: str | None) -> None: + """Stop the named runner, or every runner when *name* is None.""" + names = [name] if name else _names() + stopped = False + for runner in names: + pid = _running_pid(runner) + if pid is not None: + os.kill(pid, signal.SIGTERM) + _pid_file(runner).unlink(missing_ok=True) + print(f"Stopped {runner} (pid {pid}).") + stopped = True + if not stopped: + print("Nothing running.") + + +def status() -> None: + """Print each running runner with its PID and log file.""" + running = [(name, _running_pid(name)) for name in _names()] + running = [(name, pid) for name, pid in running if pid is not None] + if not running: + print("Nothing running.") + for name, pid in running: + print(f"{name}: running (pid {pid}) - {LOG_DIR / (name + '.log')}") + + +async def _send(name: str, command: list[str]) -> dict: + reader, writer = await asyncio.open_unix_connection(str(_socket_file(name))) + request = {"command": command[0], "args": command[1:]} + writer.write(json.dumps(request).encode() + b"\n") + await writer.drain() + try: + return json.loads(await asyncio.wait_for(reader.readline(), CALL_TIMEOUT)) + finally: + writer.close() + + +def call(name: str | None, command: list[str]) -> None: + """Run an admin command inside a running session and print the reply.""" + running = [runner for runner in _names() if _running_pid(runner) is not None] + if name is None and len(running) == 1: + name = running[0] + if name not in running: + sys.exit( + f"Pick a running session with NAME=. Running: " + f"{', '.join(running) or 'none'}." + ) + try: + reply = asyncio.run(_send(name, command)) + except (OSError, ValueError, asyncio.TimeoutError) as error: + sys.exit(f"Could not reach {name}: {error or type(error).__name__}") + if not reply.get("ok"): + sys.exit(f"{name}: {reply.get('error')}") + result = reply.get("result") + if isinstance(result, dict) and isinstance(result.get("commands"), list): + print(f"writes allowed: {result['writes_allowed']}") + result = result["commands"] + if isinstance(result, list): + print("\n".join(str(item) for item in result)) + else: + print(json.dumps(result, indent=2, default=str)) + + +def launch(args: argparse.Namespace, name: str) -> None: + """Log in if needed (foreground), then start the background worker.""" + pid = _running_pid(name) + if pid is not None: + sys.exit(f"{name} is already running (pid {pid}). Run `make stop` first.") + if args.live: + config = asyncio.run(live_config()) + else: + config = {"username": FILE_USERNAME} + + LOG_DIR.mkdir(exist_ok=True) + command = [sys.executable, __file__, "--worker", name] + command += ["--polls", str(args.polls), "--interval", str(args.interval)] + for key in args.drop: + command += ["--drop", key] + if args.debug: + command.append("--debug") + if args.allow_writes: + command.append("--allow-writes") + # Not a context manager: the worker must outlive this process. + worker = subprocess.Popen( # pylint: disable=consider-using-with + command, + stdin=subprocess.PIPE, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + text=True, + ) + # The session config (tokens, password) goes over the pipe, not argv. + if worker.stdin is not None: + worker.stdin.write(json.dumps(config)) + worker.stdin.close() + _pid_file(name).write_text(str(worker.pid), encoding="utf-8") + print(f"Started {name} in the background (pid {worker.pid}).") + print(f"Log: {LOG_DIR / (name + '.log')}") + print("`make logs` to follow it, `make stop` to stop it.") + + +def work(args: argparse.Namespace) -> None: + """Background worker: read the config from stdin and run until stopped.""" + config = json.loads(sys.stdin.read()) + LOG_DIR.mkdir(exist_ok=True) + handler = logging.handlers.RotatingFileHandler( + LOG_DIR / f"{args.worker}.log", + maxBytes=LOG_MAX_BYTES, + backupCount=LOG_BACKUPS, + encoding="utf-8", + ) + handler.setFormatter( + logging.Formatter("%(asctime)s %(levelname)s [%(name)s] %(message)s") + ) + logging.getLogger().addHandler(handler) + LOG.setLevel(logging.INFO) + logging.getLogger("apyhiveapi").setLevel( + logging.DEBUG if args.debug else logging.WARNING + ) + signal.signal(signal.SIGTERM, signal.default_int_handler) + LOG.info("=== %s started (pid %d) ===", args.worker, os.getpid()) + try: + asyncio.run(run(args, config)) + LOG.info("=== %s finished ===", args.worker) + except KeyboardInterrupt: + LOG.info("=== %s stopped ===", args.worker) + except Exception: # pylint: disable=broad-exception-caught + LOG.exception("=== %s crashed ===", args.worker) + finally: + _socket_file(args.worker).unlink(missing_ok=True) + pid_file = _pid_file(args.worker) + if pid_file.exists() and pid_file.read_text(encoding="utf-8") == str( + os.getpid() + ): + pid_file.unlink() + + +def main() -> None: + """Parse arguments and run.""" + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + mode = parser.add_mutually_exclusive_group(required=True) + mode.add_argument("--file", action="store_true", help="use the bundled fixture") + mode.add_argument("--live", action="store_true", help="use a real Hive account") + mode.add_argument( + "--forget", action="store_true", help="delete the stored device credentials" + ) + mode.add_argument("--stop", action="store_true", help="stop background runners") + mode.add_argument("--status", action="store_true", help="list background runners") + mode.add_argument( + "--call", + nargs="+", + metavar="WORD", + help="run an admin command inside a running session (try: --call help)", + ) + mode.add_argument("--worker", metavar="NAME", help=argparse.SUPPRESS) + parser.add_argument("--name", help="runner to act on (default: derived, or all)") + parser.add_argument( + "--polls", + type=int, + default=0, + help="polls to run; 0 (default) runs until stopped", + ) + parser.add_argument( + "--interval", type=float, default=120, help="seconds between polls" + ) + parser.add_argument( + "--drop", + action="append", + default=[], + metavar="KEY", + help="state key to strip from polls after the first (repeatable)", + ) + parser.add_argument("--debug", action="store_true", help="library debug logging") + parser.add_argument( + "--allow-writes", + action="store_true", + help="let admin commands change devices (default: read-only)", + ) + args = parser.parse_args() + + if args.forget: + forget_stored() + elif args.stop: + stop(args.name) + elif args.status: + status() + elif args.call: + call(args.name, args.call) + elif args.worker: + work(args) + else: + source = "live" if args.live else "file" + launch(args, args.name or (f"simulate-{source}" if args.drop else source)) + + +if __name__ == "__main__": + main() diff --git a/src/devices/heating.py b/src/devices/heating.py index de458d71..934aa562 100644 --- a/src/devices/heating.py +++ b/src/devices/heating.py @@ -164,20 +164,13 @@ async def get_mode(self, device: Device): Returns: str: Current Mode """ - state = None - final = None - device_name = device.ha_name - - try: - data = self.session.data.products[device.hive_id] - state = data["state"]["mode"] - if state == "BOOST": - state = self._get_product_state(device, "props", "previous", "mode") - final = HIVETOHA[self.heating_type].get(state, state) - except KeyError as e: - _LOGGER.error("get_mode - KeyError getting mode for %s: %s", device_name, e) - - return final + state = self._get_product_state(device, "state", "mode") + if state is None: + _LOGGER.debug("get_mode - No mode in state for %s.", device.ha_name) + return None + if state == "BOOST": + state = self._get_product_state(device, "props", "previous", "mode") + return HIVETOHA[self.heating_type].get(state, state) async def get_state(self, device: Device): """Get heating current state. diff --git a/src/devices/hotwater.py b/src/devices/hotwater.py index d4774c81..b36ca8d0 100644 --- a/src/devices/hotwater.py +++ b/src/devices/hotwater.py @@ -31,20 +31,13 @@ async def get_mode(self, device: Device): Returns: str: Return mode. """ - state = None - final = None - device_name = device.ha_name - - try: - data = self.session.data.products[device.hive_id] - state = data["state"]["mode"] - if state == "BOOST": - state = self._get_product_state(device, "props", "previous", "mode") - final = HIVETOHA[self.hotwater_type].get(state, state) - except KeyError as e: - _LOGGER.error("get_mode - KeyError getting mode for %s: %s", device_name, e) - - return final + state = self._get_product_state(device, "state", "mode") + if state is None: + _LOGGER.debug("get_mode - No mode in state for %s.", device.ha_name) + return None + if state == "BOOST": + state = self._get_product_state(device, "props", "previous", "mode") + return HIVETOHA[self.hotwater_type].get(state, state) @staticmethod async def get_operation_modes(): diff --git a/src/session/__init__.py b/src/session/__init__.py index bb9f1920..a36503bf 100644 --- a/src/session/__init__.py +++ b/src/session/__init__.py @@ -72,6 +72,7 @@ def __init__( self.hub_id = None self._last_poll_slow = False self._slow_poll_threshold = 3 + self._stale_state_polls: dict[str, int] = {} self._refresh_threshold = 0.90 self._update_task: asyncio.Task | None = None diff --git a/src/session/polling.py b/src/session/polling.py index e9e97849..2cfc254d 100644 --- a/src/session/polling.py +++ b/src/session/polling.py @@ -15,14 +15,24 @@ _LOGGER = logging.getLogger(__name__) +# Consecutive polls for which state keys missing from a response are carried +# over from the previous poll before the response is accepted as it is. +MAX_STALE_STATE_POLLS = 5 + +# State keys Hive always sends for a product that has them. Only these are +# carried over: other keys (``boost``, the colour keys) are legitimately +# omitted at times, so keeping their old value would report stale state. +REQUIRED_STATE_KEYS = frozenset({"mode", "name", "status", "target"}) + class PollingMixin: """Device polling, rate-limiting, and entity-cache methods. Expects ``self.config``, ``self.tokens``, ``self.api``, ``self.update_lock``, ``self._update_task``, - ``self._last_poll_slow``, and ``self._slow_poll_threshold`` - to be set up by the owning class's ``__init__``. + ``self._last_poll_slow``, ``self._slow_poll_threshold``, and + ``self._stale_state_polls`` to be set up by the owning class's + ``__init__``. """ # Attributes provided by HiveSession.__init__ @@ -35,6 +45,7 @@ class PollingMixin: _update_task: asyncio.Task | None _last_poll_slow: bool _slow_poll_threshold: int + _stale_state_polls: dict[str, int] @staticmethod def _entity_cache_key(device) -> str: @@ -70,6 +81,69 @@ def should_use_cached_data(self): return self._update_task is None or current_task is not self._update_task return False + def _keep_missing_state(self, new_products: dict) -> None: + """Carry over required state keys that a poll response has dropped. + + Hive sometimes answers a poll with products whose ``state`` block is + missing keys it normally has. Keep the last known value for the keys + in ``REQUIRED_STATE_KEYS``, for up to ``MAX_STALE_STATE_POLLS`` + consecutive polls. + + Args: + new_products (dict): Products from the latest response, keyed by ID. + """ + incomplete = [] + for product_id, product in new_products.items(): + previous = self.data.products.get(product_id) or {} + previous_state = previous.get("state") + new_state = product.get("state") + if not isinstance(new_state, dict): + new_state = {} + missing: list[str] = [] + if isinstance(previous_state, dict): + missing = sorted( + (previous_state.keys() & REQUIRED_STATE_KEYS) - new_state.keys() + ) + if not isinstance(previous_state, dict) or not missing: + self._stale_state_polls.pop(product_id, None) + continue + name = previous_state.get("name", product_id) + product_type = product.get("type") + _LOGGER.debug( + "Hive API state for %s %s arrived with keys: %s", + product_type, + product_id, + ", ".join(sorted(new_state)), + ) + polls = self._stale_state_polls.get(product_id, 0) + 1 + self._stale_state_polls[product_id] = polls + if polls > MAX_STALE_STATE_POLLS: + _LOGGER.warning( + "Hive API state for %s (%s %s) is still missing %s after %d " + "polls — no longer keeping the last known values.", + name, + product_type, + product_id, + ", ".join(missing), + MAX_STALE_STATE_POLLS, + ) + continue + product["state"] = { + **{key: previous_state[key] for key in missing}, + **new_state, + } + incomplete.append( + f"{name} ({product_type} {product_id}) missing {', '.join(missing)}" + ) + for product_id in self._stale_state_polls.keys() - new_products.keys(): + del self._stale_state_polls[product_id] + if incomplete: + _LOGGER.warning( + "Hive API returned incomplete state — keeping the last known " + "values: %s", + "; ".join(sorted(incomplete)), + ) + async def _poll_devices(self) -> bool: """Fetch latest device state from the Hive API.""" return await self.get_devices("No_ID") @@ -205,6 +279,7 @@ async def get_devices(self, _n_id: str): # pylint: disable=too-many-locals,too- len(tmp_actions), ) if tmp_products: + self._keep_missing_state(tmp_products) self.data.products = tmp_products if tmp_devices: self.data.devices = tmp_devices diff --git a/tests/module/test_session_polling.py b/tests/module/test_session_polling.py index dac27d5e..2a2abd03 100644 --- a/tests/module/test_session_polling.py +++ b/tests/module/test_session_polling.py @@ -30,6 +30,7 @@ class StubPolling(PollingMixin): p._update_task = None p._last_poll_slow = False p._slow_poll_threshold = 3 + p._stale_state_polls = {} p._poll_devices = AsyncMock(return_value=True) return p diff --git a/tests/unit/test_heating.py b/tests/unit/test_heating.py index 56c13ca0..181d2237 100644 --- a/tests/unit/test_heating.py +++ b/tests/unit/test_heating.py @@ -493,3 +493,28 @@ async def test_cached_none_falls_through_to_normal_path(self): result = await climate.get_climate(_make_device()) assert result is not None climate.session.attr.online_offline.assert_called_once() + + +# --------------------------------------------------------------------------- +# get_mode — state block present but mode missing (incomplete API response) +# --------------------------------------------------------------------------- + + +class TestGetModeMissingMode: + """get_mode must not log an error per call when state has no mode.""" + + async def test_missing_mode_returns_none_without_error_log(self): + """state without 'mode' returns None and logs nothing at ERROR.""" + climate = _make_climate({"heat-1": {"state": {"target": 7}}}) + with patch("apyhiveapi.devices.heating._LOGGER") as mock_log: + result = await climate.get_mode(_make_device()) + assert result is None + mock_log.error.assert_not_called() + + async def test_schedule_lookup_with_missing_mode_returns_none(self): + """get_schedule_now_next_later returns None quietly when mode is missing.""" + climate = _make_climate({"heat-1": {"state": {"schedule": {}}}}) + with patch("apyhiveapi.devices.heating._LOGGER") as mock_log: + result = await climate.get_schedule_now_next_later(_make_device()) + assert result is None + mock_log.error.assert_not_called() diff --git a/tests/unit/test_hotwater.py b/tests/unit/test_hotwater.py index 5da4e336..e308695d 100644 --- a/tests/unit/test_hotwater.py +++ b/tests/unit/test_hotwater.py @@ -326,3 +326,17 @@ async def test_cached_none_falls_through(self): result = await hw.get_water_heater(_make_device()) assert result is not None hw.session.attr.online_offline.assert_called_once() + + +class TestHotwaterGetModeMissingMode: + """get_mode must not log an error per call when state has no mode.""" + + async def test_missing_mode_returns_none_without_error_log(self): + """state without 'mode' returns None and logs nothing at ERROR.""" + from unittest.mock import patch + + hw = _make_hotwater({"hw-1": {"state": {"status": "OFF"}}}) + with patch("apyhiveapi.devices.hotwater._LOGGER") as mock_log: + result = await hw.get_mode(_make_device()) + assert result is None + mock_log.error.assert_not_called() diff --git a/tests/unit/test_polling.py b/tests/unit/test_polling.py index 77062a0b..b32df7c1 100644 --- a/tests/unit/test_polling.py +++ b/tests/unit/test_polling.py @@ -31,6 +31,7 @@ class StubPolling(PollingMixin): p._update_task = None p._last_poll_slow = False p._slow_poll_threshold = 3 + p._stale_state_polls = {} return p diff --git a/tests/unit/test_session_get_devices.py b/tests/unit/test_session_get_devices.py index b284b1ec..7ac4b37c 100644 --- a/tests/unit/test_session_get_devices.py +++ b/tests/unit/test_session_get_devices.py @@ -35,6 +35,7 @@ class StubPolling(PollingMixin): p._update_task = None p._last_poll_slow = False p._slow_poll_threshold = 3 + p._stale_state_polls = {} # External dependencies (provided by HiveSession in real code) p.api = MagicMock() @@ -423,3 +424,188 @@ async def poll_that_replaces_task(): # finally block left _update_task alone because it no longer matched # current_task (the False branch — line 113->116) assert p._update_task is sentinel + + +# --------------------------------------------------------------------------- +# get_devices — incomplete product state in an otherwise successful response +# --------------------------------------------------------------------------- + + +def _products_response(*products): + return {"original": 200, "parsed": {"products": list(products)}} + + +class TestGetDevicesIncompleteState: + """A poll that drops state keys must not wipe the last known values.""" + + def _stub(self, state): + p = _make_stub() + p.tokens = MagicMock() + p.data.products = {"heat-1": {"id": "heat-1", "state": dict(state)}} + return p + + async def test_missing_state_key_keeps_previous_value(self): + """state.mode absent from the response keeps the previous mode.""" + p = self._stub({"name": "Hallway", "mode": "OFF", "target": 7}) + p.api.get_all = AsyncMock( + return_value=_products_response( + { + "id": "heat-1", + "type": "heating", + "state": {"name": "Hallway", "target": 20}, + } + ) + ) + with patch("apyhiveapi.session.polling._LOGGER") as mock_log: + assert await p.get_devices("No_ID") is True + assert p.data.products["heat-1"]["state"] == { + "name": "Hallway", + "mode": "OFF", + "target": 20, + } + mock_log.warning.assert_called_once() + assert mock_log.warning.call_args.args[1] == ( + "Hallway (heating heat-1) missing mode" + ) + + async def test_warning_lists_every_product_and_missing_key(self): + """One warning per poll names each product and its missing keys.""" + p = self._stub({"mode": "OFF", "target": 7, "boost": None}) + p.data.products["hw-1"] = {"id": "hw-1", "state": {"mode": "OFF"}} + p.api.get_all = AsyncMock( + return_value=_products_response( + {"id": "heat-1", "type": "heating", "state": {"boost": None}}, + {"id": "hw-1", "type": "hotwater"}, + ) + ) + with patch("apyhiveapi.session.polling._LOGGER") as mock_log: + await p.get_devices("No_ID") + mock_log.warning.assert_called_once() + assert mock_log.warning.call_args.args[1] == ( + "heat-1 (heating heat-1) missing mode, target; " + "hw-1 (hotwater hw-1) missing mode" + ) + + async def test_debug_logs_the_state_keys_that_arrived(self): + """Debug shows the keys received for an incomplete product, not values.""" + p = self._stub({"mode": "OFF", "target": 7, "boost": None}) + p.api.get_all = AsyncMock( + return_value=_products_response( + {"id": "heat-1", "type": "heating", "state": {"target": 20, "boost": 5}} + ) + ) + with patch("apyhiveapi.session.polling._LOGGER") as mock_log: + await p.get_devices("No_ID") + shape_calls = [ + c.args for c in mock_log.debug.call_args_list if "arrived" in c.args[0] + ] + assert len(shape_calls) == 1 + assert shape_calls[0][1:] == ("heating", "heat-1", "boost, target") + + async def test_missing_state_block_keeps_previous_state(self): + """A product with no state block at all keeps the previous state.""" + p = self._stub({"mode": "OFF", "target": 7}) + p.api.get_all = AsyncMock(return_value=_products_response({"id": "heat-1"})) + await p.get_devices("No_ID") + assert p.data.products["heat-1"]["state"] == {"mode": "OFF", "target": 7} + + async def test_null_value_is_not_treated_as_missing(self): + """A key present with a None value replaces the previous value.""" + p = self._stub({"mode": "OFF", "boost": 30}) + p.api.get_all = AsyncMock( + return_value=_products_response( + {"id": "heat-1", "state": {"mode": "OFF", "boost": None}} + ) + ) + with patch("apyhiveapi.session.polling._LOGGER") as mock_log: + await p.get_devices("No_ID") + assert p.data.products["heat-1"]["state"] == {"mode": "OFF", "boost": None} + mock_log.warning.assert_not_called() + + async def test_new_product_is_accepted_as_is(self): + """A product with no previous record is stored unchanged.""" + p = self._stub({"mode": "OFF"}) + p.api.get_all = AsyncMock( + return_value=_products_response( + {"id": "heat-1", "state": {"mode": "OFF"}}, + {"id": "heat-2", "state": {"target": 7}}, + ) + ) + await p.get_devices("No_ID") + assert p.data.products["heat-2"]["state"] == {"target": 7} + + async def test_gives_up_after_max_consecutive_incomplete_polls(self): + """A key that stays missing is eventually dropped, not kept forever.""" + from apyhiveapi.session.polling import MAX_STALE_STATE_POLLS + + p = self._stub({"mode": "OFF", "target": 7}) + p.api.get_all = AsyncMock( + side_effect=lambda: _products_response( + {"id": "heat-1", "state": {"target": 7}} + ) + ) + for _ in range(MAX_STALE_STATE_POLLS): + await p.get_devices("No_ID") + assert p.data.products["heat-1"]["state"]["mode"] == "OFF" + await p.get_devices("No_ID") + assert p.data.products["heat-1"]["state"] == {"target": 7} + # The shorter state is now the baseline, so the next poll is complete. + with patch("apyhiveapi.session.polling._LOGGER") as mock_log: + await p.get_devices("No_ID") + mock_log.warning.assert_not_called() + assert not p._stale_state_polls + + async def test_complete_response_resets_the_count(self): + """A complete response between incomplete ones restarts the count.""" + p = self._stub({"mode": "OFF"}) + p.api.get_all = AsyncMock( + side_effect=[ + _products_response({"id": "heat-1", "state": {}}), + _products_response({"id": "heat-1", "state": {"mode": "MANUAL"}}), + _products_response({"id": "heat-1", "state": {}}), + ] + ) + await p.get_devices("No_ID") + assert p._stale_state_polls == {"heat-1": 1} + await p.get_devices("No_ID") + assert not p._stale_state_polls + await p.get_devices("No_ID") + assert p._stale_state_polls == {"heat-1": 1} + assert p.data.products["heat-1"]["state"] == {"mode": "MANUAL"} + + async def test_optional_key_dropped_by_hive_is_not_carried_over(self): + """Keys Hive may legitimately omit (e.g. boost) are never kept stale.""" + p = self._stub({"mode": "MANUAL", "target": 20, "boost": 30, "hue": 10}) + p.api.get_all = AsyncMock( + return_value=_products_response( + {"id": "heat-1", "state": {"mode": "MANUAL", "target": 20}} + ) + ) + with patch("apyhiveapi.session.polling._LOGGER") as mock_log: + await p.get_devices("No_ID") + assert p.data.products["heat-1"]["state"] == {"mode": "MANUAL", "target": 20} + mock_log.warning.assert_not_called() + assert not p._stale_state_polls + + async def test_only_required_keys_are_carried_over(self): + """A truncated state keeps required keys but not optional ones.""" + p = self._stub({"mode": "MANUAL", "target": 20, "boost": 30}) + p.api.get_all = AsyncMock( + return_value=_products_response({"id": "heat-1", "state": {}}) + ) + await p.get_devices("No_ID") + assert p.data.products["heat-1"]["state"] == {"mode": "MANUAL", "target": 20} + + async def test_count_is_dropped_for_products_no_longer_returned(self): + """A product that leaves the response does not leave a count behind.""" + p = self._stub({"mode": "OFF"}) + p.api.get_all = AsyncMock( + side_effect=[ + _products_response({"id": "heat-1", "state": {}}), + _products_response({"id": "heat-2", "state": {"mode": "OFF"}}), + ] + ) + await p.get_devices("No_ID") + assert p._stale_state_polls == {"heat-1": 1} + await p.get_devices("No_ID") + assert not p._stale_state_polls