"""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 import os import secrets from pathlib import Path from typing import Any import voluptuous as vol from homeassistant.components import websocket_api from homeassistant.core import HomeAssistant, callback from homeassistant.helpers import config_validation as cv 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", } 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]) -> 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", "") 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 "" 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]) -> 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) 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) try: config = await hass.data["lovelace"]["dashboards"][url_path].async_load(False) except KeyError: # Try the default dashboard try: config = await hass.data["lovelace"]["dashboards"][None].async_load(False) _LOGGER.warning( "Dashboard '%s' not found, falling back to default", url_path ) except (KeyError, Exception) as err: connection.send_error(msg["id"], "not_found", f"Dashboard not found: {err}") return views = config.get("views", []) rooms = [_map_view(v) for v in views] result = { "version": CONFIG_SCHEMA_VERSION, "rooms": rooms, } connection.send_result(msg["id"], result) @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", "")}, )