From ff67387c4223853788da0fa774d707adf2eff703 Mon Sep 17 00:00:00 2001 From: Gabriel Brown Date: Mon, 17 Aug 2026 16:01:50 -0400 Subject: [PATCH] Add Home Assistant light catalog and dimming --- .../quickshell/scripts/panama-home-assistant | 125 ++++++++++---- .../dot/quickshell/services/HomeAssistant.qml | 4 +- .../home-assistant-helper-contract.sh | 28 +--- .../quickshell/home_assistant_bridge_test.py | 155 ++++++++++++++---- 4 files changed, 225 insertions(+), 87 deletions(-) diff --git a/config/dot/quickshell/scripts/panama-home-assistant b/config/dot/quickshell/scripts/panama-home-assistant index 8087bbf..d8c297f 100755 --- a/config/dot/quickshell/scripts/panama-home-assistant +++ b/config/dot/quickshell/scripts/panama-home-assistant @@ -18,7 +18,6 @@ import urllib.parse import urllib.request from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass -from typing import Any ENV_KEYS = ( @@ -49,7 +48,7 @@ class Config: @property def configured(self) -> bool: - return bool(self.base_url and self.token and self.entity_ids) + return bool(self.base_url and self.token) def compact_json(value: dict[str, object]) -> str: @@ -226,7 +225,7 @@ def request_json( config: Config, method: str, path: str, - payload: dict[str, str] | None = None, + payload: Mapping[str, object] | None = None, ) -> object: if not config.base_url or not config.token: raise BridgeError("not-configured") @@ -266,71 +265,74 @@ def fallback_name(entity_id: str) -> str: return entity_id.split(".", 1)[1].replace("_", " ").title() -def normalize_entities( - raw: list[dict[str, Any]], - configured: Sequence[str], -) -> list[dict[str, object]]: - by_id = { - str(item.get("entity_id", "")): item - for item in raw - if isinstance(item, dict) - } +def normalize_catalog(raw: Sequence[object]) -> list[dict[str, object]]: result: list[dict[str, object]] = [] - for entity_id in configured: - item = by_id.get(entity_id) - if not item: + for item in raw: + if not isinstance(item, dict): continue + entity_id = item.get("entity_id") attributes = item.get("attributes") - if not isinstance(attributes, dict): + if not isinstance(entity_id, str) or not entity_id.startswith("light."): + continue + if not ENTITY_ID.fullmatch(entity_id) or not isinstance(attributes, dict): continue state = str(item.get("state", "unavailable")) available = state not in {"unknown", "unavailable"} - friendly_name = attributes.get("friendly_name") - name = ( - friendly_name.strip() - if isinstance(friendly_name, str) and friendly_name.strip() - else fallback_name(entity_id) + active = available and state == "on" + raw_brightness = attributes.get("brightness") + brightness_pct = ( + round(max(0, min(255, raw_brightness)) * 100 / 255) + if active + and isinstance(raw_brightness, (int, float)) + and not isinstance(raw_brightness, bool) + else 0 ) + modes = attributes.get("supported_color_modes", []) + dimmable = ( + isinstance(modes, list) and any(mode != "onoff" for mode in modes) + ) or isinstance(raw_brightness, (int, float)) + source_name = attributes.get("friendly_name") result.append( { "id": entity_id, - "name": name, - "domain": entity_id.split(".", 1)[0], + "sourceName": ( + source_name.strip() + if isinstance(source_name, str) and source_name.strip() + else fallback_name(entity_id) + ), "state": state, "available": available, - "active": available - and state not in {"off", "closed", "idle", "standby"}, + "active": active, + "dimmable": dimmable, + "brightnessPct": brightness_pct, } ) return result -def ensure_configured(entity_id: str, configured: Sequence[str]) -> str: - if entity_id not in configured: - raise ValueError("entity-not-configured") - return entity_id - - -def collect_snapshot(config: Config) -> dict[str, object]: +def collect_catalog(config: Config) -> dict[str, object]: + legacy_entity_ids = list(config.entity_ids) if not config.configured: return { "ok": False, "configured": False, "generatedAt": int(time.time()), "entities": [], + "legacyEntityIds": legacy_entity_ids, "error": "not-configured", } try: raw = request_json(config, "GET", "/api/states") if not isinstance(raw, list): raise BridgeError("invalid-response") - entities = normalize_entities(raw, config.entity_ids) + entities = normalize_catalog(raw) except BridgeError as error: return { "ok": False, "configured": True, "generatedAt": int(time.time()), "entities": [], + "legacyEntityIds": legacy_entity_ids, "error": str(error), } return { @@ -338,10 +340,28 @@ def collect_snapshot(config: Config) -> dict[str, object]: "configured": True, "generatedAt": int(time.time()), "entities": entities, + "legacyEntityIds": legacy_entity_ids, "error": "", } +def collect_snapshot(config: Config) -> dict[str, object]: + return collect_catalog(config) + + +def discovered_light_ids(config: Config) -> set[str]: + raw = request_json(config, "GET", "/api/states") + if not isinstance(raw, list): + raise BridgeError("invalid-response") + return {item["id"] for item in normalize_catalog(raw) if isinstance(item["id"], str)} + + +def ensure_discovered(config: Config, entity_id: str) -> str: + if entity_id not in discovered_light_ids(config): + raise ValueError("entity-not-discovered") + return entity_id + + def probe(config: Config) -> dict[str, object]: if not config.configured: return { @@ -368,7 +388,7 @@ def probe(config: Config) -> dict[str, object]: def toggle(config: Config, entity_id: str) -> dict[str, object]: - entity_id = ensure_configured(entity_id, config.entity_ids) + entity_id = ensure_discovered(config, entity_id) request_json( config, "POST", @@ -378,6 +398,31 @@ def toggle(config: Config, entity_id: str) -> dict[str, object]: return {"ok": True, "entityId": entity_id, "error": ""} +def set_brightness( + config: Config, entity_id: str, percent: int +) -> dict[str, object]: + if isinstance(percent, bool) or not isinstance(percent, int) or not 0 <= percent <= 100: + raise ValueError("invalid-brightness") + ensure_discovered(config, entity_id) + if percent == 0: + path = "/api/services/light/turn_off" + payload = {"entity_id": entity_id} + else: + path = "/api/services/light/turn_on" + payload = {"entity_id": entity_id, "brightness_pct": percent} + request_json(config, "POST", path, payload) + return {"ok": True, "entityId": entity_id, "brightnessPct": percent, "error": ""} + + +def parse_brightness(value: str) -> int: + if not re.fullmatch(r"(?:0|[1-9][0-9]{0,2})", value): + raise ValueError("invalid-brightness") + percent = int(value) + if percent > 100: + raise ValueError("invalid-brightness") + return percent + + def open_home(config: Config) -> dict[str, object]: if not config.base_url: return {"ok": False, "error": "not-configured"} @@ -409,8 +454,8 @@ def main(argv: list[str]) -> int: if command == "probe" and len(argv) <= 1: result = probe(config) success = bool(result["reachable"]) - elif command == "snapshot" and len(argv) == 1: - result = collect_snapshot(config) + elif command in {"catalog", "snapshot"} and len(argv) == 1: + result = collect_catalog(config) success = bool(result["ok"]) elif command == "toggle" and len(argv) == 2: try: @@ -420,6 +465,14 @@ def main(argv: list[str]) -> int: except BridgeError as error: result = {"ok": False, "error": str(error)} success = bool(result["ok"]) + elif command == "brightness" and len(argv) == 3: + try: + result = set_brightness(config, argv[1], parse_brightness(argv[2])) + except ValueError as error: + result = {"ok": False, "error": str(error)} + except BridgeError as error: + result = {"ok": False, "error": str(error)} + success = bool(result["ok"]) elif command == "open" and len(argv) == 1: result = open_home(config) success = bool(result["ok"]) diff --git a/config/dot/quickshell/services/HomeAssistant.qml b/config/dot/quickshell/services/HomeAssistant.qml index 123fe4b..f73f77e 100644 --- a/config/dot/quickshell/services/HomeAssistant.qml +++ b/config/dot/quickshell/services/HomeAssistant.qml @@ -1,7 +1,7 @@ pragma Singleton // Home Assistant state for the Control Center. Credentials and REST details -// remain behind the helper; QML receives configured favourites only. +// remain behind the helper; QML receives a normalized light catalog only. import Quickshell import Quickshell.Io @@ -151,7 +151,7 @@ Singleton { Process { id: refreshProc - command: [root.helperPath, "snapshot"] + command: [root.helperPath, "catalog"] stdout: StdioCollector { onStreamFinished: root.consumeSnapshot(this.text) } diff --git a/tests/quickshell/home-assistant-helper-contract.sh b/tests/quickshell/home-assistant-helper-contract.sh index 3d4cf1a..0662d58 100755 --- a/tests/quickshell/home-assistant-helper-contract.sh +++ b/tests/quickshell/home-assistant-helper-contract.sh @@ -12,24 +12,14 @@ helper="$project_root/config/dot/quickshell/scripts/panama-home-assistant" [[ -x "$helper" ]] || fail 'helper is missing or not executable' -probe="$($helper probe)" +catalog="$($helper catalog)" jq -e ' - .configured == true and - .reachable == true and - (.entityCount | type == "number" and . > 0) and - .error == "" -' <<<"$probe" >/dev/null || fail 'live API probe failed' + .ok == true and .configured == true and .error == "" and + (.entities | type == "array" and length > 0) and + ([.entities[] | (keys | sort) == (["active", "available", "brightnessPct", "dimmable", "id", "sourceName", "state"] | sort)] | all) and + ([.entities[] | (.id | startswith("light.")) and (.brightnessPct >= 0 and .brightnessPct <= 100)] | all) and + (.legacyEntityIds | type == "array") +' <<<"$catalog" >/dev/null || fail 'live catalog shape is invalid' -snapshot="$($helper snapshot)" -jq -e --argjson expected "$(jq '.entityCount' <<<"$probe")" ' - .ok == true and - .configured == true and - .error == "" and - (.generatedAt | type == "number") and - (.entities | type == "array" and length == $expected) and - ([.entities[] | (keys | sort) == (["active", "available", "domain", "id", "name", "state"] | sort)] | all) and - ([.entities[] | (.active | type == "boolean") and (.available | type == "boolean")] | all) -' <<<"$snapshot" >/dev/null || fail 'live snapshot shape is invalid' - -printf 'Home Assistant helper contract: PASS (configured=true, %s favourites; contents redacted)\n' \ - "$(jq '.entities | length' <<<"$snapshot")" +printf 'Home Assistant helper contract: PASS (configured=true, %s lights; contents redacted)\n' \ + "$(jq '.entities | length' <<<"$catalog")" diff --git a/tests/quickshell/home_assistant_bridge_test.py b/tests/quickshell/home_assistant_bridge_test.py index 8e2bc97..b1e1fb6 100644 --- a/tests/quickshell/home_assistant_bridge_test.py +++ b/tests/quickshell/home_assistant_bridge_test.py @@ -14,6 +14,7 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer ROOT = pathlib.Path(__file__).resolve().parents[2] HELPER = ROOT / "config/dot/quickshell/scripts/panama-home-assistant" +QML_SERVICE = ROOT / "config/dot/quickshell/services/HomeAssistant.qml" loader = importlib.machinery.SourceFileLoader("panama_home_assistant", str(HELPER)) spec = importlib.util.spec_from_loader(loader.name, loader) @@ -54,20 +55,36 @@ class FakeHomeAssistant(BaseHTTPRequestHandler): if self.path == "/api/states": self._json( [ - { - "entity_id": "light.hall", - "state": "off", - "attributes": {"friendly_name": "Hall"}, - }, - { - "entity_id": "sensor.private", - "state": "1", - "attributes": {"friendly_name": "Private"}, - }, { "entity_id": "light.kitchen", "state": "on", - "attributes": {"friendly_name": "Kitchen"}, + "attributes": { + "friendly_name": "Kitchen", + "brightness": 128, + "supported_color_modes": ["brightness"], + }, + }, + { + "entity_id": "light.hall", + "state": "off", + "attributes": { + "friendly_name": "Hall", + "supported_color_modes": ["color_temp"], + }, + }, + {"entity_id": "sensor.private", "state": "1", "attributes": {"token": "never-return"}}, + { + "entity_id": "light.malformed", + "state": "on", + "attributes": "invalid", + }, + { + "entity_id": "light.corner", + "state": "unavailable", + "attributes": { + "friendly_name": "Corner", + "supported_color_modes": ["brightness"], + }, }, ] ) @@ -78,7 +95,11 @@ class FakeHomeAssistant(BaseHTTPRequestHandler): length = int(self.headers.get("Content-Length", "0")) body = self.rfile.read(length) self._record(body) - if self.path == "/api/services/homeassistant/toggle": + if self.path in { + "/api/services/homeassistant/toggle", + "/api/services/light/turn_on", + "/api/services/light/turn_off", + }: self._json([]) return self._json({"message": "not found"}, 404) @@ -123,33 +144,71 @@ class HomeAssistantBridgeTest(unittest.TestCase): self.assertEqual(config.base_url, "https://home.example") self.assertEqual(config.entity_ids, ("light.kitchen", "light.hall")) - def test_snapshot_filters_and_preserves_configured_order(self) -> None: - snapshot = bridge.collect_snapshot(self.config()) - - self.assertTrue(snapshot["ok"]) - self.assertEqual( - [item["name"] for item in snapshot["entities"]], - ["Kitchen", "Hall"], + def test_config_does_not_require_legacy_entity_ids(self) -> None: + self.assertTrue( + bridge.Config( + base_url="https://home.example", + token="fixture-token", + entity_ids=(), + ).configured ) - self.assertNotIn("sensor.private", json.dumps(snapshot)) + + def test_legacy_entity_ids_remain_migration_metadata(self) -> None: + config = bridge.resolve_config( + { + "PANAMA_HOME_ASSISTANT_URL": "https://home.example", + "PANAMA_HOME_ASSISTANT_TOKEN": "fixture-token", + }, + legacy=lambda: bridge.Config( + base_url="https://legacy.example", + token="legacy-token", + entity_ids=("light.kitchen",), + ), + ) + + self.assertEqual(config.base_url, "https://home.example") + self.assertEqual(config.token, "fixture-token") + self.assertEqual(config.entity_ids, ("light.kitchen",)) + + def test_catalog_filters_and_normalizes_discovered_lights(self) -> None: + catalog = bridge.collect_catalog(self.config()) + + self.assertTrue(catalog["ok"]) + self.assertEqual( + [item["sourceName"] for item in catalog["entities"]], + ["Kitchen", "Hall", "Corner"], + ) + self.assertEqual(catalog["entities"][0]["brightnessPct"], 50) + self.assertEqual(catalog["entities"][1]["brightnessPct"], 0) + self.assertTrue(all(item["dimmable"] for item in catalog["entities"])) + rendered = json.dumps(catalog) + self.assertNotIn("sensor.private", rendered) + self.assertNotIn("light.malformed", rendered) + self.assertNotIn("token", rendered) def test_snapshot_uses_bearer_authentication(self) -> None: - bridge.collect_snapshot(self.config()) + bridge.collect_catalog(self.config()) request = FakeHomeAssistant.requests[-1] self.assertEqual(request["method"], "GET") self.assertEqual(request["path"], "/api/states") self.assertEqual(request["authorization"], "Bearer fixture-token") - def test_toggle_rejects_an_unconfigured_entity(self) -> None: - with self.assertRaisesRegex(ValueError, "entity-not-configured"): - bridge.ensure_configured( - "light.office", - ("light.kitchen",), - ) + def test_snapshot_remains_a_catalog_compatibility_alias(self) -> None: + snapshot = bridge.collect_snapshot(self.config()) - def test_toggle_calls_the_homeassistant_service(self) -> None: - result = bridge.toggle(self.config(), "light.kitchen") + self.assertTrue(snapshot["ok"]) + self.assertEqual( + [item["sourceName"] for item in snapshot["entities"]], + ["Kitchen", "Hall", "Corner"], + ) + + def test_action_rejects_an_entity_not_in_the_live_catalog(self) -> None: + with self.assertRaisesRegex(ValueError, "entity-not-discovered"): + bridge.toggle(self.config(), "light.office") + + def test_toggle_authorizes_against_discovered_catalog(self) -> None: + result = bridge.toggle(self.config(), "light.corner") self.assertTrue(result["ok"]) request = FakeHomeAssistant.requests[-1] @@ -157,9 +216,45 @@ class HomeAssistantBridgeTest(unittest.TestCase): self.assertEqual(request["path"], "/api/services/homeassistant/toggle") self.assertEqual( json.loads(request["body"]), - {"entity_id": "light.kitchen"}, + {"entity_id": "light.corner"}, ) + def test_brightness_uses_turn_on_for_positive_percent(self) -> None: + bridge.set_brightness(self.config(), "light.kitchen", 62) + + request = FakeHomeAssistant.requests[-1] + self.assertEqual(request["path"], "/api/services/light/turn_on") + self.assertEqual( + json.loads(request["body"]), + {"entity_id": "light.kitchen", "brightness_pct": 62}, + ) + + def test_brightness_zero_uses_turn_off(self) -> None: + bridge.set_brightness(self.config(), "light.hall", 0) + + request = FakeHomeAssistant.requests[-1] + self.assertEqual(request["path"], "/api/services/light/turn_off") + self.assertEqual( + json.loads(request["body"]), + {"entity_id": "light.hall"}, + ) + + def test_brightness_rejects_invalid_values_before_any_request(self) -> None: + for value in (-1, 101, 1.5, "bright"): + with self.subTest(value=value): + with self.assertRaisesRegex(ValueError, "invalid-brightness"): + bridge.set_brightness(self.config(), "light.kitchen", value) + self.assertEqual(FakeHomeAssistant.requests, []) + + def test_parse_brightness_rejects_invalid_cli_values(self) -> None: + for value in ("-1", "101", "1.5", "bright"): + with self.subTest(value=value): + with self.assertRaisesRegex(ValueError, "invalid-brightness"): + bridge.parse_brightness(value) + + def test_qml_refreshes_through_the_catalog_command(self) -> None: + self.assertIn('command: [root.helperPath, "catalog"]', QML_SERVICE.read_text()) + def test_authentication_error_is_redacted(self) -> None: self.assertEqual( bridge.public_http_error(401, "sensitive response"),