Decouple device type from Matter device

Author: balloob

custom_components/matter_experimental/adapter.py

@@ -150,10 +150,9 @@ async def setup_node(self, node: MatterNode) -> None:
 
         for device in node.devices:
             created = False
-            device_type = type(device)
 
             for platform, devices in DEVICE_PLATFORM.items():
-                device_mappings = devices.get(device_type)
+                device_mappings = devices.get(device.device_type)
 
                 if device_mappings is None:
                     continue
@@ -167,7 +166,7 @@ async def setup_node(self, node: MatterNode) -> None:
                         "Creating %s entity for %s (%s)",
                         platform,
                         type(device),
-                        hex(device.device_type),
+                        hex(device.device_type.device_type),
                     )
                     entities.append(device_mapping.entity_cls(device, device_mapping))
 
@@ -178,7 +177,7 @@ async def setup_node(self, node: MatterNode) -> None:
                 self.logger.warning(
                     "Found unsupported device %s (%s)",
                     type(device).__name__,
-                    hex(device.device_type),
+                    hex(device.device_type.device_type),
                 )
 
     async def handle_server_disconnected(self, should_reload: bool) -> None:

custom_components/matter_experimental/binary_sensor.py

@@ -13,7 +13,8 @@
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from matter_server.client.model import devices as matter_devices
+from matter_server.client.model.device import MatterDevice
+from matter_server.vendor import device_types
 from matter_server.vendor.chip.clusters import Objects as clusters
 
 from .const import DOMAIN
