evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix support for Bridge(d) and composed devices in Matter (#88662)

Browse source 
* Refactor discovery of entities to support composed and bridged devices

* Bump library version to 3.1.0

* move discovery schemas to platforms

* optimize a tiny bit

* simplify even more

* fixed bug in light platform

* fix color control logic

* fix some issues

* Update homeassistant/components/matter/discovery.py

Co-authored-by: Paulus Schoutsen <balloob@gmail.com>

* fix some tests

* fix light test

---------

Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
This commit is contained in:
Marcel van der Veldt 2023-02-23 20:58:37 +01:00 • committed by GitHub
parent e0601530a0
commit 301144993c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
18 changed files with 582 additions and 526 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
homeassistant/components/matter/__init__.py
View file

@ -27,7 +27,7 @@ from .adapter import MatterAdapter
from .addon import get_addon_manager
from .api import async_register_api
from .const import CONF_INTEGRATION_CREATED_ADDON, CONF_USE_ADDON, DOMAIN, LOGGER
from .device_platform import DEVICE_PLATFORM
from .discovery import SUPPORTED_PLATFORMS
from .helpers import MatterEntryData, get_matter, get_node_from_device_entry
CONNECT_TIMEOUT = 10
@ -101,12 +101,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
matter = MatterAdapter(hass, matter_client, entry)
hass.data[DOMAIN][entry.entry_id] = MatterEntryData(matter, listen_task)
await hass.config_entries.async_forward_entry_setups(entry, DEVICE_PLATFORM)
await hass.config_entries.async_forward_entry_setups(entry, SUPPORTED_PLATFORMS)
await matter.setup_nodes()
# If the listen task is already failed, we need to raise ConfigEntryNotReady
if listen_task.done() and (listen_error := listen_task.exception()) is not None:
await hass.config_entries.async_unload_platforms(entry, DEVICE_PLATFORM)
await hass.config_entries.async_unload_platforms(entry, SUPPORTED_PLATFORMS)
hass.data[DOMAIN].pop(entry.entry_id)
try:
await matter_client.disconnect()
@ -142,7 +142,9 @@ async def _client_listen(
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, DEVICE_PLATFORM)
unload_ok = await hass.config_entries.async_unload_platforms(
entry, SUPPORTED_PLATFORMS
)
if unload_ok:
matter_entry_data: MatterEntryData = hass.data[DOMAIN].pop(entry.entry_id)

122
homeassistant/components/matter/adapter.py
View file

@ -3,11 +3,6 @@ from __future__ import annotations
from typing import TYPE_CHECKING, cast
from chip.clusters import Objects as all_clusters
from matter_server.client.models.node_device import (
AbstractMatterNodeDevice,
MatterBridgedNodeDevice,
)
from matter_server.common.models import EventType, ServerInfoMessage
from homeassistant.config_entries import ConfigEntry
@ -17,12 +12,12 @@ from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN, ID_TYPE_DEVICE_ID, ID_TYPE_SERIAL, LOGGER
from .device_platform import DEVICE_PLATFORM
from .discovery import async_discover_entities
from .helpers import get_device_id
if TYPE_CHECKING:
from matter_server.client import MatterClient
from matter_server.client.models.node import MatterNode
from matter_server.client.models.node import MatterEndpoint, MatterNode
class MatterAdapter:
@ -51,12 +46,8 @@ class MatterAdapter:
for node in await self.matter_client.get_nodes():
self._setup_node(node)
def node_added_callback(event: EventType, node: MatterNode | None) -> None:
def node_added_callback(event: EventType, node: MatterNode) -> None:
"""Handle node added event."""
if node is None:
# We can clean this up when we've improved the typing in the library.
# https://github.com/home-assistant-libs/python-matter-server/pull/153
raise RuntimeError("Node added event without node")
self._setup_node(node)
self.config_entry.async_on_unload(
@ -67,48 +58,32 @@ class MatterAdapter:
"""Set up an node."""
LOGGER.debug("Setting up entities for node %s", node.node_id)
bridge_unique_id: str | None = None
if (
node.aggregator_device_type_instance is not None
and node.root_device_type_instance is not None
and node.root_device_type_instance.get_cluster(
all_clusters.BasicInformation
)
):
# create virtual (parent) device for bridge node device
bridge_device = MatterBridgedNodeDevice(
node.aggregator_device_type_instance
)
self._create_device_registry(bridge_device)
server_info = cast(ServerInfoMessage, self.matter_client.server_info)
bridge_unique_id = get_device_id(server_info, bridge_device)
for node_device in node.node_devices:
self._setup_node_device(node_device, bridge_unique_id)
for endpoint in node.endpoints.values():
# Node endpoints are translated into HA devices
self._setup_endpoint(endpoint)
def _create_device_registry(
self,
node_device: AbstractMatterNodeDevice,
bridge_unique_id: str | None = None,
endpoint: MatterEndpoint,
) -> None:
"""Create a device registry entry."""
"""Create a device registry entry for a MatterNode."""
server_info = cast(ServerInfoMessage, self.matter_client.server_info)
basic_info = node_device.device_info()
device_type_instances = node_device.device_type_instances()
basic_info = endpoint.device_info
name = basic_info.nodeLabel or basic_info.productLabel or basic_info.productName
name = basic_info.nodeLabel
if not name and isinstance(node_device, MatterBridgedNodeDevice):
# fallback name for Bridge
name = "Hub device"
elif not name and device_type_instances:
# use the productName if no node label is present
name = basic_info.productName
# handle bridged devices
bridge_device_id = None
if endpoint.is_bridged_device:
bridge_device_id = get_device_id(
server_info,
endpoint.node.endpoints[0],
)
bridge_device_id = f"{ID_TYPE_DEVICE_ID}_{bridge_device_id}"
node_device_id = get_device_id(
server_info,
node_device,
endpoint,
)
identifiers = {(DOMAIN, f"{ID_TYPE_DEVICE_ID}_{node_device_id}")}
# if available, we also add the serialnumber as identifier
@ -124,50 +99,21 @@ class MatterAdapter:
sw_version=basic_info.softwareVersionString,
manufacturer=basic_info.vendorName,
model=basic_info.productName,
via_device=(DOMAIN, bridge_unique_id) if bridge_unique_id else None,
via_device=(DOMAIN, bridge_device_id) if bridge_device_id else None,
)
def _setup_node_device(
self, node_device: AbstractMatterNodeDevice, bridge_unique_id: str | None
) -> None:
"""Set up a node device."""
self._create_device_registry(node_device, bridge_unique_id)
def _setup_endpoint(self, endpoint: MatterEndpoint) -> None:
"""Set up a MatterEndpoint as HA Device."""
# pre-create device registry entry
self._create_device_registry(endpoint)
# run platform discovery from device type instances
for instance in node_device.device_type_instances():
created = False
for platform, devices in DEVICE_PLATFORM.items():
entity_descriptions = devices.get(instance.device_type)
if entity_descriptions is None:
continue
if not isinstance(entity_descriptions, list):
entity_descriptions = [entity_descriptions]
entities = []
for entity_description in entity_descriptions:
LOGGER.debug(
"Creating %s entity for %s (%s)",
platform,
instance.device_type.__name__,
hex(instance.device_type.device_type),
)
entities.append(
entity_description.entity_cls(
self.matter_client,
node_device,
instance,
entity_description,
)
)
self.platform_handlers[platform](entities)
created = True
if not created:
LOGGER.warning(
"Found unsupported device %s (%s)",
type(instance).__name__,
hex(instance.device_type.device_type),
)
for entity_info in async_discover_entities(endpoint):
LOGGER.debug(
"Creating %s entity for %s",
entity_info.platform,
entity_info.primary_attribute,
)
new_entity = entity_info.entity_class(
self.matter_client, endpoint, entity_info
)
self.platform_handlers[entity_info.platform]([new_entity])

117
homeassistant/components/matter/binary_sensor.py
View file

@ -1,11 +1,9 @@
"""Matter binary sensors."""
from __future__ import annotations
from dataclasses import dataclass
from functools import partial
from chip.clusters import Objects as clusters
from matter_server.client.models import device_types
from chip.clusters.Objects import uint
from chip.clusters.Types import Nullable, NullValue
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
@ -17,8 +15,9 @@ from homeassistant.const import Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .entity import MatterEntity, MatterEntityDescriptionBaseClass
from .entity import MatterEntity
from .helpers import get_matter
from .models import MatterDiscoverySchema
async def async_setup_entry(
@ -34,60 +33,70 @@ async def async_setup_entry(
class MatterBinarySensor(MatterEntity, BinarySensorEntity):
"""Representation of a Matter binary sensor."""
entity_description: MatterBinarySensorEntityDescription
@callback
def _update_from_device(self) -> None:
"""Update from device."""
self._attr_is_on = self.get_matter_attribute_value(
# We always subscribe to a single value
self.entity_description.subscribe_attributes[0],
)
value: bool | uint | int | Nullable | None
value = self.get_matter_attribute_value(self._entity_info.primary_attribute)
if value in (None, NullValue):
value = None
elif value_convert := self._entity_info.measurement_to_ha:
value = value_convert(value)
self._attr_is_on = value
class MatterOccupancySensor(MatterBinarySensor):
"""Representation of a Matter occupancy sensor."""
_attr_device_class = BinarySensorDeviceClass.OCCUPANCY
@callback
def _update_from_device(self) -> None:
"""Update from device."""
value = self.get_matter_attribute_value(
# We always subscribe to a single value
self.entity_description.subscribe_attributes[0],
)
# Discovery schema(s) to map Matter Attributes to HA entities
DISCOVERY_SCHEMAS = [
# device specific: translate Hue motion to sensor to HA Motion sensor
# instead of generic occupancy sensor
MatterDiscoverySchema(
platform=Platform.BINARY_SENSOR,
entity_description=BinarySensorEntityDescription(
key="HueMotionSensor",
device_class=BinarySensorDeviceClass.MOTION,
name="Motion",
),
entity_class=MatterBinarySensor,
required_attributes=(clusters.OccupancySensing.Attributes.Occupancy,),
vendor_id=(4107,),
product_name=("Hue motion sensor",),
measurement_to_ha=lambda x: (x & 1 == 1) if x is not None else None,
),
MatterDiscoverySchema(
platform=Platform.BINARY_SENSOR,
entity_description=BinarySensorEntityDescription(
key="ContactSensor",
device_class=BinarySensorDeviceClass.DOOR,
name="Contact",
),
entity_class=MatterBinarySensor,
required_attributes=(clusters.BooleanState.Attributes.StateValue,),
# value is inverted on matter to what we expect
measurement_to_ha=lambda x: not x,
),
MatterDiscoverySchema(
platform=Platform.BINARY_SENSOR,
entity_description=BinarySensorEntityDescription(
key="OccupancySensor",
device_class=BinarySensorDeviceClass.OCCUPANCY,
name="Occupancy",
),
entity_class=MatterBinarySensor,
required_attributes=(clusters.OccupancySensing.Attributes.Occupancy,),
# The first bit = if occupied
self._attr_is_on = (value & 1 == 1) if value is not None else None
@dataclass
class MatterBinarySensorEntityDescription(
BinarySensorEntityDescription,
MatterEntityDescriptionBaseClass,
):
"""Matter Binary Sensor entity description."""
# You can't set default values on inherited data classes
MatterSensorEntityDescriptionFactory = partial(
MatterBinarySensorEntityDescription, entity_cls=MatterBinarySensor
)
DEVICE_ENTITY: dict[
type[device_types.DeviceType],
MatterEntityDescriptionBaseClass | list[MatterEntityDescriptionBaseClass],
] = {
device_types.ContactSensor: MatterSensorEntityDescriptionFactory(
key=device_types.ContactSensor,
name="Contact",
subscribe_attributes=(clusters.BooleanState.Attributes.StateValue,),
device_class=BinarySensorDeviceClass.DOOR,
measurement_to_ha=lambda x: (x & 1 == 1) if x is not None else None,
),
device_types.OccupancySensor: MatterSensorEntityDescriptionFactory(
key=device_types.OccupancySensor,
name="Occupancy",
entity_cls=MatterOccupancySensor,
subscribe_attributes=(clusters.OccupancySensing.Attributes.Occupancy,),
MatterDiscoverySchema(
platform=Platform.BINARY_SENSOR,
entity_description=BinarySensorEntityDescription(
key="BatteryChargeLevel",
device_class=BinarySensorDeviceClass.BATTERY,
name="Battery Status",
),
entity_class=MatterBinarySensor,
required_attributes=(clusters.PowerSource.Attributes.BatChargeLevel,),
# only add binary battery sensor if a regular percentage based is not available
absent_attributes=(clusters.PowerSource.Attributes.BatPercentRemaining,),
measurement_to_ha=lambda x: x != clusters.PowerSource.Enums.BatChargeLevel.kOk,
),
}
]

30
homeassistant/components/matter/device_platform.py
View file

@ -1,30 +0,0 @@
"""All mappings of Matter devices to Home Assistant platforms."""
from __future__ import annotations
from typing import TYPE_CHECKING
from homeassistant.const import Platform
from .binary_sensor import DEVICE_ENTITY as BINARY_SENSOR_DEVICE_ENTITY
from .light import DEVICE_ENTITY as LIGHT_DEVICE_ENTITY
from .sensor import DEVICE_ENTITY as SENSOR_DEVICE_ENTITY
from .switch import DEVICE_ENTITY as SWITCH_DEVICE_ENTITY
if TYPE_CHECKING:
from matter_server.client.models.device_types import DeviceType
from .entity import MatterEntityDescriptionBaseClass
DEVICE_PLATFORM: dict[
Platform,
dict[
type[DeviceType],
MatterEntityDescriptionBaseClass | list[MatterEntityDescriptionBaseClass],
],
] = {
Platform.BINARY_SENSOR: BINARY_SENSOR_DEVICE_ENTITY,
Platform.LIGHT: LIGHT_DEVICE_ENTITY,
Platform.SENSOR: SENSOR_DEVICE_ENTITY,
Platform.SWITCH: SWITCH_DEVICE_ENTITY,
}

115
homeassistant/components/matter/discovery.py Normal file
View file

@ -0,0 +1,115 @@
"""Map Matter Nodes and Attributes to Home Assistant entities."""
from __future__ import annotations
from collections.abc import Generator
from chip.clusters.Objects import ClusterAttributeDescriptor
from matter_server.client.models.node import MatterEndpoint
from homeassistant.const import Platform
from homeassistant.core import callback
from .binary_sensor import DISCOVERY_SCHEMAS as BINARY_SENSOR_SCHEMAS
from .light import DISCOVERY_SCHEMAS as LIGHT_SCHEMAS
from .models import MatterDiscoverySchema, MatterEntityInfo
from .sensor import DISCOVERY_SCHEMAS as SENSOR_SCHEMAS
from .switch import DISCOVERY_SCHEMAS as SWITCH_SCHEMAS
DISCOVERY_SCHEMAS: dict[Platform, list[MatterDiscoverySchema]] = {
Platform.BINARY_SENSOR: BINARY_SENSOR_SCHEMAS,
Platform.LIGHT: LIGHT_SCHEMAS,
Platform.SENSOR: SENSOR_SCHEMAS,
Platform.SWITCH: SWITCH_SCHEMAS,
}
SUPPORTED_PLATFORMS = tuple(DISCOVERY_SCHEMAS.keys())
@callback
def iter_schemas() -> Generator[MatterDiscoverySchema, None, None]:
"""Iterate over all available discovery schemas."""
for platform_schemas in DISCOVERY_SCHEMAS.values():
yield from platform_schemas
@callback
def async_discover_entities(
endpoint: MatterEndpoint,
) -> Generator[MatterEntityInfo, None, None]:
"""Run discovery on MatterEndpoint and return matching MatterEntityInfo(s)."""
discovered_attributes: set[type[ClusterAttributeDescriptor]] = set()
device_info = endpoint.device_info
for schema in iter_schemas():
# abort if attribute(s) already discovered
if any(x in schema.required_attributes for x in discovered_attributes):
continue
# check vendor_id
if (
schema.vendor_id is not None
and device_info.vendorID not in schema.vendor_id
):
continue
# check product_name
if (
schema.product_name is not None
and device_info.productName not in schema.product_name
):
continue
# check required device_type
if schema.device_type is not None and not any(
x in schema.device_type for x in endpoint.device_types
):
continue
# check absent device_type
if schema.not_device_type is not None and any(
x in schema.not_device_type for x in endpoint.device_types
):
continue
# check endpoint_id
if (
schema.endpoint_id is not None
and endpoint.endpoint_id not in schema.endpoint_id
):
continue
# check required attributes
if schema.required_attributes is not None and not all(
endpoint.has_attribute(None, val_schema)
for val_schema in schema.required_attributes
):
continue
# check for values that may not be present
if schema.absent_attributes is not None and any(
endpoint.has_attribute(None, val_schema)
for val_schema in schema.absent_attributes
):
continue
# all checks passed, this value belongs to an entity
attributes_to_watch = list(schema.required_attributes)
if schema.optional_attributes:
# check optional attributes
for optional_attribute in schema.optional_attributes:
if optional_attribute in attributes_to_watch:
continue
if endpoint.has_attribute(None, optional_attribute):
attributes_to_watch.append(optional_attribute)
yield MatterEntityInfo(
endpoint=endpoint,
platform=schema.platform,
attributes_to_watch=attributes_to_watch,
entity_description=schema.entity_description,
entity_class=schema.entity_class,
measurement_to_ha=schema.measurement_to_ha,
)
# prevent re-discovery of the same attributes
if not schema.allow_multi:
discovered_attributes.update(attributes_to_watch)

60
homeassistant/components/matter/entity.py
View file

@ -3,90 +3,77 @@ from __future__ import annotations
from abc import abstractmethod
from collections.abc import Callable
from dataclasses import dataclass
import logging
from typing import TYPE_CHECKING, Any, cast
from chip.clusters.Objects import ClusterAttributeDescriptor
from matter_server.client.models.device_type_instance import MatterDeviceTypeInstance
from matter_server.client.models.node_device import AbstractMatterNodeDevice
from matter_server.common.helpers.util import create_attribute_path
from matter_server.common.models import EventType, ServerInfoMessage
from homeassistant.core import callback
from homeassistant.helpers.entity import DeviceInfo, Entity, EntityDescription
from homeassistant.helpers.entity import DeviceInfo, Entity
from .const import DOMAIN, ID_TYPE_DEVICE_ID
from .helpers import get_device_id, get_operational_instance_id
from .helpers import get_device_id
if TYPE_CHECKING:
from matter_server.client import MatterClient
from matter_server.client.models.node import MatterEndpoint
from .discovery import MatterEntityInfo
LOGGER = logging.getLogger(__name__)
@dataclass
class MatterEntityDescription:
"""Mixin to map a matter device to a Home Assistant entity."""
entity_cls: type[MatterEntity]
subscribe_attributes: tuple
@dataclass
class MatterEntityDescriptionBaseClass(EntityDescription, MatterEntityDescription):
"""For typing a base class that inherits from both entity descriptions."""
class MatterEntity(Entity):
"""Entity class for Matter devices."""
entity_description: MatterEntityDescriptionBaseClass
_attr_should_poll = False
_attr_has_entity_name = True
def __init__(
self,
matter_client: MatterClient,
node_device: AbstractMatterNodeDevice,
device_type_instance: MatterDeviceTypeInstance,
entity_description: MatterEntityDescriptionBaseClass,
endpoint: MatterEndpoint,
entity_info: MatterEntityInfo,
) -> None:
"""Initialize the entity."""
self.matter_client = matter_client
self._node_device = node_device
self._device_type_instance = device_type_instance
self.entity_description = entity_description
self._endpoint = endpoint
self._entity_info = entity_info
self.entity_description = entity_info.entity_description
self._unsubscribes: list[Callable] = []
# for fast lookups we create a mapping to the attribute paths
self._attributes_map: dict[type, str] = {}
# The server info is set when the client connects to the server.
server_info = cast(ServerInfoMessage, self.matter_client.server_info)
# create unique_id based on "Operational Instance Name" and endpoint/device type
node_device_id = get_device_id(server_info, endpoint)
self._attr_unique_id = (
f"{get_operational_instance_id(server_info, self._node_device.node())}-"
f"{device_type_instance.endpoint.endpoint_id}-"
f"{device_type_instance.device_type.device_type}"
f"{node_device_id}-"
f"{endpoint.endpoint_id}-"
f"{entity_info.entity_description.key}-"
f"{entity_info.primary_attribute.cluster_id}-"
f"{entity_info.primary_attribute.attribute_id}"
)
node_device_id = get_device_id(server_info, node_device)
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, f"{ID_TYPE_DEVICE_ID}_{node_device_id}")}
)
self._attr_available = self._node_device.node().available
self._attr_available = self._endpoint.node.available
async def async_added_to_hass(self) -> None:
"""Handle being added to Home Assistant."""
await super().async_added_to_hass()
# Subscribe to attribute updates.
for attr_cls in self.entity_description.subscribe_attributes:
for attr_cls in self._entity_info.attributes_to_watch:
attr_path = self.get_matter_attribute_path(attr_cls)
self._attributes_map[attr_cls] = attr_path
self._unsubscribes.append(
self.matter_client.subscribe(
callback=self._on_matter_event,
event_filter=EventType.ATTRIBUTE_UPDATED,
node_filter=self._device_type_instance.node.node_id,
node_filter=self._endpoint.node.node_id,
attr_path_filter=attr_path,
)
)
@ -95,7 +82,7 @@ class MatterEntity(Entity):
self.matter_client.subscribe(
callback=self._on_matter_event,
event_filter=EventType.NODE_UPDATED,
node_filter=self._device_type_instance.node.node_id,
node_filter=self._endpoint.node.node_id,
)
)
@ -110,7 +97,7 @@ class MatterEntity(Entity):
@callback
def _on_matter_event(self, event: EventType, data: Any = None) -> None:
"""Call on update."""
self._attr_available = self._device_type_instance.node.available
self._attr_available = self._endpoint.node.available
self._update_from_device()
self.async_write_ha_state()
@ -124,14 +111,13 @@ class MatterEntity(Entity):
self, attribute: type[ClusterAttributeDescriptor]
) -> Any:
"""Get current value for given attribute."""
return self._device_type_instance.get_attribute_value(None, attribute)
return self._endpoint.get_attribute_value(None, attribute)
@callback
def get_matter_attribute_path(
self, attribute: type[ClusterAttributeDescriptor]
) -> str:
"""Return AttributePath by providing the endpoint and Attribute class."""
endpoint = self._device_type_instance.endpoint.endpoint_id
return create_attribute_path(
endpoint, attribute.cluster_id, attribute.attribute_id
self._endpoint.endpoint_id, attribute.cluster_id, attribute.attribute_id
)

