Add support for config_manager.lookup_device (#1189)

* Add support for `config_manager.lookup_device`

* fix test

* lint

* PR comments

* fix type

Author: MindFreeze

test/conftest.py

@@ -127,6 +127,12 @@ def lock_ultraloq_ubolt_pro_state_fixture():
     return json.loads(load_fixture("lock_ultraloq_ubolt_pro_state.json"))
 
 
+@pytest.fixture(name="device_config", scope="session")
+def device_config_fixture() -> dict[str, Any]:
+    """Load the device config fixture data."""
+    return json.loads(load_fixture("device_config.json"))
+
+
 @pytest.fixture(name="client_session")
 def client_session_fixture(ws_client: AsyncMock) -> AsyncMock:
     """Mock an aiohttp client session."""

test/fixtures/device_config.json

@@ -0,0 +1,22 @@
+{
+  "filename": "test_device.json",
+  "manufacturer": "Test Manufacturer",
+  "manufacturerId": "0x1234",
+  "label": "Test Device",
+  "description": "Test Device Description",
+  "devices": [
+    {
+      "productType": "0x5678",
+      "productId": "0x9ABC"
+    }
+  ],
+  "firmwareVersion": {
+    "min": "1.0",
+    "max": "2.0"
+  },
+  "associations": {},
+  "paramInformation": {},
+  "supportsZWavePlus": true,
+  "proprietary": {},
+  "compat": {}
+}

test/model/test_config_manager.py

@@ -0,0 +1,73 @@
+"""Test the config manager."""
+
+from zwave_js_server.client import Client
+from zwave_js_server.model.config_manager import ConfigManager
+from zwave_js_server.model.device_config import DeviceConfigDataType
+
+from ..common import MockCommandProtocol
+
+
+async def test_lookup_device(
+    client: Client,
+    mock_command: MockCommandProtocol,
+    device_config: DeviceConfigDataType,
+) -> None:
+    """Test the lookup_device command."""
+    # Test successful lookup
+    mock_command(
+        {
+            "command": "config_manager.lookup_device",
+            "manufacturerId": 0x1234,
+            "productType": 0x5678,
+            "productId": 0x9ABC,
+        },
+        {"config": device_config},
+    )
+
+    config_manager = ConfigManager(client)
+    result = await config_manager.lookup_device(0x1234, 0x5678, 0x9ABC)
+
+    assert result is not None
+    assert result.manufacturer == "Test Manufacturer"
+    assert result.manufacturer_id == "0x1234"
+    assert result.label == "Test Device"
+    assert result.description == "Test Device Description"
+    assert len(result.devices) == 1
+    assert result.devices[0].product_type == "0x5678"
+    assert result.devices[0].product_id == "0x9ABC"
+    assert result.firmware_version.min == "1.0"
+    assert result.firmware_version.max == "2.0"
+
+    # Test lookup with firmware version
+    mock_command(
+        {
+            "command": "config_manager.lookup_device",
+            "manufacturerId": 0x1234,
+            "productType": 0x5678,
+            "productId": 0x9ABC,
+            "firmwareVersion": "1.5",
+        },
+        {"config": device_config},
+    )
+
+    result = await config_manager.lookup_device(0x1234, 0x5678, 0x9ABC, "1.5")
+
+    assert result is not None
+    assert result.manufacturer == "Test Manufacturer"
+
+    assert result.to_dict() == device_config
+
+    # Test device not found
+    mock_command(
+        {
+            "command": "config_manager.lookup_device",
+            "manufacturerId": 0x4321,
+            "productType": 0x8765,
+            "productId": 0xCBA9,
+        },
+        {"config": None},
+    )
+
+    result = await config_manager.lookup_device(0x4321, 0x8765, 0xCBA9)
+
+    assert result is None

test/model/test_driver.py

@@ -390,3 +390,9 @@ async def test_all_nodes_ready_event(driver):
     """Test that the all nodes ready event is succesfully validated by pydantic."""
     event = Event("all nodes ready", {"source": "driver", "event": "all nodes ready"})
     driver.receive_event(event)
+
+
+def test_config_manager(driver):
+    """Test the driver has the config manager property."""
+    assert driver.config_manager is not None
+    assert driver.config_manager._client is driver.client

zwave_js_server/model/config_manager/__init__.py

@@ -0,0 +1,47 @@
+"""
+Model for Z-Wave JS config manager.
+
+https://zwave-js.github.io/node-zwave-js/#/api/config-manager
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Optional
+
+from ..device_config import DeviceConfig
+
+if TYPE_CHECKING:
+    from ...client import Client
+
+
+class ConfigManager:
+    """Model for the Z-Wave JS config manager."""
+
+    def __init__(self, client: Client) -> None:
+        """Initialize."""
+        self._client = client
+
+    async def lookup_device(
+        self,
+        manufacturer_id: int,
+        product_type: int,
+        product_id: int,
+        firmware_version: Optional[str] = None,
+    ) -> DeviceConfig | None:
+        """Look up the definition of a given device in the configuration DB."""
+        cmd: dict[str, Any] = {
+            "command": "config_manager.lookup_device",
+            "manufacturerId": manufacturer_id,
+            "productType": product_type,
+            "productId": product_id,
+        }
+
+        if firmware_version is not None:
+            cmd["firmwareVersion"] = firmware_version
+
+        data = await self._client.async_send_command(cmd)
+
+        if not data or not (config := data.get("config")):
+            return None
+
+        return DeviceConfig(config)

zwave_js_server/model/device_config.py

@@ -222,3 +222,7 @@ def metadata(self) -> DeviceMetadata:
     def is_embedded(self) -> bool | None:
         """Return whether device config is embedded in zwave-js-server."""
         return self.data.get("isEmbedded")
+
+    def to_dict(self) -> DeviceConfigDataType:
+        """Return dict representation of device config."""
+        return self.data.copy()

zwave_js_server/model/driver.py

@@ -5,6 +5,7 @@
 from typing import TYPE_CHECKING, Any, Literal, cast
 
 from ..event import BaseEventModel, Event, EventBase
+from .config_manager import ConfigManager
 from .controller import Controller
 from .log_config import LogConfig, LogConfigDataType
 from .log_message import LogMessage, LogMessageDataType
@@ -79,6 +80,7 @@ def __init__(
         self.client = client
         self.controller = Controller(client, state)
         self.log_config = LogConfig.from_dict(log_config)
+        self.config_manager = ConfigManager(client)
 
     def __hash__(self) -> int:
         """Return the hash."""