From 74a60929e422ea8df20289f1df72c13c2b563580 Mon Sep 17 00:00:00 2001 From: "David F. Mulcahey" Date: Thu, 25 Jan 2024 08:47:26 -0500 Subject: [PATCH] Use Zigpy definition objects in ZHA cluster handlers (#108383) * use zigpy def objects in ZHA cluster handlers * shorten with direct imports * shorten with rename due to clash --- .../zha/core/cluster_handlers/__init__.py | 2 +- .../zha/core/cluster_handlers/closures.py | 14 +- .../zha/core/cluster_handlers/general.py | 211 +++++++++++++----- .../core/cluster_handlers/homeautomation.py | 130 ++++++++--- .../zha/core/cluster_handlers/hvac.py | 172 +++++++++----- .../zha/core/cluster_handlers/lighting.py | 101 +++++---- .../zha/core/cluster_handlers/measurement.py | 41 ++-- .../zha/core/cluster_handlers/security.py | 61 +++-- .../zha/core/cluster_handlers/smartenergy.py | 69 ++++-- 9 files changed, 538 insertions(+), 263 deletions(-) diff --git a/homeassistant/components/zha/core/cluster_handlers/__init__.py b/homeassistant/components/zha/core/cluster_handlers/__init__.py index 00439343e81..c72d84adecd 100644 --- a/homeassistant/components/zha/core/cluster_handlers/__init__.py +++ b/homeassistant/components/zha/core/cluster_handlers/__init__.py @@ -553,7 +553,7 @@ class ClusterHandler(LogMixin): class ZDOClusterHandler(LogMixin): """Cluster handler for ZDO events.""" - def __init__(self, device): + def __init__(self, device) -> None: """Initialize ZDOClusterHandler.""" self.name = CLUSTER_HANDLER_ZDO self._cluster = device.device.endpoints[0] diff --git a/homeassistant/components/zha/core/cluster_handlers/closures.py b/homeassistant/components/zha/core/cluster_handlers/closures.py index 16c7aef89ad..46fb6d5a538 100644 --- a/homeassistant/components/zha/core/cluster_handlers/closures.py +++ b/homeassistant/components/zha/core/cluster_handlers/closures.py @@ -22,15 +22,23 @@ class DoorLockClusterHandler(ClusterHandler): _value_attribute = 0 REPORT_CONFIG = ( - AttrReportConfig(attr="lock_state", config=REPORT_CONFIG_IMMEDIATE), + AttrReportConfig( + attr=closures.DoorLock.AttributeDefs.lock_state.name, + config=REPORT_CONFIG_IMMEDIATE, + ), ) async def async_update(self): """Retrieve latest state.""" - result = await self.get_attribute_value("lock_state", from_cache=True) + result = await self.get_attribute_value( + closures.DoorLock.AttributeDefs.lock_state.name, from_cache=True + ) if result is not None: self.async_send_signal( - f"{self.unique_id}_{SIGNAL_ATTR_UPDATED}", 0, "lock_state", result + f"{self.unique_id}_{SIGNAL_ATTR_UPDATED}", + closures.DoorLock.AttributeDefs.lock_state.id, + closures.DoorLock.AttributeDefs.lock_state.name, + result, ) @callback diff --git a/homeassistant/components/zha/core/cluster_handlers/general.py b/homeassistant/components/zha/core/cluster_handlers/general.py index 8bc6902b4ff..aee66748461 100644 --- a/homeassistant/components/zha/core/cluster_handlers/general.py +++ b/homeassistant/components/zha/core/cluster_handlers/general.py @@ -50,7 +50,10 @@ class AnalogInput(ClusterHandler): """Analog Input cluster handler.""" REPORT_CONFIG = ( - AttrReportConfig(attr="present_value", config=REPORT_CONFIG_DEFAULT), + AttrReportConfig( + attr=general.AnalogInput.AttributeDefs.present_value.name, + config=REPORT_CONFIG_DEFAULT, + ), ) @@ -60,61 +63,76 @@ class AnalogOutput(ClusterHandler): """Analog Output cluster handler.""" REPORT_CONFIG = ( - AttrReportConfig(attr="present_value", config=REPORT_CONFIG_DEFAULT), + AttrReportConfig( + attr=general.AnalogOutput.AttributeDefs.present_value.name, + config=REPORT_CONFIG_DEFAULT, + ), ) ZCL_INIT_ATTRS = { - "min_present_value": True, - "max_present_value": True, - "resolution": True, - "relinquish_default": True, - "description": True, - "engineering_units": True, - "application_type": True, + general.AnalogOutput.AttributeDefs.min_present_value.name: True, + general.AnalogOutput.AttributeDefs.max_present_value.name: True, + general.AnalogOutput.AttributeDefs.resolution.name: True, + general.AnalogOutput.AttributeDefs.relinquish_default.name: True, + general.AnalogOutput.AttributeDefs.description.name: True, + general.AnalogOutput.AttributeDefs.engineering_units.name: True, + general.AnalogOutput.AttributeDefs.application_type.name: True, } @property def present_value(self) -> float | None: """Return cached value of present_value.""" - return self.cluster.get("present_value") + return self.cluster.get(general.AnalogOutput.AttributeDefs.present_value.name) @property def min_present_value(self) -> float | None: """Return cached value of min_present_value.""" - return self.cluster.get("min_present_value") + return self.cluster.get( + general.AnalogOutput.AttributeDefs.min_present_value.name + ) @property def max_present_value(self) -> float | None: """Return cached value of max_present_value.""" - return self.cluster.get("max_present_value") + return self.cluster.get( + general.AnalogOutput.AttributeDefs.max_present_value.name + ) @property def resolution(self) -> float | None: """Return cached value of resolution.""" - return self.cluster.get("resolution") + return self.cluster.get(general.AnalogOutput.AttributeDefs.resolution.name) @property def relinquish_default(self) -> float | None: """Return cached value of relinquish_default.""" - return self.cluster.get("relinquish_default") + return self.cluster.get( + general.AnalogOutput.AttributeDefs.relinquish_default.name + ) @property def description(self) -> str | None: """Return cached value of description.""" - return self.cluster.get("description") + return self.cluster.get(general.AnalogOutput.AttributeDefs.description.name) @property def engineering_units(self) -> int | None: """Return cached value of engineering_units.""" - return self.cluster.get("engineering_units") + return self.cluster.get( + general.AnalogOutput.AttributeDefs.engineering_units.name + ) @property def application_type(self) -> int | None: """Return cached value of application_type.""" - return self.cluster.get("application_type") + return self.cluster.get( + general.AnalogOutput.AttributeDefs.application_type.name + ) async def async_set_present_value(self, value: float) -> None: """Update present_value.""" - await self.write_attributes_safe({"present_value": value}) + await self.write_attributes_safe( + {general.AnalogOutput.AttributeDefs.present_value.name: value} + ) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.AnalogValue.cluster_id) @@ -122,7 +140,10 @@ class AnalogValue(ClusterHandler): """Analog Value cluster handler.""" REPORT_CONFIG = ( - AttrReportConfig(attr="present_value", config=REPORT_CONFIG_DEFAULT), + AttrReportConfig( + attr=general.AnalogValue.AttributeDefs.present_value.name, + config=REPORT_CONFIG_DEFAULT, + ), ) @@ -171,7 +192,10 @@ class BinaryInput(ClusterHandler): """Binary Input cluster handler.""" REPORT_CONFIG = ( - AttrReportConfig(attr="present_value", config=REPORT_CONFIG_DEFAULT), + AttrReportConfig( + attr=general.BinaryInput.AttributeDefs.present_value.name, + config=REPORT_CONFIG_DEFAULT, + ), ) @@ -180,7 +204,10 @@ class BinaryOutput(ClusterHandler): """Binary Output cluster handler.""" REPORT_CONFIG = ( - AttrReportConfig(attr="present_value", config=REPORT_CONFIG_DEFAULT), + AttrReportConfig( + attr=general.BinaryOutput.AttributeDefs.present_value.name, + config=REPORT_CONFIG_DEFAULT, + ), ) @@ -189,7 +216,10 @@ class BinaryValue(ClusterHandler): """Binary Value cluster handler.""" REPORT_CONFIG = ( - AttrReportConfig(attr="present_value", config=REPORT_CONFIG_DEFAULT), + AttrReportConfig( + attr=general.BinaryValue.AttributeDefs.present_value.name, + config=REPORT_CONFIG_DEFAULT, + ), ) @@ -206,7 +236,7 @@ class DeviceTemperature(ClusterHandler): REPORT_CONFIG = ( { - "attr": "current_temperature", + "attr": general.DeviceTemperature.AttributeDefs.current_temperature.name, "config": (REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 50), }, ) @@ -237,7 +267,7 @@ class Identify(ClusterHandler): """Handle commands received to this cluster.""" cmd = parse_and_log_command(self, tsn, command_id, args) - if cmd == "trigger_effect": + if cmd == general.Identify.ServerCommandDefs.trigger_effect.name: self.async_send_signal(f"{self.unique_id}_{cmd}", args[0]) @@ -252,35 +282,49 @@ class LevelControlClusterHandler(ClusterHandler): """Cluster handler for the LevelControl Zigbee cluster.""" CURRENT_LEVEL = 0 - REPORT_CONFIG = (AttrReportConfig(attr="current_level", config=REPORT_CONFIG_ASAP),) + REPORT_CONFIG = ( + AttrReportConfig( + attr=general.LevelControl.AttributeDefs.current_level.name, + config=REPORT_CONFIG_ASAP, + ), + ) ZCL_INIT_ATTRS = { - "on_off_transition_time": True, - "on_level": True, - "on_transition_time": True, - "off_transition_time": True, - "default_move_rate": True, - "start_up_current_level": True, + general.LevelControl.AttributeDefs.on_off_transition_time.name: True, + general.LevelControl.AttributeDefs.on_level.name: True, + general.LevelControl.AttributeDefs.on_transition_time.name: True, + general.LevelControl.AttributeDefs.off_transition_time.name: True, + general.LevelControl.AttributeDefs.default_move_rate.name: True, + general.LevelControl.AttributeDefs.start_up_current_level.name: True, } @property def current_level(self) -> int | None: """Return cached value of the current_level attribute.""" - return self.cluster.get("current_level") + return self.cluster.get(general.LevelControl.AttributeDefs.current_level.name) @callback def cluster_command(self, tsn, command_id, args): """Handle commands received to this cluster.""" cmd = parse_and_log_command(self, tsn, command_id, args) - if cmd in ("move_to_level", "move_to_level_with_on_off"): + if cmd in ( + general.LevelControl.ServerCommandDefs.move_to_level.name, + general.LevelControl.ServerCommandDefs.move_to_level_with_on_off.name, + ): self.dispatch_level_change(SIGNAL_SET_LEVEL, args[0]) - elif cmd in ("move", "move_with_on_off"): + elif cmd in ( + general.LevelControl.ServerCommandDefs.move.name, + general.LevelControl.ServerCommandDefs.move_with_on_off.name, + ): # We should dim slowly -- for now, just step once rate = args[1] if args[0] == 0xFF: rate = 10 # Should read default move rate self.dispatch_level_change(SIGNAL_MOVE_LEVEL, -rate if args[0] else rate) - elif cmd in ("step", "step_with_on_off"): + elif cmd in ( + general.LevelControl.ServerCommandDefs.step.name, + general.LevelControl.ServerCommandDefs.step_with_on_off.name, + ): # Step (technically may change on/off) self.dispatch_level_change( SIGNAL_MOVE_LEVEL, -args[1] if args[0] else args[1] @@ -303,7 +347,10 @@ class MultistateInput(ClusterHandler): """Multistate Input cluster handler.""" REPORT_CONFIG = ( - AttrReportConfig(attr="present_value", config=REPORT_CONFIG_DEFAULT), + AttrReportConfig( + attr=general.MultistateInput.AttributeDefs.present_value.name, + config=REPORT_CONFIG_DEFAULT, + ), ) @@ -314,7 +361,10 @@ class MultistateOutput(ClusterHandler): """Multistate Output cluster handler.""" REPORT_CONFIG = ( - AttrReportConfig(attr="present_value", config=REPORT_CONFIG_DEFAULT), + AttrReportConfig( + attr=general.MultistateOutput.AttributeDefs.present_value.name, + config=REPORT_CONFIG_DEFAULT, + ), ) @@ -323,7 +373,10 @@ class MultistateValue(ClusterHandler): """Multistate Value cluster handler.""" REPORT_CONFIG = ( - AttrReportConfig(attr="present_value", config=REPORT_CONFIG_DEFAULT), + AttrReportConfig( + attr=general.MultistateValue.AttributeDefs.present_value.name, + config=REPORT_CONFIG_DEFAULT, + ), ) @@ -337,10 +390,13 @@ class OnOffClientClusterHandler(ClientClusterHandler): class OnOffClusterHandler(ClusterHandler): """Cluster handler for the OnOff Zigbee cluster.""" - ON_OFF = general.OnOff.attributes_by_name["on_off"].id - REPORT_CONFIG = (AttrReportConfig(attr="on_off", config=REPORT_CONFIG_IMMEDIATE),) + REPORT_CONFIG = ( + AttrReportConfig( + attr=general.OnOff.AttributeDefs.on_off.name, config=REPORT_CONFIG_IMMEDIATE + ), + ) ZCL_INIT_ATTRS = { - "start_up_on_off": True, + general.OnOff.AttributeDefs.start_up_on_off.name: True, } def __init__(self, cluster: zigpy.zcl.Cluster, endpoint: Endpoint) -> None: @@ -366,32 +422,46 @@ class OnOffClusterHandler(ClusterHandler): @property def on_off(self) -> bool | None: """Return cached value of on/off attribute.""" - return self.cluster.get("on_off") + return self.cluster.get(general.OnOff.AttributeDefs.on_off.name) async def turn_on(self) -> None: """Turn the on off cluster on.""" result = await self.on() if result[1] is not Status.SUCCESS: raise HomeAssistantError(f"Failed to turn on: {result[1]}") - self.cluster.update_attribute(self.ON_OFF, t.Bool.true) + self.cluster.update_attribute( + general.OnOff.AttributeDefs.on_off.id, t.Bool.true + ) async def turn_off(self) -> None: """Turn the on off cluster off.""" result = await self.off() if result[1] is not Status.SUCCESS: raise HomeAssistantError(f"Failed to turn off: {result[1]}") - self.cluster.update_attribute(self.ON_OFF, t.Bool.false) + self.cluster.update_attribute( + general.OnOff.AttributeDefs.on_off.id, t.Bool.false + ) @callback def cluster_command(self, tsn, command_id, args): """Handle commands received to this cluster.""" cmd = parse_and_log_command(self, tsn, command_id, args) - if cmd in ("off", "off_with_effect"): - self.cluster.update_attribute(self.ON_OFF, t.Bool.false) - elif cmd in ("on", "on_with_recall_global_scene"): - self.cluster.update_attribute(self.ON_OFF, t.Bool.true) - elif cmd == "on_with_timed_off": + if cmd in ( + general.OnOff.ServerCommandDefs.off.name, + general.OnOff.ServerCommandDefs.off_with_effect.name, + ): + self.cluster.update_attribute( + general.OnOff.AttributeDefs.on_off.id, t.Bool.false + ) + elif cmd in ( + general.OnOff.ServerCommandDefs.on.name, + general.OnOff.ServerCommandDefs.on_with_recall_global_scene.name, + ): + self.cluster.update_attribute( + general.OnOff.AttributeDefs.on_off.id, t.Bool.true + ) + elif cmd == general.OnOff.ServerCommandDefs.on_with_timed_off.name: should_accept = args[0] on_time = args[1] # 0 is always accept 1 is only accept when already on @@ -399,7 +469,9 @@ class OnOffClusterHandler(ClusterHandler): if self._off_listener is not None: self._off_listener() self._off_listener = None - self.cluster.update_attribute(self.ON_OFF, t.Bool.true) + self.cluster.update_attribute( + general.OnOff.AttributeDefs.on_off.id, t.Bool.true + ) if on_time > 0: self._off_listener = async_call_later( self._endpoint.device.hass, @@ -407,20 +479,27 @@ class OnOffClusterHandler(ClusterHandler): self.set_to_off, ) elif cmd == "toggle": - self.cluster.update_attribute(self.ON_OFF, not bool(self.on_off)) + self.cluster.update_attribute( + general.OnOff.AttributeDefs.on_off.id, not bool(self.on_off) + ) @callback def set_to_off(self, *_): """Set the state to off.""" self._off_listener = None - self.cluster.update_attribute(self.ON_OFF, t.Bool.false) + self.cluster.update_attribute( + general.OnOff.AttributeDefs.on_off.id, t.Bool.false + ) @callback def attribute_updated(self, attrid: int, value: Any, _: Any) -> None: """Handle attribute updates on this cluster.""" - if attrid == self.ON_OFF: + if attrid == general.OnOff.AttributeDefs.on_off.id: self.async_send_signal( - f"{self.unique_id}_{SIGNAL_ATTR_UPDATED}", attrid, "on_off", value + f"{self.unique_id}_{SIGNAL_ATTR_UPDATED}", + attrid, + general.OnOff.AttributeDefs.on_off.name, + value, ) async def async_update(self): @@ -429,7 +508,9 @@ class OnOffClusterHandler(ClusterHandler): return from_cache = not self._endpoint.device.is_mains_powered self.debug("attempting to update onoff state - from cache: %s", from_cache) - await self.get_attribute_value(self.ON_OFF, from_cache=from_cache) + await self.get_attribute_value( + general.OnOff.AttributeDefs.on_off.id, from_cache=from_cache + ) await super().async_update() @@ -482,7 +563,11 @@ class PollControl(ClusterHandler): async def async_configure_cluster_handler_specific(self) -> None: """Configure cluster handler: set check-in interval.""" - await self.write_attributes_safe({"checkin_interval": self.CHECKIN_INTERVAL}) + await self.write_attributes_safe( + { + general.PollControl.AttributeDefs.checkin_interval.name: self.CHECKIN_INTERVAL + } + ) @callback def cluster_command( @@ -496,7 +581,7 @@ class PollControl(ClusterHandler): self.debug("Received %s tsn command '%s': %s", tsn, cmd_name, args) self.zha_send_event(cmd_name, args) - if cmd_name == "checkin": + if cmd_name == general.PollControl.ClientCommandDefs.checkin.name: self.cluster.create_catching_task(self.check_in_response(tsn)) async def check_in_response(self, tsn: int) -> None: @@ -519,17 +604,21 @@ class PowerConfigurationClusterHandler(ClusterHandler): """Cluster handler for the zigbee power configuration cluster.""" REPORT_CONFIG = ( - AttrReportConfig(attr="battery_voltage", config=REPORT_CONFIG_BATTERY_SAVE), AttrReportConfig( - attr="battery_percentage_remaining", config=REPORT_CONFIG_BATTERY_SAVE + attr=general.PowerConfiguration.AttributeDefs.battery_voltage.name, + config=REPORT_CONFIG_BATTERY_SAVE, + ), + AttrReportConfig( + attr=general.PowerConfiguration.AttributeDefs.battery_percentage_remaining.name, + config=REPORT_CONFIG_BATTERY_SAVE, ), ) def async_initialize_cluster_handler_specific(self, from_cache: bool) -> Coroutine: """Initialize cluster handler specific attrs.""" attributes = [ - "battery_size", - "battery_quantity", + general.PowerConfiguration.AttributeDefs.battery_size.name, + general.PowerConfiguration.AttributeDefs.battery_quantity.name, ] return self.get_attributes( attributes, from_cache=from_cache, only_cache=from_cache diff --git a/homeassistant/components/zha/core/cluster_handlers/homeautomation.py b/homeassistant/components/zha/core/cluster_handlers/homeautomation.py index a379db54dac..484ec9f423e 100644 --- a/homeassistant/components/zha/core/cluster_handlers/homeautomation.py +++ b/homeassistant/components/zha/core/cluster_handlers/homeautomation.py @@ -4,6 +4,7 @@ from __future__ import annotations import enum from zigpy.zcl.clusters import homeautomation +from zigpy.zcl.clusters.homeautomation import ElectricalMeasurement from .. import registries from ..const import ( @@ -43,9 +44,7 @@ class Diagnostic(ClusterHandler): """Diagnostic cluster handler.""" -@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( - homeautomation.ElectricalMeasurement.cluster_id -) +@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(ElectricalMeasurement.cluster_id) class ElectricalMeasurementClusterHandler(ClusterHandler): """Cluster handler that polls active power level.""" @@ -65,29 +64,56 @@ class ElectricalMeasurementClusterHandler(ClusterHandler): POWER_QUALITY_MEASUREMENT = 256 REPORT_CONFIG = ( - AttrReportConfig(attr="active_power", config=REPORT_CONFIG_OP), - AttrReportConfig(attr="active_power_max", config=REPORT_CONFIG_DEFAULT), - AttrReportConfig(attr="apparent_power", config=REPORT_CONFIG_OP), - AttrReportConfig(attr="rms_current", config=REPORT_CONFIG_OP), - AttrReportConfig(attr="rms_current_max", config=REPORT_CONFIG_DEFAULT), - AttrReportConfig(attr="rms_voltage", config=REPORT_CONFIG_OP), - AttrReportConfig(attr="rms_voltage_max", config=REPORT_CONFIG_DEFAULT), - AttrReportConfig(attr="ac_frequency", config=REPORT_CONFIG_OP), - AttrReportConfig(attr="ac_frequency_max", config=REPORT_CONFIG_DEFAULT), + AttrReportConfig( + attr=ElectricalMeasurement.AttributeDefs.active_power.name, + config=REPORT_CONFIG_OP, + ), + AttrReportConfig( + attr=ElectricalMeasurement.AttributeDefs.active_power_max.name, + config=REPORT_CONFIG_DEFAULT, + ), + AttrReportConfig( + attr=ElectricalMeasurement.AttributeDefs.apparent_power.name, + config=REPORT_CONFIG_OP, + ), + AttrReportConfig( + attr=ElectricalMeasurement.AttributeDefs.rms_current.name, + config=REPORT_CONFIG_OP, + ), + AttrReportConfig( + attr=ElectricalMeasurement.AttributeDefs.rms_current_max.name, + config=REPORT_CONFIG_DEFAULT, + ), + AttrReportConfig( + attr=ElectricalMeasurement.AttributeDefs.rms_voltage.name, + config=REPORT_CONFIG_OP, + ), + AttrReportConfig( + attr=ElectricalMeasurement.AttributeDefs.rms_voltage_max.name, + config=REPORT_CONFIG_DEFAULT, + ), + AttrReportConfig( + attr=ElectricalMeasurement.AttributeDefs.ac_frequency.name, + config=REPORT_CONFIG_OP, + ), + AttrReportConfig( + attr=ElectricalMeasurement.AttributeDefs.ac_frequency_max.name, + config=REPORT_CONFIG_DEFAULT, + ), ) ZCL_INIT_ATTRS = { - "ac_current_divisor": True, - "ac_current_multiplier": True, - "ac_power_divisor": True, - "ac_power_multiplier": True, - "ac_voltage_divisor": True, - "ac_voltage_multiplier": True, - "ac_frequency_divisor": True, - "ac_frequency_multiplier": True, - "measurement_type": True, - "power_divisor": True, - "power_multiplier": True, - "power_factor": True, + ElectricalMeasurement.AttributeDefs.ac_current_divisor.name: True, + ElectricalMeasurement.AttributeDefs.ac_current_multiplier.name: True, + ElectricalMeasurement.AttributeDefs.ac_power_divisor.name: True, + ElectricalMeasurement.AttributeDefs.ac_power_multiplier.name: True, + ElectricalMeasurement.AttributeDefs.ac_voltage_divisor.name: True, + ElectricalMeasurement.AttributeDefs.ac_voltage_multiplier.name: True, + ElectricalMeasurement.AttributeDefs.ac_frequency_divisor.name: True, + ElectricalMeasurement.AttributeDefs.ac_frequency_multiplier.name: True, + ElectricalMeasurement.AttributeDefs.measurement_type.name: True, + ElectricalMeasurement.AttributeDefs.power_divisor.name: True, + ElectricalMeasurement.AttributeDefs.power_multiplier.name: True, + ElectricalMeasurement.AttributeDefs.power_factor.name: True, } async def async_update(self): @@ -113,51 +139,89 @@ class ElectricalMeasurementClusterHandler(ClusterHandler): @property def ac_current_divisor(self) -> int: """Return ac current divisor.""" - return self.cluster.get("ac_current_divisor") or 1 + return ( + self.cluster.get( + ElectricalMeasurement.AttributeDefs.ac_current_divisor.name + ) + or 1 + ) @property def ac_current_multiplier(self) -> int: """Return ac current multiplier.""" - return self.cluster.get("ac_current_multiplier") or 1 + return ( + self.cluster.get( + ElectricalMeasurement.AttributeDefs.ac_current_multiplier.name + ) + or 1 + ) @property def ac_voltage_divisor(self) -> int: """Return ac voltage divisor.""" - return self.cluster.get("ac_voltage_divisor") or 1 + return ( + self.cluster.get( + ElectricalMeasurement.AttributeDefs.ac_voltage_divisor.name + ) + or 1 + ) @property def ac_voltage_multiplier(self) -> int: """Return ac voltage multiplier.""" - return self.cluster.get("ac_voltage_multiplier") or 1 + return ( + self.cluster.get( + ElectricalMeasurement.AttributeDefs.ac_voltage_multiplier.name + ) + or 1 + ) @property def ac_frequency_divisor(self) -> int: """Return ac frequency divisor.""" - return self.cluster.get("ac_frequency_divisor") or 1 + return ( + self.cluster.get( + ElectricalMeasurement.AttributeDefs.ac_frequency_divisor.name + ) + or 1 + ) @property def ac_frequency_multiplier(self) -> int: """Return ac frequency multiplier.""" - return self.cluster.get("ac_frequency_multiplier") or 1 + return ( + self.cluster.get( + ElectricalMeasurement.AttributeDefs.ac_frequency_multiplier.name + ) + or 1 + ) @property def ac_power_divisor(self) -> int: """Return active power divisor.""" return self.cluster.get( - "ac_power_divisor", self.cluster.get("power_divisor") or 1 + ElectricalMeasurement.AttributeDefs.ac_power_divisor.name, + self.cluster.get(ElectricalMeasurement.AttributeDefs.power_divisor.name) + or 1, ) @property def ac_power_multiplier(self) -> int: """Return active power divisor.""" return self.cluster.get( - "ac_power_multiplier", self.cluster.get("power_multiplier") or 1 + ElectricalMeasurement.AttributeDefs.ac_power_multiplier.name, + self.cluster.get(ElectricalMeasurement.AttributeDefs.power_multiplier.name) + or 1, ) @property def measurement_type(self) -> str | None: """Return Measurement type.""" - if (meas_type := self.cluster.get("measurement_type")) is None: + if ( + meas_type := self.cluster.get( + ElectricalMeasurement.AttributeDefs.measurement_type.name + ) + ) is None: return None meas_type = self.MeasurementType(meas_type) diff --git a/homeassistant/components/zha/core/cluster_handlers/hvac.py b/homeassistant/components/zha/core/cluster_handlers/hvac.py index 5e41785a6d8..f5b70798c2d 100644 --- a/homeassistant/components/zha/core/cluster_handlers/hvac.py +++ b/homeassistant/components/zha/core/cluster_handlers/hvac.py @@ -8,6 +8,7 @@ from __future__ import annotations from typing import Any from zigpy.zcl.clusters import hvac +from zigpy.zcl.clusters.hvac import Fan, Thermostat from homeassistant.core import callback @@ -30,32 +31,36 @@ class Dehumidification(ClusterHandler): """Dehumidification cluster handler.""" -@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(hvac.Fan.cluster_id) +@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Fan.cluster_id) class FanClusterHandler(ClusterHandler): """Fan cluster handler.""" _value_attribute = 0 - REPORT_CONFIG = (AttrReportConfig(attr="fan_mode", config=REPORT_CONFIG_OP),) - ZCL_INIT_ATTRS = {"fan_mode_sequence": True} + REPORT_CONFIG = ( + AttrReportConfig(attr=Fan.AttributeDefs.fan_mode.name, config=REPORT_CONFIG_OP), + ) + ZCL_INIT_ATTRS = {Fan.AttributeDefs.fan_mode_sequence.name: True} @property def fan_mode(self) -> int | None: """Return current fan mode.""" - return self.cluster.get("fan_mode") + return self.cluster.get(Fan.AttributeDefs.fan_mode.name) @property def fan_mode_sequence(self) -> int | None: """Return possible fan mode speeds.""" - return self.cluster.get("fan_mode_sequence") + return self.cluster.get(Fan.AttributeDefs.fan_mode_sequence.name) async def async_set_speed(self, value) -> None: """Set the speed of the fan.""" - await self.write_attributes_safe({"fan_mode": value}) + await self.write_attributes_safe({Fan.AttributeDefs.fan_mode.name: value}) async def async_update(self) -> None: """Retrieve latest state.""" - await self.get_attribute_value("fan_mode", from_cache=False) + await self.get_attribute_value( + Fan.AttributeDefs.fan_mode.name, from_cache=False + ) @callback def attribute_updated(self, attrid: int, value: Any, _: Any) -> None: @@ -75,73 +80,110 @@ class Pump(ClusterHandler): """Pump cluster handler.""" -@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(hvac.Thermostat.cluster_id) +@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Thermostat.cluster_id) class ThermostatClusterHandler(ClusterHandler): """Thermostat cluster handler.""" REPORT_CONFIG = ( - AttrReportConfig(attr="local_temperature", config=REPORT_CONFIG_CLIMATE), AttrReportConfig( - attr="occupied_cooling_setpoint", config=REPORT_CONFIG_CLIMATE + attr=Thermostat.AttributeDefs.local_temperature.name, + config=REPORT_CONFIG_CLIMATE, ), AttrReportConfig( - attr="occupied_heating_setpoint", config=REPORT_CONFIG_CLIMATE + attr=Thermostat.AttributeDefs.occupied_cooling_setpoint.name, + config=REPORT_CONFIG_CLIMATE, ), AttrReportConfig( - attr="unoccupied_cooling_setpoint", config=REPORT_CONFIG_CLIMATE + attr=Thermostat.AttributeDefs.occupied_heating_setpoint.name, + config=REPORT_CONFIG_CLIMATE, ), AttrReportConfig( - attr="unoccupied_heating_setpoint", config=REPORT_CONFIG_CLIMATE + attr=Thermostat.AttributeDefs.unoccupied_cooling_setpoint.name, + config=REPORT_CONFIG_CLIMATE, + ), + AttrReportConfig( + attr=Thermostat.AttributeDefs.unoccupied_heating_setpoint.name, + config=REPORT_CONFIG_CLIMATE, + ), + AttrReportConfig( + attr=Thermostat.AttributeDefs.running_mode.name, + config=REPORT_CONFIG_CLIMATE, + ), + AttrReportConfig( + attr=Thermostat.AttributeDefs.running_state.name, + config=REPORT_CONFIG_CLIMATE_DEMAND, + ), + AttrReportConfig( + attr=Thermostat.AttributeDefs.system_mode.name, + config=REPORT_CONFIG_CLIMATE, + ), + AttrReportConfig( + attr=Thermostat.AttributeDefs.occupancy.name, + config=REPORT_CONFIG_CLIMATE_DISCRETE, + ), + AttrReportConfig( + attr=Thermostat.AttributeDefs.pi_cooling_demand.name, + config=REPORT_CONFIG_CLIMATE_DEMAND, + ), + AttrReportConfig( + attr=Thermostat.AttributeDefs.pi_heating_demand.name, + config=REPORT_CONFIG_CLIMATE_DEMAND, ), - AttrReportConfig(attr="running_mode", config=REPORT_CONFIG_CLIMATE), - AttrReportConfig(attr="running_state", config=REPORT_CONFIG_CLIMATE_DEMAND), - AttrReportConfig(attr="system_mode", config=REPORT_CONFIG_CLIMATE), - AttrReportConfig(attr="occupancy", config=REPORT_CONFIG_CLIMATE_DISCRETE), - AttrReportConfig(attr="pi_cooling_demand", config=REPORT_CONFIG_CLIMATE_DEMAND), - AttrReportConfig(attr="pi_heating_demand", config=REPORT_CONFIG_CLIMATE_DEMAND), ) ZCL_INIT_ATTRS: dict[str, bool] = { - "abs_min_heat_setpoint_limit": True, - "abs_max_heat_setpoint_limit": True, - "abs_min_cool_setpoint_limit": True, - "abs_max_cool_setpoint_limit": True, - "ctrl_sequence_of_oper": False, - "max_cool_setpoint_limit": True, - "max_heat_setpoint_limit": True, - "min_cool_setpoint_limit": True, - "min_heat_setpoint_limit": True, - "local_temperature_calibration": True, + Thermostat.AttributeDefs.abs_min_heat_setpoint_limit.name: True, + Thermostat.AttributeDefs.abs_max_heat_setpoint_limit.name: True, + Thermostat.AttributeDefs.abs_min_cool_setpoint_limit.name: True, + Thermostat.AttributeDefs.abs_max_cool_setpoint_limit.name: True, + Thermostat.AttributeDefs.ctrl_sequence_of_oper.name: False, + Thermostat.AttributeDefs.max_cool_setpoint_limit.name: True, + Thermostat.AttributeDefs.max_heat_setpoint_limit.name: True, + Thermostat.AttributeDefs.min_cool_setpoint_limit.name: True, + Thermostat.AttributeDefs.min_heat_setpoint_limit.name: True, + Thermostat.AttributeDefs.local_temperature_calibration.name: True, } @property def abs_max_cool_setpoint_limit(self) -> int: """Absolute maximum cooling setpoint.""" - return self.cluster.get("abs_max_cool_setpoint_limit", 3200) + return self.cluster.get( + Thermostat.AttributeDefs.abs_max_cool_setpoint_limit.name, 3200 + ) @property def abs_min_cool_setpoint_limit(self) -> int: """Absolute minimum cooling setpoint.""" - return self.cluster.get("abs_min_cool_setpoint_limit", 1600) + return self.cluster.get( + Thermostat.AttributeDefs.abs_min_cool_setpoint_limit.name, 1600 + ) @property def abs_max_heat_setpoint_limit(self) -> int: """Absolute maximum heating setpoint.""" - return self.cluster.get("abs_max_heat_setpoint_limit", 3000) + return self.cluster.get( + Thermostat.AttributeDefs.abs_max_heat_setpoint_limit.name, 3000 + ) @property def abs_min_heat_setpoint_limit(self) -> int: """Absolute minimum heating setpoint.""" - return self.cluster.get("abs_min_heat_setpoint_limit", 700) + return self.cluster.get( + Thermostat.AttributeDefs.abs_min_heat_setpoint_limit.name, 700 + ) @property def ctrl_sequence_of_oper(self) -> int: """Control Sequence of operations attribute.""" - return self.cluster.get("ctrl_sequence_of_oper", 0xFF) + return self.cluster.get( + Thermostat.AttributeDefs.ctrl_sequence_of_oper.name, 0xFF + ) @property def max_cool_setpoint_limit(self) -> int: """Maximum cooling setpoint.""" - sp_limit = self.cluster.get("max_cool_setpoint_limit") + sp_limit = self.cluster.get( + Thermostat.AttributeDefs.max_cool_setpoint_limit.name + ) if sp_limit is None: return self.abs_max_cool_setpoint_limit return sp_limit @@ -149,7 +191,9 @@ class ThermostatClusterHandler(ClusterHandler): @property def min_cool_setpoint_limit(self) -> int: """Minimum cooling setpoint.""" - sp_limit = self.cluster.get("min_cool_setpoint_limit") + sp_limit = self.cluster.get( + Thermostat.AttributeDefs.min_cool_setpoint_limit.name + ) if sp_limit is None: return self.abs_min_cool_setpoint_limit return sp_limit @@ -157,7 +201,9 @@ class ThermostatClusterHandler(ClusterHandler): @property def max_heat_setpoint_limit(self) -> int: """Maximum heating setpoint.""" - sp_limit = self.cluster.get("max_heat_setpoint_limit") + sp_limit = self.cluster.get( + Thermostat.AttributeDefs.max_heat_setpoint_limit.name + ) if sp_limit is None: return self.abs_max_heat_setpoint_limit return sp_limit @@ -165,7 +211,9 @@ class ThermostatClusterHandler(ClusterHandler): @property def min_heat_setpoint_limit(self) -> int: """Minimum heating setpoint.""" - sp_limit = self.cluster.get("min_heat_setpoint_limit") + sp_limit = self.cluster.get( + Thermostat.AttributeDefs.min_heat_setpoint_limit.name + ) if sp_limit is None: return self.abs_min_heat_setpoint_limit return sp_limit @@ -173,57 +221,61 @@ class ThermostatClusterHandler(ClusterHandler): @property def local_temperature(self) -> int | None: """Thermostat temperature.""" - return self.cluster.get("local_temperature") + return self.cluster.get(Thermostat.AttributeDefs.local_temperature.name) @property def occupancy(self) -> int | None: """Is occupancy detected.""" - return self.cluster.get("occupancy") + return self.cluster.get(Thermostat.AttributeDefs.occupancy.name) @property def occupied_cooling_setpoint(self) -> int | None: """Temperature when room is occupied.""" - return self.cluster.get("occupied_cooling_setpoint") + return self.cluster.get(Thermostat.AttributeDefs.occupied_cooling_setpoint.name) @property def occupied_heating_setpoint(self) -> int | None: """Temperature when room is occupied.""" - return self.cluster.get("occupied_heating_setpoint") + return self.cluster.get(Thermostat.AttributeDefs.occupied_heating_setpoint.name) @property def pi_cooling_demand(self) -> int: """Cooling demand.""" - return self.cluster.get("pi_cooling_demand") + return self.cluster.get(Thermostat.AttributeDefs.pi_cooling_demand.name) @property def pi_heating_demand(self) -> int: """Heating demand.""" - return self.cluster.get("pi_heating_demand") + return self.cluster.get(Thermostat.AttributeDefs.pi_heating_demand.name) @property def running_mode(self) -> int | None: """Thermostat running mode.""" - return self.cluster.get("running_mode") + return self.cluster.get(Thermostat.AttributeDefs.running_mode.name) @property def running_state(self) -> int | None: """Thermostat running state, state of heat, cool, fan relays.""" - return self.cluster.get("running_state") + return self.cluster.get(Thermostat.AttributeDefs.running_state.name) @property def system_mode(self) -> int | None: """System mode.""" - return self.cluster.get("system_mode") + return self.cluster.get(Thermostat.AttributeDefs.system_mode.name) @property def unoccupied_cooling_setpoint(self) -> int | None: """Temperature when room is not occupied.""" - return self.cluster.get("unoccupied_cooling_setpoint") + return self.cluster.get( + Thermostat.AttributeDefs.unoccupied_cooling_setpoint.name + ) @property def unoccupied_heating_setpoint(self) -> int | None: """Temperature when room is not occupied.""" - return self.cluster.get("unoccupied_heating_setpoint") + return self.cluster.get( + Thermostat.AttributeDefs.unoccupied_heating_setpoint.name + ) @callback def attribute_updated(self, attrid: int, value: Any, _: Any) -> None: @@ -241,14 +293,20 @@ class ThermostatClusterHandler(ClusterHandler): async def async_set_operation_mode(self, mode) -> bool: """Set Operation mode.""" - await self.write_attributes_safe({"system_mode": mode}) + await self.write_attributes_safe( + {Thermostat.AttributeDefs.system_mode.name: mode} + ) return True async def async_set_heating_setpoint( self, temperature: int, is_away: bool = False ) -> bool: """Set heating setpoint.""" - attr = "unoccupied_heating_setpoint" if is_away else "occupied_heating_setpoint" + attr = ( + Thermostat.AttributeDefs.unoccupied_heating_setpoint.name + if is_away + else Thermostat.AttributeDefs.occupied_heating_setpoint.name + ) await self.write_attributes_safe({attr: temperature}) return True @@ -256,15 +314,21 @@ class ThermostatClusterHandler(ClusterHandler): self, temperature: int, is_away: bool = False ) -> bool: """Set cooling setpoint.""" - attr = "unoccupied_cooling_setpoint" if is_away else "occupied_cooling_setpoint" + attr = ( + Thermostat.AttributeDefs.unoccupied_cooling_setpoint.name + if is_away + else Thermostat.AttributeDefs.occupied_cooling_setpoint.name + ) await self.write_attributes_safe({attr: temperature}) return True async def get_occupancy(self) -> bool | None: """Get unreportable occupancy attribute.""" - res, fail = await self.read_attributes(["occupancy"]) + res, fail = await self.read_attributes( + [Thermostat.AttributeDefs.occupancy.name] + ) self.debug("read 'occupancy' attr, success: %s, fail: %s", res, fail) - if "occupancy" not in res: + if Thermostat.AttributeDefs.occupancy.name not in res: return None return bool(self.occupancy) diff --git a/homeassistant/components/zha/core/cluster_handlers/lighting.py b/homeassistant/components/zha/core/cluster_handlers/lighting.py index 5f1e52fa241..515c2e88d10 100644 --- a/homeassistant/components/zha/core/cluster_handlers/lighting.py +++ b/homeassistant/components/zha/core/cluster_handlers/lighting.py @@ -2,6 +2,7 @@ from __future__ import annotations from zigpy.zcl.clusters import lighting +from zigpy.zcl.clusters.lighting import Color from homeassistant.backports.functools import cached_property @@ -15,88 +16,107 @@ class Ballast(ClusterHandler): """Ballast cluster handler.""" -@registries.CLIENT_CLUSTER_HANDLER_REGISTRY.register(lighting.Color.cluster_id) +@registries.CLIENT_CLUSTER_HANDLER_REGISTRY.register(Color.cluster_id) class ColorClientClusterHandler(ClientClusterHandler): """Color client cluster handler.""" -@registries.BINDABLE_CLUSTERS.register(lighting.Color.cluster_id) -@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(lighting.Color.cluster_id) +@registries.BINDABLE_CLUSTERS.register(Color.cluster_id) +@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Color.cluster_id) class ColorClusterHandler(ClusterHandler): """Color cluster handler.""" REPORT_CONFIG = ( - AttrReportConfig(attr="current_x", config=REPORT_CONFIG_DEFAULT), - AttrReportConfig(attr="current_y", config=REPORT_CONFIG_DEFAULT), - AttrReportConfig(attr="current_hue", config=REPORT_CONFIG_DEFAULT), - AttrReportConfig(attr="current_saturation", config=REPORT_CONFIG_DEFAULT), - AttrReportConfig(attr="color_temperature", config=REPORT_CONFIG_DEFAULT), + AttrReportConfig( + attr=Color.AttributeDefs.current_x.name, + config=REPORT_CONFIG_DEFAULT, + ), + AttrReportConfig( + attr=Color.AttributeDefs.current_y.name, + config=REPORT_CONFIG_DEFAULT, + ), + AttrReportConfig( + attr=Color.AttributeDefs.current_hue.name, + config=REPORT_CONFIG_DEFAULT, + ), + AttrReportConfig( + attr=Color.AttributeDefs.current_saturation.name, + config=REPORT_CONFIG_DEFAULT, + ), + AttrReportConfig( + attr=Color.AttributeDefs.color_temperature.name, + config=REPORT_CONFIG_DEFAULT, + ), ) MAX_MIREDS: int = 500 MIN_MIREDS: int = 153 ZCL_INIT_ATTRS = { - "color_mode": False, - "color_temp_physical_min": True, - "color_temp_physical_max": True, - "color_capabilities": True, - "color_loop_active": False, - "enhanced_current_hue": False, - "start_up_color_temperature": True, - "options": True, + Color.AttributeDefs.color_mode.name: False, + Color.AttributeDefs.color_temp_physical_min.name: True, + Color.AttributeDefs.color_temp_physical_max.name: True, + Color.AttributeDefs.color_capabilities.name: True, + Color.AttributeDefs.color_loop_active.name: False, + Color.AttributeDefs.enhanced_current_hue.name: False, + Color.AttributeDefs.start_up_color_temperature.name: True, + Color.AttributeDefs.options.name: True, } @cached_property - def color_capabilities(self) -> lighting.Color.ColorCapabilities: + def color_capabilities(self) -> Color.ColorCapabilities: """Return ZCL color capabilities of the light.""" - color_capabilities = self.cluster.get("color_capabilities") + color_capabilities = self.cluster.get( + Color.AttributeDefs.color_capabilities.name + ) if color_capabilities is None: - return lighting.Color.ColorCapabilities.XY_attributes - return lighting.Color.ColorCapabilities(color_capabilities) + return Color.ColorCapabilities.XY_attributes + return Color.ColorCapabilities(color_capabilities) @property def color_mode(self) -> int | None: """Return cached value of the color_mode attribute.""" - return self.cluster.get("color_mode") + return self.cluster.get(Color.AttributeDefs.color_mode.name) @property def color_loop_active(self) -> int | None: """Return cached value of the color_loop_active attribute.""" - return self.cluster.get("color_loop_active") + return self.cluster.get(Color.AttributeDefs.color_loop_active.name) @property def color_temperature(self) -> int | None: """Return cached value of color temperature.""" - return self.cluster.get("color_temperature") + return self.cluster.get(Color.AttributeDefs.color_temperature.name) @property def current_x(self) -> int | None: """Return cached value of the current_x attribute.""" - return self.cluster.get("current_x") + return self.cluster.get(Color.AttributeDefs.current_x.name) @property def current_y(self) -> int | None: """Return cached value of the current_y attribute.""" - return self.cluster.get("current_y") + return self.cluster.get(Color.AttributeDefs.current_y.name) @property def current_hue(self) -> int | None: """Return cached value of the current_hue attribute.""" - return self.cluster.get("current_hue") + return self.cluster.get(Color.AttributeDefs.current_hue.name) @property def enhanced_current_hue(self) -> int | None: """Return cached value of the enhanced_current_hue attribute.""" - return self.cluster.get("enhanced_current_hue") + return self.cluster.get(Color.AttributeDefs.enhanced_current_hue.name) @property def current_saturation(self) -> int | None: """Return cached value of the current_saturation attribute.""" - return self.cluster.get("current_saturation") + return self.cluster.get(Color.AttributeDefs.current_saturation.name) @property def min_mireds(self) -> int: """Return the coldest color_temp that this cluster handler supports.""" - min_mireds = self.cluster.get("color_temp_physical_min", self.MIN_MIREDS) + min_mireds = self.cluster.get( + Color.AttributeDefs.color_temp_physical_min.name, self.MIN_MIREDS + ) if min_mireds == 0: self.warning( ( @@ -111,7 +131,9 @@ class ColorClusterHandler(ClusterHandler): @property def max_mireds(self) -> int: """Return the warmest color_temp that this cluster handler supports.""" - max_mireds = self.cluster.get("color_temp_physical_max", self.MAX_MIREDS) + max_mireds = self.cluster.get( + Color.AttributeDefs.color_temp_physical_max.name, self.MAX_MIREDS + ) if max_mireds == 0: self.warning( ( @@ -128,8 +150,7 @@ class ColorClusterHandler(ClusterHandler): """Return True if the cluster handler supports hue and saturation.""" return ( self.color_capabilities is not None - and lighting.Color.ColorCapabilities.Hue_and_saturation - in self.color_capabilities + and Color.ColorCapabilities.Hue_and_saturation in self.color_capabilities ) @property @@ -137,7 +158,7 @@ class ColorClusterHandler(ClusterHandler): """Return True if the cluster handler supports enhanced hue and saturation.""" return ( self.color_capabilities is not None - and lighting.Color.ColorCapabilities.Enhanced_hue in self.color_capabilities + and Color.ColorCapabilities.Enhanced_hue in self.color_capabilities ) @property @@ -145,8 +166,7 @@ class ColorClusterHandler(ClusterHandler): """Return True if the cluster handler supports xy.""" return ( self.color_capabilities is not None - and lighting.Color.ColorCapabilities.XY_attributes - in self.color_capabilities + and Color.ColorCapabilities.XY_attributes in self.color_capabilities ) @property @@ -154,8 +174,7 @@ class ColorClusterHandler(ClusterHandler): """Return True if the cluster handler supports color temperature.""" return ( self.color_capabilities is not None - and lighting.Color.ColorCapabilities.Color_temperature - in self.color_capabilities + and Color.ColorCapabilities.Color_temperature in self.color_capabilities ) or self.color_temperature is not None @property @@ -163,15 +182,15 @@ class ColorClusterHandler(ClusterHandler): """Return True if the cluster handler supports color loop.""" return ( self.color_capabilities is not None - and lighting.Color.ColorCapabilities.Color_loop in self.color_capabilities + and Color.ColorCapabilities.Color_loop in self.color_capabilities ) @property - def options(self) -> lighting.Color.Options: + def options(self) -> Color.Options: """Return ZCL options of the cluster handler.""" - return lighting.Color.Options(self.cluster.get("options", 0)) + return Color.Options(self.cluster.get(Color.AttributeDefs.options.name, 0)) @property def execute_if_off_supported(self) -> bool: """Return True if the cluster handler can execute commands when off.""" - return lighting.Color.Options.Execute_if_off in self.options + return Color.Options.Execute_if_off in self.options diff --git a/homeassistant/components/zha/core/cluster_handlers/measurement.py b/homeassistant/components/zha/core/cluster_handlers/measurement.py index 5249c196864..4df24c32fad 100644 --- a/homeassistant/components/zha/core/cluster_handlers/measurement.py +++ b/homeassistant/components/zha/core/cluster_handlers/measurement.py @@ -27,7 +27,10 @@ class FlowMeasurement(ClusterHandler): """Flow Measurement cluster handler.""" REPORT_CONFIG = ( - AttrReportConfig(attr="measured_value", config=REPORT_CONFIG_DEFAULT), + AttrReportConfig( + attr=measurement.FlowMeasurement.AttributeDefs.measured_value.name, + config=REPORT_CONFIG_DEFAULT, + ), ) @@ -38,7 +41,10 @@ class IlluminanceLevelSensing(ClusterHandler): """Illuminance Level Sensing cluster handler.""" REPORT_CONFIG = ( - AttrReportConfig(attr="level_status", config=REPORT_CONFIG_DEFAULT), + AttrReportConfig( + attr=measurement.IlluminanceLevelSensing.AttributeDefs.level_status.name, + config=REPORT_CONFIG_DEFAULT, + ), ) @@ -49,7 +55,10 @@ class IlluminanceMeasurement(ClusterHandler): """Illuminance Measurement cluster handler.""" REPORT_CONFIG = ( - AttrReportConfig(attr="measured_value", config=REPORT_CONFIG_DEFAULT), + AttrReportConfig( + attr=measurement.IlluminanceMeasurement.AttributeDefs.measured_value.name, + config=REPORT_CONFIG_DEFAULT, + ), ) @@ -60,7 +69,10 @@ class OccupancySensing(ClusterHandler): """Occupancy Sensing cluster handler.""" REPORT_CONFIG = ( - AttrReportConfig(attr="occupancy", config=REPORT_CONFIG_IMMEDIATE), + AttrReportConfig( + attr=measurement.OccupancySensing.AttributeDefs.occupancy.name, + config=REPORT_CONFIG_IMMEDIATE, + ), ) def __init__(self, cluster: zigpy.zcl.Cluster, endpoint: Endpoint) -> None: @@ -82,7 +94,10 @@ class PressureMeasurement(ClusterHandler): """Pressure measurement cluster handler.""" REPORT_CONFIG = ( - AttrReportConfig(attr="measured_value", config=REPORT_CONFIG_DEFAULT), + AttrReportConfig( + attr=measurement.PressureMeasurement.AttributeDefs.measured_value.name, + config=REPORT_CONFIG_DEFAULT, + ), ) @@ -94,7 +109,7 @@ class RelativeHumidity(ClusterHandler): REPORT_CONFIG = ( AttrReportConfig( - attr="measured_value", + attr=measurement.RelativeHumidity.AttributeDefs.measured_value.name, config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 100), ), ) @@ -108,7 +123,7 @@ class SoilMoisture(ClusterHandler): REPORT_CONFIG = ( AttrReportConfig( - attr="measured_value", + attr=measurement.SoilMoisture.AttributeDefs.measured_value.name, config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 100), ), ) @@ -120,7 +135,7 @@ class LeafWetness(ClusterHandler): REPORT_CONFIG = ( AttrReportConfig( - attr="measured_value", + attr=measurement.LeafWetness.AttributeDefs.measured_value.name, config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 100), ), ) @@ -134,7 +149,7 @@ class TemperatureMeasurement(ClusterHandler): REPORT_CONFIG = ( AttrReportConfig( - attr="measured_value", + attr=measurement.TemperatureMeasurement.AttributeDefs.measured_value.name, config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 50), ), ) @@ -148,7 +163,7 @@ class CarbonMonoxideConcentration(ClusterHandler): REPORT_CONFIG = ( AttrReportConfig( - attr="measured_value", + attr=measurement.CarbonMonoxideConcentration.AttributeDefs.measured_value.name, config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 0.000001), ), ) @@ -162,7 +177,7 @@ class CarbonDioxideConcentration(ClusterHandler): REPORT_CONFIG = ( AttrReportConfig( - attr="measured_value", + attr=measurement.CarbonDioxideConcentration.AttributeDefs.measured_value.name, config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 0.000001), ), ) @@ -174,7 +189,7 @@ class PM25(ClusterHandler): REPORT_CONFIG = ( AttrReportConfig( - attr="measured_value", + attr=measurement.PM25.AttributeDefs.measured_value.name, config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 0.1), ), ) @@ -188,7 +203,7 @@ class FormaldehydeConcentration(ClusterHandler): REPORT_CONFIG = ( AttrReportConfig( - attr="measured_value", + attr=measurement.FormaldehydeConcentration.AttributeDefs.measured_value.name, config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 0.000001), ), ) diff --git a/homeassistant/components/zha/core/cluster_handlers/security.py b/homeassistant/components/zha/core/cluster_handlers/security.py index 9c74a14daa8..ac28c5a72da 100644 --- a/homeassistant/components/zha/core/cluster_handlers/security.py +++ b/homeassistant/components/zha/core/cluster_handlers/security.py @@ -29,19 +29,6 @@ from . import ClusterHandler, ClusterHandlerStatus if TYPE_CHECKING: from ..endpoint import Endpoint -IAS_ACE_ARM = 0x0000 # ("arm", (t.enum8, t.CharacterString, t.uint8_t), False), -IAS_ACE_BYPASS = 0x0001 # ("bypass", (t.LVList(t.uint8_t), t.CharacterString), False), -IAS_ACE_EMERGENCY = 0x0002 # ("emergency", (), False), -IAS_ACE_FIRE = 0x0003 # ("fire", (), False), -IAS_ACE_PANIC = 0x0004 # ("panic", (), False), -IAS_ACE_GET_ZONE_ID_MAP = 0x0005 # ("get_zone_id_map", (), False), -IAS_ACE_GET_ZONE_INFO = 0x0006 # ("get_zone_info", (t.uint8_t,), False), -IAS_ACE_GET_PANEL_STATUS = 0x0007 # ("get_panel_status", (), False), -IAS_ACE_GET_BYPASSED_ZONE_LIST = 0x0008 # ("get_bypassed_zone_list", (), False), -IAS_ACE_GET_ZONE_STATUS = ( - 0x0009 # ("get_zone_status", (t.uint8_t, t.uint8_t, t.Bool, t.bitmap16), False) -) -NAME = 0 SIGNAL_ARMED_STATE_CHANGED = "zha_armed_state_changed" SIGNAL_ALARM_TRIGGERED = "zha_armed_triggered" @@ -54,16 +41,16 @@ class IasAce(ClusterHandler): """Initialize IAS Ancillary Control Equipment cluster handler.""" super().__init__(cluster, endpoint) self.command_map: dict[int, Callable[..., Any]] = { - IAS_ACE_ARM: self.arm, - IAS_ACE_BYPASS: self._bypass, - IAS_ACE_EMERGENCY: self._emergency, - IAS_ACE_FIRE: self._fire, - IAS_ACE_PANIC: self._panic, - IAS_ACE_GET_ZONE_ID_MAP: self._get_zone_id_map, - IAS_ACE_GET_ZONE_INFO: self._get_zone_info, - IAS_ACE_GET_PANEL_STATUS: self._send_panel_status_response, - IAS_ACE_GET_BYPASSED_ZONE_LIST: self._get_bypassed_zone_list, - IAS_ACE_GET_ZONE_STATUS: self._get_zone_status, + AceCluster.ServerCommandDefs.arm.id: self.arm, + AceCluster.ServerCommandDefs.bypass.id: self._bypass, + AceCluster.ServerCommandDefs.emergency.id: self._emergency, + AceCluster.ServerCommandDefs.fire.id: self._fire, + AceCluster.ServerCommandDefs.panic.id: self._panic, + AceCluster.ServerCommandDefs.get_zone_id_map.id: self._get_zone_id_map, + AceCluster.ServerCommandDefs.get_zone_info.id: self._get_zone_info, + AceCluster.ServerCommandDefs.get_panel_status.id: self._send_panel_status_response, + AceCluster.ServerCommandDefs.get_bypassed_zone_list.id: self._get_bypassed_zone_list, + AceCluster.ServerCommandDefs.get_zone_status.id: self._get_zone_status, } self.arm_map: dict[AceCluster.ArmMode, Callable[..., Any]] = { AceCluster.ArmMode.Disarm: self._disarm, @@ -95,7 +82,7 @@ class IasAce(ClusterHandler): mode = AceCluster.ArmMode(arm_mode) self.zha_send_event( - self._cluster.server_commands[IAS_ACE_ARM].name, + AceCluster.ServerCommandDefs.arm.name, { "arm_mode": mode.value, "arm_mode_description": mode.name, @@ -191,7 +178,7 @@ class IasAce(ClusterHandler): def _bypass(self, zone_list, code) -> None: """Handle the IAS ACE bypass command.""" self.zha_send_event( - self._cluster.server_commands[IAS_ACE_BYPASS].name, + AceCluster.ServerCommandDefs.bypass.name, {"zone_list": zone_list, "code": code}, ) @@ -336,19 +323,23 @@ class IasWd(ClusterHandler): class IASZoneClusterHandler(ClusterHandler): """Cluster handler for the IASZone Zigbee cluster.""" - ZCL_INIT_ATTRS = {"zone_status": False, "zone_state": True, "zone_type": True} + ZCL_INIT_ATTRS = { + IasZone.AttributeDefs.zone_status.name: False, + IasZone.AttributeDefs.zone_state.name: True, + IasZone.AttributeDefs.zone_type.name: True, + } @callback def cluster_command(self, tsn, command_id, args): """Handle commands received to this cluster.""" - if command_id == 0: + if command_id == IasZone.ClientCommandDefs.status_change_notification.id: zone_status = args[0] # update attribute cache with new zone status self.cluster.update_attribute( - IasZone.attributes_by_name["zone_status"].id, zone_status + IasZone.AttributeDefs.zone_status.id, zone_status ) self.debug("Updated alarm state: %s", zone_status) - elif command_id == 1: + elif command_id == IasZone.ClientCommandDefs.enroll.id: self.debug("Enroll requested") self._cluster.create_catching_task( self.enroll_response( @@ -358,7 +349,9 @@ class IASZoneClusterHandler(ClusterHandler): async def async_configure(self): """Configure IAS device.""" - await self.get_attribute_value("zone_type", from_cache=False) + await self.get_attribute_value( + IasZone.AttributeDefs.zone_type.name, from_cache=False + ) if self._endpoint.device.skip_configuration: self.debug("skipping IASZoneClusterHandler configuration") return @@ -369,7 +362,9 @@ class IASZoneClusterHandler(ClusterHandler): ieee = self.cluster.endpoint.device.application.state.node_info.ieee try: - await self.write_attributes_safe({"cie_addr": ieee}) + await self.write_attributes_safe( + {IasZone.AttributeDefs.cie_addr.name: ieee} + ) self.debug( "wrote cie_addr: %s to '%s' cluster", str(ieee), @@ -396,10 +391,10 @@ class IASZoneClusterHandler(ClusterHandler): @callback def attribute_updated(self, attrid: int, value: Any, _: Any) -> None: """Handle attribute updates on this cluster.""" - if attrid == IasZone.attributes_by_name["zone_status"].id: + if attrid == IasZone.AttributeDefs.zone_status.id: self.async_send_signal( f"{self.unique_id}_{SIGNAL_ATTR_UPDATED}", attrid, - "zone_status", + IasZone.AttributeDefs.zone_status.name, value, ) diff --git a/homeassistant/components/zha/core/cluster_handlers/smartenergy.py b/homeassistant/components/zha/core/cluster_handlers/smartenergy.py index 32e7899d413..d52d62897bc 100644 --- a/homeassistant/components/zha/core/cluster_handlers/smartenergy.py +++ b/homeassistant/components/zha/core/cluster_handlers/smartenergy.py @@ -67,41 +67,62 @@ class Messaging(ClusterHandler): """Messaging cluster handler.""" +SEAttrs = smartenergy.Metering.AttributeDefs + + @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(smartenergy.Metering.cluster_id) class Metering(ClusterHandler): """Metering cluster handler.""" REPORT_CONFIG = ( - AttrReportConfig(attr="instantaneous_demand", config=REPORT_CONFIG_OP), - AttrReportConfig(attr="current_summ_delivered", config=REPORT_CONFIG_DEFAULT), AttrReportConfig( - attr="current_tier1_summ_delivered", config=REPORT_CONFIG_DEFAULT + attr=SEAttrs.instantaneous_demand.name, + config=REPORT_CONFIG_OP, ), AttrReportConfig( - attr="current_tier2_summ_delivered", config=REPORT_CONFIG_DEFAULT + attr=SEAttrs.current_summ_delivered.name, + config=REPORT_CONFIG_DEFAULT, ), AttrReportConfig( - attr="current_tier3_summ_delivered", config=REPORT_CONFIG_DEFAULT + attr=SEAttrs.current_tier1_summ_delivered.name, + config=REPORT_CONFIG_DEFAULT, ), AttrReportConfig( - attr="current_tier4_summ_delivered", config=REPORT_CONFIG_DEFAULT + attr=SEAttrs.current_tier2_summ_delivered.name, + config=REPORT_CONFIG_DEFAULT, ), AttrReportConfig( - attr="current_tier5_summ_delivered", config=REPORT_CONFIG_DEFAULT + attr=SEAttrs.current_tier3_summ_delivered.name, + config=REPORT_CONFIG_DEFAULT, ), AttrReportConfig( - attr="current_tier6_summ_delivered", config=REPORT_CONFIG_DEFAULT + attr=SEAttrs.current_tier4_summ_delivered.name, + config=REPORT_CONFIG_DEFAULT, + ), + AttrReportConfig( + attr=SEAttrs.current_tier5_summ_delivered.name, + config=REPORT_CONFIG_DEFAULT, + ), + AttrReportConfig( + attr=SEAttrs.current_tier6_summ_delivered.name, + config=REPORT_CONFIG_DEFAULT, + ), + AttrReportConfig( + attr=SEAttrs.current_summ_received.name, + config=REPORT_CONFIG_DEFAULT, + ), + AttrReportConfig( + attr=SEAttrs.status.name, + config=REPORT_CONFIG_ASAP, ), - AttrReportConfig(attr="current_summ_received", config=REPORT_CONFIG_DEFAULT), - AttrReportConfig(attr="status", config=REPORT_CONFIG_ASAP), ) ZCL_INIT_ATTRS = { - "demand_formatting": True, - "divisor": True, - "metering_device_type": True, - "multiplier": True, - "summation_formatting": True, - "unit_of_measure": True, + SEAttrs.demand_formatting.name: True, + SEAttrs.divisor.name: True, + SEAttrs.metering_device_type.name: True, + SEAttrs.multiplier.name: True, + SEAttrs.summation_formatting.name: True, + SEAttrs.unit_of_measure.name: True, } metering_device_type = { @@ -153,12 +174,12 @@ class Metering(ClusterHandler): @property def divisor(self) -> int: """Return divisor for the value.""" - return self.cluster.get("divisor") or 1 + return self.cluster.get(SEAttrs.divisor.name) or 1 @property def device_type(self) -> str | int | None: """Return metering device type.""" - dev_type = self.cluster.get("metering_device_type") + dev_type = self.cluster.get(SEAttrs.metering_device_type.name) if dev_type is None: return None return self.metering_device_type.get(dev_type, dev_type) @@ -166,14 +187,14 @@ class Metering(ClusterHandler): @property def multiplier(self) -> int: """Return multiplier for the value.""" - return self.cluster.get("multiplier") or 1 + return self.cluster.get(SEAttrs.multiplier.name) or 1 @property def status(self) -> int | None: """Return metering device status.""" - if (status := self.cluster.get("status")) is None: + if (status := self.cluster.get(SEAttrs.status.name)) is None: return None - if self.cluster.get("metering_device_type") == 0: + if self.cluster.get(SEAttrs.metering_device_type.name) == 0: # Electric metering device type return self.DeviceStatusElectric(status) return self.DeviceStatusDefault(status) @@ -181,18 +202,18 @@ class Metering(ClusterHandler): @property def unit_of_measurement(self) -> int: """Return unit of measurement.""" - return self.cluster.get("unit_of_measure") + return self.cluster.get(SEAttrs.unit_of_measure.name) async def async_initialize_cluster_handler_specific(self, from_cache: bool) -> None: """Fetch config from device and updates format specifier.""" fmting = self.cluster.get( - "demand_formatting", 0xF9 + SEAttrs.demand_formatting.name, 0xF9 ) # 1 digit to the right, 15 digits to the left self._format_spec = self.get_formatting(fmting) fmting = self.cluster.get( - "summation_formatting", 0xF9 + SEAttrs.summation_formatting.name, 0xF9 ) # 1 digit to the right, 15 digits to the left self._summa_format = self.get_formatting(fmting)