Panama/config/dot/quickshell/scripts/panama-home-assistant

#!/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


ENV_KEYS = (
    "PANAMA_HOME_ASSISTANT_URL",
    "PANAMA_HOME_ASSISTANT_TOKEN",
    "PANAMA_HOME_ASSISTANT_ENTITIES",
)
ENTITY_ID = re.compile(r"^[a-z_]+\.[a-z0-9_]+$")
# One template renders the whole area list, so a room grouping costs a single
# request. Home Assistant answers /api/template with the rendered text, which
# `tojson` makes a JSON document the caller can parse like any other response.
AREAS_TEMPLATE = (
    "{% set ns = namespace(items=[]) %}"
    "{% for area in areas() %}"
    "{% set ns.items = ns.items + ["
    "{'id': area, 'name': area_name(area), 'entities': area_entities(area)}"
    "] %}"
    "{% endfor %}"
    "{{ ns.items | tojson }}"
)
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"
# The migrated personal env (outside the checkout) wins; the in-repo file is
# only read on machines that have not migrated. See panama-home-assistant-config.
LEGACY_ENV = pathlib.Path(__file__).resolve().parents[3] / "bash/env"
_config_home = pathlib.Path(os.environ.get("XDG_CONFIG_HOME", str(pathlib.Path.home() / ".config")))
_migrated_env = _config_home / "panama" / "env"
PANAMA_ENV = _migrated_env if _migrated_env.exists() else LEGACY_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)


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:
    # An explicitly supplied environment is the WHOLE environment. Reading the
    # user's private env file underneath it makes callers -- tests especially --
    # depend on whatever happens to be in that file: adding a real
    # PANAMA_HOME_ASSISTANT_ENTITIES to it silently overrode a fixture that was
    # asserting the legacy fallback. Production passes env=None and still gets
    # the file.
    if env is None:
        private_env = read_panama_env()
        private_env.update(dict(os.environ))
    else:
        private_env = dict(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: Mapping[str, object] | 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 = ""
        finally:
            error.close()
        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_catalog(raw: Sequence[object]) -> list[dict[str, object]]:
    result: list[dict[str, object]] = []
    for item in raw:
        if not isinstance(item, dict):
            continue
        entity_id = item.get("entity_id")
        attributes = item.get("attributes")
        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"}
        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,
                "sourceName": (
                    source_name.strip()
                    if isinstance(source_name, str) and source_name.strip()
                    else fallback_name(entity_id)
                ),
                "state": state,
                "available": available,
                "active": active,
                "dimmable": dimmable,
                "brightnessPct": brightness_pct,
            }
        )
    return result


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_catalog(raw)
    except BridgeError as error:
        return {
            "ok": False,
            "configured": True,
            "generatedAt": int(time.time()),
            "entities": [],
            "legacyEntityIds": legacy_entity_ids,
            "error": str(error),
        }
    return {
        "ok": True,
        "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 normalize_areas(raw: Sequence[object]) -> list[dict[str, object]]:
    # A template renders whatever Home Assistant happens to hold, so every entry
    # is checked on its own: one malformed area is dropped rather than costing
    # the caller the whole grouping. Only lights are kept, because lights are
    # all the catalog holds today, and an area left with none is not a room the
    # UI can draw.
    result: list[dict[str, object]] = []
    for item in raw:
        if not isinstance(item, dict):
            continue
        area_id = item.get("id")
        name = item.get("name")
        entities = item.get("entities")
        if not isinstance(area_id, str) or not isinstance(name, str):
            continue
        if not area_id.strip() or not name.strip() or not isinstance(entities, list):
            continue
        lights = [
            entity_id
            for entity_id in entities
            if isinstance(entity_id, str)
            and entity_id.startswith("light.")
            and ENTITY_ID.fullmatch(entity_id)
        ]
        if not lights:
            continue
        result.append(
            {"id": area_id.strip(), "name": name.strip(), "entities": lights}
        )
    return result


def collect_areas(config: Config) -> dict[str, object]:
    if not config.configured:
        return {"ok": False, "areas": [], "error": "not-configured"}
    try:
        raw = request_json(
            config, "POST", "/api/template", {"template": AREAS_TEMPLATE}
        )
        if not isinstance(raw, list):
            raise BridgeError("invalid-response")
        areas = normalize_areas(raw)
    except BridgeError as error:
        return {"ok": False, "areas": [], "error": str(error)}
    return {"ok": True, "areas": areas, "error": ""}


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 {
            "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_discovered(config, entity_id)
    request_json(
        config,
        "POST",
        "/api/services/homeassistant/toggle",
        {"entity_id": entity_id},
    )
    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"}
    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 in {"catalog", "snapshot"} and len(argv) == 1:
        result = collect_catalog(config)
        success = bool(result["ok"])
    elif command == "areas" and len(argv) == 1:
        result = collect_areas(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 == "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"])
    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:]))