Cache device class data (#958)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

test/fixtures/climate_radio_thermostat_ct100_plus_state.json

@@ -10,7 +10,7 @@
     "generic": { "key": 2, "label": "Thermostat" },
     "specific": { "key": 3, "label": "Thermostat General V2" },
     "mandatorySupportedCCs": [],
-    "mandatoryControlCCs": []
+    "mandatoryControlledCCs": []
   },
   "isListening": true,
   "isFrequentListening": false,

test/fixtures/cover_qubino_shutter_state.json

@@ -10,7 +10,7 @@
     "generic": { "key": 2, "label": "Multilevel Switch" },
     "specific": { "key": 3, "label": "Motor Control Class C" },
     "mandatorySupportedCCs": [],
-    "mandatoryControlCCs": []
+    "mandatoryControlledCCs": []
   },
   "isListening": true,
   "isFrequentListening": false,

test/fixtures/idl_101_lock_state.json

@@ -6,9 +6,9 @@
   "status": 4,
   "ready": true,
   "deviceClass": {
-    "basic": "Routing Slave",
-    "generic": "Entry Control",
-    "specific": "Secure Keypad Door Lock",
+    "basic": { "key": 4, "label": "Routing Slave" },
+    "generic": { "key": 2, "label": "Entry Control" },
+    "specific": { "key": 3, "label": "Secure Keypad Door Lock" },
     "mandatorySupportedCCs": [
       "Basic",
       "Door Lock",
@@ -17,7 +17,7 @@
       "Security",
       "Version"
     ],
-    "mandatoryControlCCs": []
+    "mandatoryControlledCCs": []
   },
   "isListening": false,
   "isFrequentListening": true,

test/fixtures/lock_schlage_be469_state.json

@@ -8,7 +8,7 @@
     "generic": { "key": 2, "label": "Entry Control" },
     "specific": { "key": 3, "label": "Secure Keypad Door Lock" },
     "mandatorySupportedCCs": [],
-    "mandatoryControlCCs": []
+    "mandatoryControlledCCs": []
   },
   "isListening": false,
   "isFrequentListening": true,

test/fixtures/unparseable_json_string_value_state.json

@@ -8,7 +8,7 @@
     "generic": { "key": 2, "label": "Entry Control" },
     "specific": { "key": 3, "label": "Secure Keypad Door Lock" },
     "mandatorySupportedCCs": [],
-    "mandatoryControlCCs": []
+    "mandatoryControlledCCs": []
   },
   "isListening": false,
   "isFrequentListening": true,

test/model/test_node.py

@@ -611,6 +611,57 @@ def test_node_inclusion(multisensor_6_state):
     assert node.device_config.manufacturer == "AEON Labs"
     assert len(node.values) > 0
 
+    new_state = deepcopy(multisensor_6_state)
+    new_state["values"].append(
+        {
+            "commandClassName": "Binary Sensor",
+            "commandClass": 48,
+            "endpoint": 0,
+            "property": "test",
+            "propertyName": "test",
+            "metadata": {
+                "type": "boolean",
+                "readable": True,
+                "writeable": False,
+                "label": "Any",
+                "ccSpecific": {"sensorType": 255},
+            },
+            "value": False,
+        }
+    )
+    new_state["endpoints"].append(
+        {"nodeId": 52, "index": 1, "installerIcon": 3079, "userIcon": 3079}
+    )
+
+    event = Event(
+        "ready",
+        {
+            "event": "ready",
+            "source": "node",
+            "nodeId": node.node_id,
+            "nodeState": new_state,
+            "result": [],
+        },
+    )
+    node.receive_event(event)
+    assert "52-48-0-test" in node.values
+    assert 1 in node.endpoints
+
+    new_state = deepcopy(new_state)
+    new_state["endpoints"].pop(1)
+    event = Event(
+        "ready",
+        {
+            "event": "ready",
+            "source": "node",
+            "nodeId": node.node_id,
+            "nodeState": multisensor_6_state,
+            "result": [],
+        },
+    )
+    node.receive_event(event)
+    assert 1 not in node.endpoints
+
 
 def test_node_ready_event(switch_enbrighten_zw3010_state):
     """Emulate a node ready event."""