29
homeassistant/components/matter/helpers.py
View file

@ -11,8 +11,7 @@ from homeassistant.helpers import device_registry as dr
from .const import DOMAIN, ID_TYPE_DEVICE_ID
if TYPE_CHECKING:
from matter_server.client.models.node import MatterNode
from matter_server.client.models.node_device import AbstractMatterNodeDevice
from matter_server.client.models.node import MatterEndpoint, MatterNode
from matter_server.common.models import ServerInfoMessage
from .adapter import MatterAdapter
@ -50,15 +49,21 @@ def get_operational_instance_id(
def get_device_id(
server_info: ServerInfoMessage,
node_device: AbstractMatterNodeDevice,
endpoint: MatterEndpoint,
) -> str:
"""Return HA device_id for the given MatterNodeDevice."""
operational_instance_id = get_operational_instance_id(
server_info, node_device.node()
)
# Append nodedevice(type) to differentiate between a root node
# and bridge within Home Assistant devices.
return f"{operational_instance_id}-{node_device.__class__.__name__}"
"""Return HA device_id for the given MatterEndpoint."""
operational_instance_id = get_operational_instance_id(server_info, endpoint.node)
# Append endpoint ID if this endpoint is a bridged or composed device
if endpoint.is_composed_device:
compose_parent = endpoint.node.get_compose_parent(endpoint.endpoint_id)
assert compose_parent is not None
postfix = str(compose_parent.endpoint_id)
elif endpoint.is_bridged_device:
postfix = str(endpoint.endpoint_id)
else:
# this should be compatible with previous versions
postfix = "MatterNodeDevice"
return f"{operational_instance_id}-{postfix}"
async def get_node_from_device_entry(
@ -91,8 +96,8 @@ async def get_node_from_device_entry(
(
node
for node in await matter_client.get_nodes()
for node_device in node.node_devices
if get_device_id(server_info, node_device) == device_id
for endpoint in node.endpoints.values()
if get_device_id(server_info, endpoint) == device_id
),
None,
)

273
homeassistant/components/matter/light.py
View file

@ -1,9 +1,6 @@
"""Matter light."""
from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
from functools import partial
from typing import Any
from chip.clusters import Objects as clusters
@ -24,8 +21,9 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import LOGGER
from .entity import MatterEntity, MatterEntityDescriptionBaseClass
from .entity import MatterEntity
from .helpers import get_matter
from .models import MatterDiscoverySchema
from .util import (
convert_to_hass_hs,
convert_to_hass_xy,
@ -34,32 +32,13 @@ from .util import (
renormalize,
)
class MatterColorMode(Enum):
"""Matter color mode."""
HS = 0
XY = 1
COLOR_TEMP = 2
COLOR_MODE_MAP = {
MatterColorMode.HS: ColorMode.HS,
MatterColorMode.XY: ColorMode.XY,
MatterColorMode.COLOR_TEMP: ColorMode.COLOR_TEMP,
clusters.ColorControl.Enums.ColorMode.kCurrentHueAndCurrentSaturation: ColorMode.HS,
clusters.ColorControl.Enums.ColorMode.kCurrentXAndCurrentY: ColorMode.XY,
clusters.ColorControl.Enums.ColorMode.kColorTemperature: ColorMode.COLOR_TEMP,
}
class MatterColorControlFeatures(Enum):
"""Matter color control features."""
HS = 0 # Hue and saturation (Optional if device is color capable)
EHUE = 1 # Enhanced hue and saturation (Optional if device is color capable)
COLOR_LOOP = 2 # Color loop (Optional if device is color capable)
XY = 3 # XY (Mandatory if device is color capable)
COLOR_TEMP = 4 # Color temperature (Mandatory if device is color capable)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
@ -73,63 +52,37 @@ async def async_setup_entry(
class MatterLight(MatterEntity, LightEntity):
"""Representation of a Matter light."""
entity_description: MatterLightEntityDescription
def _supports_feature(
self, feature_map: int, feature: MatterColorControlFeatures
) -> bool:
"""Return if device supports given feature."""
return (feature_map & (1 << feature.value)) != 0
def _supports_color_mode(self, color_feature: MatterColorControlFeatures) -> bool:
"""Return if device supports given color mode."""
feature_map = self.get_matter_attribute_value(
clusters.ColorControl.Attributes.FeatureMap,
)
assert isinstance(feature_map, int)
return self._supports_feature(feature_map, color_feature)
def _supports_hs_color(self) -> bool:
"""Return if device supports hs color."""
return self._supports_color_mode(MatterColorControlFeatures.HS)
def _supports_xy_color(self) -> bool:
"""Return if device supports xy color."""
return self._supports_color_mode(MatterColorControlFeatures.XY)
def _supports_color_temperature(self) -> bool:
"""Return if device supports color temperature."""
return self._supports_color_mode(MatterColorControlFeatures.COLOR_TEMP)
def _supports_brightness(self) -> bool:
"""Return if device supports brightness."""
entity_description: LightEntityDescription
@property
def supports_color(self) -> bool:
"""Return if the device supports color control."""
if not self._attr_supported_color_modes:
return False
return (
clusters.LevelControl.Attributes.CurrentLevel
in self.entity_description.subscribe_attributes
ColorMode.HS in self._attr_supported_color_modes
or ColorMode.XY in self._attr_supported_color_modes
)
def _supports_color(self) -> bool:
"""Return if device supports color."""
@property
def supports_color_temperature(self) -> bool:
"""Return if the device supports color temperature control."""
if not self._attr_supported_color_modes:
return False
return ColorMode.COLOR_TEMP in self._attr_supported_color_modes
return (
clusters.ColorControl.Attributes.ColorMode
in self.entity_description.subscribe_attributes
)
@property
def supports_brightness(self) -> bool:
"""Return if the device supports bridghtness control."""
if not self._attr_supported_color_modes:
return False
return ColorMode.BRIGHTNESS in self._attr_supported_color_modes
async def _set_xy_color(self, xy_color: tuple[float, float]) -> None:
"""Set xy color."""
matter_xy = convert_to_matter_xy(xy_color)
LOGGER.debug("Setting xy color to %s", matter_xy)
await self.send_device_command(
clusters.ColorControl.Commands.MoveToColor(
colorX=int(matter_xy[0]),
@ -144,7 +97,6 @@ class MatterLight(MatterEntity, LightEntity):
matter_hs = convert_to_matter_hs(hs_color)
LOGGER.debug("Setting hs color to %s", matter_hs)
await self.send_device_command(
clusters.ColorControl.Commands.MoveToHueAndSaturation(
hue=int(matter_hs[0]),
@ -157,7 +109,6 @@ class MatterLight(MatterEntity, LightEntity):
async def _set_color_temp(self, color_temp: int) -> None:
"""Set color temperature."""
LOGGER.debug("Setting color temperature to %s", color_temp)
await self.send_device_command(
clusters.ColorControl.Commands.MoveToColorTemperature(
colorTemperature=color_temp,
@ -169,8 +120,7 @@ class MatterLight(MatterEntity, LightEntity):
async def _set_brightness(self, brightness: int) -> None:
"""Set brightness."""
LOGGER.debug("Setting brightness to %s", brightness)
level_control = self._device_type_instance.get_cluster(clusters.LevelControl)
level_control = self._endpoint.get_cluster(clusters.LevelControl)
assert level_control is not None
@ -207,7 +157,7 @@ class MatterLight(MatterEntity, LightEntity):
LOGGER.debug(
"Got xy color %s for %s",
xy_color,
self._device_type_instance,
self.entity_id,
)
return xy_color
@ -231,7 +181,7 @@ class MatterLight(MatterEntity, LightEntity):
LOGGER.debug(
"Got hs color %s for %s",
hs_color,
self._device_type_instance,
self.entity_id,
)
return hs_color
@ -248,7 +198,7 @@ class MatterLight(MatterEntity, LightEntity):
LOGGER.debug(
"Got color temperature %s for %s",
color_temp,
self._device_type_instance,
self.entity_id,
)
return int(color_temp)
@ -256,7 +206,7 @@ class MatterLight(MatterEntity, LightEntity):
def _get_brightness(self) -> int:
"""Get brightness from matter."""
level_control = self._device_type_instance.get_cluster(clusters.LevelControl)
level_control = self._endpoint.get_cluster(clusters.LevelControl)
# We should not get here if brightness is not supported.
assert level_control is not None
@ -264,7 +214,7 @@ class MatterLight(MatterEntity, LightEntity):
LOGGER.debug( # type: ignore[unreachable]
"Got brightness %s for %s",
level_control.currentLevel,
self._device_type_instance,
self.entity_id,
)
return round(
@ -284,10 +234,12 @@ class MatterLight(MatterEntity, LightEntity):
assert color_mode is not None
ha_color_mode = COLOR_MODE_MAP[MatterColorMode(color_mode)]
ha_color_mode = COLOR_MODE_MAP[color_mode]
LOGGER.debug(
"Got color mode (%s) for %s", ha_color_mode, self._device_type_instance
"Got color mode (%s) for %s",
ha_color_mode,
self.entity_id,
)
return ha_color_mode
@ -295,8 +247,8 @@ class MatterLight(MatterEntity, LightEntity):
async def send_device_command(self, command: Any) -> None:
"""Send device command."""
await self.matter_client.send_device_command(
node_id=self._device_type_instance.node.node_id,
endpoint_id=self._device_type_instance.endpoint_id,
node_id=self._endpoint.node.node_id,
endpoint_id=self._endpoint.endpoint_id,
command=command,
)
@ -308,15 +260,14 @@ class MatterLight(MatterEntity, LightEntity):
color_temp = kwargs.get(ATTR_COLOR_TEMP)
brightness = kwargs.get(ATTR_BRIGHTNESS)
if self._supports_color():
if hs_color is not None and self._supports_hs_color():
await self._set_hs_color(hs_color)
elif xy_color is not None and self._supports_xy_color():
await self._set_xy_color(xy_color)
elif color_temp is not None and self._supports_color_temperature():
await self._set_color_temp(color_temp)
if hs_color is not None and self.supports_color:
await self._set_hs_color(hs_color)
elif xy_color is not None:
await self._set_xy_color(xy_color)
elif color_temp is not None and self.supports_color_temperature:
await self._set_color_temp(color_temp)
if brightness is not None and self._supports_brightness():
if brightness is not None and self.supports_brightness:
await self._set_brightness(brightness)
return
@ -334,106 +285,80 @@ class MatterLight(MatterEntity, LightEntity):
def _update_from_device(self) -> None:
"""Update from device."""
supports_color = self._supports_color()
supports_color_temperature = (
self._supports_color_temperature() if supports_color else False
)
supports_brightness = self._supports_brightness()
if self._attr_supported_color_modes is None:
supported_color_modes = set()
if supports_color:
supported_color_modes.add(ColorMode.XY)
if self._supports_hs_color():
supported_color_modes.add(ColorMode.HS)
if supports_color_temperature:
supported_color_modes.add(ColorMode.COLOR_TEMP)
if supports_brightness:
# work out what (color)features are supported
supported_color_modes: set[ColorMode] = set()
# brightness support
if self._entity_info.endpoint.has_attribute(
None, clusters.LevelControl.Attributes.CurrentLevel
):
supported_color_modes.add(ColorMode.BRIGHTNESS)
# colormode(s)
if self._entity_info.endpoint.has_attribute(
None, clusters.ColorControl.Attributes.ColorMode
):
# device has some color support, check which color modes
# are supported with the featuremap on the ColorControl cluster
color_feature_map = self.get_matter_attribute_value(
clusters.ColorControl.Attributes.FeatureMap,
)
if (
color_feature_map
& clusters.ColorControl.Attributes.CurrentHue.attribute_id
):
supported_color_modes.add(ColorMode.HS)
if (
color_feature_map
& clusters.ColorControl.Attributes.CurrentX.attribute_id
):
supported_color_modes.add(ColorMode.XY)
self._attr_supported_color_modes = (
supported_color_modes if supported_color_modes else None
# color temperature support detection using the featuremap is not reliable
# (temporary?) fallback to checking the value
if (
self.get_matter_attribute_value(
clusters.ColorControl.Attributes.ColorTemperatureMireds
)
is not None
):
supported_color_modes.add(ColorMode.COLOR_TEMP)
self._attr_supported_color_modes = supported_color_modes
LOGGER.debug(
"Supported color modes: %s for %s",
self._attr_supported_color_modes,
self.entity_id,
)
LOGGER.debug(
"Supported color modes: %s for %s",
self._attr_supported_color_modes,
self._device_type_instance,
)
# set current values
if supports_color:
if self.supports_color:
self._attr_color_mode = self._get_color_mode()
if self._attr_color_mode == ColorMode.HS:
self._attr_hs_color = self._get_hs_color()
else:
self._attr_xy_color = self._get_xy_color()
if supports_color_temperature:
if self.supports_color_temperature:
self._attr_color_temp = self._get_color_temperature()
self._attr_is_on = self.get_matter_attribute_value(
clusters.OnOff.Attributes.OnOff
)
if supports_brightness:
if self.supports_brightness:
self._attr_brightness = self._get_brightness()
@dataclass
class MatterLightEntityDescription(
LightEntityDescription,
MatterEntityDescriptionBaseClass,
):
"""Matter light entity description."""
# You can't set default values on inherited data classes
MatterLightEntityDescriptionFactory = partial(
MatterLightEntityDescription, entity_cls=MatterLight
)
# Mapping of a Matter Device type to Light Entity Description.
# A Matter device type (instance) can consist of multiple attributes.
# For example a Color Light which has an attribute to control brightness
# but also for color.
DEVICE_ENTITY: dict[
type[device_types.DeviceType],
MatterEntityDescriptionBaseClass | list[MatterEntityDescriptionBaseClass],
] = {
device_types.OnOffLight: MatterLightEntityDescriptionFactory(
key=device_types.OnOffLight,
subscribe_attributes=(clusters.OnOff.Attributes.OnOff,),
),
device_types.DimmableLight: MatterLightEntityDescriptionFactory(
key=device_types.DimmableLight,
subscribe_attributes=(
clusters.OnOff.Attributes.OnOff,
clusters.LevelControl.Attributes.CurrentLevel,
),
),
device_types.DimmablePlugInUnit: MatterLightEntityDescriptionFactory(
key=device_types.DimmablePlugInUnit,
subscribe_attributes=(
clusters.OnOff.Attributes.OnOff,
clusters.LevelControl.Attributes.CurrentLevel,
),
),
device_types.ColorTemperatureLight: MatterLightEntityDescriptionFactory(
key=device_types.ColorTemperatureLight,
subscribe_attributes=(
clusters.OnOff.Attributes.OnOff,
clusters.LevelControl.Attributes.CurrentLevel,
clusters.ColorControl.Attributes.ColorMode,
clusters.ColorControl.Attributes.ColorTemperatureMireds,
),
),
device_types.ExtendedColorLight: MatterLightEntityDescriptionFactory(
key=device_types.ExtendedColorLight,
subscribe_attributes=(
clusters.OnOff.Attributes.OnOff,
# Discovery schema(s) to map Matter Attributes to HA entities
DISCOVERY_SCHEMAS = [
MatterDiscoverySchema(
platform=Platform.LIGHT,
entity_description=LightEntityDescription(key="ExtendedMatterLight"),
entity_class=MatterLight,
required_attributes=(clusters.OnOff.Attributes.OnOff,),
optional_attributes=(
clusters.LevelControl.Attributes.CurrentLevel,
clusters.ColorControl.Attributes.ColorMode,
clusters.ColorControl.Attributes.CurrentHue,
@ -442,5 +367,7 @@ DEVICE_ENTITY: dict[
clusters.ColorControl.Attributes.CurrentY,
clusters.ColorControl.Attributes.ColorTemperatureMireds,
),
# restrict device type to prevent discovery in switch platform
not_device_type=(device_types.OnOffPlugInUnit,),
),
}
]

2
homeassistant/components/matter/manifest.json
View file

@ -6,5 +6,5 @@
"dependencies": ["websocket_api"],
"documentation": "https://www.home-assistant.io/integrations/matter",
"iot_class": "local_push",
"requirements": ["python-matter-server==3.0.0"]
"requirements": ["python-matter-server==3.1.0"]
}

109
homeassistant/components/matter/models.py Normal file
View file

@ -0,0 +1,109 @@
"""Models used for the Matter integration."""
from collections.abc import Callable
from dataclasses import asdict, dataclass
from typing import Any
from chip.clusters import Objects as clusters
from chip.clusters.Objects import ClusterAttributeDescriptor
from matter_server.client.models.device_types import DeviceType
from matter_server.client.models.node import MatterEndpoint
from homeassistant.const import Platform
from homeassistant.helpers.entity import EntityDescription
class DataclassMustHaveAtLeastOne:
"""A dataclass that must have at least one input parameter that is not None."""
def __post_init__(self) -> None:
"""Post dataclass initialization."""
if all(val is None for val in asdict(self).values()):
raise ValueError("At least one input parameter must not be None")
SensorValueTypes = type[
clusters.uint | int | clusters.Nullable | clusters.float32 | float
]
@dataclass
class MatterEntityInfo:
"""Info discovered from (primary) Matter Attribute to create entity."""
# MatterEndpoint to which the value(s) belongs
endpoint: MatterEndpoint
# the home assistant platform for which an entity should be created
platform: Platform
# All attributes that need to be watched by entity (incl. primary)
attributes_to_watch: list[type[ClusterAttributeDescriptor]]
# the entity description to use
entity_description: EntityDescription
# entity class to use to instantiate the entity
entity_class: type
# [optional] function to call to convert the value from the primary attribute
measurement_to_ha: Callable[[SensorValueTypes], SensorValueTypes] | None = None
@property
def primary_attribute(self) -> type[ClusterAttributeDescriptor]:
"""Return Primary Attribute belonging to the entity."""
return self.attributes_to_watch[0]
@dataclass
class MatterDiscoverySchema:
"""Matter discovery schema.
The Matter endpoint and it's (primary) Attribute for an entity must match these conditions.
"""
# specify the hass platform for which this scheme applies (e.g. light, sensor)
platform: Platform
# platform-specific entity description
entity_description: EntityDescription
# entity class to use to instantiate the entity
entity_class: type
# DISCOVERY OPTIONS
# [required] attributes that ALL need to be present
# on the node for this scheme to pass (minimal one == primary)
required_attributes: tuple[type[ClusterAttributeDescriptor], ...]
# [optional] the value's endpoint must contain this devicetype(s)
device_type: tuple[type[DeviceType] | DeviceType, ...] | None = None
# [optional] the value's endpoint must NOT contain this devicetype(s)
not_device_type: tuple[type[DeviceType] | DeviceType, ...] | None = None
# [optional] the endpoint's vendor_id must match ANY of these values
vendor_id: tuple[int, ...] | None = None
# [optional] the endpoint's product_name must match ANY of these values
product_name: tuple[str, ...] | None = None
# [optional] the attribute's endpoint_id must match ANY of these values
endpoint_id: tuple[int, ...] | None = None
# [optional] additional attributes that MAY NOT be present
# on the node for this scheme to pass
absent_attributes: tuple[type[ClusterAttributeDescriptor], ...] | None = None
# [optional] additional attributes that may be present
# these attributes are copied over to attributes_to_watch and
# are not discovered by other entities
optional_attributes: tuple[type[ClusterAttributeDescriptor], ...] | None = None
# [optional] bool to specify if this primary value may be discovered
# by multiple platforms
allow_multi: bool = False
# [optional] function to call to convert the value from the primary attribute
measurement_to_ha: Callable[[Any], Any] | None = None

164
homeassistant/components/matter/sensor.py
View file

@ -1,13 +1,8 @@
"""Matter sensors."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from functools import partial
from chip.clusters import Objects as clusters
from chip.clusters.Types import Nullable, NullValue
from matter_server.client.models import device_types
from homeassistant.components.sensor import (
SensorDeviceClass,
@ -27,8 +22,9 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .entity import MatterEntity, MatterEntityDescriptionBaseClass
from .entity import MatterEntity
from .helpers import get_matter
from .models import MatterDiscoverySchema
async def async_setup_entry(
@ -45,94 +41,94 @@ class MatterSensor(MatterEntity, SensorEntity):
"""Representation of a Matter sensor."""
_attr_state_class = SensorStateClass.MEASUREMENT
entity_description: MatterSensorEntityDescription
@callback
def _update_from_device(self) -> None:
"""Update from device."""
measurement: Nullable | float | None
measurement = self.get_matter_attribute_value(
# We always subscribe to a single value
self.entity_description.subscribe_attributes[0],
)
if measurement == NullValue or measurement is None:
measurement = None
else:
measurement = self.entity_description.measurement_to_ha(measurement)
self._attr_native_value = measurement
value: Nullable | float | None
value = self.get_matter_attribute_value(self._entity_info.primary_attribute)
if value in (None, NullValue):
value = None
elif value_convert := self._entity_info.measurement_to_ha:
value = value_convert(value)
self._attr_native_value = value
@dataclass
class MatterSensorEntityDescriptionMixin:
"""Required fields for sensor device mapping."""
measurement_to_ha: Callable[[float], float]
@dataclass
class MatterSensorEntityDescription(
SensorEntityDescription,
MatterEntityDescriptionBaseClass,
MatterSensorEntityDescriptionMixin,
):
"""Matter Sensor entity description."""
# You can't set default values on inherited data classes
MatterSensorEntityDescriptionFactory = partial(
MatterSensorEntityDescription, entity_cls=MatterSensor
)
DEVICE_ENTITY: dict[
type[device_types.DeviceType],
MatterEntityDescriptionBaseClass | list[MatterEntityDescriptionBaseClass],
] = {
device_types.TemperatureSensor: MatterSensorEntityDescriptionFactory(
key=device_types.TemperatureSensor,
name="Temperature",
measurement_to_ha=lambda x: x / 100,
subscribe_attributes=(
clusters.TemperatureMeasurement.Attributes.MeasuredValue,
# Discovery schema(s) to map Matter Attributes to HA entities
DISCOVERY_SCHEMAS = [
MatterDiscoverySchema(
platform=Platform.SENSOR,
entity_description=SensorEntityDescription(
key="TemperatureSensor",
name="Temperature",
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
),
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
),
device_types.PressureSensor: MatterSensorEntityDescriptionFactory(
key=device_types.PressureSensor,
name="Pressure",
measurement_to_ha=lambda x: x / 10,
subscribe_attributes=(clusters.PressureMeasurement.Attributes.MeasuredValue,),
native_unit_of_measurement=UnitOfPressure.KPA,
device_class=SensorDeviceClass.PRESSURE,
),
device_types.FlowSensor: MatterSensorEntityDescriptionFactory(
key=device_types.FlowSensor,
name="Flow",
measurement_to_ha=lambda x: x / 10,
subscribe_attributes=(clusters.FlowMeasurement.Attributes.MeasuredValue,),
native_unit_of_measurement=UnitOfVolumeFlowRate.CUBIC_METERS_PER_HOUR,
),
device_types.HumiditySensor: MatterSensorEntityDescriptionFactory(
key=device_types.HumiditySensor,
name="Humidity",
entity_class=MatterSensor,
required_attributes=(clusters.TemperatureMeasurement.Attributes.MeasuredValue,),
measurement_to_ha=lambda x: x / 100,
subscribe_attributes=(
),
MatterDiscoverySchema(
platform=Platform.SENSOR,
entity_description=SensorEntityDescription(
key="PressureSensor",
name="Pressure",
native_unit_of_measurement=UnitOfPressure.KPA,
device_class=SensorDeviceClass.PRESSURE,
),
entity_class=MatterSensor,
required_attributes=(clusters.PressureMeasurement.Attributes.MeasuredValue,),
measurement_to_ha=lambda x: x / 10,
),
MatterDiscoverySchema(
platform=Platform.SENSOR,
entity_description=SensorEntityDescription(
key="FlowSensor",
name="Flow",
native_unit_of_measurement=UnitOfVolumeFlowRate.CUBIC_METERS_PER_HOUR,
device_class=SensorDeviceClass.WATER, # what is the device class here ?
),
entity_class=MatterSensor,
required_attributes=(clusters.FlowMeasurement.Attributes.MeasuredValue,),
measurement_to_ha=lambda x: x / 10,
),
MatterDiscoverySchema(
platform=Platform.SENSOR,
entity_description=SensorEntityDescription(
key="HumiditySensor",
name="Humidity",
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.HUMIDITY,
),
entity_class=MatterSensor,
required_attributes=(
clusters.RelativeHumidityMeasurement.Attributes.MeasuredValue,
),
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.HUMIDITY,
measurement_to_ha=lambda x: x / 100,
),
device_types.LightSensor: MatterSensorEntityDescriptionFactory(
key=device_types.LightSensor,
name="Light",
measurement_to_ha=lambda x: round(pow(10, ((x - 1) / 10000)), 1),
subscribe_attributes=(
clusters.IlluminanceMeasurement.Attributes.MeasuredValue,
MatterDiscoverySchema(
platform=Platform.SENSOR,
entity_description=SensorEntityDescription(
key="LightSensor",
name="Illuminance",
native_unit_of_measurement=LIGHT_LUX,
device_class=SensorDeviceClass.ILLUMINANCE,
),
native_unit_of_measurement=LIGHT_LUX,
device_class=SensorDeviceClass.ILLUMINANCE,
entity_class=MatterSensor,
required_attributes=(clusters.IlluminanceMeasurement.Attributes.MeasuredValue,),
measurement_to_ha=lambda x: round(pow(10, ((x - 1) / 10000)), 1),
),
}
MatterDiscoverySchema(
platform=Platform.SENSOR,
entity_description=SensorEntityDescription(
key="PowerSource",
name="Battery",
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.BATTERY,
),
entity_class=MatterSensor,
required_attributes=(clusters.PowerSource.Attributes.BatPercentRemaining,),
# value has double precision
measurement_to_ha=lambda x: int(x / 2),
),
]

53
homeassistant/components/matter/switch.py
View file

@ -1,8 +1,6 @@
"""Matter switches."""
from __future__ import annotations
from dataclasses import dataclass
from functools import partial
from typing import Any
from chip.clusters import Objects as clusters
@ -18,8 +16,9 @@ from homeassistant.const import Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .entity import MatterEntity, MatterEntityDescriptionBaseClass
from .entity import MatterEntity
from .helpers import get_matter
from .models import MatterDiscoverySchema
async def async_setup_entry(
@ -35,21 +34,19 @@ async def async_setup_entry(
class MatterSwitch(MatterEntity, SwitchEntity):
"""Representation of a Matter switch."""
entity_description: MatterSwitchEntityDescription
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn switch on."""
await self.matter_client.send_device_command(
node_id=self._device_type_instance.node.node_id,
endpoint_id=self._device_type_instance.endpoint_id,
node_id=self._endpoint.node.node_id,
endpoint_id=self._endpoint.endpoint_id,
command=clusters.OnOff.Commands.On(),
)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn switch off."""
await self.matter_client.send_device_command(
node_id=self._device_type_instance.node.node_id,
endpoint_id=self._device_type_instance.endpoint_id,
node_id=self._endpoint.node.node_id,
endpoint_id=self._endpoint.endpoint_id,
command=clusters.OnOff.Commands.Off(),
)
@ -57,31 +54,21 @@ class MatterSwitch(MatterEntity, SwitchEntity):
def _update_from_device(self) -> None:
"""Update from device."""
self._attr_is_on = self.get_matter_attribute_value(
clusters.OnOff.Attributes.OnOff
self._entity_info.primary_attribute
)
@dataclass
class MatterSwitchEntityDescription(
SwitchEntityDescription,
MatterEntityDescriptionBaseClass,
):
"""Matter Switch entity description."""
# You can't set default values on inherited data classes
MatterSwitchEntityDescriptionFactory = partial(
MatterSwitchEntityDescription, entity_cls=MatterSwitch
)
DEVICE_ENTITY: dict[
type[device_types.DeviceType],
MatterEntityDescriptionBaseClass | list[MatterEntityDescriptionBaseClass],
] = {
device_types.OnOffPlugInUnit: MatterSwitchEntityDescriptionFactory(
key=device_types.OnOffPlugInUnit,
subscribe_attributes=(clusters.OnOff.Attributes.OnOff,),
device_class=SwitchDeviceClass.OUTLET,
# Discovery schema(s) to map Matter Attributes to HA entities
DISCOVERY_SCHEMAS = [
MatterDiscoverySchema(
platform=Platform.SWITCH,
entity_description=SwitchEntityDescription(
key="MatterPlug", device_class=SwitchDeviceClass.OUTLET
),
entity_class=MatterSwitch,
required_attributes=(clusters.OnOff.Attributes.OnOff,),
# restrict device type to prevent discovery by light
# platform which also uses OnOff cluster
not_device_type=(device_types.OnOffLight, device_types.DimmableLight),
),
}
]

2
requirements_all.txt
View file

@ -2081,7 +2081,7 @@ python-kasa==0.5.1
# python-lirc==1.2.3
# homeassistant.components.matter
python-matter-server==3.0.0
python-matter-server==3.1.0
# homeassistant.components.xiaomi_miio
python-miio==0.5.12

2
requirements_test_all.txt
View file

@ -1480,7 +1480,7 @@ python-juicenet==1.1.0
python-kasa==0.5.1
# homeassistant.components.matter
python-matter-server==3.0.0
python-matter-server==3.1.0
# homeassistant.components.xiaomi_miio
python-miio==0.5.12

4
tests/components/matter/test_binary_sensor.py
View file

@ -31,7 +31,7 @@ async def test_contact_sensor(
"""Test contact sensor."""
state = hass.states.get("binary_sensor.mock_contact_sensor_contact")
assert state
assert state.state == "on"
assert state.state == "off"
set_node_attribute(contact_sensor_node, 1, 69, 0, False)
await trigger_subscription_callback(
@ -40,7 +40,7 @@ async def test_contact_sensor(
state = hass.states.get("binary_sensor.mock_contact_sensor_contact")
assert state
assert state.state == "off"
assert state.state == "on"
@pytest.fixture(name="occupancy_sensor_node")

2
tests/components/matter/test_helpers.py
View file

@ -26,7 +26,7 @@ async def test_get_device_id(
node = await setup_integration_with_node_fixture(
hass, "device_diagnostics", matter_client
)
device_id = get_device_id(matter_client.server_info, node.node_devices[0])
device_id = get_device_id(matter_client.server_info, node.endpoints[0])
assert device_id == "00000000000004D2-0000000000000005-MatterNodeDevice"

10
tests/components/matter/test_light.py
View file

@ -297,10 +297,14 @@ async def test_extended_color_light(
matter_client.send_device_command.assert_has_calls(
[
call(
node_id=light_node.node_id,
node_id=1,
endpoint_id=1,
command=clusters.ColorControl.Commands.MoveToHueAndSaturation(
hue=0, saturation=0, transitionTime=0
command=clusters.ColorControl.Commands.MoveToColor(
colorX=21168,
colorY=21561,
transitionTime=0,
optionsMask=0,
optionsOverride=0,
),
),
call(

4
tests/components/matter/test_sensor.py
View file

@ -121,14 +121,14 @@ async def test_light_sensor(
light_sensor_node: MatterNode,
) -> None:
"""Test light sensor."""
state = hass.states.get("sensor.mock_light_sensor_light")
state = hass.states.get("sensor.mock_light_sensor_illuminance")
assert state
assert state.state == "1.3"
set_node_attribute(light_sensor_node, 1, 1024, 0, 3000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("sensor.mock_light_sensor_light")
state = hass.states.get("sensor.mock_light_sensor_illuminance")
assert state
assert state.state == "2.0"