Add the Panama Home Assistant bridge

2026-08-17 13:27:11 -04:00
parent 3a903934c0
commit 173c00fca7
3 changed files with 651 additions and 0 deletions
@@ -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:]))