Add support for ZHA entities exposed by Zigpy quirks (#111176)

* Add counter entities to the ZHA coordinator device

* rework to prepare for non coordinator device counters

* Initial scaffolding to support quirks v2 entities

* update for zigpy changes

* add assertion error message

* clean up test

* update group entity discovery kwargs

* constants and clearer names

* apply custom device configuration

* quirks switches

* quirks select entities

* quirks sensor entities

* update discovery

* move call to super

* add complex quirks v2 discovery test

* remove duplicate replaces

* add quirks v2 button entity support

* add quirks v2 binary sensor entity support

* fix exception in counter entitiy discovery

* oops

* update formatting

* support custom on and off values

* logging

* don't filter out entities quirks says should be created

* fix type alias warnings

* sync up with zigpy changes and additions

* add a binary sensor test

* button coverage

* switch coverage

* initial select coverage

* number coverage

* sensor coverage

* update discovery after rebase

* coverage

* single line

* line lengths

* fix double underscore

* review comments

* set category from quirks in base entity

* line lengths

* move comment

* imports

* simplify

* simplify
This commit is contained in:
David F. Mulcahey 2024-02-29 10:38:21 -05:00 committed by GitHub
parent f44b759a99
commit 73b6e2bac8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
16 changed files with 1340 additions and 54 deletions

View File

@ -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."""

View File

@ -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."""

View File

@ -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"

View File

@ -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,

View File

@ -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)

View File

@ -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

View File

@ -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."""

View File

@ -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."""

View File

@ -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."""

View File

@ -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

View File

@ -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:

View File

@ -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

View File

@ -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, <ClusterType.Server: 0>)"
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}, <ClusterType.Server: 0>)"
assert f"{m1}{m2}{m3}" in caplog.text
# fmt: off
entity_details = (
"{'cluster_details': (1, 6, <ClusterType.Server: 0>), "
"'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=<EntityPlatform."
"SENSOR: 'sensor'>, entity_type=<EntityType.CONFIG: 'config'>, "
"cluster_id=6, endpoint_id=1, cluster_type=<ClusterType.Server: 0>, "
"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

View File

@ -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"

View File

@ -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."""

View File

@ -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