From fba540868a85d37364bc036040918af5deed34ba Mon Sep 17 00:00:00 2001 From: Khole Jones <29937485+KJonline@users.noreply.github.com> Date: Sat, 10 Oct 2026 11:31:24 +0100 Subject: [PATCH 1/4] refactor: handle incomplete Hive API state responses without error logs Replace try/except KeyError blocks in heating/hotwater get_mode with early returns when state.mode is missing. Add _keep_missing_state to PollingMixin that carries over dropped state keys from the previous poll for up to MAX_STALE_STATE_POLLS (5) consecutive incomplete responses, logging a warning per poll listing affected products and missing keys, plus a debug log showing the keys that did arrive. After the threshold, accept --- src/devices/heating.py | 21 ++-- src/devices/hotwater.py | 21 ++-- src/session/__init__.py | 1 + src/session/polling.py | 66 ++++++++++- tests/module/test_session_polling.py | 1 + tests/unit/test_heating.py | 25 +++++ tests/unit/test_hotwater.py | 14 +++ tests/unit/test_polling.py | 1 + tests/unit/test_session_get_devices.py | 149 +++++++++++++++++++++++++ 9 files changed, 269 insertions(+), 30 deletions(-) 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..01b82605 100644 --- a/src/session/polling.py +++ b/src/session/polling.py @@ -15,14 +15,19 @@ _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 + 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 +40,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 +76,61 @@ 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 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 those + 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() - 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"] = {**previous_state, **new_state} + incomplete.append( + f"{name} ({product_type} {product_id}) missing {', '.join(missing)}" + ) + 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 +266,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..5711a1da 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,151 @@ 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"} From 7ad54e7d89c46dba755615b5444e040430e8b705 Mon Sep 17 00:00:00 2001 From: Khole Jones <29937485+KJonline@users.noreply.github.com> Date: Sat, 10 Oct 2026 11:36:16 +0100 Subject: [PATCH 2/4] fix: only carry over required state keys from an incomplete poll Keys Hive may legitimately omit (boost, the colour keys) were kept from the previous poll for up to five polls, which could report stale state such as boost ON. Carry over only mode, name, status and target, and drop the per-product count for products that leave the response. Co-Authored-By: Claude Opus 5.5 --- src/session/polling.py | 23 ++++++++++++---- tests/unit/test_session_get_devices.py | 37 ++++++++++++++++++++++++++ 2 files changed, 55 insertions(+), 5 deletions(-) diff --git a/src/session/polling.py b/src/session/polling.py index 01b82605..2cfc254d 100644 --- a/src/session/polling.py +++ b/src/session/polling.py @@ -19,6 +19,11 @@ # 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. @@ -77,11 +82,12 @@ def should_use_cached_data(self): return False def _keep_missing_state(self, new_products: dict) -> None: - """Carry over state keys that a poll response has dropped. + """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 those - keys, for up to ``MAX_STALE_STATE_POLLS`` consecutive polls. + 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. @@ -95,7 +101,9 @@ def _keep_missing_state(self, new_products: dict) -> None: new_state = {} missing: list[str] = [] if isinstance(previous_state, dict): - missing = sorted(previous_state.keys() - new_state.keys()) + 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 @@ -120,10 +128,15 @@ def _keep_missing_state(self, new_products: dict) -> None: MAX_STALE_STATE_POLLS, ) continue - product["state"] = {**previous_state, **new_state} + 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 " diff --git a/tests/unit/test_session_get_devices.py b/tests/unit/test_session_get_devices.py index 5711a1da..7ac4b37c 100644 --- a/tests/unit/test_session_get_devices.py +++ b/tests/unit/test_session_get_devices.py @@ -572,3 +572,40 @@ async def test_complete_response_resets_the_count(self): 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 From 680983dd0c31596010ec1f7850dca5f0bf8891d2 Mon Sep 17 00:00:00 2001 From: Khole Jones <29937485+KJonline@users.noreply.github.com> Date: Sat, 10 Oct 2026 23:52:25 +0100 Subject: [PATCH 3/4] ci: add Makefile targets for local session runner and ignore logs/ Add `run-file`, `run-live`, `simulate-file`, `simulate-live` targets that start background polling sessions (fixture or real account) and write to logs/. Add `logs` (tail -F), `status` (list runners), `stop` (kill runners), and `live-forget` (clear keychain credentials). Ignore logs/ in .gitignore. --- .gitignore | 5 +- Makefile | 48 ++++- scripts/run_session.py | 400 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 451 insertions(+), 2 deletions(-) create mode 100644 scripts/run_session.py 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..0621a58f 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: setup test lint sync verify-dist +.PHONY: setup test lint sync verify-dist run-file run-live simulate-file simulate-live live-forget logs status stop setup: pip install -e ".[dev]" @@ -22,3 +22,49 @@ verify-dist: rm -rf dist python -m build python scripts/verify_dist.py dist + +# --- Local session runner (scripts/run_session.py) -------------------------- +# Read-only: each run-*/simulate-* target starts a background process that +# polls until `make stop`, and writes to logs/ (gitignored). They never change +# a device. Options: INTERVAL=, DROP=, +# DEBUG=1 (library debug logging), POLLS= (stop after n polls). +RUN_SESSION = python3 scripts/run_session.py +RUN_OPTS = $(if $(POLLS),--polls $(POLLS)) $(if $(INTERVAL),--interval $(INTERVAL)) \ + $(if $(DEBUG),--debug) + +# 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) \ + $(if $(POLLS),--polls $(POLLS)) $(if $(DEBUG),--debug) + +simulate-live: + $(RUN_SESSION) --live --drop $(or $(DROP),mode) --interval $(or $(INTERVAL),30) \ + $(if $(POLLS),--polls $(POLLS)) $(if $(DEBUG),--debug) + +# 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)) + +# 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..1ac5732a --- /dev/null +++ b/scripts/run_session.py @@ -0,0 +1,400 @@ +"""Run a read-only 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/``. +It never calls a setter, so it cannot change any device. + +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 json +import logging +import logging.handlers +import os +import shutil +import signal +import subprocess +import sys +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 + +LOG = logging.getLogger("run_session") + + +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") + + if device_data: + print("Logging in with the stored device credentials (no SMS)...") + auth = Auth(username, password, *device_data) + try: + result = await auth.device_login() + except HiveInvalidDeviceAuthentication: + sys.exit( + "Hive rejected the stored device credentials. " + "Run `make live-forget`, then try again." + ) + else: + 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) + 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, keys: list[str], switch: dict) -> None: + """Strip *keys* from every product's state while ``switch["on"]`` is set. + + Simulates the incomplete responses Hive occasionally returns. + """ + + def _strip(response): + if switch["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 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)), + ) + + +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", "")) + switch = {"on": False} + async with hive: + if args.drop: + install_drop(hive, args.drop, switch) + 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: + switch["on"] = True + LOG.info("Dropping state keys from later polls: %s", ", ".join(args.drop)) + poll = 1 + while args.polls == 0 or poll < args.polls: + poll += 1 + await asyncio.sleep(args.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 _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')}") + + +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") + # 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: + 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("--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") + args = parser.parse_args() + + if args.forget: + forget_stored() + elif args.stop: + stop(args.name) + elif args.status: + status() + 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() From f10025118184a7c7fe6b85f16b212d7e97626108 Mon Sep 17 00:00:00 2001 From: Khole Jones <29937485+KJonline@users.noreply.github.com> Date: Thu, 15 Oct 2026 18:59:21 +0100 Subject: [PATCH 4/4] feat: add interactive admin commands to local session runner Add `make call` target and `--call` flag that send commands to a running session over a Unix socket. Commands include poll, mode/state inspection, drop control, debug/interval adjustment, and device control (set-mode, set-temp, boost, turn-on/off). Device control requires `--allow-writes` flag at session start. Add `make help` target documenting all targets and options. Refactor Makefile to use RUN_FLAGS for shared options and add detailed --- Makefile | 89 ++++++++++- scripts/run_session.py | 337 ++++++++++++++++++++++++++++++++++++++--- 2 files changed, 396 insertions(+), 30 deletions(-) diff --git a/Makefile b/Makefile index 0621a58f..29fd58aa 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,69 @@ -.PHONY: setup test lint sync verify-dist run-file run-live simulate-file simulate-live live-forget logs status stop +.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]" @@ -24,13 +89,15 @@ verify-dist: python scripts/verify_dist.py dist # --- Local session runner (scripts/run_session.py) -------------------------- -# Read-only: each run-*/simulate-* target starts a background process that -# polls until `make stop`, and writes to logs/ (gitignored). They never change -# a device. Options: INTERVAL=, DROP=, +# 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_OPTS = $(if $(POLLS),--polls $(POLLS)) $(if $(INTERVAL),--interval $(INTERVAL)) \ - $(if $(DEBUG),--debug) +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: @@ -45,11 +112,11 @@ run-live: # from every poll after the first. simulate-file: $(RUN_SESSION) --file --drop $(or $(DROP),mode) --interval $(or $(INTERVAL),5) \ - $(if $(POLLS),--polls $(POLLS)) $(if $(DEBUG),--debug) + $(RUN_FLAGS) simulate-live: $(RUN_SESSION) --live --drop $(or $(DROP),mode) --interval $(or $(INTERVAL),30) \ - $(if $(POLLS),--polls $(POLLS)) $(if $(DEBUG),--debug) + $(RUN_FLAGS) # Follow the logs of every runner (Ctrl-C stops following, not the runners). # NAME= follows just one. @@ -65,6 +132,12 @@ status: 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 index 1ac5732a..f2384993 100644 --- a/scripts/run_session.py +++ b/scripts/run_session.py @@ -1,9 +1,12 @@ -"""Run a read-only Hive session in the background for local testing. +"""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/``. -It never calls a setter, so it cannot change any device. + +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 @@ -22,6 +25,7 @@ import argparse import asyncio import getpass +import inspect import json import logging import logging.handlers @@ -30,6 +34,8 @@ 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 @@ -43,9 +49,23 @@ 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 @@ -134,23 +154,36 @@ async def live_config() -> dict: 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: - result = await auth.device_login() + # 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: - sys.exit( - "Hive rejected the stored device credentials. " - "Run `make live-forget`, then try again." - ) - else: + 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.") @@ -163,18 +196,18 @@ async def live_config() -> dict: return config -def install_drop(hive: Hive, keys: list[str], switch: dict) -> None: - """Strip *keys* from every product's state while ``switch["on"]`` is set. +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 switch["on"] and isinstance(response, dict): + 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 keys: + for key in settings.drop_keys: state.pop(key, None) return response @@ -206,33 +239,240 @@ def log_summary(hive: Hive, label: str) -> None: ) +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", "")) - switch = {"on": False} + settings = Settings( + interval=args.interval, + allow_writes=args.allow_writes, + drop_keys=list(args.drop), + ) async with hive: - if args.drop: - install_drop(hive, args.drop, switch) + 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: - switch["on"] = True + settings.drop_on = True LOG.info("Dropping state keys from later polls: %s", ", ".join(args.drop)) - poll = 1 - while args.polls == 0 or poll < args.polls: - poll += 1 - await asyncio.sleep(args.interval) - ok = await hive.force_update() - log_summary(hive, f"poll {poll}{'' if ok else ' (FAILED)'}") + 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: @@ -280,6 +520,43 @@ def status() -> None: 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) @@ -297,6 +574,8 @@ def launch(args: argparse.Namespace, name: str) -> None: 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, @@ -344,6 +623,7 @@ def work(args: argparse.Namespace) -> None: 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() @@ -362,6 +642,12 @@ def main() -> None: ) 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( @@ -381,6 +667,11 @@ def main() -> None: 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: @@ -389,6 +680,8 @@ def main() -> None: stop(args.name) elif args.status: status() + elif args.call: + call(args.name, args.call) elif args.worker: work(args) else: