Add esphome/get_device_capabilities websocket command

homeassistant/components/esphome/websocket_api.py

@@ -1,26 +1,44 @@
 """ESPHome websocket API."""
 
-import logging
-from typing import Any
+from typing import Any, cast
 
+from aioesphomeapi.model import SerialProxyPortType
 import voluptuous as vol
 
 from homeassistant.components import websocket_api
+from homeassistant.config_entries import ConfigEntryState
 from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers import device_registry as dr
 
-from .const import CONF_NOISE_PSK
-
-_LOGGER = logging.getLogger(__name__)
-
+from .const import CONF_NOISE_PSK, DOMAIN
+from .entry_data import ESPHomeConfigEntry
+from .serial_proxy import build_url
 
 TYPE = "type"
 ENTRY_ID = "entry_id"
+DEVICE_ID = "device_id"
+
+_UNAVAILABLE_CAPABILITIES: dict[str, Any] = {
+    "available": False,
+    "bluetooth_proxy": {"supported": False},
+    "zwave_proxy": {"supported": False, "home_id": 0},
+    "serial_proxies": [],
+}
 
 
 @callback
 def async_setup(hass: HomeAssistant) -> None:
     """Set up the websocket API."""
     websocket_api.async_register_command(hass, get_encryption_key)
+    websocket_api.async_register_command(hass, get_device_capabilities)
+
+
+def _serial_port_type_name(port_type: SerialProxyPortType | int) -> str | None:
+    """Return the SerialProxyPortType name, or None if unknown."""
+    try:
+        return SerialProxyPortType(port_type).name
+    except ValueError:
+        return None
 
 
 @callback
@@ -50,3 +68,76 @@
             "encryption_key": entry.data.get(CONF_NOISE_PSK),
         },
     )
+
+
+@callback
+@websocket_api.require_admin
+@websocket_api.websocket_command(
+    {
+        vol.Required(TYPE): "esphome/get_device_capabilities",
+        vol.Required(DEVICE_ID): str,
+    }
+)
+def get_device_capabilities(
+    hass: HomeAssistant,
+    connection: websocket_api.connection.ActiveConnection,
+    msg: dict[str, Any],
+) -> None:
+    """Return cached ESPHome DeviceInfo capabilities for the device page."""
+    device = dr.async_get(hass).async_get(msg[DEVICE_ID])
+    if device is None:
+        connection.send_error(
+            msg["id"], websocket_api.ERR_NOT_FOUND, "Device not found"
+        )
+        return
+
+    entry: ESPHomeConfigEntry | None = None
+    for entry_id in device.config_entries:
+        candidate = hass.config_entries.async_get_entry(entry_id)
+        if candidate is not None and candidate.domain == DOMAIN:
+            entry = cast(ESPHomeConfigEntry, candidate)
+            break
+
+    if entry is None:
+        connection.send_error(
+            msg["id"],
+            websocket_api.ERR_NOT_FOUND,
+            "Device is not an ESPHome device",
+        )
+        return
+
+    if entry.state is not ConfigEntryState.LOADED:
+        connection.send_result(msg["id"], _UNAVAILABLE_CAPABILITIES)
+        return
+
+    entry_data = entry.runtime_data
+    device_info = entry_data.device_info
+    if device_info is None:
+        connection.send_result(msg["id"], _UNAVAILABLE_CAPABILITIES)
+        return
+
+    connection.send_result(
+        msg["id"],
+        {
+            "available": entry_data.available,
+            "bluetooth_proxy": {
+                "supported": bool(
+                    device_info.bluetooth_proxy_feature_flags_compat(
+                        entry_data.api_version
+                    )
+                ),
+            },
+            "zwave_proxy": {
+                "supported": bool(device_info.zwave_proxy_feature_flags),
+                "home_id": device_info.zwave_home_id or 0,
+            },
+            "serial_proxies": [
+                {
+                    "name": proxy.name,
+                    "port_type": _serial_port_type_name(proxy.port_type),
+                    "url": str(build_url(entry.entry_id, proxy.name)),
+                }
+                for proxy in device_info.serial_proxies
+            ],
+        },
+    )

tests/components/esphome/test_websocket_api.py

@@ -1,14 +1,33 @@
 """Tests for ESPHome websocket API."""
 
 from aioesphomeapi import APIClient
+from aioesphomeapi.model import SerialProxyInfo, SerialProxyPortType
 
 from homeassistant.components.esphome.const import CONF_NOISE_PSK
-from homeassistant.components.esphome.websocket_api import ENTRY_ID, TYPE
+from homeassistant.components.esphome.serial_proxy import build_url
+from homeassistant.components.esphome.websocket_api import DEVICE_ID, ENTRY_ID, TYPE
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import device_registry as dr
+
+from .conftest import MockESPHomeDeviceType
 
 from tests.common import MockConfigEntry
 from tests.typing import WebSocketGenerator
 
 
