luma-home/integration/custom_components/luma_home/websocket_api.py

"""WebSocket API for Luma Home.

Exposes:
  luma/get_config  — returns v2 config derived from a Lovelace dashboard
  luma/pair         — writes a pairing file for TV token provisioning
"""

from __future__ import annotations

import json
import logging
from pathlib import Path
from typing import Any

import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.core import HomeAssistant

from .const import DOMAIN, DEFAULT_DASHBOARD_URL_PATH, PAIRING_DIR

_LOGGER = logging.getLogger(__name__)

CONFIG_SCHEMA_VERSION = 2

LUMA_CARD_TYPE_MAP = {
    "custom:luma-range-card": "range",
    "custom:luma-toggle-card": "toggle",
    "custom:luma-climate-card": "climate",
    "custom:luma-media-card": "media",
    "custom:luma-curtain-card": "curtain",
    "custom:luma-action-card": "action",
    "custom:luma-vacuum-card": "vacuum",
    "custom:luma-weather-card": "weather",
    "custom:luma-led-strip-card": "led_strip",
    "custom:luma-govee-tv-card": "govee_tv",
    "custom:luma-ac-control-card": "ac_control",
}

HA_CARD_TYPE_MAP = {
    "light": "range",
    "thermostat": "climate",
    "media-control": "media",
    "weather-forecast": "weather",
    "button": "action",
}

CONTAINER_CARD_TYPES = {
    "vertical-stack",
    "horizontal-stack",
    "grid",
    "conditional",
    "entities",
    "sections",
}


def async_register_commands(hass: HomeAssistant) -> None:
    """Register WS commands."""
    websocket_api.async_register_command(hass, handle_get_config)
    websocket_api.async_register_command(hass, handle_pair)
    websocket_api.async_register_command(hass, handle_pair_status)


def _get_dashboard_path(hass: HomeAssistant) -> str:
    """Get the configured dashboard URL path."""
    for entry_data in hass.data.get(DOMAIN, {}).values():
        if isinstance(entry_data, dict):
            return entry_data.get("dashboard_url_path", DEFAULT_DASHBOARD_URL_PATH)
    return DEFAULT_DASHBOARD_URL_PATH