zwave_js_server/model/device_class.py

@@ -40,29 +40,33 @@ class DeviceClass:
 
     def __init__(self, data: DeviceClassDataType) -> None:
         """Initialize."""
-        self.data = data
+        self._basic = DeviceClassItem(**data["basic"])
+        self._generic = DeviceClassItem(**data["generic"])
+        self._specific = DeviceClassItem(**data["specific"])
+        self._mandatory_supported_ccs: list[int] = data["mandatorySupportedCCs"]
+        self._mandatory_controlled_ccs: list[int] = data["mandatoryControlledCCs"]
 
     @property
     def basic(self) -> DeviceClassItem:
         """Return basic DeviceClass."""
-        return DeviceClassItem(**self.data["basic"])
+        return self._basic
 
     @property
     def generic(self) -> DeviceClassItem:
         """Return generic DeviceClass."""
-        return DeviceClassItem(**self.data["generic"])
+        return self._generic
 
     @property
     def specific(self) -> DeviceClassItem:
         """Return specific DeviceClass."""
-        return DeviceClassItem(**self.data["specific"])
+        return self._specific
 
     @property
     def mandatory_supported_ccs(self) -> list[int]:
         """Return list of mandatory Supported CC id's."""
-        return self.data["mandatorySupportedCCs"]
+        return self._mandatory_supported_ccs
 
     @property
     def mandatory_controlled_ccs(self) -> list[int]:
         """Return list of mandatory Controlled CC id's."""
-        return self.data["mandatoryControlledCCs"]
+        return self._mandatory_controlled_ccs

zwave_js_server/model/endpoint.py

