Add Matter network topology WebSocket API (#177114)

homeassistant/components/matter/api.py

@@ -4,17 +4,25 @@ from collections.abc import Callable, Coroutine
 from functools import wraps
 from typing import Any, Concatenate
 
+from matter_server.client.exceptions import ServerVersionTooOld
 from matter_server.client.models.node import MatterNode
 from matter_server.common.errors import MatterError
 from matter_server.common.helpers.util import dataclass_to_dict
+from matter_server.common.models import EventType, NetworkTopology
 import voluptuous as vol
 
 from homeassistant.components import websocket_api
-from homeassistant.components.websocket_api import ActiveConnection
+from homeassistant.components.websocket_api import ERR_NOT_SUPPORTED, ActiveConnection
 from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers import device_registry as dr
 
 from .adapter import MatterAdapter
-from .helpers import MissingNode, get_matter, node_from_ha_device_id
+from .helpers import (
+    MissingNode,
+    get_matter,
+    get_node_device_identifier,
+    node_from_ha_device_id,
+)
 
 ID = "id"
 TYPE = "type"
@@ -23,6 +31,9 @@ DEVICE_ID = "device_id"
 
 ERROR_NODE_NOT_FOUND = "node_not_found"
 
+# minimum server schema version that provides network topology
+TOPOLOGY_SCHEMA_VERSION = 13
+
 
 @callback
 def async_register_api(hass: HomeAssistant) -> None:
@@ -36,6 +47,8 @@ def async_register_api(hass: HomeAssistant) -> None:
     websocket_api.async_register_command(hass, websocket_open_commissioning_window)
     websocket_api.async_register_command(hass, websocket_remove_matter_fabric)
     websocket_api.async_register_command(hass, websocket_interview_node)
+    websocket_api.async_register_command(hass, websocket_network_topology)
+    websocket_api.async_register_command(hass, websocket_subscribe_network_topology)
 
 
 def async_get_node(
@@ -115,6 +128,8 @@ def async_handle_failed_command[**_P](
             connection.send_error(msg[ID], str(err.error_code), err.args[0])
         except MissingNode as err:
             connection.send_error(msg[ID], ERROR_NODE_NOT_FOUND, err.args[0])
+        except ServerVersionTooOld as err:
+            connection.send_error(msg[ID], ERR_NOT_SUPPORTED, err.args[0])
 
     return async_handle_failed_command_func
 
@@ -328,3 +343,124 @@ async def websocket_interview_node(
     """Interview a node."""
     await matter.matter_client.interview_node(node_id=node.node_id)
     connection.send_result(msg[ID])
+
+
+@callback
+def _topology_supported(
+    connection: ActiveConnection, msg: dict[str, Any], matter: MatterAdapter
+) -> bool:
+    """Check if the server supports network topology, send an error if not."""
+    server_info = matter.matter_client.server_info
+    if server_info is None or server_info.schema_version < TOPOLOGY_SCHEMA_VERSION:
+        connection.send_error(
+            msg[ID],
+            ERR_NOT_SUPPORTED,
+            "The Matter server does not support network topology "
+            f"(requires schema version {TOPOLOGY_SCHEMA_VERSION}).",
+        )
+        return False
+    return True
+
+
+@callback
+def _serialize_topology(
+    hass: HomeAssistant, matter: MatterAdapter, topology: NetworkTopology
+) -> dict[str, Any]:
+    """Serialize a topology snapshot, annotating nodes with HA device ids."""
+    server_info = matter.matter_client.server_info
+    dev_reg = dr.async_get(hass)
+    result: dict[str, Any] = dataclass_to_dict(topology)
+    for node in result["nodes"]:
+        device = None
+        if (node_id := node.get("node_id")) is not None and server_info is not None:
+            device = dev_reg.async_get_device_by_identifier(
+                get_node_device_identifier(server_info, node_id),
+                matter.config_entry.entry_id,
+            )
+        node["ha_device_id"] = device.id if device else None
+    return result
+
+
+@websocket_api.require_admin
+@websocket_api.websocket_command(
+    {
+        vol.Required(TYPE): "matter/network_topology",
+        vol.Optional("refresh", default=False): bool,
+    }
+)
+@websocket_api.async_response
+@async_handle_failed_command
+@async_get_matter_adapter
+async def websocket_network_topology(
+    hass: HomeAssistant,
+    connection: ActiveConnection,
+    msg: dict[str, Any],
+    matter: MatterAdapter,
+) -> None:
+    """Get the network topology graph."""
+    if not _topology_supported(connection, msg, matter):
+        return
+    topology = await matter.matter_client.get_network_topology(refresh=msg["refresh"])
+    connection.send_result(msg[ID], _serialize_topology(hass, matter, topology))
+
+
+@websocket_api.require_admin
+@websocket_api.websocket_command(
+    {
+        vol.Required(TYPE): "matter/subscribe_network_topology",
+    }
+)
+@websocket_api.async_response
+@async_handle_failed_command
+@async_get_matter_adapter
+async def websocket_subscribe_network_topology(
+    hass: HomeAssistant,
+    connection: ActiveConnection,
+    msg: dict[str, Any],
+    matter: MatterAdapter,
+) -> None:
+    """Subscribe to network topology updates."""
+    if not _topology_supported(connection, msg, matter):
+        return
+
+    initial_sent = False
+    # updates are full snapshots, so only the newest buffered one matters
+    buffered: NetworkTopology | None = None
+
+    @callback
+    def forward_topology(event: EventType, topology: NetworkTopology) -> None:
+        nonlocal buffered
+        if not initial_sent:
+            buffered = topology
+            return
+        connection.send_message(
+            websocket_api.event_message(
+                msg[ID], _serialize_topology(hass, matter, topology)
+            )
+        )
+
+    # subscribe before the fetch: the fetch opts this client in server-side,
+    # and an update may arrive before the command result does
+    unsubscribe = matter.matter_client.subscribe_events(
+        callback=forward_topology,
+        event_filter=EventType.NETWORK_TOPOLOGY_UPDATED,
+    )
+    try:
+        topology = await matter.matter_client.get_network_topology()
+    except Exception:
+        unsubscribe()
+        raise
+    connection.subscriptions[msg[ID]] = unsubscribe
+    connection.send_result(msg[ID])
+    connection.send_message(
+        websocket_api.event_message(
+            msg[ID], _serialize_topology(hass, matter, topology)
+        )
+    )
+    if buffered is not None:
+        connection.send_message(
+            websocket_api.event_message(
+                msg[ID], _serialize_topology(hass, matter, buffered)
+            )
+        )
+    initial_sent = True

homeassistant/components/matter/helpers.py

@@ -83,6 +83,18 @@ def get_device_id(
     return f"{operational_instance_id}-{postfix}"
 
 
+def get_node_device_identifier(
+    server_info: ServerInfoMessage, node_id: int
+) -> tuple[str, str]:
+    """Return the device registry identifier for the node-level device of a node."""
+    fabric_id_hex = f"{server_info.compressed_fabric_id:016X}"
+    node_id_hex = f"{node_id:016X}"
+    return (
+        DOMAIN,
+        f"{ID_TYPE_DEVICE_ID}_{fabric_id_hex}-{node_id_hex}-MatterNodeDevice",
+    )
+
+
 @callback
 def node_from_ha_device_id(hass: HomeAssistant, ha_device_id: str) -> MatterNode | None:
     """Get node id from ha device id."""

tests/components/matter/test_api.py

@@ -1,7 +1,9 @@
 """Test the api module."""
 
+from collections.abc import Callable
 from unittest.mock import AsyncMock, MagicMock, call
 
+from matter_server.client.exceptions import ServerVersionTooOld
 from matter_server.client.models.node import (
     MatterFabricData,
     NetworkType,
@@ -10,7 +12,14 @@ from matter_server.client.models.node import (
 )
 from matter_server.common.errors import InvalidCommand, NodeCommissionFailed
 from matter_server.common.helpers.util import dataclass_to_dict
-from matter_server.common.models import CommissioningParameters
+from matter_server.common.models import (
+    CommissioningParameters,
+    EventType,
+    NetworkTopology,
+    NetworkTopologyConnection,
+    NetworkTopologyNode,
+    TopologyDirectionInfo,
+)
 import pytest
 
 from homeassistant.components.matter.api import (
@@ -460,3 +469,238 @@ async def test_interview_node(
 
     assert not msg["success"]
     assert msg["error"]["code"] == ERROR_NODE_NOT_FOUND
+
+
+def _mock_topology() -> NetworkTopology:
+    """Return a mock topology with a known node, an unknown node and a border router."""
+    return NetworkTopology(
+        collected_at=1767888000000,
+        nodes=[
+            NetworkTopologyNode(
+                id="30",
+                kind="matter",
+                network_type="thread",
+                node_id=30,
+                role="router",
+                available=True,
+            ),
+            NetworkTopologyNode(
+                id="99",
+                kind="matter",
+                network_type="thread",
+                node_id=99,
+                role="end_device",
+                available=True,
+            ),
+            NetworkTopologyNode(
+                id="br_1122AABBCC334455",
+                kind="border_router",
+                network_type="thread",
+                role="router",
+                ext_address="1122AABBCC334455",
+                vendor_name="Apple",
+            ),
+        ],
+        connections=[
+            NetworkTopologyConnection(
+                source="30",
+                target="br_1122AABBCC334455",
+                network="thread",
+                strength="strong",
+                source_to_target=TopologyDirectionInfo(strength="strong", lqi=3),
+            ),
+        ],
+    )
+
+
+def _expected_topology(
+    topology: NetworkTopology, ha_device_ids: list[str | None]
+) -> dict:
+    """Return the expected ws payload for the given topology."""
+    expected = dataclass_to_dict(topology)
+    for node, ha_device_id in zip(expected["nodes"], ha_device_ids, strict=True):
+        node["ha_device_id"] = ha_device_id
+    return expected
+
+
+@pytest.mark.usefixtures("matter_node")
+@pytest.mark.parametrize("node_fixture", ["mock_onoff_light"])
+async def test_network_topology(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    device_registry: dr.DeviceRegistry,
+    matter_client: MagicMock,
+) -> None:
+    """Test the network_topology command."""
+    matter_client.server_info.schema_version = 13
+    entry = device_registry.async_get_device_by_identifier(
+        (DOMAIN, "deviceid_00000000000004D2-000000000000001E-MatterNodeDevice"),
+        hass.config_entries.async_entries(DOMAIN)[0].entry_id,
+    )
+    assert entry is not None
+
+    topology = _mock_topology()
+    matter_client.get_network_topology = AsyncMock(return_value=topology)
+
+    ws_client = await hass_ws_client(hass)
+    await ws_client.send_json({ID: 1, TYPE: "matter/network_topology"})
+    msg = await ws_client.receive_json()
+
+    assert msg["success"]
+    # node 30 maps to the registry device, node 99 and the border router do not
+    assert msg["result"] == _expected_topology(topology, [entry.id, None, None])
+    matter_client.get_network_topology.assert_called_once_with(refresh=False)
+
+    matter_client.get_network_topology.reset_mock()
+    await ws_client.send_json({ID: 2, TYPE: "matter/network_topology", "refresh": True})
+    msg = await ws_client.receive_json()
+
+    assert msg["success"]
+    matter_client.get_network_topology.assert_called_once_with(refresh=True)
+
+
+@pytest.mark.parametrize(
+    "command", ["matter/network_topology", "matter/subscribe_network_topology"]
+)
+@pytest.mark.usefixtures("integration")
+async def test_network_topology_not_supported(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    matter_client: MagicMock,
+    command: str,
+) -> None:
+    """Test the topology commands against a server without topology support."""
+    # the conftest default schema version (1) predates network topology
+    matter_client.get_network_topology = AsyncMock()
+
+    ws_client = await hass_ws_client(hass)
+    await ws_client.send_json({ID: 1, TYPE: command})
+    msg = await ws_client.receive_json()
+
+    assert not msg["success"]
+    assert msg["error"]["code"] == "not_supported"
+    matter_client.get_network_topology.assert_not_called()
+
+    # a version mismatch raised by the client also maps to not_supported
+    matter_client.server_info.schema_version = 13
+    matter_client.get_network_topology.side_effect = ServerVersionTooOld(
+        "Command not available due to too old server version"
+    )
+    await ws_client.send_json({ID: 2, TYPE: command})
+    msg = await ws_client.receive_json()
+
+    assert not msg["success"]
+    assert msg["error"]["code"] == "not_supported"
+
+
+@pytest.mark.usefixtures("matter_node")
+@pytest.mark.parametrize("node_fixture", ["mock_onoff_light"])
+async def test_subscribe_network_topology(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    device_registry: dr.DeviceRegistry,
+    matter_client: MagicMock,
+) -> None:
+    """Test the subscribe_network_topology command."""
+    matter_client.server_info.schema_version = 13
+    entry = device_registry.async_get_device_by_identifier(
+        (DOMAIN, "deviceid_00000000000004D2-000000000000001E-MatterNodeDevice"),
+        hass.config_entries.async_entries(DOMAIN)[0].entry_id,
+    )
+    assert entry is not None
+
+    topology = _mock_topology()
+
+    subscription_callback: Callable[[EventType, NetworkTopology], None] | None = None
+    unsubscribe = MagicMock()
+
+    def capture_subscription(
+        callback: Callable[[EventType, NetworkTopology], None],
+        event_filter: EventType | None = None,
+        node_filter: int | None = None,
+        attr_path_filter: str | None = None,
+    ) -> MagicMock:
+        nonlocal subscription_callback
+        assert event_filter is EventType.NETWORK_TOPOLOGY_UPDATED
+        subscription_callback = callback
+        return unsubscribe
+
+    matter_client.subscribe_events.side_effect = capture_subscription
+
+    during_fetch = _mock_topology()
+    during_fetch.collected_at = 1767888030000
+
+    async def fetch_topology() -> NetworkTopology:
+        # an update arriving while the initial fetch is in flight is buffered
+        assert subscription_callback is not None
+        subscription_callback(EventType.NETWORK_TOPOLOGY_UPDATED, during_fetch)
+        return topology
+
+    matter_client.get_network_topology = AsyncMock(side_effect=fetch_topology)
+
+    ws_client = await hass_ws_client(hass)
+    await ws_client.send_json({ID: 1, TYPE: "matter/subscribe_network_topology"})
+    msg = await ws_client.receive_json()
+
+    assert msg["success"]
+    matter_client.get_network_topology.assert_called_once_with()
+    assert subscription_callback is not None
+
+    # the initial snapshot is pushed as the first event
+    msg = await ws_client.receive_json()
+    assert msg["type"] == "event"
+    assert msg["event"] == _expected_topology(topology, [entry.id, None, None])
+
+    # the update buffered during the fetch is flushed right after
+    msg = await ws_client.receive_json()
+    assert msg["type"] == "event"
+    assert msg["event"] == _expected_topology(during_fetch, [entry.id, None, None])
+
+    # a topology update from the server is forwarded to the subscription
+    updated = _mock_topology()
+    updated.collected_at = 1767888060000
+    updated.nodes = topology.nodes[:1]
+    updated.connections = []
+    subscription_callback(EventType.NETWORK_TOPOLOGY_UPDATED, updated)
+    msg = await ws_client.receive_json()
+
+    assert msg["type"] == "event"
+    assert msg["event"] == _expected_topology(updated, [entry.id])
+
+    await ws_client.send_json({ID: 2, TYPE: "unsubscribe_events", "subscription": 1})
+    msg = await ws_client.receive_json()
+
+    assert msg["success"]
+    unsubscribe.assert_called_once_with()
+
+
+@pytest.mark.usefixtures("integration")
+async def test_subscribe_network_topology_fetch_failure(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    matter_client: MagicMock,
+) -> None:
+    """Test the event subscription is cleaned up when the initial fetch fails."""
+    matter_client.server_info.schema_version = 13
+    unsubscribe = MagicMock()
+
+    def capture_subscription(
+        callback: Callable[[EventType, NetworkTopology], None],
+        event_filter: EventType | None = None,
+        node_filter: int | None = None,
+        attr_path_filter: str | None = None,
+    ) -> MagicMock:
+        return unsubscribe
+
+    matter_client.subscribe_events.side_effect = capture_subscription
+    matter_client.get_network_topology = AsyncMock(
+        side_effect=ServerVersionTooOld("Command not available")
+    )
+
+    ws_client = await hass_ws_client(hass)
+    await ws_client.send_json({ID: 1, TYPE: "matter/subscribe_network_topology"})
+    msg = await ws_client.receive_json()
+
+    assert not msg["success"]
+    assert msg["error"]["code"] == "not_supported"
+    unsubscribe.assert_called_once_with()