From 173c00fca7043940e3f6ef813483798732059062 Mon Sep 17 00:00:00 2001 From: Gabriel Brown Date: Mon, 17 Aug 2026 13:27:11 -0400 Subject: [PATCH] Add the Panama Home Assistant bridge --- .../quickshell/scripts/panama-home-assistant | 434 ++++++++++++++++++ .../home-assistant-helper-contract.sh | 35 ++ .../quickshell/home_assistant_bridge_test.py | 182 ++++++++ 3 files changed, 651 insertions(+) create mode 100755 config/dot/quickshell/scripts/panama-home-assistant create mode 100755 tests/quickshell/home-assistant-helper-contract.sh create mode 100644 tests/quickshell/home_assistant_bridge_test.py diff --git a/config/dot/quickshell/scripts/panama-home-assistant b/config/dot/quickshell/scripts/panama-home-assistant new file mode 100755 index 0000000..8087bbf --- /dev/null +++ b/config/dot/quickshell/scripts/panama-home-assistant @@ -0,0 +1,434 @@ +#!/usr/bin/env python3 + +"""Private Home Assistant REST boundary for the Panama Control Center.""" + +from __future__ import annotations + +import ast +import json +import os +import pathlib +import re +import shlex +import subprocess +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from typing import Any + + +ENV_KEYS = ( + "PANAMA_HOME_ASSISTANT_URL", + "PANAMA_HOME_ASSISTANT_TOKEN", + "PANAMA_HOME_ASSISTANT_ENTITIES", +) +ENTITY_ID = re.compile(r"^[a-z_]+\.[a-z0-9_]+$") +GNOME_EXTENSION = ( + pathlib.Path.home() + / ".local/share/gnome-shell/extensions/hass-gshell@geoph9-on-github" +) +GNOME_SCHEMA_DIR = GNOME_EXTENSION / "schemas" +GNOME_SCHEMA = "org.gnome.shell.extensions.hass-data" +PANAMA_ENV = pathlib.Path(__file__).resolve().parents[3] / "bash/env" +Runner = Callable[..., subprocess.CompletedProcess[str]] + + +class BridgeError(RuntimeError): + """An error code safe to pass across the shell boundary.""" + + +@dataclass(frozen=True) +class Config: + base_url: str + token: str + entity_ids: tuple[str, ...] + + @property + def configured(self) -> bool: + return bool(self.base_url and self.token and self.entity_ids) + + +def compact_json(value: dict[str, object]) -> str: + return json.dumps(value, separators=(",", ":")) + + +def read_panama_env(path: pathlib.Path = PANAMA_ENV) -> dict[str, str]: + values: dict[str, str] = {} + try: + lines = path.read_text(encoding="utf-8").splitlines() + except OSError: + return values + + for line in lines: + stripped = line.strip() + if not stripped or stripped.startswith("#"): + continue + if stripped.startswith("export "): + stripped = stripped[7:].lstrip() + if "=" not in stripped: + continue + key, raw_value = stripped.split("=", 1) + key = key.strip() + if key not in ENV_KEYS: + continue + try: + parsed = shlex.split(raw_value, comments=True, posix=True) + except ValueError: + continue + if len(parsed) == 1: + values[key] = parsed[0] + return values + + +def normalize_url(value: str) -> str: + value = value.strip().rstrip("/") + if not value: + return "" + parsed = urllib.parse.urlsplit(value) + if ( + parsed.scheme not in {"http", "https"} + or not parsed.hostname + or parsed.username + or parsed.password + ): + raise ValueError("invalid-url") + return value + + +def parse_entity_ids(value: str | Sequence[str]) -> tuple[str, ...]: + candidates = value.split(",") if isinstance(value, str) else value + result: list[str] = [] + for candidate in candidates: + entity_id = str(candidate).strip() + if not entity_id: + continue + if not ENTITY_ID.fullmatch(entity_id): + raise ValueError("invalid-entity") + if entity_id not in result: + result.append(entity_id) + return tuple(result) + + +def parse_gsettings_string(output: str) -> str: + try: + value = ast.literal_eval(output.strip()) + except (SyntaxError, ValueError): + return "" + return value if isinstance(value, str) else "" + + +def parse_gsettings_array(output: str) -> tuple[str, ...]: + stripped = output.strip() + if stripped.startswith("@as "): + stripped = stripped[4:].strip() + try: + value = ast.literal_eval(stripped) + except (SyntaxError, ValueError): + return () + if not isinstance(value, (list, tuple)): + return () + return tuple(item for item in value if isinstance(item, str)) + + +def run_command( + command: list[str], + *, + runner: Runner = subprocess.run, + timeout: float = 5, +) -> subprocess.CompletedProcess[str]: + return runner( + command, + capture_output=True, + text=True, + timeout=timeout, + check=False, + ) + + +def gsettings_value(key: str, runner: Runner = subprocess.run) -> str: + if not GNOME_SCHEMA_DIR.is_dir(): + return "" + try: + result = run_command( + [ + "gsettings", + "--schemadir", + str(GNOME_SCHEMA_DIR), + "get", + GNOME_SCHEMA, + key, + ], + runner=runner, + ) + except (FileNotFoundError, subprocess.TimeoutExpired): + return "" + return result.stdout.strip() if result.returncode == 0 else "" + + +def secret_service_token(runner: Runner = subprocess.run) -> str: + try: + result = run_command( + ["secret-tool", "lookup", "token_string", "user_token"], + runner=runner, + ) + except (FileNotFoundError, subprocess.TimeoutExpired): + return "" + return result.stdout.strip() if result.returncode == 0 else "" + + +def load_legacy_config(runner: Runner = subprocess.run) -> Config: + base_url = parse_gsettings_string(gsettings_value("hass-url", runner)) + entity_ids = parse_gsettings_array( + gsettings_value("hass-enabled-entities", runner) + ) + token = secret_service_token(runner) + return Config(base_url=base_url, token=token, entity_ids=entity_ids) + + +def resolve_config( + env: Mapping[str, str] | None = None, + legacy: Callable[[], Config] = load_legacy_config, +) -> Config: + private_env = read_panama_env() + private_env.update(dict(os.environ if env is None else env)) + + url_value = private_env.get("PANAMA_HOME_ASSISTANT_URL", "").strip() + token_value = private_env.get("PANAMA_HOME_ASSISTANT_TOKEN", "").strip() + entities_value = private_env.get("PANAMA_HOME_ASSISTANT_ENTITIES", "").strip() + + legacy_config = Config("", "", ()) + if not (url_value and token_value and entities_value): + try: + legacy_config = legacy() + except (OSError, subprocess.SubprocessError, ValueError): + legacy_config = Config("", "", ()) + + base_url = normalize_url(url_value or legacy_config.base_url) + token = token_value or legacy_config.token + entity_ids = ( + parse_entity_ids(entities_value) + if entities_value + else parse_entity_ids(legacy_config.entity_ids) + ) + return Config(base_url=base_url, token=token, entity_ids=entity_ids) + + +def public_http_error(status: int, body: str) -> str: + del body + return "authentication-required" if status in {401, 403} else "request-failed" + + +def request_json( + config: Config, + method: str, + path: str, + payload: dict[str, str] | None = None, +) -> object: + if not config.base_url or not config.token: + raise BridgeError("not-configured") + body = None + if payload is not None: + body = json.dumps(payload, separators=(",", ":")).encode() + request = urllib.request.Request( + config.base_url + path, + data=body, + method=method, + headers={ + "Authorization": f"Bearer {config.token}", + "Content-Type": "application/json", + }, + ) + try: + with urllib.request.urlopen(request, timeout=5) as response: + response_body = response.read() + except urllib.error.HTTPError as error: + try: + error_body = error.read().decode(errors="replace") + except OSError: + error_body = "" + raise BridgeError(public_http_error(error.code, error_body)) from None + except (TimeoutError, urllib.error.URLError, OSError): + raise BridgeError("unreachable") from None + + if not response_body: + return None + try: + return json.loads(response_body) + except (json.JSONDecodeError, UnicodeDecodeError): + raise BridgeError("invalid-response") from None + + +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) + } + result: list[dict[str, object]] = [] + for entity_id in configured: + item = by_id.get(entity_id) + if not item: + continue + attributes = item.get("attributes") + if 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) + ) + result.append( + { + "id": entity_id, + "name": name, + "domain": entity_id.split(".", 1)[0], + "state": state, + "available": available, + "active": available + and state not in {"off", "closed", "idle", "standby"}, + } + ) + 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]: + if not config.configured: + return { + "ok": False, + "configured": False, + "generatedAt": int(time.time()), + "entities": [], + "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) + except BridgeError as error: + return { + "ok": False, + "configured": True, + "generatedAt": int(time.time()), + "entities": [], + "error": str(error), + } + return { + "ok": True, + "configured": True, + "generatedAt": int(time.time()), + "entities": entities, + "error": "", + } + + +def probe(config: Config) -> dict[str, object]: + if not config.configured: + return { + "configured": False, + "reachable": False, + "entityCount": len(config.entity_ids), + "error": "not-configured", + } + try: + request_json(config, "GET", "/api/") + except BridgeError as error: + return { + "configured": True, + "reachable": False, + "entityCount": len(config.entity_ids), + "error": str(error), + } + return { + "configured": True, + "reachable": True, + "entityCount": len(config.entity_ids), + "error": "", + } + + +def toggle(config: Config, entity_id: str) -> dict[str, object]: + entity_id = ensure_configured(entity_id, config.entity_ids) + request_json( + config, + "POST", + "/api/services/homeassistant/toggle", + {"entity_id": entity_id}, + ) + return {"ok": True, "entityId": entity_id, "error": ""} + + +def open_home(config: Config) -> dict[str, object]: + if not config.base_url: + return {"ok": False, "error": "not-configured"} + try: + subprocess.Popen( + ["xdg-open", config.base_url], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + except OSError: + return {"ok": False, "error": "open-failed"} + return {"ok": True, "error": ""} + + +def print_result(value: dict[str, object]) -> None: + sys.stdout.write(compact_json(value) + "\n") + + +def main(argv: list[str]) -> int: + try: + config = resolve_config() + except ValueError as error: + print_result({"ok": False, "configured": False, "error": str(error)}) + return 2 + + command = argv[0] if argv else "probe" + 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) + success = bool(result["ok"]) + elif command == "toggle" and len(argv) == 2: + try: + result = toggle(config, argv[1]) + 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"]) + else: + result = {"ok": False, "error": "invalid-command"} + success = False + print_result(result) + return 0 if success else 2 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/tests/quickshell/home-assistant-helper-contract.sh b/tests/quickshell/home-assistant-helper-contract.sh new file mode 100755 index 0000000..3d4cf1a --- /dev/null +++ b/tests/quickshell/home-assistant-helper-contract.sh @@ -0,0 +1,35 @@ +#!/usr/bin/env bash + +set -euo pipefail + +fail() { + printf 'Home Assistant helper contract: %s\n' "$1" >&2 + exit 1 +} + +project_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +helper="$project_root/config/dot/quickshell/scripts/panama-home-assistant" + +[[ -x "$helper" ]] || fail 'helper is missing or not executable' + +probe="$($helper probe)" +jq -e ' + .configured == true and + .reachable == true and + (.entityCount | type == "number" and . > 0) and + .error == "" +' <<<"$probe" >/dev/null || fail 'live API probe failed' + +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")" diff --git a/tests/quickshell/home_assistant_bridge_test.py b/tests/quickshell/home_assistant_bridge_test.py new file mode 100644 index 0000000..8e2bc97 --- /dev/null +++ b/tests/quickshell/home_assistant_bridge_test.py @@ -0,0 +1,182 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import importlib.machinery +import importlib.util +import json +import pathlib +import sys +import threading +import unittest +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + + +ROOT = pathlib.Path(__file__).resolve().parents[2] +HELPER = ROOT / "config/dot/quickshell/scripts/panama-home-assistant" + +loader = importlib.machinery.SourceFileLoader("panama_home_assistant", str(HELPER)) +spec = importlib.util.spec_from_loader(loader.name, loader) +bridge = importlib.util.module_from_spec(spec) +sys.modules[spec.name] = bridge +loader.exec_module(bridge) + + +class FakeHomeAssistant(BaseHTTPRequestHandler): + requests: list[dict[str, object]] = [] + + def log_message(self, _format: str, *args: object) -> None: + del args + + def _record(self, body: bytes = b"") -> None: + self.requests.append( + { + "method": self.command, + "path": self.path, + "authorization": self.headers.get("Authorization"), + "body": body, + } + ) + + def _json(self, value: object, status: int = 200) -> None: + payload = json.dumps(value).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def do_GET(self) -> None: + self._record() + if self.path == "/api/": + self._json({"message": "API running."}) + return + 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"}, + }, + ] + ) + return + self._json({"message": "not found"}, 404) + + def do_POST(self) -> None: + length = int(self.headers.get("Content-Length", "0")) + body = self.rfile.read(length) + self._record(body) + if self.path == "/api/services/homeassistant/toggle": + self._json([]) + return + self._json({"message": "not found"}, 404) + + +class HomeAssistantBridgeTest(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.server = ThreadingHTTPServer(("127.0.0.1", 0), FakeHomeAssistant) + cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True) + cls.thread.start() + cls.base_url = f"http://127.0.0.1:{cls.server.server_port}" + + @classmethod + def tearDownClass(cls) -> None: + cls.server.shutdown() + cls.server.server_close() + cls.thread.join(timeout=2) + + def setUp(self) -> None: + FakeHomeAssistant.requests.clear() + + def config(self) -> object: + return bridge.Config( + base_url=self.base_url, + token="fixture-token", + entity_ids=("light.kitchen", "light.hall"), + ) + + def test_environment_configuration_wins_over_legacy_sources(self) -> None: + env = { + "PANAMA_HOME_ASSISTANT_URL": "https://home.example", + "PANAMA_HOME_ASSISTANT_TOKEN": "fixture-token", + "PANAMA_HOME_ASSISTANT_ENTITIES": "light.kitchen,light.hall", + } + + config = bridge.resolve_config( + env, + legacy=lambda: (_ for _ in ()).throw(AssertionError("legacy called")), + ) + + 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"], + ) + self.assertNotIn("sensor.private", json.dumps(snapshot)) + + def test_snapshot_uses_bearer_authentication(self) -> None: + bridge.collect_snapshot(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_toggle_calls_the_homeassistant_service(self) -> None: + result = bridge.toggle(self.config(), "light.kitchen") + + self.assertTrue(result["ok"]) + request = FakeHomeAssistant.requests[-1] + self.assertEqual(request["method"], "POST") + self.assertEqual(request["path"], "/api/services/homeassistant/toggle") + self.assertEqual( + json.loads(request["body"]), + {"entity_id": "light.kitchen"}, + ) + + def test_authentication_error_is_redacted(self) -> None: + self.assertEqual( + bridge.public_http_error(401, "sensitive response"), + "authentication-required", + ) + self.assertEqual( + bridge.public_http_error(500, "sensitive response"), + "request-failed", + ) + + def test_gsettings_array_parser_handles_empty_and_values(self) -> None: + self.assertEqual(bridge.parse_gsettings_array("@as []"), ()) + self.assertEqual( + bridge.parse_gsettings_array("['light.kitchen', 'light.hall']"), + ("light.kitchen", "light.hall"), + ) + + +if __name__ == "__main__": + unittest.main()