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
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:]))