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
This commit is contained in:
David F. Mulcahey 2024-01-25 08:47:26 -05:00 committed by GitHub
parent 4138b5c308
commit 74a60929e4
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 538 additions and 263 deletions

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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