def _flatten_cards(cards: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Recursively unwrap container cards to get leaf cards."""
    result = []
    for card in cards:
        card_type = card.get("type", "")
        if card_type in CONTAINER_CARD_TYPES:
            children = card.get("cards", [])
            if card_type == "conditional":
                children = card.get("card", [])
                if isinstance(children, dict):
                    children = [children]
            elif card_type == "entities":
                # entities card has entity rows, not sub-cards — skip it
                # unless it wraps actual card children
                children = card.get("cards", [])
            result.extend(_flatten_cards(children))
        else:
            result.append(card)
    return result


def _map_card(card: dict[str, Any], hass: HomeAssistant | None = None) -> dict[str, Any] | None:
    """Map a Lovelace card config to a Luma v2 card config.

    Returns None if the card type is not recognized.
    """
    card_type = card.get("type", "")

    # Custom Luma cards — direct passthrough
    if card_type in LUMA_CARD_TYPE_MAP:
        luma_type = LUMA_CARD_TYPE_MAP[card_type]
        return {
            "type": luma_type,
            "entity_id": card.get("entity", card.get("entity_id", "")),
            "name": card.get("name", ""),
            "size": card.get("size", "1x1"),
            "accent_color": card.get("accent_color", ""),
            "glow_color": card.get("glow_color", ""),
            "icon": card.get("icon", ""),
            "tap_action": card.get("tap_action"),
            "hold_action": card.get("hold_action"),
            "confirmation": card.get("confirmation", False),
        }

    # Standard HA cards — best-effort mapping
    if card_type in HA_CARD_TYPE_MAP:
        luma_type = HA_CARD_TYPE_MAP[card_type]
        entity = card.get("entity", "")

        # Override: detect LED strip lights for the "light" HA card type
        if card_type == "light" and hass is not None:
            state = hass.states.get(entity)
            if state is not None:
                attrs = state.attributes
                has_effects = bool(attrs.get("effect_list"))
                color_modes = attrs.get("supported_color_modes", [])
                has_rgb = "rgb" in color_modes or "rgbw" in color_modes or "rgbww" in color_modes
                if has_effects and has_rgb:
                    luma_type = "led_strip"

        return {
            "type": luma_type,
            "entity_id": entity,
            "name": card.get("name", ""),
            "size": "1x1",
            "accent_color": "",
            "glow_color": "",
            "icon": "",
            "tap_action": card.get("tap_action"),
            "hold_action": card.get("hold_action"),
            "confirmation": False,
        }

    # Standard HA entity-domain based fallback for 'button' cards etc.
    entity = card.get("entity", "")
    if entity:
        domain = entity.split(".")[0] if "." in entity else ""

        # Detect LED strip lights: lights with effect_list and rgb_color support
        if domain == "light" and hass is not None:
            state = hass.states.get(entity)
            if state is not None:
                attrs = state.attributes
                has_effects = bool(attrs.get("effect_list"))
                color_modes = attrs.get("supported_color_modes", [])
                has_rgb = "rgb" in color_modes or "rgbw" in color_modes or "rgbww" in color_modes
                if has_effects and has_rgb:
                    return {
                        "type": "led_strip",
                        "entity_id": entity,
                        "name": card.get("name", ""),
                        "size": "1x1",
                        "accent_color": "",
                        "glow_color": "",
                        "icon": card.get("icon", ""),
                        "tap_action": card.get("tap_action"),
                        "hold_action": card.get("hold_action"),
                        "confirmation": False,
                    }

        domain_map = {
            "light": "range",
            "switch": "toggle",
            "input_boolean": "toggle",
            "climate": "climate",
            "media_player": "media",
            "cover": "curtain",
            "scene": "action",
            "script": "action",
            "vacuum": "vacuum",
            "weather": "weather",
        }
        if domain in domain_map:
            return {
                "type": domain_map[domain],
                "entity_id": entity,
                "name": card.get("name", ""),
                "size": "1x1",
                "accent_color": "",
                "glow_color": "",
                "icon": card.get("icon", ""),
                "tap_action": card.get("tap_action"),
                "hold_action": card.get("hold_action"),
                "confirmation": False,
            }

    return None


def _map_view(view: dict[str, Any], hass: HomeAssistant | None = None) -> dict[str, Any]:
    """Map a Lovelace view to a Luma v2 room."""
    raw_cards = view.get("cards", [])
    flat_cards = _flatten_cards(raw_cards)
    mapped = [c for c in (_map_card(card, hass) for card in flat_cards) if c is not None]

    return {
        "id": view.get("path", view.get("title", "").lower().replace(" ", "_")),
        "name": view.get("title", ""),
        "icon": view.get("icon", ""),
        "theme": view.get("theme"),
        "badges": view.get("badges", []),
        "subview": view.get("subview", False),
        "cards": mapped,
    }


@websocket_api.websocket_command(
    {
        vol.Required("type"): "luma/get_config",
        vol.Optional("dashboard_url_path"): str,
    }
)
@websocket_api.async_response
async def handle_get_config(
    hass: HomeAssistant,
    connection: websocket_api.ActiveConnection,
    msg: dict[str, Any],
) -> None:
    """Return the Luma v2 config derived from a Lovelace dashboard."""
    url_path = msg.get("dashboard_url_path") or _get_dashboard_path(hass)

    config = await _load_lovelace_config(hass, url_path)

    if config is None:
        connection.send_error(msg["id"], "not_found", f"Dashboard '{url_path}' not found")
        return

    views = config.get("views", [])
    rooms = [_map_view(v, hass) for v in views]

    result = {
        "version": CONFIG_SCHEMA_VERSION,
        "rooms": rooms,
    }

    connection.send_result(msg["id"], result)


async def _load_lovelace_config(hass: HomeAssistant, url_path: str) -> dict | None:
    """Load a Lovelace dashboard config by reading the .storage file."""
    storage_key = url_path.replace("-", "_")
    storage_path = Path(hass.config.path(f".storage/lovelace.{storage_key}"))
    _LOGGER.debug("Looking for dashboard at %s", storage_path)

    try:
        if not await hass.async_add_executor_job(storage_path.exists):
            _LOGGER.warning("Dashboard storage file not found: %s", storage_path)
            return None
        raw = await hass.async_add_executor_job(storage_path.read_text, "utf-8")
        stored = json.loads(raw)
        config = stored.get("data", {}).get("config")
        if config is None:
            config = stored.get("data", {})
        _LOGGER.debug("Loaded dashboard with %d views", len(config.get("views", [])))
        return config
    except Exception as err:
        _LOGGER.error("Failed to load dashboard '%s': %s", url_path, err)
        return None


@websocket_api.websocket_command(
    {
        vol.Required("type"): "luma/pair",
        vol.Required("pin"): str,
        vol.Required("token"): str,
        vol.Optional("ha_url"): str,
    }
)
@websocket_api.async_response
async def handle_pair(
    hass: HomeAssistant,
    connection: websocket_api.ActiveConnection,
    msg: dict[str, Any],
) -> None:
    """Write a pairing file for TV token provisioning."""
    pin = msg["pin"]
    token = msg["token"]
    ha_url = msg.get("ha_url", "")

    if not pin.isdigit() or len(pin) != 6:
        connection.send_error(msg["id"], "invalid_pin", "PIN must be 6 digits")
        return

    www_dir = Path(hass.config.path("www"))
    pairing_dir = www_dir / PAIRING_DIR
    pairing_dir.mkdir(parents=True, exist_ok=True)

    pairing_file = pairing_dir / f"{pin}.json"
    payload = {"token": token, "ha_url": ha_url, "pin": pin}

    await hass.async_add_executor_job(
        _write_pairing_file, pairing_file, payload
    )

    connection.send_result(msg["id"], {"success": True})
    _LOGGER.info("Pairing file written for PIN %s", pin)


def _write_pairing_file(path: Path, data: dict) -> None:
    """Write pairing JSON to disk."""
    path.write_text(json.dumps(data), encoding="utf-8")


@websocket_api.websocket_command(
    {
        vol.Required("type"): "luma/pair_status",
        vol.Required("pin"): str,
    }
)
@websocket_api.async_response
async def handle_pair_status(
    hass: HomeAssistant,
    connection: websocket_api.ActiveConnection,
    msg: dict[str, Any],
) -> None:
    """Check if a pairing file exists and return/clean it up.

    Called by the TV after it polls the HTTP endpoint and finds the file.
    This command lets authenticated clients clean up the pairing file.
    """
    pin = msg["pin"]
    www_dir = Path(hass.config.path("www"))
    pairing_file = www_dir / PAIRING_DIR / f"{pin}.json"

    if not await hass.async_add_executor_job(pairing_file.exists):
        connection.send_result(msg["id"], {"paired": False})
        return

    data = json.loads(
        await hass.async_add_executor_job(pairing_file.read_text, "utf-8")
    )

    # Clean up immediately
    await hass.async_add_executor_job(pairing_file.unlink, True)

    connection.send_result(
        msg["id"],
        {"paired": True, "token": data.get("token", ""), "ha_url": data.get("ha_url", "")},
    )