@@ -37,9 +38,7 @@ async def async_setup_entry(
 class MatterBinarySensor(MatterEntity, BinarySensorEntity):
     """Representation of a Matter binary sensor."""
 
-    def __init__(
-        self, device: matter_devices.MatterDevice, mapping: DeviceMapping
-    ) -> None:
+    def __init__(self, device: MatterDevice, mapping: DeviceMapping) -> None:
         """Initialize the sensor."""
         super().__init__(device, mapping)
         self._attr_name = device.node.name or f"Matter Sensor {device.node.node_id}"
@@ -64,17 +63,17 @@ def _update_from_device(self) -> None:
 
 
 DEVICE_ENTITY: dict[
-    matter_devices.MatterDevice, DeviceMapping | list[DeviceMapping]
+    type[device_types.DeviceType], DeviceMapping | list[DeviceMapping]
 ] = {
-    matter_devices.ContactSensor: DeviceMapping(
+    device_types.ContactSensor: DeviceMapping(
         entity_cls=MatterBinarySensor,
         subscribe_attributes=(clusters.BooleanState.Attributes.StateValue,),
         entity_description=BinarySensorEntityDescription(
             key=None,
             device_class=BinarySensorDeviceClass.DOOR,
         ),
     ),
-    matter_devices.OccupancySensor: DeviceMapping(
+    device_types.OccupancySensor: DeviceMapping(
         entity_cls=MatterOccupancySensor,
         subscribe_attributes=(clusters.OccupancySensing.Attributes.Occupancy,),
     ),

custom_components/matter_experimental/device_platform.py

@@ -11,13 +11,13 @@
 from .switch import DEVICE_ENTITY as SWITCH_DEVICE_ENTITY
 
 if TYPE_CHECKING:
-    from matter_server.client.model import device as matter_devices
+    from matter_server.vendor.device_types import DeviceType
 
     from .device_platform_helper import DeviceMapping
 
 
 DEVICE_PLATFORM: dict[
-    Platform, dict[matter_devices.MatterDevice, DeviceMapping | list[DeviceMapping]]
+    Platform, dict[type[DeviceType], DeviceMapping | list[DeviceMapping]]
 ] = {
     Platform.BINARY_SENSOR: BINARY_SENSOR_DEVICE_ENTITY,
     Platform.LIGHT: LIGHT_DEVICE_ENTITY,

custom_components/matter_experimental/entity.py

@@ -14,14 +14,13 @@
 
 class MatterEntity(entity.Entity):
 
+    _attr_should_poll = False
     _unsubscribe: Callable[..., Coroutine[Any, Any, None]] | None = None
 
     def __init__(self, device: MatterDevice, mapping: DeviceMapping) -> None:
         self._device = device
         self._device_mapping = mapping
-        self._attr_unique_id = (
-            f"{device.node.unique_id}-{device.endpoint_id}-{device.device_type}"
-        )
+        self._attr_unique_id = f"{device.node.unique_id}-{device.endpoint_id}-{device.device_type.device_type}"
 
     @property
     def device_info(self) -> entity.DeviceInfo | None:
@@ -38,7 +37,7 @@ async def async_added_to_hass(self) -> None:
 
         # Subscribe to updates.
         self._unsubscribe = await self._device.subscribe_updates(
-            self._device_mapping.subscribe_attributes, self._update_from_device
+            self._device_mapping.subscribe_attributes, self._subscription_update
         )
 
         # Fetch latest info from the device.

custom_components/matter_experimental/light.py

@@ -9,7 +9,8 @@
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from matter_server.client.model import devices as matter_devices
+from matter_server.client.model.device import MatterDevice
+from matter_server.vendor import device_types
 from matter_server.vendor.chip.clusters import Objects as clusters
 
 from .const import DOMAIN
@@ -34,9 +35,7 @@ async def async_setup_entry(
 class MatterLight(MatterEntity, LightEntity):
     """Representation of a Matter light."""
 
-    def __init__(
-        self, device: matter_devices.MatterDevice, mapping: DeviceMapping
-    ) -> None:
+    def __init__(self, device: MatterDevice, mapping: DeviceMapping) -> None:
         """Initialize the light."""
         super().__init__(device, mapping)
         self._attr_name = device.node.name or f"Matter Light {device.node.node_id}"
@@ -99,20 +98,20 @@ def _update_from_device(self) -> None:
 
 
 DEVICE_ENTITY: dict[
-    matter_devices.MatterDevice, DeviceMapping | list[DeviceMapping]
+    type[device_types.DeviceType], DeviceMapping | list[DeviceMapping]
 ] = {
-    matter_devices.OnOffLight: DeviceMapping(
+    device_types.OnOffLight: DeviceMapping(
         entity_cls=MatterLight,
         subscribe_attributes=(clusters.OnOff.Attributes.OnOff,),
     ),
-    matter_devices.DimmableLight: DeviceMapping(
+    device_types.DimmableLight: DeviceMapping(
         entity_cls=MatterLight,
         subscribe_attributes=(
             clusters.OnOff.Attributes.OnOff,
             clusters.LevelControl.Attributes.CurrentLevel,
         ),
     ),
-    matter_devices.DimmablePlugInUnit: DeviceMapping(
+    device_types.DimmablePlugInUnit: DeviceMapping(
         entity_cls=MatterLight,
         subscribe_attributes=(
             clusters.OnOff.Attributes.OnOff,

custom_components/matter_experimental/sensor.py

@@ -20,7 +20,8 @@
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from matter_server.client.model import devices as matter_devices
+from matter_server.client.model.device import MatterDevice
+from matter_server.vendor import device_types
 from matter_server.vendor.chip.clusters import Objects as clusters
 from matter_server.vendor.chip.clusters.Types import NullValue
 
@@ -47,9 +48,7 @@ class MatterSensor(MatterEntity, SensorEntity):
 
     _attr_state_class = SensorStateClass.MEASUREMENT
 
-    def __init__(
-        self, device: matter_devices.MatterDevice, mapping: DeviceMapping
-    ) -> None:
+    def __init__(self, device: MatterDevice, mapping: DeviceMapping) -> None:
         """Initialize the sensor."""
         super().__init__(device, mapping)
         self._attr_name = device.node.name or f"Matter Sensor {device.node.node_id}"
@@ -158,29 +157,29 @@ def _update_from_device(self) -> None:
 
 
 DEVICE_ENTITY: dict[
-    matter_devices.MatterDevice, DeviceMapping | list[DeviceMapping]
+    type[device_types.DeviceType], DeviceMapping | list[DeviceMapping]
 ] = {
-    matter_devices.TemperatureSensor: DeviceMapping(
+    device_types.TemperatureSensor: DeviceMapping(
         entity_cls=MatterTemperatureSensor,
         subscribe_attributes=(
             clusters.TemperatureMeasurement.Attributes.MeasuredValue,
         ),
     ),
-    matter_devices.PressureSensor: DeviceMapping(
+    device_types.PressureSensor: DeviceMapping(
         entity_cls=MatterPressureSensor,
         subscribe_attributes=(clusters.PressureMeasurement.Attributes.MeasuredValue,),
     ),
-    matter_devices.FlowSensor: DeviceMapping(
+    device_types.FlowSensor: DeviceMapping(
         entity_cls=MatterFlowSensor,
         subscribe_attributes=(clusters.FlowMeasurement.Attributes.MeasuredValue,),
     ),
-    matter_devices.HumiditySensor: DeviceMapping(
+    device_types.HumiditySensor: DeviceMapping(
         entity_cls=MatterHumiditySensor,
         subscribe_attributes=(
             clusters.RelativeHumidityMeasurement.Attributes.MeasuredValue,
         ),
     ),
-    matter_devices.LightSensor: DeviceMapping(
+    device_types.LightSensor: DeviceMapping(
         entity_cls=MatterLightSensor,
         subscribe_attributes=(
             clusters.IlluminanceMeasurement.Attributes.MeasuredValue,

custom_components/matter_experimental/switch.py

@@ -13,7 +13,8 @@
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from matter_server.client.model import devices as matter_devices
+from matter_server.client.model.device import MatterDevice
+from matter_server.vendor import device_types
 from matter_server.vendor.chip.clusters import Objects as clusters
 
 from .const import DOMAIN
@@ -37,9 +38,7 @@ async def async_setup_entry(
 class MatterSwitch(MatterEntity, SwitchEntity):
     """Representation of a Matter switch."""
 
-    def __init__(
-        self, device: matter_devices.MatterDevice, mapping: DeviceMapping
-    ) -> None:
+    def __init__(self, device: MatterDevice, mapping: DeviceMapping) -> None:
         """Initialize the switch."""
         super().__init__(device, mapping)
         self._attr_name = device.node.name or f"Matter Switch {device.node.node_id}"
@@ -63,9 +62,9 @@ def _update_from_device(self) -> None:
 
 
 DEVICE_ENTITY: dict[
-    matter_devices.MatterDevice, DeviceMapping | list[DeviceMapping]
+    type[device_types.DeviceType], DeviceMapping | list[DeviceMapping]
 ] = {
-    matter_devices.OnOffPlugInUnit: DeviceMapping(
+    device_types.OnOffPlugInUnit: DeviceMapping(
         entity_cls=MatterSwitch,
         subscribe_attributes=(clusters.OnOff.Attributes.OnOff,),
         entity_description=SwitchEntityDescription(

matter_server/client/model/device.py

@@ -1,36 +1,32 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, Callable, Coroutine, TypeVar
+from typing import TYPE_CHECKING, Callable, Coroutine, Generic, TypeVar
 
+from matter_server.vendor import device_types
 from matter_server.vendor.chip.clusters import Objects as all_clusters
 
 if TYPE_CHECKING:
     from .node import MatterNode
 
-DEVICE_TYPES = {}
-
 SubscriberType = Callable[[], None]
 
 
+_DEVICE_TYPE_T = TypeVar("_DEVICE_TYPE_T", bound=device_types.DeviceType)
 _CLUSTER_T = TypeVar("_CLUSTER_T", bound=all_clusters.Cluster)
 
 
-class MatterDevice:
+class MatterDevice(Generic[_DEVICE_TYPE_T]):
     """Base class for Matter devices."""
 
-    device_type: int
-    clusters: set[type[all_clusters.Cluster]] = set()
-
-    def __init_subclass__(cls, *, device_type: int, **kwargs: Any) -> None:
-        """Initialize a subclass, register if possible."""
-        super().__init_subclass__(**kwargs)
-        cls.device_type = device_type
-        DEVICE_TYPES[device_type] = cls
-
     def __init__(
-        self, node: MatterNode, endpoint_id: int, device_revision: int
+        self,
+        node: MatterNode,
+        device_type: _DEVICE_TYPE_T,
+        endpoint_id: int,
+        device_revision: int,
     ) -> None:
         self.node = node
+        self.device_type = device_type
         self.device_revision = device_revision
         self.endpoint_id = endpoint_id
         self._on_update_listener: SubscriberType | None = None
@@ -39,9 +35,9 @@ def __init__(
     def data(self) -> dict:
         return self.node.raw_data["attributes"][str(self.endpoint_id)]
 
-    def has_cluster(self, cluster: type[clusters.Cluster]) -> bool:
+    def has_cluster(self, cluster: type[all_clusters.Cluster]) -> bool:
         """Check if device has a specific cluster."""
-        return cluster in self.clusters and cluster.__name__ in self.data
+        return cluster in self.device_type.clusters and cluster.__name__ in self.data
 
     def get_cluster(self, cluster: type[_CLUSTER_T]) -> _CLUSTER_T | None:
         """Get the cluster object."""

matter_server/client/model/node.py

@@ -3,10 +3,11 @@
 
 from typing import TYPE_CHECKING
 
-from .device import DEVICE_TYPES, MatterDevice
-from .devices import RootNode
-
 from matter_server.vendor.chip.clusters import Objects as all_clusters
+from matter_server.vendor import device_types
+
+from .device import MatterDevice
+
 
 if TYPE_CHECKING:
     from ..matter import Matter
@@ -15,7 +16,7 @@
 class MatterNode:
     """Matter node."""
 
-    root_device: RootNode
+    root_device: MatterDevice[device_types.RootNode]
 
     def __init__(self, matter: Matter, node_info: dict) -> None:
         self.matter = matter
@@ -26,16 +27,18 @@ def __init__(self, matter: Matter, node_info: dict) -> None:
         for endpoint_id, endpoint_info in node_info["attributes"].items():
             descriptor: all_clusters.Descriptor = endpoint_info["Descriptor"]
             for device_info in descriptor.deviceList:
-                device_cls = DEVICE_TYPES.get(device_info.type)
+                device_type = device_types.ALL_TYPES.get(device_info.type)
 
-                if device_cls is None:
+                if device_type is None:
                     matter.adapter.logger.warning(
                         "Found unknown device type %s", device_info.type
                     )
                     continue
 
-                device = device_cls(self, int(endpoint_id), device_info.revision)
-                if isinstance(device, RootNode):
+                device = MatterDevice(
+                    self, device_type, int(endpoint_id), device_info.revision
+                )
+                if device_type is device_types.RootNode:
                     self.root_device = device
                 else:
                     devices.append(device)