"""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", "")}, )