@@ -57,6 +57,7 @@ def __init__(
         self.node = node
         self.data: EndpointDataType = data
         self.values: dict[str, ConfigurationValue | Value] = {}
+        self._device_class: DeviceClass | None = None
         self.update(data, values)
 
     def __repr__(self) -> str:
@@ -90,9 +91,7 @@ def index(self) -> int:
     @property
     def device_class(self) -> DeviceClass | None:
         """Return the device_class."""
-        if (device_class := self.data.get("deviceClass")) is None:
-            return None
-        return DeviceClass(device_class)
+        return self._device_class
 
     @property
     def installer_icon(self) -> int | None:
@@ -119,6 +118,10 @@ def update(
     ) -> None:
         """Update the endpoint data."""
         self.data = data
+        if (device_class := self.data.get("deviceClass")) is None:
+            self._device_class = None
+        else:
+            self._device_class = DeviceClass(device_class)
 
         # Remove stale values
         self.values = {

zwave_js_server/model/node/__init__.py

@@ -27,7 +27,7 @@
 from ..command_class import CommandClassInfo
 from ..device_class import DeviceClass
 from ..device_config import DeviceConfig
-from ..endpoint import Endpoint
+from ..endpoint import Endpoint, EndpointDataType
 from ..notification import (
     EntryControlNotification,
     EntryControlNotificationDataType,
@@ -115,6 +115,7 @@ def __init__(self, client: Client, data: NodeDataType) -> None:
             client, data.get("statistics", DEFAULT_NODE_STATISTICS)
         )
         self._firmware_update_progress: NodeFirmwareUpdateProgress | None = None
+        self._device_class: DeviceClass | None = None
         self._last_seen: datetime | None = None
         self.values: dict[str, ConfigurationValue | Value] = {}
         self.endpoints: dict[int, Endpoint] = {}
@@ -150,9 +151,7 @@ def index(self) -> int:
     @property
     def device_class(self) -> DeviceClass | None:
         """Return the device_class."""
-        if (device_class := self.data.get("deviceClass")) is None:
-            return None
-        return DeviceClass(device_class)
+        return self._device_class
 
     @property
     def installer_icon(self) -> int | None:
@@ -372,21 +371,35 @@ def default_transition_duration(self) -> int | float | None:
         """Return the default transition duration."""
         return self.data.get("defaultTransitionDuration")
 
-    def update(self, data: NodeDataType) -> None:
-        """Update the internal state data."""
-        self.data = copy.deepcopy(data)
-        self._device_config = DeviceConfig(self.data.get("deviceConfig", {}))
-        self._statistics = NodeStatistics(
-            self.client, self.data.get("statistics", DEFAULT_NODE_STATISTICS)
-        )
-        if last_seen := data.get("lastSeen"):
-            self._last_seen = datetime.fromisoformat(last_seen)
-        if not self._statistics.last_seen:
-            self._statistics.last_seen = self.last_seen
+    def _update_endpoints(self, endpoints: list[EndpointDataType]) -> None:
+        """Update the endpoints data."""
+        new_endpoints_data = {endpoint["index"]: endpoint for endpoint in endpoints}
+        new_endpoint_idxs = set(new_endpoints_data)
+        stale_endpoint_idxs = set(self.endpoints) - new_endpoint_idxs
+
+        # Remove stale endpoints
+        for endpoint_idx in stale_endpoint_idxs:
+            self.endpoints.pop(endpoint_idx)
+
+        # Add new endpoints or update existing ones
+        for endpoint_idx in new_endpoint_idxs:
+            endpoint = new_endpoints_data[endpoint_idx]
+            values = {
+                value_id: value
+                for value_id, value in self.values.items()
+                if self.index == value.endpoint
+            }
+            if endpoint_idx in self.endpoints:
+                self.endpoints[endpoint_idx].update(endpoint, values)
+            else:
+                self.endpoints[endpoint_idx] = Endpoint(
+                    self.client, self, endpoint, values
+                )
 
+    def _update_values(self, values: list[ValueDataType]) -> None:
+        """Update the values data."""
         new_values_data = {
-            _get_value_id_str_from_dict(self, val): val
-            for val in self.data.pop("values")
+            _get_value_id_str_from_dict(self, val): val for val in values
         }
         new_value_ids = set(new_values_data)
         stale_value_ids = set(self.values) - new_value_ids
@@ -413,30 +426,25 @@ def update(self, data: NodeDataType) -> None:
                 # If we can't parse the value, don't store it
                 pass
 
-        new_endpoints_data = {
-            endpoint["index"]: endpoint for endpoint in self.data.pop("endpoints")
-        }
-        new_endpoint_idxs = set(new_endpoints_data)
-        stale_endpoint_idxs = set(self.endpoints) - new_endpoint_idxs
+    def update(self, data: NodeDataType) -> None:
+        """Update the internal state data."""
+        self.data = copy.deepcopy(data)
+        self._device_config = DeviceConfig(self.data.get("deviceConfig", {}))
+        if (device_class := self.data.get("deviceClass")) is None:
+            self._device_class = None
+        else:
+            self._device_class = DeviceClass(device_class)
 
-        # Remove stale endpoints
-        for endpoint_idx in stale_endpoint_idxs:
-            self.endpoints.pop(endpoint_idx)
+        self._statistics = NodeStatistics(
+            self.client, self.data.get("statistics", DEFAULT_NODE_STATISTICS)
+        )
+        if last_seen := data.get("lastSeen"):
+            self._last_seen = datetime.fromisoformat(last_seen)
+        if not self._statistics.last_seen:
+            self._statistics.last_seen = self.last_seen
 
-        # Add new endpoints or update existing ones
-        for endpoint_idx in new_endpoint_idxs - stale_endpoint_idxs:
-            endpoint = new_endpoints_data[endpoint_idx]
-            values = {
-                value_id: value
-                for value_id, value in self.values.items()
-                if self.index == value.endpoint
-            }
-            if endpoint_idx in self.endpoints:
-                self.endpoints[endpoint_idx].update(endpoint, values)
-            else:
-                self.endpoints[endpoint_idx] = Endpoint(
-                    self.client, self, endpoint, values
-                )
+        self._update_values(self.data.pop("values"))
+        self._update_endpoints(self.data.pop("endpoints"))
 
     def get_command_class_values(
         self, command_class: CommandClass, endpoint: int | None = None