Add node_status websocket API command to zwave_js

homeassistant/components/zwave_js/websocket_api.py

@@ -19,13 +19,15 @@
 
 ID = "id"
 ENTRY_ID = "entry_id"
+NODE_ID = "node_id"
 TYPE = "type"
 
 
 @callback
 def async_register_api(hass: HomeAssistant) -> None:
     """Register all of our api endpoints."""
     websocket_api.async_register_command(hass, websocket_network_status)
+    websocket_api.async_register_command(hass, websocket_node_status)
     websocket_api.async_register_command(hass, websocket_add_node)
     websocket_api.async_register_command(hass, websocket_stop_inclusion)
     websocket_api.async_register_command(hass, websocket_remove_node)
@@ -61,6 +63,35 @@ def websocket_network_status(
     )
 
 
+@websocket_api.websocket_command(
+    {
+        vol.Required(TYPE): "zwave_js/node_status",
+        vol.Required(ENTRY_ID): str,
+        vol.Required(NODE_ID): int,
+    }
+)
+@callback
+def websocket_node_status(
+    hass: HomeAssistant, connection: ActiveConnection, msg: dict
+) -> None:
+    """Get the status of a Z-Wave JS node."""
+    entry_id = msg[ENTRY_ID]
+    client = hass.data[DOMAIN][entry_id][DATA_CLIENT]
+    node_id = msg[NODE_ID]
+    node = client.driver.controller.nodes[node_id]
+    data = {
+        "node_id": node.node_id,
+        "is_routing": node.is_routing,
+        "status": node.status,
+        "is_secure": node.is_secure,
+        "ready": node.ready,
+    }
+    connection.send_result(
+        msg[ID],
+        data,
+    )
+
+
 @websocket_api.require_admin  # type: ignore
 @websocket_api.async_response
 @websocket_api.websocket_command(

tests/components/zwave_js/test_websocket_api.py

@@ -2,12 +2,12 @@
 from zwave_js_server.event import Event
 
 from homeassistant.components.zwave_js.const import DOMAIN
-from homeassistant.components.zwave_js.websocket_api import ENTRY_ID, ID, TYPE
+from homeassistant.components.zwave_js.websocket_api import ENTRY_ID, ID, NODE_ID, TYPE
 from homeassistant.helpers.device_registry import async_get_registry
 
 
-async def test_websocket_api(hass, integration, hass_ws_client):
-    """Test the network_status websocket command."""
+async def test_websocket_api(hass, integration, multisensor_6, hass_ws_client):
+    """Test the network and node status websocket commands."""
     entry = integration
     ws_client = await hass_ws_client(hass)
 
@@ -20,6 +20,24 @@ async def test_websocket_api(hass, integration, hass_ws_client):
     assert result["client"]["ws_server_url"] == "ws://test:3000/zjs"
     assert result["client"]["server_version"] == "1.0.0"
 
+    node = multisensor_6
+    await ws_client.send_json(
+        {
+            ID: 3,
+            TYPE: "zwave_js/node_status",
+            ENTRY_ID: entry.entry_id,
+            NODE_ID: node.node_id,
+        }
+    )
+    msg = await ws_client.receive_json()
+    result = msg["result"]
+
+    assert result[NODE_ID] == 52
+    assert result["ready"]
+    assert result["is_routing"]
+    assert not result["is_secure"]
+    assert result["status"] == 1
+
 
 async def test_add_node(
     hass, integration, client, hass_ws_client, nortek_thermostat_added_event