diff --git a/homeassistant/components/zha/binary_sensor.py b/homeassistant/components/zha/binary_sensor.py index 5ec829fcb05..aed0a16a681 100644 --- a/homeassistant/components/zha/binary_sensor.py +++ b/homeassistant/components/zha/binary_sensor.py @@ -4,6 +4,7 @@ from __future__ import annotations import functools from typing import Any +from zigpy.quirks.v2 import BinarySensorMetadata, EntityMetadata import zigpy.types as t from zigpy.zcl.clusters.general import OnOff from zigpy.zcl.clusters.security import IasZone @@ -26,6 +27,7 @@ from .core.const import ( CLUSTER_HANDLER_OCCUPANCY, CLUSTER_HANDLER_ON_OFF, CLUSTER_HANDLER_ZONE, + QUIRK_METADATA, SIGNAL_ADD_ENTITIES, SIGNAL_ATTR_UPDATED, ) @@ -76,8 +78,16 @@ class BinarySensor(ZhaEntity, BinarySensorEntity): def __init__(self, unique_id, zha_device, cluster_handlers, **kwargs) -> None: """Initialize the ZHA binary sensor.""" - super().__init__(unique_id, zha_device, cluster_handlers, **kwargs) self._cluster_handler = cluster_handlers[0] + if QUIRK_METADATA in kwargs: + self._init_from_quirks_metadata(kwargs[QUIRK_METADATA]) + super().__init__(unique_id, zha_device, cluster_handlers, **kwargs) + + def _init_from_quirks_metadata(self, entity_metadata: EntityMetadata) -> None: + """Init this entity from the quirks metadata.""" + super()._init_from_quirks_metadata(entity_metadata) + binary_sensor_metadata: BinarySensorMetadata = entity_metadata.entity_metadata + self._attribute_name = binary_sensor_metadata.attribute_name async def async_added_to_hass(self) -> None: """Run when about to be added to hass.""" diff --git a/homeassistant/components/zha/button.py b/homeassistant/components/zha/button.py index e16ae082eda..2c0028cd3d1 100644 --- a/homeassistant/components/zha/button.py +++ b/homeassistant/components/zha/button.py @@ -1,11 +1,16 @@ """Support for ZHA button.""" from __future__ import annotations -import abc import functools import logging from typing import TYPE_CHECKING, Any, Self +from zigpy.quirks.v2 import ( + EntityMetadata, + WriteAttributeButtonMetadata, + ZCLCommandButtonMetadata, +) + from homeassistant.components.button import ButtonDeviceClass, ButtonEntity from homeassistant.config_entries import ConfigEntry from homeassistant.const import EntityCategory, Platform @@ -14,7 +19,7 @@ from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity_platform import AddEntitiesCallback from .core import discovery -from .core.const import CLUSTER_HANDLER_IDENTIFY, SIGNAL_ADD_ENTITIES +from .core.const import CLUSTER_HANDLER_IDENTIFY, QUIRK_METADATA, SIGNAL_ADD_ENTITIES from .core.helpers import get_zha_data from .core.registries import ZHA_ENTITIES from .entity import ZhaEntity @@ -58,6 +63,8 @@ class ZHAButton(ZhaEntity, ButtonEntity): """Defines a ZHA button.""" _command_name: str + _args: list[Any] + _kwargs: dict[str, Any] def __init__( self, @@ -67,18 +74,33 @@ class ZHAButton(ZhaEntity, ButtonEntity): **kwargs: Any, ) -> None: """Init this button.""" - super().__init__(unique_id, zha_device, cluster_handlers, **kwargs) self._cluster_handler: ClusterHandler = cluster_handlers[0] + if QUIRK_METADATA in kwargs: + self._init_from_quirks_metadata(kwargs[QUIRK_METADATA]) + super().__init__(unique_id, zha_device, cluster_handlers, **kwargs) + + def _init_from_quirks_metadata(self, entity_metadata: EntityMetadata) -> None: + """Init this entity from the quirks metadata.""" + super()._init_from_quirks_metadata(entity_metadata) + button_metadata: ZCLCommandButtonMetadata = entity_metadata.entity_metadata + self._command_name = button_metadata.command_name + self._args = button_metadata.args + self._kwargs = button_metadata.kwargs - @abc.abstractmethod def get_args(self) -> list[Any]: """Return the arguments to use in the command.""" + return list(self._args) if self._args else [] + + def get_kwargs(self) -> dict[str, Any]: + """Return the keyword arguments to use in the command.""" + return self._kwargs async def async_press(self) -> None: """Send out a update command.""" command = getattr(self._cluster_handler, self._command_name) - arguments = self.get_args() - await command(*arguments) + arguments = self.get_args() or [] + kwargs = self.get_kwargs() or {} + await command(*arguments, **kwargs) @MULTI_MATCH(cluster_handler_names=CLUSTER_HANDLER_IDENTIFY) @@ -106,11 +128,8 @@ class ZHAIdentifyButton(ZHAButton): _attr_device_class = ButtonDeviceClass.IDENTIFY _attr_entity_category = EntityCategory.DIAGNOSTIC _command_name = "identify" - - def get_args(self) -> list[Any]: - """Return the arguments to use in the command.""" - - return [DEFAULT_DURATION] + _kwargs = {} + _args = [DEFAULT_DURATION] class ZHAAttributeButton(ZhaEntity, ButtonEntity): @@ -127,8 +146,17 @@ class ZHAAttributeButton(ZhaEntity, ButtonEntity): **kwargs: Any, ) -> None: """Init this button.""" - super().__init__(unique_id, zha_device, cluster_handlers, **kwargs) self._cluster_handler: ClusterHandler = cluster_handlers[0] + if QUIRK_METADATA in kwargs: + self._init_from_quirks_metadata(kwargs[QUIRK_METADATA]) + super().__init__(unique_id, zha_device, cluster_handlers, **kwargs) + + def _init_from_quirks_metadata(self, entity_metadata: EntityMetadata) -> None: + """Init this entity from the quirks metadata.""" + super()._init_from_quirks_metadata(entity_metadata) + button_metadata: WriteAttributeButtonMetadata = entity_metadata.entity_metadata + self._attribute_name = button_metadata.attribute_name + self._attribute_value = button_metadata.attribute_value async def async_press(self) -> None: """Write attribute with defined value.""" diff --git a/homeassistant/components/zha/core/const.py b/homeassistant/components/zha/core/const.py index cb0aa466046..fd54351739e 100644 --- a/homeassistant/components/zha/core/const.py +++ b/homeassistant/components/zha/core/const.py @@ -64,6 +64,8 @@ ATTR_WARNING_DEVICE_STROBE_INTENSITY = "intensity" BAUD_RATES = [2400, 4800, 9600, 14400, 19200, 38400, 57600, 115200, 128000, 256000] BINDINGS = "bindings" +CLUSTER_DETAILS = "cluster_details" + CLUSTER_HANDLER_ACCELEROMETER = "accelerometer" CLUSTER_HANDLER_BINARY_INPUT = "binary_input" CLUSTER_HANDLER_ANALOG_INPUT = "analog_input" @@ -230,6 +232,10 @@ PRESET_SCHEDULE = "Schedule" PRESET_COMPLEX = "Complex" PRESET_TEMP_MANUAL = "Temporary manual" +QUIRK_METADATA = "quirk_metadata" + +ZCL_INIT_ATTRS = "ZCL_INIT_ATTRS" + ZHA_ALARM_OPTIONS = "zha_alarm_options" ZHA_OPTIONS = "zha_options" diff --git a/homeassistant/components/zha/core/device.py b/homeassistant/components/zha/core/device.py index 0d473bf0810..f1b7ec60728 100644 --- a/homeassistant/components/zha/core/device.py +++ b/homeassistant/components/zha/core/device.py @@ -15,6 +15,7 @@ from zigpy.device import Device as ZigpyDevice import zigpy.exceptions from zigpy.profiles import PROFILES import zigpy.quirks +from zigpy.quirks.v2 import CustomDeviceV2 from zigpy.types.named import EUI64, NWK from zigpy.zcl.clusters import Cluster from zigpy.zcl.clusters.general import Groups, Identify @@ -582,6 +583,9 @@ class ZHADevice(LogMixin): await asyncio.gather( *(endpoint.async_configure() for endpoint in self._endpoints.values()) ) + if isinstance(self._zigpy_device, CustomDeviceV2): + self.debug("applying quirks v2 custom device configuration") + await self._zigpy_device.apply_custom_configuration() async_dispatcher_send( self.hass, const.ZHA_CLUSTER_HANDLER_MSG, diff --git a/homeassistant/components/zha/core/discovery.py b/homeassistant/components/zha/core/discovery.py index 5575d633593..221c601827e 100644 --- a/homeassistant/components/zha/core/discovery.py +++ b/homeassistant/components/zha/core/discovery.py @@ -4,10 +4,22 @@ from __future__ import annotations from collections import Counter from collections.abc import Callable import logging -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING, Any, cast from slugify import slugify +from zigpy.quirks.v2 import ( + BinarySensorMetadata, + CustomDeviceV2, + EntityType, + NumberMetadata, + SwitchMetadata, + WriteAttributeButtonMetadata, + ZCLCommandButtonMetadata, + ZCLEnumMetadata, + ZCLSensorMetadata, +) from zigpy.state import State +from zigpy.zcl import ClusterType from zigpy.zcl.clusters.general import Ota from homeassistant.const import CONF_TYPE, Platform @@ -66,6 +78,59 @@ if TYPE_CHECKING: _LOGGER = logging.getLogger(__name__) +QUIRKS_ENTITY_META_TO_ENTITY_CLASS = { + ( + Platform.BUTTON, + WriteAttributeButtonMetadata, + EntityType.CONFIG, + ): button.ZHAAttributeButton, + (Platform.BUTTON, ZCLCommandButtonMetadata, EntityType.CONFIG): button.ZHAButton, + ( + Platform.BUTTON, + ZCLCommandButtonMetadata, + EntityType.DIAGNOSTIC, + ): button.ZHAButton, + ( + Platform.BINARY_SENSOR, + BinarySensorMetadata, + EntityType.CONFIG, + ): binary_sensor.BinarySensor, + ( + Platform.BINARY_SENSOR, + BinarySensorMetadata, + EntityType.DIAGNOSTIC, + ): binary_sensor.BinarySensor, + ( + Platform.BINARY_SENSOR, + BinarySensorMetadata, + EntityType.STANDARD, + ): binary_sensor.BinarySensor, + (Platform.SENSOR, ZCLEnumMetadata, EntityType.DIAGNOSTIC): sensor.EnumSensor, + (Platform.SENSOR, ZCLEnumMetadata, EntityType.STANDARD): sensor.EnumSensor, + (Platform.SENSOR, ZCLSensorMetadata, EntityType.DIAGNOSTIC): sensor.Sensor, + (Platform.SENSOR, ZCLSensorMetadata, EntityType.STANDARD): sensor.Sensor, + (Platform.SELECT, ZCLEnumMetadata, EntityType.CONFIG): select.ZCLEnumSelectEntity, + ( + Platform.SELECT, + ZCLEnumMetadata, + EntityType.DIAGNOSTIC, + ): select.ZCLEnumSelectEntity, + ( + Platform.NUMBER, + NumberMetadata, + EntityType.CONFIG, + ): number.ZHANumberConfigurationEntity, + (Platform.NUMBER, NumberMetadata, EntityType.DIAGNOSTIC): number.ZhaNumber, + (Platform.NUMBER, NumberMetadata, EntityType.STANDARD): number.ZhaNumber, + ( + Platform.SWITCH, + SwitchMetadata, + EntityType.CONFIG, + ): switch.ZHASwitchConfigurationEntity, + (Platform.SWITCH, SwitchMetadata, EntityType.STANDARD): switch.Switch, +} + + @callback async def async_add_entities( _async_add_entities: AddEntitiesCallback, @@ -73,6 +138,7 @@ async def async_add_entities( tuple[ type[ZhaEntity], tuple[str, ZHADevice, list[ClusterHandler]], + dict[str, Any], ] ], **kwargs, @@ -80,7 +146,11 @@ async def async_add_entities( """Add entities helper.""" if not entities: return - to_add = [ent_cls.create_entity(*args, **kwargs) for ent_cls, args in entities] + + to_add = [ + ent_cls.create_entity(*args, **{**kwargs, **kw_args}) + for ent_cls, args, kw_args in entities + ] entities_to_add = [entity for entity in to_add if entity is not None] _async_add_entities(entities_to_add, update_before_add=False) entities.clear() @@ -118,6 +188,129 @@ class ProbeEndpoint: if device.is_coordinator: self.discover_coordinator_device_entities(device) + return + + self.discover_quirks_v2_entities(device) + zha_regs.ZHA_ENTITIES.clean_up() + + @callback + def discover_quirks_v2_entities(self, device: ZHADevice) -> None: + """Discover entities for a ZHA device exposed by quirks v2.""" + _LOGGER.debug( + "Attempting to discover quirks v2 entities for device: %s-%s", + str(device.ieee), + device.name, + ) + + if not isinstance(device.device, CustomDeviceV2): + _LOGGER.debug( + "Device: %s-%s is not a quirks v2 device - skipping " + "discover_quirks_v2_entities", + str(device.ieee), + device.name, + ) + return + + zigpy_device: CustomDeviceV2 = device.device + + if not zigpy_device.exposes_metadata: + _LOGGER.debug( + "Device: %s-%s does not expose any quirks v2 entities", + str(device.ieee), + device.name, + ) + return + + for ( + cluster_details, + quirk_metadata_list, + ) in zigpy_device.exposes_metadata.items(): + endpoint_id, cluster_id, cluster_type = cluster_details + + if endpoint_id not in device.endpoints: + _LOGGER.warning( + "Device: %s-%s does not have an endpoint with id: %s - unable to " + "create entity with cluster details: %s", + str(device.ieee), + device.name, + endpoint_id, + cluster_details, + ) + continue + + endpoint: Endpoint = device.endpoints[endpoint_id] + cluster = ( + endpoint.zigpy_endpoint.in_clusters.get(cluster_id) + if cluster_type is ClusterType.Server + else endpoint.zigpy_endpoint.out_clusters.get(cluster_id) + ) + + if cluster is None: + _LOGGER.warning( + "Device: %s-%s does not have a cluster with id: %s - " + "unable to create entity with cluster details: %s", + str(device.ieee), + device.name, + cluster_id, + cluster_details, + ) + continue + + cluster_handler_id = f"{endpoint.id}:0x{cluster.cluster_id:04x}" + cluster_handler = ( + endpoint.all_cluster_handlers.get(cluster_handler_id) + if cluster_type is ClusterType.Server + else endpoint.client_cluster_handlers.get(cluster_handler_id) + ) + assert cluster_handler + + for quirk_metadata in quirk_metadata_list: + platform = Platform(quirk_metadata.entity_platform.value) + metadata_type = type(quirk_metadata.entity_metadata) + entity_class = QUIRKS_ENTITY_META_TO_ENTITY_CLASS.get( + (platform, metadata_type, quirk_metadata.entity_type) + ) + + if entity_class is None: + _LOGGER.warning( + "Device: %s-%s has an entity with details: %s that does not" + " have an entity class mapping - unable to create entity", + str(device.ieee), + device.name, + { + zha_const.CLUSTER_DETAILS: cluster_details, + zha_const.QUIRK_METADATA: quirk_metadata, + }, + ) + continue + + # automatically add the attribute to ZCL_INIT_ATTRS for the cluster + # handler if it is not already in the list + if ( + hasattr(quirk_metadata.entity_metadata, "attribute_name") + and quirk_metadata.entity_metadata.attribute_name + not in cluster_handler.ZCL_INIT_ATTRS + ): + init_attrs = cluster_handler.ZCL_INIT_ATTRS.copy() + init_attrs[ + quirk_metadata.entity_metadata.attribute_name + ] = quirk_metadata.attribute_initialized_from_cache + cluster_handler.__dict__[zha_const.ZCL_INIT_ATTRS] = init_attrs + + endpoint.async_new_entity( + platform, + entity_class, + endpoint.unique_id, + [cluster_handler], + quirk_metadata=quirk_metadata, + ) + + _LOGGER.debug( + "'%s' platform -> '%s' using %s", + platform, + entity_class.__name__, + [cluster_handler.name], + ) @callback def discover_coordinator_device_entities(self, device: ZHADevice) -> None: @@ -144,14 +337,20 @@ class ProbeEndpoint: counter_group, counter, ), + {}, ) ) + _LOGGER.debug( + "'%s' platform -> '%s' using %s", + Platform.SENSOR, + sensor.DeviceCounterSensor.__name__, + f"counter groups[{counter_groups}] counter group[{counter_group}] counter[{counter}]", + ) process_counters("counters") process_counters("broadcast_counters") process_counters("device_counters") process_counters("group_counters") - zha_regs.ZHA_ENTITIES.clean_up() @callback def discover_by_device_type(self, endpoint: Endpoint) -> None: @@ -309,7 +508,7 @@ class ProbeEndpoint: for platform, ent_n_handler_list in matches.items(): for entity_and_handler in ent_n_handler_list: _LOGGER.debug( - "'%s' component -> '%s' using %s", + "'%s' platform -> '%s' using %s", platform, entity_and_handler.entity_class.__name__, [ch.name for ch in entity_and_handler.claimed_cluster_handlers], @@ -317,7 +516,8 @@ class ProbeEndpoint: for platform, ent_n_handler_list in matches.items(): for entity_and_handler in ent_n_handler_list: if platform == cmpt_by_dev_type: - # for well known device types, like thermostats we'll take only 1st class + # for well known device types, + # like thermostats we'll take only 1st class endpoint.async_new_entity( platform, entity_and_handler.entity_class, @@ -405,6 +605,7 @@ class GroupProbe: group.group_id, zha_gateway.coordinator_zha_device, ), + {}, ) ) async_dispatcher_send(self._hass, zha_const.SIGNAL_ADD_ENTITIES) diff --git a/homeassistant/components/zha/core/endpoint.py b/homeassistant/components/zha/core/endpoint.py index 490a4e05ea2..37a2c951a7f 100644 --- a/homeassistant/components/zha/core/endpoint.py +++ b/homeassistant/components/zha/core/endpoint.py @@ -7,8 +7,6 @@ import functools import logging from typing import TYPE_CHECKING, Any, Final, TypeVar -from zigpy.typing import EndpointType as ZigpyEndpointType - from homeassistant.const import Platform from homeassistant.core import callback from homeassistant.helpers.dispatcher import async_dispatcher_send @@ -19,6 +17,8 @@ from .cluster_handlers import ClusterHandler from .helpers import get_zha_data if TYPE_CHECKING: + from zigpy import Endpoint as ZigpyEndpoint + from .cluster_handlers import ClientClusterHandler from .device import ZHADevice @@ -34,11 +34,11 @@ CALLABLE_T = TypeVar("CALLABLE_T", bound=Callable) class Endpoint: """Endpoint for a zha device.""" - def __init__(self, zigpy_endpoint: ZigpyEndpointType, device: ZHADevice) -> None: + def __init__(self, zigpy_endpoint: ZigpyEndpoint, device: ZHADevice) -> None: """Initialize instance.""" assert zigpy_endpoint is not None assert device is not None - self._zigpy_endpoint: ZigpyEndpointType = zigpy_endpoint + self._zigpy_endpoint: ZigpyEndpoint = zigpy_endpoint self._device: ZHADevice = device self._all_cluster_handlers: dict[str, ClusterHandler] = {} self._claimed_cluster_handlers: dict[str, ClusterHandler] = {} @@ -66,7 +66,7 @@ class Endpoint: return self._client_cluster_handlers @property - def zigpy_endpoint(self) -> ZigpyEndpointType: + def zigpy_endpoint(self) -> ZigpyEndpoint: """Return endpoint of zigpy device.""" return self._zigpy_endpoint @@ -104,7 +104,7 @@ class Endpoint: ) @classmethod - def new(cls, zigpy_endpoint: ZigpyEndpointType, device: ZHADevice) -> Endpoint: + def new(cls, zigpy_endpoint: ZigpyEndpoint, device: ZHADevice) -> Endpoint: """Create new endpoint and populate cluster handlers.""" endpoint = cls(zigpy_endpoint, device) endpoint.add_all_cluster_handlers() @@ -211,6 +211,7 @@ class Endpoint: entity_class: CALLABLE_T, unique_id: str, cluster_handlers: list[ClusterHandler], + **kwargs: Any, ) -> None: """Create a new entity.""" from .device import DeviceStatus # pylint: disable=import-outside-toplevel @@ -220,7 +221,7 @@ class Endpoint: zha_data = get_zha_data(self.device.hass) zha_data.platforms[platform].append( - (entity_class, (unique_id, self.device, cluster_handlers)) + (entity_class, (unique_id, self.device, cluster_handlers), kwargs or {}) ) @callback diff --git a/homeassistant/components/zha/entity.py b/homeassistant/components/zha/entity.py index ef1b89f1095..3f127c74c0e 100644 --- a/homeassistant/components/zha/entity.py +++ b/homeassistant/components/zha/entity.py @@ -7,7 +7,9 @@ import functools import logging from typing import TYPE_CHECKING, Any, Self -from homeassistant.const import ATTR_NAME +from zigpy.quirks.v2 import EntityMetadata, EntityType + +from homeassistant.const import ATTR_NAME, EntityCategory from homeassistant.core import CALLBACK_TYPE, callback from homeassistant.helpers import entity from homeassistant.helpers.debounce import Debouncer @@ -175,6 +177,31 @@ class ZhaEntity(BaseZhaEntity, RestoreEntity): """ return cls(unique_id, zha_device, cluster_handlers, **kwargs) + def _init_from_quirks_metadata(self, entity_metadata: EntityMetadata) -> None: + """Init this entity from the quirks metadata.""" + if entity_metadata.initially_disabled: + self._attr_entity_registry_enabled_default = False + + if entity_metadata.translation_key: + self._attr_translation_key = entity_metadata.translation_key + + if hasattr(entity_metadata.entity_metadata, "attribute_name"): + if not entity_metadata.translation_key: + self._attr_translation_key = ( + entity_metadata.entity_metadata.attribute_name + ) + self._unique_id_suffix = entity_metadata.entity_metadata.attribute_name + elif hasattr(entity_metadata.entity_metadata, "command_name"): + if not entity_metadata.translation_key: + self._attr_translation_key = ( + entity_metadata.entity_metadata.command_name + ) + self._unique_id_suffix = entity_metadata.entity_metadata.command_name + if entity_metadata.entity_type is EntityType.CONFIG: + self._attr_entity_category = EntityCategory.CONFIG + elif entity_metadata.entity_type is EntityType.DIAGNOSTIC: + self._attr_entity_category = EntityCategory.DIAGNOSTIC + @property def available(self) -> bool: """Return entity availability.""" diff --git a/homeassistant/components/zha/number.py b/homeassistant/components/zha/number.py index a4568b5a14c..c452752f14b 100644 --- a/homeassistant/components/zha/number.py +++ b/homeassistant/components/zha/number.py @@ -5,6 +5,7 @@ import functools import logging from typing import TYPE_CHECKING, Any, Self +from zigpy.quirks.v2 import EntityMetadata, NumberMetadata from zigpy.zcl.clusters.hvac import Thermostat from homeassistant.components.number import NumberEntity, NumberMode @@ -24,6 +25,7 @@ from .core.const import ( CLUSTER_HANDLER_LEVEL, CLUSTER_HANDLER_OCCUPANCY, CLUSTER_HANDLER_THERMOSTAT, + QUIRK_METADATA, SIGNAL_ADD_ENTITIES, SIGNAL_ATTR_UPDATED, ) @@ -400,7 +402,7 @@ class ZHANumberConfigurationEntity(ZhaEntity, NumberEntity): Return entity if it is a supported configuration, otherwise return None """ cluster_handler = cluster_handlers[0] - if ( + if QUIRK_METADATA not in kwargs and ( cls._attribute_name in cluster_handler.cluster.unsupported_attributes or cls._attribute_name not in cluster_handler.cluster.attributes_by_name or cluster_handler.cluster.get(cls._attribute_name) is None @@ -423,8 +425,27 @@ class ZHANumberConfigurationEntity(ZhaEntity, NumberEntity): ) -> None: """Init this number configuration entity.""" self._cluster_handler: ClusterHandler = cluster_handlers[0] + if QUIRK_METADATA in kwargs: + self._init_from_quirks_metadata(kwargs[QUIRK_METADATA]) super().__init__(unique_id, zha_device, cluster_handlers, **kwargs) + def _init_from_quirks_metadata(self, entity_metadata: EntityMetadata) -> None: + """Init this entity from the quirks metadata.""" + super()._init_from_quirks_metadata(entity_metadata) + number_metadata: NumberMetadata = entity_metadata.entity_metadata + self._attribute_name = number_metadata.attribute_name + + if number_metadata.min is not None: + self._attr_native_min_value = number_metadata.min + if number_metadata.max is not None: + self._attr_native_max_value = number_metadata.max + if number_metadata.step is not None: + self._attr_native_step = number_metadata.step + if number_metadata.unit is not None: + self._attr_native_unit_of_measurement = number_metadata.unit + if number_metadata.multiplier is not None: + self._attr_multiplier = number_metadata.multiplier + @property def native_value(self) -> float: """Return the current value.""" diff --git a/homeassistant/components/zha/select.py b/homeassistant/components/zha/select.py index 3736858d599..53acc5cdd02 100644 --- a/homeassistant/components/zha/select.py +++ b/homeassistant/components/zha/select.py @@ -10,6 +10,7 @@ from zhaquirks.quirk_ids import TUYA_PLUG_MANUFACTURER, TUYA_PLUG_ONOFF from zhaquirks.xiaomi.aqara.magnet_ac01 import OppleCluster as MagnetAC01OppleCluster from zhaquirks.xiaomi.aqara.switch_acn047 import OppleCluster as T2RelayOppleCluster from zigpy import types +from zigpy.quirks.v2 import EntityMetadata, ZCLEnumMetadata from zigpy.zcl.clusters.general import OnOff from zigpy.zcl.clusters.security import IasWd @@ -27,6 +28,7 @@ from .core.const import ( CLUSTER_HANDLER_INOVELLI, CLUSTER_HANDLER_OCCUPANCY, CLUSTER_HANDLER_ON_OFF, + QUIRK_METADATA, SIGNAL_ADD_ENTITIES, SIGNAL_ATTR_UPDATED, Strobe, @@ -82,9 +84,9 @@ class ZHAEnumSelectEntity(ZhaEntity, SelectEntity): **kwargs: Any, ) -> None: """Init this select entity.""" + self._cluster_handler: ClusterHandler = cluster_handlers[0] self._attribute_name = self._enum.__name__ self._attr_options = [entry.name.replace("_", " ") for entry in self._enum] - self._cluster_handler: ClusterHandler = cluster_handlers[0] super().__init__(unique_id, zha_device, cluster_handlers, **kwargs) @property @@ -176,7 +178,7 @@ class ZCLEnumSelectEntity(ZhaEntity, SelectEntity): Return entity if it is a supported configuration, otherwise return None """ cluster_handler = cluster_handlers[0] - if ( + if QUIRK_METADATA not in kwargs and ( cls._attribute_name in cluster_handler.cluster.unsupported_attributes or cls._attribute_name not in cluster_handler.cluster.attributes_by_name or cluster_handler.cluster.get(cls._attribute_name) is None @@ -198,10 +200,19 @@ class ZCLEnumSelectEntity(ZhaEntity, SelectEntity): **kwargs: Any, ) -> None: """Init this select entity.""" - self._attr_options = [entry.name.replace("_", " ") for entry in self._enum] self._cluster_handler: ClusterHandler = cluster_handlers[0] + if QUIRK_METADATA in kwargs: + self._init_from_quirks_metadata(kwargs[QUIRK_METADATA]) + self._attr_options = [entry.name.replace("_", " ") for entry in self._enum] super().__init__(unique_id, zha_device, cluster_handlers, **kwargs) + def _init_from_quirks_metadata(self, entity_metadata: EntityMetadata) -> None: + """Init this entity from the quirks metadata.""" + super()._init_from_quirks_metadata(entity_metadata) + zcl_enum_metadata: ZCLEnumMetadata = entity_metadata.entity_metadata + self._attribute_name = zcl_enum_metadata.attribute_name + self._enum = zcl_enum_metadata.enum + @property def current_option(self) -> str | None: """Return the selected entity option to represent the entity state.""" diff --git a/homeassistant/components/zha/sensor.py b/homeassistant/components/zha/sensor.py index c4e620a8b0e..6a68b55a8be 100644 --- a/homeassistant/components/zha/sensor.py +++ b/homeassistant/components/zha/sensor.py @@ -6,11 +6,13 @@ from dataclasses import dataclass from datetime import timedelta import enum import functools +import logging import numbers import random from typing import TYPE_CHECKING, Any, Self from zigpy import types +from zigpy.quirks.v2 import EntityMetadata, ZCLEnumMetadata, ZCLSensorMetadata from zigpy.state import Counter, State from zigpy.zcl.clusters.closures import WindowCovering from zigpy.zcl.clusters.general import Basic @@ -68,6 +70,7 @@ from .core.const import ( CLUSTER_HANDLER_TEMPERATURE, CLUSTER_HANDLER_THERMOSTAT, DATA_ZHA, + QUIRK_METADATA, SIGNAL_ADD_ENTITIES, SIGNAL_ATTR_UPDATED, ) @@ -95,6 +98,8 @@ BATTERY_SIZES = { 255: "Unknown", } +_LOGGER = logging.getLogger(__name__) + CLUSTER_HANDLER_ST_HUMIDITY_CLUSTER = ( f"cluster_handler_0x{SMARTTHINGS_HUMIDITY_CLUSTER:04x}" ) @@ -135,17 +140,6 @@ class Sensor(ZhaEntity, SensorEntity): _divisor: int = 1 _multiplier: int | float = 1 - def __init__( - self, - unique_id: str, - zha_device: ZHADevice, - cluster_handlers: list[ClusterHandler], - **kwargs: Any, - ) -> None: - """Init this sensor.""" - super().__init__(unique_id, zha_device, cluster_handlers, **kwargs) - self._cluster_handler: ClusterHandler = cluster_handlers[0] - @classmethod def create_entity( cls, @@ -159,14 +153,44 @@ class Sensor(ZhaEntity, SensorEntity): Return entity if it is a supported configuration, otherwise return None """ cluster_handler = cluster_handlers[0] - if ( + if QUIRK_METADATA not in kwargs and ( cls._attribute_name in cluster_handler.cluster.unsupported_attributes or cls._attribute_name not in cluster_handler.cluster.attributes_by_name ): + _LOGGER.debug( + "%s is not supported - skipping %s entity creation", + cls._attribute_name, + cls.__name__, + ) return None return cls(unique_id, zha_device, cluster_handlers, **kwargs) + def __init__( + self, + unique_id: str, + zha_device: ZHADevice, + cluster_handlers: list[ClusterHandler], + **kwargs: Any, + ) -> None: + """Init this sensor.""" + self._cluster_handler: ClusterHandler = cluster_handlers[0] + if QUIRK_METADATA in kwargs: + self._init_from_quirks_metadata(kwargs[QUIRK_METADATA]) + super().__init__(unique_id, zha_device, cluster_handlers, **kwargs) + + def _init_from_quirks_metadata(self, entity_metadata: EntityMetadata) -> None: + """Init this entity from the quirks metadata.""" + super()._init_from_quirks_metadata(entity_metadata) + sensor_metadata: ZCLSensorMetadata = entity_metadata.entity_metadata + self._attribute_name = sensor_metadata.attribute_name + if sensor_metadata.divisor is not None: + self._divisor = sensor_metadata.divisor + if sensor_metadata.multiplier is not None: + self._multiplier = sensor_metadata.multiplier + if sensor_metadata.unit is not None: + self._attr_native_unit_of_measurement = sensor_metadata.unit + async def async_added_to_hass(self) -> None: """Run when about to be added to hass.""" await super().async_added_to_hass() @@ -330,6 +354,13 @@ class EnumSensor(Sensor): _attr_device_class: SensorDeviceClass = SensorDeviceClass.ENUM _enum: type[enum.Enum] + def _init_from_quirks_metadata(self, entity_metadata: EntityMetadata) -> None: + """Init this entity from the quirks metadata.""" + ZhaEntity._init_from_quirks_metadata(self, entity_metadata) # pylint: disable=protected-access + sensor_metadata: ZCLEnumMetadata = entity_metadata.entity_metadata + self._attribute_name = sensor_metadata.attribute_name + self._enum = sensor_metadata.enum + def formatter(self, value: int) -> str | None: """Use name of enum.""" assert self._enum is not None diff --git a/homeassistant/components/zha/switch.py b/homeassistant/components/zha/switch.py index afc73baca70..960124c4a8a 100644 --- a/homeassistant/components/zha/switch.py +++ b/homeassistant/components/zha/switch.py @@ -6,6 +6,7 @@ import logging from typing import TYPE_CHECKING, Any, Self from zhaquirks.quirk_ids import TUYA_PLUG_ONOFF +from zigpy.quirks.v2 import EntityMetadata, SwitchMetadata from zigpy.zcl.clusters.closures import ConfigStatus, WindowCovering, WindowCoveringMode from zigpy.zcl.clusters.general import OnOff from zigpy.zcl.foundation import Status @@ -23,6 +24,7 @@ from .core.const import ( CLUSTER_HANDLER_COVER, CLUSTER_HANDLER_INOVELLI, CLUSTER_HANDLER_ON_OFF, + QUIRK_METADATA, SIGNAL_ADD_ENTITIES, SIGNAL_ATTR_UPDATED, ) @@ -173,6 +175,8 @@ class ZHASwitchConfigurationEntity(ZhaEntity, SwitchEntity): _attribute_name: str _inverter_attribute_name: str | None = None _force_inverted: bool = False + _off_value: int = 0 + _on_value: int = 1 @classmethod def create_entity( @@ -187,7 +191,7 @@ class ZHASwitchConfigurationEntity(ZhaEntity, SwitchEntity): Return entity if it is a supported configuration, otherwise return None """ cluster_handler = cluster_handlers[0] - if ( + if QUIRK_METADATA not in kwargs and ( cls._attribute_name in cluster_handler.cluster.unsupported_attributes or cls._attribute_name not in cluster_handler.cluster.attributes_by_name or cluster_handler.cluster.get(cls._attribute_name) is None @@ -210,8 +214,22 @@ class ZHASwitchConfigurationEntity(ZhaEntity, SwitchEntity): ) -> None: """Init this number configuration entity.""" self._cluster_handler: ClusterHandler = cluster_handlers[0] + if QUIRK_METADATA in kwargs: + self._init_from_quirks_metadata(kwargs[QUIRK_METADATA]) super().__init__(unique_id, zha_device, cluster_handlers, **kwargs) + def _init_from_quirks_metadata(self, entity_metadata: EntityMetadata) -> None: + """Init this entity from the quirks metadata.""" + super()._init_from_quirks_metadata(entity_metadata) + switch_metadata: SwitchMetadata = entity_metadata.entity_metadata + self._attribute_name = switch_metadata.attribute_name + if switch_metadata.invert_attribute_name: + self._inverter_attribute_name = switch_metadata.invert_attribute_name + if switch_metadata.force_inverted: + self._force_inverted = switch_metadata.force_inverted + self._off_value = switch_metadata.off_value + self._on_value = switch_metadata.on_value + async def async_added_to_hass(self) -> None: """Run when about to be added to hass.""" await super().async_added_to_hass() @@ -236,14 +254,25 @@ class ZHASwitchConfigurationEntity(ZhaEntity, SwitchEntity): @property def is_on(self) -> bool: """Return if the switch is on based on the statemachine.""" - val = bool(self._cluster_handler.cluster.get(self._attribute_name)) + if self._on_value != 1: + val = self._cluster_handler.cluster.get(self._attribute_name) + val = val == self._on_value + else: + val = bool(self._cluster_handler.cluster.get(self._attribute_name)) return (not val) if self.inverted else val async def async_turn_on_off(self, state: bool) -> None: """Turn the entity on or off.""" - await self._cluster_handler.write_attributes_safe( - {self._attribute_name: not state if self.inverted else state} - ) + if self.inverted: + state = not state + if state: + await self._cluster_handler.write_attributes_safe( + {self._attribute_name: self._on_value} + ) + else: + await self._cluster_handler.write_attributes_safe( + {self._attribute_name: self._off_value} + ) self.async_write_ha_state() async def async_turn_on(self, **kwargs: Any) -> None: diff --git a/tests/components/zha/test_button.py b/tests/components/zha/test_button.py index cc0b5079fd3..9eab72b435b 100644 --- a/tests/components/zha/test_button.py +++ b/tests/components/zha/test_button.py @@ -1,4 +1,5 @@ """Test ZHA button.""" +from typing import Final from unittest.mock import call, patch from freezegun import freeze_time @@ -15,6 +16,7 @@ from zigpy.const import SIG_EP_PROFILE from zigpy.exceptions import ZigbeeException import zigpy.profiles.zha as zha from zigpy.quirks import CustomCluster, CustomDevice +from zigpy.quirks.v2 import add_to_registry_v2 import zigpy.types as t import zigpy.zcl.clusters.general as general from zigpy.zcl.clusters.manufacturer_specific import ManufacturerSpecificCluster @@ -33,7 +35,7 @@ from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers import entity_registry as er -from .common import find_entity_id +from .common import find_entity_id, update_attribute_cache from .conftest import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_TYPE @@ -56,7 +58,9 @@ def button_platform_only(): @pytest.fixture -async def contact_sensor(hass, zigpy_device_mock, zha_device_joined_restored): +async def contact_sensor( + hass: HomeAssistant, zigpy_device_mock, zha_device_joined_restored +): """Contact sensor fixture.""" zigpy_device = zigpy_device_mock( @@ -102,7 +106,9 @@ class FrostLockQuirk(CustomDevice): @pytest.fixture -async def tuya_water_valve(hass, zigpy_device_mock, zha_device_joined_restored): +async def tuya_water_valve( + hass: HomeAssistant, zigpy_device_mock, zha_device_joined_restored +): """Tuya Water Valve fixture.""" zigpy_device = zigpy_device_mock( @@ -224,3 +230,141 @@ async def test_frost_unlock(hass: HomeAssistant, tuya_water_valve) -> None: call({"frost_lock_reset": 0}, manufacturer=None), call({"frost_lock_reset": 0}, manufacturer=None), ] + + +class FakeManufacturerCluster(CustomCluster, ManufacturerSpecificCluster): + """Fake manufacturer cluster.""" + + cluster_id: Final = 0xFFF3 + ep_attribute: Final = "mfg_identify" + + class AttributeDefs(zcl_f.BaseAttributeDefs): + """Attribute definitions.""" + + feed: Final = zcl_f.ZCLAttributeDef( + id=0x0000, type=t.uint8_t, access="rw", is_manufacturer_specific=True + ) + + class ServerCommandDefs(zcl_f.BaseCommandDefs): + """Server command definitions.""" + + self_test: Final = zcl_f.ZCLCommandDef( + id=0x00, schema={"identify_time": t.uint16_t}, direction=False + ) + + +( + add_to_registry_v2("Fake_Model", "Fake_Manufacturer") + .replaces(FakeManufacturerCluster) + .command_button( + FakeManufacturerCluster.ServerCommandDefs.self_test.name, + FakeManufacturerCluster.cluster_id, + command_args=(5,), + ) + .write_attr_button( + FakeManufacturerCluster.AttributeDefs.feed.name, + 2, + FakeManufacturerCluster.cluster_id, + ) +) + + +@pytest.fixture +async def custom_button_device( + hass: HomeAssistant, zigpy_device_mock, zha_device_joined_restored +): + """Button device fixture for quirks button tests.""" + + zigpy_device = zigpy_device_mock( + { + 1: { + SIG_EP_INPUT: [ + general.Basic.cluster_id, + FakeManufacturerCluster.cluster_id, + ], + SIG_EP_OUTPUT: [], + SIG_EP_TYPE: zha.DeviceType.REMOTE_CONTROL, + SIG_EP_PROFILE: zha.PROFILE_ID, + } + }, + manufacturer="Fake_Model", + model="Fake_Manufacturer", + ) + + zigpy_device.endpoints[1].mfg_identify.PLUGGED_ATTR_READS = { + FakeManufacturerCluster.AttributeDefs.feed.name: 0, + } + update_attribute_cache(zigpy_device.endpoints[1].mfg_identify) + zha_device = await zha_device_joined_restored(zigpy_device) + return zha_device, zigpy_device.endpoints[1].mfg_identify + + +@freeze_time("2021-11-04 17:37:00", tz_offset=-1) +async def test_quirks_command_button(hass: HomeAssistant, custom_button_device) -> None: + """Test ZHA button platform.""" + + zha_device, cluster = custom_button_device + assert cluster is not None + entity_id = find_entity_id(DOMAIN, zha_device, hass, qualifier="self_test") + assert entity_id is not None + + state = hass.states.get(entity_id) + assert state + assert state.state == STATE_UNKNOWN + + with patch( + "zigpy.zcl.Cluster.request", + return_value=[0x00, zcl_f.Status.SUCCESS], + ): + await hass.services.async_call( + DOMAIN, + SERVICE_PRESS, + {ATTR_ENTITY_ID: entity_id}, + blocking=True, + ) + await hass.async_block_till_done() + assert len(cluster.request.mock_calls) == 1 + assert cluster.request.call_args[0][0] is False + assert cluster.request.call_args[0][1] == 0 + assert cluster.request.call_args[0][3] == 5 # duration in seconds + + state = hass.states.get(entity_id) + assert state + assert state.state == "2021-11-04T16:37:00+00:00" + + +@freeze_time("2021-11-04 17:37:00", tz_offset=-1) +async def test_quirks_write_attr_button( + hass: HomeAssistant, custom_button_device +) -> None: + """Test ZHA button platform.""" + + zha_device, cluster = custom_button_device + assert cluster is not None + entity_id = find_entity_id(DOMAIN, zha_device, hass, qualifier="feed") + assert entity_id is not None + + state = hass.states.get(entity_id) + assert state + assert state.state == STATE_UNKNOWN + assert cluster.get(cluster.AttributeDefs.feed.name) == 0 + + with patch( + "zigpy.zcl.Cluster.request", + return_value=[0x00, zcl_f.Status.SUCCESS], + ): + await hass.services.async_call( + DOMAIN, + SERVICE_PRESS, + {ATTR_ENTITY_ID: entity_id}, + blocking=True, + ) + await hass.async_block_till_done() + assert cluster.write_attributes.mock_calls == [ + call({cluster.AttributeDefs.feed.name: 2}, manufacturer=None) + ] + + state = hass.states.get(entity_id) + assert state + assert state.state == "2021-11-04T16:37:00+00:00" + assert cluster.get(cluster.AttributeDefs.feed.name) == 2 diff --git a/tests/components/zha/test_discover.py b/tests/components/zha/test_discover.py index 1491b46005b..c8eba90a372 100644 --- a/tests/components/zha/test_discover.py +++ b/tests/components/zha/test_discover.py @@ -6,10 +6,23 @@ from unittest import mock from unittest.mock import AsyncMock, Mock, patch import pytest +from zhaquirks.ikea import PowerConfig1CRCluster, ScenesCluster +from zhaquirks.xiaomi import ( + BasicCluster, + LocalIlluminanceMeasurementCluster, + XiaomiPowerConfigurationPercent, +) +from zhaquirks.xiaomi.aqara.driver_curtain_e1 import ( + WindowCoveringE1, + XiaomiAqaraDriverE1, +) from zigpy.const import SIG_ENDPOINTS, SIG_MANUFACTURER, SIG_MODEL, SIG_NODE_DESC import zigpy.profiles.zha import zigpy.quirks +from zigpy.quirks.v2 import EntityType, add_to_registry_v2 +from zigpy.quirks.v2.homeassistant import UnitOfTime import zigpy.types +from zigpy.zcl import ClusterType import zigpy.zcl.clusters.closures import zigpy.zcl.clusters.general import zigpy.zcl.clusters.security @@ -22,11 +35,12 @@ import homeassistant.components.zha.core.discovery as disc from homeassistant.components.zha.core.endpoint import Endpoint from homeassistant.components.zha.core.helpers import get_zha_gateway import homeassistant.components.zha.core.registries as zha_regs -from homeassistant.const import Platform +from homeassistant.const import STATE_OFF, Platform from homeassistant.core import HomeAssistant from homeassistant.helpers import entity_registry as er from homeassistant.helpers.entity_platform import EntityPlatform +from .common import find_entity_id, update_attribute_cache from .conftest import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_PROFILE, SIG_EP_TYPE from .zha_devices_list import ( DEV_SIG_ATTRIBUTES, @@ -147,7 +161,9 @@ async def test_devices( for (platform, unique_id), ent_info in device[DEV_SIG_ENT_MAP].items(): no_tail_id = NO_TAIL_ID.sub("", ent_info[DEV_SIG_ENT_MAP_ID]) ha_entity_id = entity_registry.async_get_entity_id(platform, "zha", unique_id) - assert ha_entity_id is not None + message1 = f"No entity found for platform[{platform}] unique_id[{unique_id}]" + message2 = f"no_tail_id[{no_tail_id}] with entity_id[{ha_entity_id}]" + assert ha_entity_id is not None, f"{message1} {message2}" assert ha_entity_id.startswith(no_tail_id) entity = created_entities[ha_entity_id] @@ -461,3 +477,332 @@ async def test_group_probe_cleanup_called( await config_entry.async_unload(hass_disable_services) await hass_disable_services.async_block_till_done() disc.GROUP_PROBE.cleanup.assert_called() + + +async def test_quirks_v2_entity_discovery( + hass, + zigpy_device_mock, + zha_device_joined, +) -> None: + """Test quirks v2 discovery.""" + + zigpy_device = zigpy_device_mock( + { + 1: { + SIG_EP_INPUT: [ + zigpy.zcl.clusters.general.PowerConfiguration.cluster_id, + zigpy.zcl.clusters.general.Groups.cluster_id, + zigpy.zcl.clusters.general.OnOff.cluster_id, + ], + SIG_EP_OUTPUT: [ + zigpy.zcl.clusters.general.Scenes.cluster_id, + ], + SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.NON_COLOR_CONTROLLER, + } + }, + ieee="01:2d:6f:00:0a:90:69:e8", + manufacturer="Ikea of Sweden", + model="TRADFRI remote control", + ) + + ( + add_to_registry_v2( + "Ikea of Sweden", "TRADFRI remote control", zigpy.quirks._DEVICE_REGISTRY + ) + .replaces(PowerConfig1CRCluster) + .replaces(ScenesCluster, cluster_type=ClusterType.Client) + .number( + zigpy.zcl.clusters.general.OnOff.AttributeDefs.off_wait_time.name, + zigpy.zcl.clusters.general.OnOff.cluster_id, + min_value=1, + max_value=100, + step=1, + unit=UnitOfTime.SECONDS, + multiplier=1, + ) + ) + + zigpy_device = zigpy.quirks._DEVICE_REGISTRY.get_device(zigpy_device) + zigpy_device.endpoints[1].power.PLUGGED_ATTR_READS = { + "battery_voltage": 3, + "battery_percentage_remaining": 100, + } + update_attribute_cache(zigpy_device.endpoints[1].power) + zigpy_device.endpoints[1].on_off.PLUGGED_ATTR_READS = { + zigpy.zcl.clusters.general.OnOff.AttributeDefs.off_wait_time.name: 3, + } + update_attribute_cache(zigpy_device.endpoints[1].on_off) + + zha_device = await zha_device_joined(zigpy_device) + + entity_id = find_entity_id( + Platform.NUMBER, + zha_device, + hass, + ) + assert entity_id is not None + + state = hass.states.get(entity_id) + assert state is not None + + +async def test_quirks_v2_entity_discovery_e1_curtain( + hass, + zigpy_device_mock, + zha_device_joined, +) -> None: + """Test quirks v2 discovery for e1 curtain motor.""" + aqara_E1_device = zigpy_device_mock( + { + 1: { + SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.WINDOW_COVERING_DEVICE, + SIG_EP_INPUT: [ + zigpy.zcl.clusters.general.Basic.cluster_id, + zigpy.zcl.clusters.general.PowerConfiguration.cluster_id, + zigpy.zcl.clusters.general.Identify.cluster_id, + zigpy.zcl.clusters.general.Time.cluster_id, + WindowCoveringE1.cluster_id, + XiaomiAqaraDriverE1.cluster_id, + ], + SIG_EP_OUTPUT: [ + zigpy.zcl.clusters.general.Identify.cluster_id, + zigpy.zcl.clusters.general.Time.cluster_id, + zigpy.zcl.clusters.general.Ota.cluster_id, + XiaomiAqaraDriverE1.cluster_id, + ], + } + }, + ieee="01:2d:6f:00:0a:90:69:e8", + manufacturer="LUMI", + model="lumi.curtain.agl006", + ) + + class AqaraE1HookState(zigpy.types.enum8): + """Aqara hook state.""" + + Unlocked = 0x00 + Locked = 0x01 + Locking = 0x02 + Unlocking = 0x03 + + class FakeXiaomiAqaraDriverE1(XiaomiAqaraDriverE1): + """Fake XiaomiAqaraDriverE1 cluster.""" + + attributes = XiaomiAqaraDriverE1.attributes.copy() + attributes.update( + { + 0x9999: ("error_detected", zigpy.types.Bool, True), + } + ) + + ( + add_to_registry_v2("LUMI", "lumi.curtain.agl006") + .adds(LocalIlluminanceMeasurementCluster) + .replaces(BasicCluster) + .replaces(XiaomiPowerConfigurationPercent) + .replaces(WindowCoveringE1) + .replaces(FakeXiaomiAqaraDriverE1) + .removes(FakeXiaomiAqaraDriverE1, cluster_type=ClusterType.Client) + .enum( + BasicCluster.AttributeDefs.power_source.name, + BasicCluster.PowerSource, + BasicCluster.cluster_id, + entity_platform=Platform.SENSOR, + entity_type=EntityType.DIAGNOSTIC, + ) + .enum( + "hooks_state", + AqaraE1HookState, + FakeXiaomiAqaraDriverE1.cluster_id, + entity_platform=Platform.SENSOR, + entity_type=EntityType.DIAGNOSTIC, + ) + .binary_sensor("error_detected", FakeXiaomiAqaraDriverE1.cluster_id) + ) + + aqara_E1_device = zigpy.quirks._DEVICE_REGISTRY.get_device(aqara_E1_device) + + aqara_E1_device.endpoints[1].opple_cluster.PLUGGED_ATTR_READS = { + "hand_open": 0, + "positions_stored": 0, + "hooks_lock": 0, + "hooks_state": AqaraE1HookState.Unlocked, + "light_level": 0, + "error_detected": 0, + } + update_attribute_cache(aqara_E1_device.endpoints[1].opple_cluster) + + aqara_E1_device.endpoints[1].basic.PLUGGED_ATTR_READS = { + BasicCluster.AttributeDefs.power_source.name: BasicCluster.PowerSource.Mains_single_phase, + } + update_attribute_cache(aqara_E1_device.endpoints[1].basic) + + WCAttrs = zigpy.zcl.clusters.closures.WindowCovering.AttributeDefs + WCT = zigpy.zcl.clusters.closures.WindowCovering.WindowCoveringType + WCCS = zigpy.zcl.clusters.closures.WindowCovering.ConfigStatus + aqara_E1_device.endpoints[1].window_covering.PLUGGED_ATTR_READS = { + WCAttrs.current_position_lift_percentage.name: 0, + WCAttrs.window_covering_type.name: WCT.Drapery, + WCAttrs.config_status.name: WCCS(~WCCS.Open_up_commands_reversed), + } + update_attribute_cache(aqara_E1_device.endpoints[1].window_covering) + + zha_device = await zha_device_joined(aqara_E1_device) + + power_source_entity_id = find_entity_id( + Platform.SENSOR, + zha_device, + hass, + qualifier=BasicCluster.AttributeDefs.power_source.name, + ) + assert power_source_entity_id is not None + state = hass.states.get(power_source_entity_id) + assert state is not None + assert state.state == BasicCluster.PowerSource.Mains_single_phase.name + + hook_state_entity_id = find_entity_id( + Platform.SENSOR, + zha_device, + hass, + qualifier="hooks_state", + ) + assert hook_state_entity_id is not None + state = hass.states.get(hook_state_entity_id) + assert state is not None + assert state.state == AqaraE1HookState.Unlocked.name + + error_detected_entity_id = find_entity_id( + Platform.BINARY_SENSOR, + zha_device, + hass, + ) + assert error_detected_entity_id is not None + state = hass.states.get(error_detected_entity_id) + assert state is not None + assert state.state == STATE_OFF + + +def _get_test_device(zigpy_device_mock, manufacturer: str, model: str): + zigpy_device = zigpy_device_mock( + { + 1: { + SIG_EP_INPUT: [ + zigpy.zcl.clusters.general.PowerConfiguration.cluster_id, + zigpy.zcl.clusters.general.Groups.cluster_id, + zigpy.zcl.clusters.general.OnOff.cluster_id, + ], + SIG_EP_OUTPUT: [ + zigpy.zcl.clusters.general.Scenes.cluster_id, + ], + SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.NON_COLOR_CONTROLLER, + } + }, + ieee="01:2d:6f:00:0a:90:69:e8", + manufacturer=manufacturer, + model=model, + ) + + ( + add_to_registry_v2(manufacturer, model, zigpy.quirks._DEVICE_REGISTRY) + .replaces(PowerConfig1CRCluster) + .replaces(ScenesCluster, cluster_type=ClusterType.Client) + .number( + zigpy.zcl.clusters.general.OnOff.AttributeDefs.off_wait_time.name, + zigpy.zcl.clusters.general.OnOff.cluster_id, + endpoint_id=3, + min_value=1, + max_value=100, + step=1, + unit=UnitOfTime.SECONDS, + multiplier=1, + ) + .number( + zigpy.zcl.clusters.general.OnOff.AttributeDefs.off_wait_time.name, + zigpy.zcl.clusters.general.Time.cluster_id, + min_value=1, + max_value=100, + step=1, + unit=UnitOfTime.SECONDS, + multiplier=1, + ) + .sensor( + zigpy.zcl.clusters.general.OnOff.AttributeDefs.off_wait_time.name, + zigpy.zcl.clusters.general.OnOff.cluster_id, + entity_type=EntityType.CONFIG, + ) + ) + + zigpy_device = zigpy.quirks._DEVICE_REGISTRY.get_device(zigpy_device) + zigpy_device.endpoints[1].power.PLUGGED_ATTR_READS = { + "battery_voltage": 3, + "battery_percentage_remaining": 100, + } + update_attribute_cache(zigpy_device.endpoints[1].power) + zigpy_device.endpoints[1].on_off.PLUGGED_ATTR_READS = { + zigpy.zcl.clusters.general.OnOff.AttributeDefs.off_wait_time.name: 3, + } + update_attribute_cache(zigpy_device.endpoints[1].on_off) + return zigpy_device + + +async def test_quirks_v2_entity_no_metadata( + hass: HomeAssistant, + zigpy_device_mock, + zha_device_joined, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test quirks v2 discovery skipped - no metadata.""" + + zigpy_device = _get_test_device( + zigpy_device_mock, "Ikea of Sweden2", "TRADFRI remote control2" + ) + setattr(zigpy_device, "_exposes_metadata", {}) + zha_device = await zha_device_joined(zigpy_device) + assert ( + f"Device: {str(zigpy_device.ieee)}-{zha_device.name} does not expose any quirks v2 entities" + in caplog.text + ) + + +async def test_quirks_v2_entity_discovery_errors( + hass: HomeAssistant, + zigpy_device_mock, + zha_device_joined, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test quirks v2 discovery skipped - errors.""" + + zigpy_device = _get_test_device( + zigpy_device_mock, "Ikea of Sweden3", "TRADFRI remote control3" + ) + zha_device = await zha_device_joined(zigpy_device) + + m1 = f"Device: {str(zigpy_device.ieee)}-{zha_device.name} does not have an" + m2 = " endpoint with id: 3 - unable to create entity with cluster" + m3 = " details: (3, 6, )" + assert f"{m1}{m2}{m3}" in caplog.text + + time_cluster_id = zigpy.zcl.clusters.general.Time.cluster_id + + m1 = f"Device: {str(zigpy_device.ieee)}-{zha_device.name} does not have a" + m2 = f" cluster with id: {time_cluster_id} - unable to create entity with " + m3 = f"cluster details: (1, {time_cluster_id}, )" + assert f"{m1}{m2}{m3}" in caplog.text + + # fmt: off + entity_details = ( + "{'cluster_details': (1, 6, ), " + "'quirk_metadata': EntityMetadata(entity_metadata=ZCLSensorMetadata(" + "attribute_name='off_wait_time', divisor=1, multiplier=1, unit=None, " + "device_class=None, state_class=None), entity_platform=, entity_type=, " + "cluster_id=6, endpoint_id=1, cluster_type=, " + "initially_disabled=False, attribute_initialized_from_cache=True, " + "translation_key=None)}" + ) + # fmt: on + + m1 = f"Device: {str(zigpy_device.ieee)}-{zha_device.name} has an entity with " + m2 = f"details: {entity_details} that does not have an entity class mapping - " + m3 = "unable to create entity" + assert f"{m1}{m2}{m3}" in caplog.text diff --git a/tests/components/zha/test_select.py b/tests/components/zha/test_select.py index d2d1b79c92f..549a123aefb 100644 --- a/tests/components/zha/test_select.py +++ b/tests/components/zha/test_select.py @@ -12,6 +12,7 @@ from zhaquirks import ( from zigpy.const import SIG_EP_PROFILE import zigpy.profiles.zha as zha from zigpy.quirks import CustomCluster, CustomDevice +from zigpy.quirks.v2 import CustomDeviceV2, add_to_registry_v2 import zigpy.types as t import zigpy.zcl.clusters.general as general from zigpy.zcl.clusters.manufacturer_specific import ManufacturerSpecificCluster @@ -349,13 +350,19 @@ class MotionSensitivityQuirk(CustomDevice): ep_attribute = "opple_cluster" attributes = { 0x010C: ("motion_sensitivity", t.uint8_t, True), + 0x020C: ("motion_sensitivity_disabled", t.uint8_t, True), } def __init__(self, *args, **kwargs): """Initialize.""" super().__init__(*args, **kwargs) # populate cache to create config entity - self._attr_cache.update({0x010C: AqaraMotionSensitivities.Medium}) + self._attr_cache.update( + { + 0x010C: AqaraMotionSensitivities.Medium, + 0x020C: AqaraMotionSensitivities.Medium, + } + ) replacement = { ENDPOINTS: { @@ -413,3 +420,79 @@ async def test_on_off_select_attribute_report( hass, cluster, {"motion_sensitivity": AqaraMotionSensitivities.Low} ) assert hass.states.get(entity_id).state == AqaraMotionSensitivities.Low.name + + +( + add_to_registry_v2("Fake_Manufacturer", "Fake_Model") + .replaces(MotionSensitivityQuirk.OppleCluster) + .enum( + "motion_sensitivity", + AqaraMotionSensitivities, + MotionSensitivityQuirk.OppleCluster.cluster_id, + ) + .enum( + "motion_sensitivity_disabled", + AqaraMotionSensitivities, + MotionSensitivityQuirk.OppleCluster.cluster_id, + translation_key="motion_sensitivity_translation_key", + initially_disabled=True, + ) +) + + +@pytest.fixture +async def zigpy_device_aqara_sensor_v2( + hass: HomeAssistant, zigpy_device_mock, zha_device_joined_restored +): + """Device tracker zigpy Aqara motion sensor device.""" + + zigpy_device = zigpy_device_mock( + { + 1: { + SIG_EP_INPUT: [ + general.Basic.cluster_id, + MotionSensitivityQuirk.OppleCluster.cluster_id, + ], + SIG_EP_OUTPUT: [], + SIG_EP_TYPE: zha.DeviceType.OCCUPANCY_SENSOR, + } + }, + manufacturer="Fake_Manufacturer", + model="Fake_Model", + ) + + zha_device = await zha_device_joined_restored(zigpy_device) + return zha_device, zigpy_device.endpoints[1].opple_cluster + + +async def test_on_off_select_attribute_report_v2( + hass: HomeAssistant, zigpy_device_aqara_sensor_v2 +) -> None: + """Test ZHA attribute report parsing for select platform.""" + + zha_device, cluster = zigpy_device_aqara_sensor_v2 + assert isinstance(zha_device.device, CustomDeviceV2) + entity_id = find_entity_id( + Platform.SELECT, zha_device, hass, qualifier="motion_sensitivity" + ) + assert entity_id is not None + + # allow traffic to flow through the gateway and device + await async_enable_traffic(hass, [zha_device]) + + # test that the state is in default medium state + assert hass.states.get(entity_id).state == AqaraMotionSensitivities.Medium.name + + # send attribute report from device + await send_attributes_report( + hass, cluster, {"motion_sensitivity": AqaraMotionSensitivities.Low} + ) + assert hass.states.get(entity_id).state == AqaraMotionSensitivities.Low.name + + entity_registry = er.async_get(hass) + # none in id because the translation key does not exist + entity_entry = entity_registry.async_get("select.fake_manufacturer_fake_model_none") + assert entity_entry + assert entity_entry.entity_category == EntityCategory.CONFIG + assert entity_entry.disabled is True + assert entity_entry.translation_key == "motion_sensitivity_translation_key" diff --git a/tests/components/zha/test_sensor.py b/tests/components/zha/test_sensor.py index 7b96d43aed3..a7047b8dcd4 100644 --- a/tests/components/zha/test_sensor.py +++ b/tests/components/zha/test_sensor.py @@ -5,8 +5,13 @@ from unittest.mock import MagicMock, patch import pytest import zigpy.profiles.zha +from zigpy.quirks import CustomCluster +from zigpy.quirks.v2 import CustomDeviceV2, add_to_registry_v2 +from zigpy.quirks.v2.homeassistant import UnitOfMass +import zigpy.types as t from zigpy.zcl.clusters import general, homeautomation, hvac, measurement, smartenergy from zigpy.zcl.clusters.hvac import Thermostat +from zigpy.zcl.clusters.manufacturer_specific import ManufacturerSpecificCluster from homeassistant.components.sensor import SensorDeviceClass from homeassistant.components.zha.core import ZHADevice @@ -1187,6 +1192,79 @@ async def test_elec_measurement_skip_unsupported_attribute( assert read_attrs == supported_attributes +class OppleCluster(CustomCluster, ManufacturerSpecificCluster): + """Aqara manufacturer specific cluster.""" + + cluster_id = 0xFCC0 + ep_attribute = "opple_cluster" + attributes = { + 0x010C: ("last_feeding_size", t.uint16_t, True), + } + + def __init__(self, *args, **kwargs) -> None: + """Initialize.""" + super().__init__(*args, **kwargs) + # populate cache to create config entity + self._attr_cache.update({0x010C: 10}) + + +( + add_to_registry_v2("Fake_Manufacturer_sensor", "Fake_Model_sensor") + .replaces(OppleCluster) + .sensor( + "last_feeding_size", + OppleCluster.cluster_id, + divisor=1, + multiplier=1, + unit=UnitOfMass.GRAMS, + ) +) + + +@pytest.fixture +async def zigpy_device_aqara_sensor_v2( + hass: HomeAssistant, zigpy_device_mock, zha_device_joined_restored +): + """Device tracker zigpy Aqara motion sensor device.""" + + zigpy_device = zigpy_device_mock( + { + 1: { + SIG_EP_INPUT: [ + general.Basic.cluster_id, + OppleCluster.cluster_id, + ], + SIG_EP_OUTPUT: [], + SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.OCCUPANCY_SENSOR, + } + }, + manufacturer="Fake_Manufacturer_sensor", + model="Fake_Model_sensor", + ) + + zha_device = await zha_device_joined_restored(zigpy_device) + return zha_device, zigpy_device.endpoints[1].opple_cluster + + +async def test_last_feeding_size_sensor_v2( + hass: HomeAssistant, zigpy_device_aqara_sensor_v2 +) -> None: + """Test quirks defined sensor.""" + + zha_device, cluster = zigpy_device_aqara_sensor_v2 + assert isinstance(zha_device.device, CustomDeviceV2) + entity_id = find_entity_id( + Platform.SENSOR, zha_device, hass, qualifier="last_feeding_size" + ) + assert entity_id is not None + + await send_attributes_report(hass, cluster, {0x010C: 1}) + assert_state(hass, entity_id, "1.0", UnitOfMass.GRAMS) + + await send_attributes_report(hass, cluster, {0x010C: 5}) + assert_state(hass, entity_id, "5.0", UnitOfMass.GRAMS) + + @pytest.fixture async def coordinator(hass: HomeAssistant, zigpy_device_mock, zha_device_joined): """Test ZHA fan platform.""" diff --git a/tests/components/zha/test_switch.py b/tests/components/zha/test_switch.py index 6bfd7e051f1..cb1d87210a7 100644 --- a/tests/components/zha/test_switch.py +++ b/tests/components/zha/test_switch.py @@ -11,7 +11,8 @@ from zhaquirks.const import ( ) from zigpy.exceptions import ZigbeeException import zigpy.profiles.zha as zha -from zigpy.quirks import CustomCluster, CustomDevice +from zigpy.quirks import _DEVICE_REGISTRY, CustomCluster, CustomDevice +from zigpy.quirks.v2 import CustomDeviceV2, add_to_registry_v2 import zigpy.types as t import zigpy.zcl.clusters.closures as closures import zigpy.zcl.clusters.general as general @@ -564,6 +565,272 @@ async def test_switch_configurable( await async_test_rejoin(hass, zigpy_device_tuya, [cluster], (0,)) +async def test_switch_configurable_custom_on_off_values( + hass: HomeAssistant, zha_device_joined_restored, zigpy_device_mock +) -> None: + """Test ZHA configurable switch platform.""" + + zigpy_device = zigpy_device_mock( + { + 1: { + SIG_EP_INPUT: [general.Basic.cluster_id], + SIG_EP_OUTPUT: [], + SIG_EP_TYPE: zha.DeviceType.ON_OFF_SWITCH, + } + }, + manufacturer="manufacturer", + model="model", + ) + + ( + add_to_registry_v2(zigpy_device.manufacturer, zigpy_device.model) + .adds(WindowDetectionFunctionQuirk.TuyaManufCluster) + .switch( + "window_detection_function", + WindowDetectionFunctionQuirk.TuyaManufCluster.cluster_id, + on_value=3, + off_value=5, + ) + ) + + zigpy_device = _DEVICE_REGISTRY.get_device(zigpy_device) + + assert isinstance(zigpy_device, CustomDeviceV2) + cluster = zigpy_device.endpoints[1].tuya_manufacturer + cluster.PLUGGED_ATTR_READS = {"window_detection_function": 5} + update_attribute_cache(cluster) + + zha_device = await zha_device_joined_restored(zigpy_device) + + entity_id = find_entity_id(Platform.SWITCH, zha_device, hass) + assert entity_id is not None + + assert hass.states.get(entity_id).state == STATE_OFF + await async_enable_traffic(hass, [zha_device], enabled=False) + # test that the switch was created and that its state is unavailable + assert hass.states.get(entity_id).state == STATE_UNAVAILABLE + + # allow traffic to flow through the gateway and device + await async_enable_traffic(hass, [zha_device]) + + # test that the state has changed from unavailable to off + assert hass.states.get(entity_id).state == STATE_OFF + + # turn on at switch + await send_attributes_report(hass, cluster, {"window_detection_function": 3}) + assert hass.states.get(entity_id).state == STATE_ON + + # turn off at switch + await send_attributes_report(hass, cluster, {"window_detection_function": 5}) + assert hass.states.get(entity_id).state == STATE_OFF + + # turn on from HA + with patch( + "zigpy.zcl.Cluster.write_attributes", + return_value=[zcl_f.WriteAttributesResponse.deserialize(b"\x00")[0]], + ): + # turn on via UI + await hass.services.async_call( + SWITCH_DOMAIN, "turn_on", {"entity_id": entity_id}, blocking=True + ) + assert cluster.write_attributes.mock_calls == [ + call({"window_detection_function": 3}, manufacturer=None) + ] + cluster.write_attributes.reset_mock() + + # turn off from HA + with patch( + "zigpy.zcl.Cluster.write_attributes", + return_value=[zcl_f.WriteAttributesResponse.deserialize(b"\x00")[0]], + ): + # turn off via UI + await hass.services.async_call( + SWITCH_DOMAIN, "turn_off", {"entity_id": entity_id}, blocking=True + ) + assert cluster.write_attributes.mock_calls == [ + call({"window_detection_function": 5}, manufacturer=None) + ] + + +async def test_switch_configurable_custom_on_off_values_force_inverted( + hass: HomeAssistant, zha_device_joined_restored, zigpy_device_mock +) -> None: + """Test ZHA configurable switch platform.""" + + zigpy_device = zigpy_device_mock( + { + 1: { + SIG_EP_INPUT: [general.Basic.cluster_id], + SIG_EP_OUTPUT: [], + SIG_EP_TYPE: zha.DeviceType.ON_OFF_SWITCH, + } + }, + manufacturer="manufacturer2", + model="model2", + ) + + ( + add_to_registry_v2(zigpy_device.manufacturer, zigpy_device.model) + .adds(WindowDetectionFunctionQuirk.TuyaManufCluster) + .switch( + "window_detection_function", + WindowDetectionFunctionQuirk.TuyaManufCluster.cluster_id, + on_value=3, + off_value=5, + force_inverted=True, + ) + ) + + zigpy_device = _DEVICE_REGISTRY.get_device(zigpy_device) + + assert isinstance(zigpy_device, CustomDeviceV2) + cluster = zigpy_device.endpoints[1].tuya_manufacturer + cluster.PLUGGED_ATTR_READS = {"window_detection_function": 5} + update_attribute_cache(cluster) + + zha_device = await zha_device_joined_restored(zigpy_device) + + entity_id = find_entity_id(Platform.SWITCH, zha_device, hass) + assert entity_id is not None + + assert hass.states.get(entity_id).state == STATE_ON + await async_enable_traffic(hass, [zha_device], enabled=False) + # test that the switch was created and that its state is unavailable + assert hass.states.get(entity_id).state == STATE_UNAVAILABLE + + # allow traffic to flow through the gateway and device + await async_enable_traffic(hass, [zha_device]) + + # test that the state has changed from unavailable to off + assert hass.states.get(entity_id).state == STATE_ON + + # turn on at switch + await send_attributes_report(hass, cluster, {"window_detection_function": 3}) + assert hass.states.get(entity_id).state == STATE_OFF + + # turn off at switch + await send_attributes_report(hass, cluster, {"window_detection_function": 5}) + assert hass.states.get(entity_id).state == STATE_ON + + # turn on from HA + with patch( + "zigpy.zcl.Cluster.write_attributes", + return_value=[zcl_f.WriteAttributesResponse.deserialize(b"\x00")[0]], + ): + # turn on via UI + await hass.services.async_call( + SWITCH_DOMAIN, "turn_on", {"entity_id": entity_id}, blocking=True + ) + assert cluster.write_attributes.mock_calls == [ + call({"window_detection_function": 5}, manufacturer=None) + ] + cluster.write_attributes.reset_mock() + + # turn off from HA + with patch( + "zigpy.zcl.Cluster.write_attributes", + return_value=[zcl_f.WriteAttributesResponse.deserialize(b"\x00")[0]], + ): + # turn off via UI + await hass.services.async_call( + SWITCH_DOMAIN, "turn_off", {"entity_id": entity_id}, blocking=True + ) + assert cluster.write_attributes.mock_calls == [ + call({"window_detection_function": 3}, manufacturer=None) + ] + + +async def test_switch_configurable_custom_on_off_values_inverter_attribute( + hass: HomeAssistant, zha_device_joined_restored, zigpy_device_mock +) -> None: + """Test ZHA configurable switch platform.""" + + zigpy_device = zigpy_device_mock( + { + 1: { + SIG_EP_INPUT: [general.Basic.cluster_id], + SIG_EP_OUTPUT: [], + SIG_EP_TYPE: zha.DeviceType.ON_OFF_SWITCH, + } + }, + manufacturer="manufacturer3", + model="model3", + ) + + ( + add_to_registry_v2(zigpy_device.manufacturer, zigpy_device.model) + .adds(WindowDetectionFunctionQuirk.TuyaManufCluster) + .switch( + "window_detection_function", + WindowDetectionFunctionQuirk.TuyaManufCluster.cluster_id, + on_value=3, + off_value=5, + invert_attribute_name="window_detection_function_inverter", + ) + ) + + zigpy_device = _DEVICE_REGISTRY.get_device(zigpy_device) + + assert isinstance(zigpy_device, CustomDeviceV2) + cluster = zigpy_device.endpoints[1].tuya_manufacturer + cluster.PLUGGED_ATTR_READS = { + "window_detection_function": 5, + "window_detection_function_inverter": t.Bool(True), + } + update_attribute_cache(cluster) + + zha_device = await zha_device_joined_restored(zigpy_device) + + entity_id = find_entity_id(Platform.SWITCH, zha_device, hass) + assert entity_id is not None + + assert hass.states.get(entity_id).state == STATE_ON + await async_enable_traffic(hass, [zha_device], enabled=False) + # test that the switch was created and that its state is unavailable + assert hass.states.get(entity_id).state == STATE_UNAVAILABLE + + # allow traffic to flow through the gateway and device + await async_enable_traffic(hass, [zha_device]) + + # test that the state has changed from unavailable to off + assert hass.states.get(entity_id).state == STATE_ON + + # turn on at switch + await send_attributes_report(hass, cluster, {"window_detection_function": 3}) + assert hass.states.get(entity_id).state == STATE_OFF + + # turn off at switch + await send_attributes_report(hass, cluster, {"window_detection_function": 5}) + assert hass.states.get(entity_id).state == STATE_ON + + # turn on from HA + with patch( + "zigpy.zcl.Cluster.write_attributes", + return_value=[zcl_f.WriteAttributesResponse.deserialize(b"\x00")[0]], + ): + # turn on via UI + await hass.services.async_call( + SWITCH_DOMAIN, "turn_on", {"entity_id": entity_id}, blocking=True + ) + assert cluster.write_attributes.mock_calls == [ + call({"window_detection_function": 5}, manufacturer=None) + ] + cluster.write_attributes.reset_mock() + + # turn off from HA + with patch( + "zigpy.zcl.Cluster.write_attributes", + return_value=[zcl_f.WriteAttributesResponse.deserialize(b"\x00")[0]], + ): + # turn off via UI + await hass.services.async_call( + SWITCH_DOMAIN, "turn_off", {"entity_id": entity_id}, blocking=True + ) + assert cluster.write_attributes.mock_calls == [ + call({"window_detection_function": 3}, manufacturer=None) + ] + + WCAttrs = closures.WindowCovering.AttributeDefs WCT = closures.WindowCovering.WindowCoveringType WCCS = closures.WindowCovering.ConfigStatus