+def _device_id_for_mac(
+    device_registry: dr.DeviceRegistry,
+    entry: MockConfigEntry,
+    mac: str = "11:22:33:44:55:aa",
+) -> str:
+    """Return the device registry id for an ESPHome MAC."""
+    device = device_registry.async_get_device_by_connection(
+        (dr.CONNECTION_NETWORK_MAC, mac), entry.entry_id
+    )
+    assert device is not None
+    return device.id
+
+
 async def test_get_encryption_key(
     mock_client: APIClient,
     init_integration: MockConfigEntry,
@@ -30,3 +49,179 @@ async def test_get_encryption_key(
     assert response["result"] == {
         "encryption_key": mock_config_entry.data.get(CONF_NOISE_PSK)
     }
+
+
+async def test_get_device_capabilities(
+    hass: HomeAssistant,
+    device_registry: dr.DeviceRegistry,
+    mock_client: APIClient,
+    mock_esphome_device: MockESPHomeDeviceType,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test capabilities from cached DeviceInfo."""
+    mock_client.connected_address = "192.168.1.2"
+    device = await mock_esphome_device(
+        mock_client=mock_client,
+        device_info={
+            "bluetooth_proxy_feature_flags": 1,
+            "zwave_proxy_feature_flags": 1,
+            "zwave_home_id": 1234567890,
+            "serial_proxies": [
+                SerialProxyInfo(name="uart0", port_type=SerialProxyPortType.TTL),
+                SerialProxyInfo(name="amp", port_type=SerialProxyPortType.RS232),
+                SerialProxyInfo(name="bus", port_type=SerialProxyPortType.RS485),
+            ],
+        },
+    )
+
+    websocket_client = await hass_ws_client()
+    await websocket_client.send_json_auto_id(
+        {
+            TYPE: "esphome/get_device_capabilities",
+            DEVICE_ID: _device_id_for_mac(device_registry, device.entry),
+        }
+    )
+
+    response = await websocket_client.receive_json()
+    assert response["success"] is True
+    assert response["result"] == {
+        "available": True,
+        "bluetooth_proxy": {"supported": True},
+        "zwave_proxy": {
+            "supported": True,
+            "home_id": 1234567890,
+        },
+        "serial_proxies": [
+            {
+                "name": "uart0",
+                "port_type": "TTL",
+                "url": str(build_url(device.entry.entry_id, "uart0")),
+            },
+            {
+                "name": "amp",
+                "port_type": "RS232",
+                "url": str(build_url(device.entry.entry_id, "amp")),
+            },
+            {
+                "name": "bus",
+                "port_type": "RS485",
+                "url": str(build_url(device.entry.entry_id, "bus")),
+            },
+        ],
+    }
+
+
+async def test_get_device_capabilities_device_not_found(
+    init_integration: MockConfigEntry,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test get_device_capabilities when the device registry id is unknown."""
+    websocket_client = await hass_ws_client()
+    await websocket_client.send_json_auto_id(
+        {
+            TYPE: "esphome/get_device_capabilities",
+            DEVICE_ID: "not-a-device",
+        }
+    )
+
+    response = await websocket_client.receive_json()
+    assert response["success"] is False
+    assert response["error"]["code"] == "not_found"
+    assert response["error"]["message"] == "Device not found"
+
+
+async def test_get_device_capabilities_wrong_domain(
+    hass: HomeAssistant,
+    init_integration: MockConfigEntry,
+    device_registry: dr.DeviceRegistry,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test get_device_capabilities when the device is not ESPHome."""
+    other_entry = MockConfigEntry(domain="switch", data={})
+    other_entry.add_to_hass(hass)
+    device = device_registry.async_get_or_create(
+        config_entry_id=other_entry.entry_id,
+        connections={(dr.CONNECTION_NETWORK_MAC, "aa:bb:cc:dd:ee:ff")},
+    )
+
+    websocket_client = await hass_ws_client()
+    await websocket_client.send_json_auto_id(
+        {
+            TYPE: "esphome/get_device_capabilities",
+            DEVICE_ID: device.id,
+        }
+    )
+
+    response = await websocket_client.receive_json()
+    assert response["success"] is False
+    assert response["error"]["code"] == "not_found"
+    assert response["error"]["message"] == "Device is not an ESPHome device"
+
+
+async def test_get_device_capabilities_unavailable(
+    hass: HomeAssistant,
+    device_registry: dr.DeviceRegistry,
+    mock_client: APIClient,
+    mock_esphome_device: MockESPHomeDeviceType,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test cached capabilities are returned when the device is unavailable."""
+    mock_client.connected_address = "192.168.1.2"
+    device = await mock_esphome_device(
+        mock_client=mock_client,
+        device_info={
+            "bluetooth_proxy_feature_flags": 1,
+            "zwave_proxy_feature_flags": 1,
+            "zwave_home_id": 1234567890,
+        },
+    )
+    await device.mock_disconnect(expected_disconnect=False)
+
+    websocket_client = await hass_ws_client()
+    await websocket_client.send_json_auto_id(
+        {
+            TYPE: "esphome/get_device_capabilities",
+            DEVICE_ID: _device_id_for_mac(device_registry, device.entry),
+        }
+    )
+
+    response = await websocket_client.receive_json()
+    assert response["success"] is True
+    assert response["result"] == {
+        "available": False,
+        "bluetooth_proxy": {"supported": True},
+        "zwave_proxy": {"supported": True, "home_id": 1234567890},
+        "serial_proxies": [],
+    }
+
+
+async def test_get_device_capabilities_no_device_info(
+    hass: HomeAssistant,
+    device_registry: dr.DeviceRegistry,
+    mock_client: APIClient,
+    mock_esphome_device: MockESPHomeDeviceType,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test a useful empty payload when cached DeviceInfo is missing."""
+    device = await mock_esphome_device(mock_client=mock_client)
+    device.entry.runtime_data.device_info = None
+
+    websocket_client = await hass_ws_client()
+    await websocket_client.send_json_auto_id(
+        {
+            TYPE: "esphome/get_device_capabilities",
+            DEVICE_ID: _device_id_for_mac(device_registry, device.entry),
+        }
+    )
+
+    response = await websocket_client.receive_json()
+    assert response["success"] is True
+    assert response["result"] == {
+        "available": False,
+        "bluetooth_proxy": {"supported": False},
+        "zwave_proxy": {
+            "supported": False,
+            "home_id": 0,
+        },
+        "serial_proxies": [],
+    }