jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-17 02:07:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Apply consistent naming to ZHA cluster handler implementations (#108851)

Browse Source 
* Apply consistent naming to ZHA cluster handler implentations

* remove import alias

* remove if statement around assert in test
This commit is contained in:
David F. Mulcahey 2024-01-25 20:09:38 -05:00 committed by GitHub
parent eb85f469e9
commit 3f31a76692
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 450 additions and 443 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
homeassistant/components/zha/alarm_control_panel.py
View File

@ -28,7 +28,7 @@ from .core import discovery
from .core.cluster_handlers.security import ( from .core.cluster_handlers.security import (
SIGNAL_ALARM_TRIGGERED, SIGNAL_ALARM_TRIGGERED,
SIGNAL_ARMED_STATE_CHANGED, SIGNAL_ARMED_STATE_CHANGED,
IasAce as AceClusterHandler, IasAceClusterHandler,
) )
from .core.const import ( from .core.const import (
CLUSTER_HANDLER_IAS_ACE, CLUSTER_HANDLER_IAS_ACE,
@ -96,7 +96,7 @@ class ZHAAlarmControlPanel(ZhaEntity, AlarmControlPanelEntity):
"""Initialize the ZHA alarm control device.""" """Initialize the ZHA alarm control device."""
super().__init__(unique_id, zha_device, cluster_handlers, **kwargs) super().__init__(unique_id, zha_device, cluster_handlers, **kwargs)
cfg_entry = zha_device.gateway.config_entry cfg_entry = zha_device.gateway.config_entry
self._cluster_handler: AceClusterHandler = cluster_handlers[0] self._cluster_handler: IasAceClusterHandler = cluster_handlers[0]
self._cluster_handler.panel_code = async_get_zha_config_value( self._cluster_handler.panel_code = async_get_zha_config_value(
cfg_entry, ZHA_ALARM_OPTIONS, CONF_ALARM_MASTER_CODE, "1234" cfg_entry, ZHA_ALARM_OPTIONS, CONF_ALARM_MASTER_CODE, "1234"
) )

38
homeassistant/components/zha/core/cluster_handlers/closures.py
View File

@ -4,7 +4,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
import zigpy.zcl import zigpy.zcl
from zigpy.zcl.clusters import closures from zigpy.zcl.clusters.closures import DoorLock, Shade, WindowCovering
from homeassistant.core import callback from homeassistant.core import callback
@ -16,14 +16,14 @@ if TYPE_CHECKING:
from ..endpoint import Endpoint from ..endpoint import Endpoint
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(closures.DoorLock.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(DoorLock.cluster_id)
class DoorLockClusterHandler(ClusterHandler): class DoorLockClusterHandler(ClusterHandler):
"""Door lock cluster handler.""" """Door lock cluster handler."""
_value_attribute = 0 _value_attribute = 0
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=closures.DoorLock.AttributeDefs.lock_state.name, attr=DoorLock.AttributeDefs.lock_state.name,
config=REPORT_CONFIG_IMMEDIATE, config=REPORT_CONFIG_IMMEDIATE,
), ),
) )
@ -31,13 +31,13 @@ class DoorLockClusterHandler(ClusterHandler):
async def async_update(self): async def async_update(self):
"""Retrieve latest state.""" """Retrieve latest state."""
result = await self.get_attribute_value( result = await self.get_attribute_value(
closures.DoorLock.AttributeDefs.lock_state.name, from_cache=True DoorLock.AttributeDefs.lock_state.name, from_cache=True
) )
if result is not None: if result is not None:
self.async_send_signal( self.async_send_signal(
f"{self.unique_id}_{SIGNAL_ATTR_UPDATED}", f"{self.unique_id}_{SIGNAL_ATTR_UPDATED}",
closures.DoorLock.AttributeDefs.lock_state.id, DoorLock.AttributeDefs.lock_state.id,
closures.DoorLock.AttributeDefs.lock_state.name, DoorLock.AttributeDefs.lock_state.name,
result, result,
) )
@ -80,20 +80,20 @@ class DoorLockClusterHandler(ClusterHandler):
await self.set_pin_code( await self.set_pin_code(
code_slot - 1, # start code slots at 1, Zigbee internals use 0 code_slot - 1, # start code slots at 1, Zigbee internals use 0
closures.DoorLock.UserStatus.Enabled, DoorLock.UserStatus.Enabled,
closures.DoorLock.UserType.Unrestricted, DoorLock.UserType.Unrestricted,
user_code, user_code,
) )
async def async_enable_user_code(self, code_slot: int) -> None: async def async_enable_user_code(self, code_slot: int) -> None:
"""Enable the code slot.""" """Enable the code slot."""
await self.set_user_status(code_slot - 1, closures.DoorLock.UserStatus.Enabled) await self.set_user_status(code_slot - 1, DoorLock.UserStatus.Enabled)
async def async_disable_user_code(self, code_slot: int) -> None: async def async_disable_user_code(self, code_slot: int) -> None:
"""Disable the code slot.""" """Disable the code slot."""
await self.set_user_status(code_slot - 1, closures.DoorLock.UserStatus.Disabled) await self.set_user_status(code_slot - 1, DoorLock.UserStatus.Disabled)
async def async_get_user_code(self, code_slot: int) -> int: async def async_get_user_code(self, code_slot: int) -> int:
"""Get the user code from the code slot.""" """Get the user code from the code slot."""
@ -123,26 +123,26 @@ class DoorLockClusterHandler(ClusterHandler):
return result return result
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(closures.Shade.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Shade.cluster_id)
class Shade(ClusterHandler): class ShadeClusterHandler(ClusterHandler):
"""Shade cluster handler.""" """Shade cluster handler."""
@registries.CLIENT_CLUSTER_HANDLER_REGISTRY.register(closures.WindowCovering.cluster_id) @registries.CLIENT_CLUSTER_HANDLER_REGISTRY.register(WindowCovering.cluster_id)
class WindowCoveringClient(ClientClusterHandler): class WindowCoveringClientClusterHandler(ClientClusterHandler):
"""Window client cluster handler.""" """Window client cluster handler."""
@registries.BINDABLE_CLUSTERS.register(closures.WindowCovering.cluster_id) @registries.BINDABLE_CLUSTERS.register(WindowCovering.cluster_id)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(closures.WindowCovering.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(WindowCovering.cluster_id)
class WindowCovering(ClusterHandler): class WindowCoveringClusterHandler(ClusterHandler):
"""Window cluster handler.""" """Window cluster handler."""
_value_attribute_lift = ( _value_attribute_lift = (
closures.WindowCovering.AttributeDefs.current_position_lift_percentage.id WindowCovering.AttributeDefs.current_position_lift_percentage.id
) )
_value_attribute_tilt = ( _value_attribute_tilt = (
closures.WindowCovering.AttributeDefs.current_position_tilt_percentage.id WindowCovering.AttributeDefs.current_position_tilt_percentage.id
) )
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(

315
homeassistant/components/zha/core/cluster_handlers/general.py
View File

@ -8,7 +8,36 @@ from zhaquirks.quirk_ids import TUYA_PLUG_ONOFF
import zigpy.exceptions import zigpy.exceptions
import zigpy.types as t import zigpy.types as t
import zigpy.zcl import zigpy.zcl
from zigpy.zcl.clusters import general from zigpy.zcl.clusters.general import (
Alarms,
AnalogInput,
AnalogOutput,
AnalogValue,
ApplianceControl,
Basic,
BinaryInput,
BinaryOutput,
BinaryValue,
Commissioning,
DeviceTemperature,
GreenPowerProxy,
Groups,
Identify,
LevelControl,
MultistateInput,
MultistateOutput,
MultistateValue,
OnOff,
OnOffConfiguration,
Ota,
Partition,
PollControl,
PowerConfiguration,
PowerProfile,
RSSILocation,
Scenes,
Time,
)
from zigpy.zcl.foundation import Status from zigpy.zcl.foundation import Status
from homeassistant.core import callback from homeassistant.core import callback
@ -40,122 +69,110 @@ if TYPE_CHECKING:
from ..endpoint import Endpoint from ..endpoint import Endpoint
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.Alarms.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Alarms.cluster_id)
class Alarms(ClusterHandler): class AlarmsClusterHandler(ClusterHandler):
"""Alarms cluster handler.""" """Alarms cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.AnalogInput.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(AnalogInput.cluster_id)
class AnalogInput(ClusterHandler): class AnalogInputClusterHandler(ClusterHandler):
"""Analog Input cluster handler.""" """Analog Input cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=general.AnalogInput.AttributeDefs.present_value.name, attr=AnalogInput.AttributeDefs.present_value.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
) )
@registries.BINDABLE_CLUSTERS.register(general.AnalogOutput.cluster_id) @registries.BINDABLE_CLUSTERS.register(AnalogOutput.cluster_id)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.AnalogOutput.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(AnalogOutput.cluster_id)
class AnalogOutput(ClusterHandler): class AnalogOutputClusterHandler(ClusterHandler):
"""Analog Output cluster handler.""" """Analog Output cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=general.AnalogOutput.AttributeDefs.present_value.name, attr=AnalogOutput.AttributeDefs.present_value.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
) )
ZCL_INIT_ATTRS = { ZCL_INIT_ATTRS = {
general.AnalogOutput.AttributeDefs.min_present_value.name: True, AnalogOutput.AttributeDefs.min_present_value.name: True,
general.AnalogOutput.AttributeDefs.max_present_value.name: True, AnalogOutput.AttributeDefs.max_present_value.name: True,
general.AnalogOutput.AttributeDefs.resolution.name: True, AnalogOutput.AttributeDefs.resolution.name: True,
general.AnalogOutput.AttributeDefs.relinquish_default.name: True, AnalogOutput.AttributeDefs.relinquish_default.name: True,
general.AnalogOutput.AttributeDefs.description.name: True, AnalogOutput.AttributeDefs.description.name: True,
general.AnalogOutput.AttributeDefs.engineering_units.name: True, AnalogOutput.AttributeDefs.engineering_units.name: True,
general.AnalogOutput.AttributeDefs.application_type.name: True, AnalogOutput.AttributeDefs.application_type.name: True,
} }
@property @property
def present_value(self) -> float | None: def present_value(self) -> float | None:
"""Return cached value of present_value.""" """Return cached value of present_value."""
return self.cluster.get(general.AnalogOutput.AttributeDefs.present_value.name) return self.cluster.get(AnalogOutput.AttributeDefs.present_value.name)
@property @property
def min_present_value(self) -> float | None: def min_present_value(self) -> float | None:
"""Return cached value of min_present_value.""" """Return cached value of min_present_value."""
return self.cluster.get( return self.cluster.get(AnalogOutput.AttributeDefs.min_present_value.name)
general.AnalogOutput.AttributeDefs.min_present_value.name
)
@property @property
def max_present_value(self) -> float | None: def max_present_value(self) -> float | None:
"""Return cached value of max_present_value.""" """Return cached value of max_present_value."""
return self.cluster.get( return self.cluster.get(AnalogOutput.AttributeDefs.max_present_value.name)
general.AnalogOutput.AttributeDefs.max_present_value.name
)
@property @property
def resolution(self) -> float | None: def resolution(self) -> float | None:
"""Return cached value of resolution.""" """Return cached value of resolution."""
return self.cluster.get(general.AnalogOutput.AttributeDefs.resolution.name) return self.cluster.get(AnalogOutput.AttributeDefs.resolution.name)
@property @property
def relinquish_default(self) -> float | None: def relinquish_default(self) -> float | None:
"""Return cached value of relinquish_default.""" """Return cached value of relinquish_default."""
return self.cluster.get( return self.cluster.get(AnalogOutput.AttributeDefs.relinquish_default.name)
general.AnalogOutput.AttributeDefs.relinquish_default.name
)
@property @property
def description(self) -> str | None: def description(self) -> str | None:
"""Return cached value of description.""" """Return cached value of description."""
return self.cluster.get(general.AnalogOutput.AttributeDefs.description.name) return self.cluster.get(AnalogOutput.AttributeDefs.description.name)
@property @property
def engineering_units(self) -> int | None: def engineering_units(self) -> int | None:
"""Return cached value of engineering_units.""" """Return cached value of engineering_units."""
return self.cluster.get( return self.cluster.get(AnalogOutput.AttributeDefs.engineering_units.name)
general.AnalogOutput.AttributeDefs.engineering_units.name
)
@property @property
def application_type(self) -> int | None: def application_type(self) -> int | None:
"""Return cached value of application_type.""" """Return cached value of application_type."""
return self.cluster.get( return self.cluster.get(AnalogOutput.AttributeDefs.application_type.name)
general.AnalogOutput.AttributeDefs.application_type.name
)
async def async_set_present_value(self, value: float) -> None: async def async_set_present_value(self, value: float) -> None:
"""Update present_value.""" """Update present_value."""
await self.write_attributes_safe( await self.write_attributes_safe(
{general.AnalogOutput.AttributeDefs.present_value.name: value} {AnalogOutput.AttributeDefs.present_value.name: value}
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.AnalogValue.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(AnalogValue.cluster_id)
class AnalogValue(ClusterHandler): class AnalogValueClusterHandler(ClusterHandler):
"""Analog Value cluster handler.""" """Analog Value cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=general.AnalogValue.AttributeDefs.present_value.name, attr=AnalogValue.AttributeDefs.present_value.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(ApplianceControl.cluster_id)
general.ApplianceControl.cluster_id class ApplianceControlClusterHandler(ClusterHandler):
)
class ApplianceContorl(ClusterHandler):
"""Appliance Control cluster handler.""" """Appliance Control cluster handler."""
@registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(general.Basic.cluster_id) @registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(Basic.cluster_id)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.Basic.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Basic.cluster_id)
class BasicClusterHandler(ClusterHandler): class BasicClusterHandler(ClusterHandler):
"""Cluster handler to interact with the basic cluster.""" """Cluster handler to interact with the basic cluster."""
@ -187,77 +204,75 @@ class BasicClusterHandler(ClusterHandler):
self.ZCL_INIT_ATTRS["transmit_power"] = True self.ZCL_INIT_ATTRS["transmit_power"] = True
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.BinaryInput.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(BinaryInput.cluster_id)
class BinaryInput(ClusterHandler): class BinaryInputClusterHandler(ClusterHandler):
"""Binary Input cluster handler.""" """Binary Input cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=general.BinaryInput.AttributeDefs.present_value.name, attr=BinaryInput.AttributeDefs.present_value.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.BinaryOutput.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(BinaryOutput.cluster_id)
class BinaryOutput(ClusterHandler): class BinaryOutputClusterHandler(ClusterHandler):
"""Binary Output cluster handler.""" """Binary Output cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=general.BinaryOutput.AttributeDefs.present_value.name, attr=BinaryOutput.AttributeDefs.present_value.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.BinaryValue.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(BinaryValue.cluster_id)
class BinaryValue(ClusterHandler): class BinaryValueClusterHandler(ClusterHandler):
"""Binary Value cluster handler.""" """Binary Value cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=general.BinaryValue.AttributeDefs.present_value.name, attr=BinaryValue.AttributeDefs.present_value.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.Commissioning.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Commissioning.cluster_id)
class Commissioning(ClusterHandler): class CommissioningClusterHandler(ClusterHandler):
"""Commissioning cluster handler.""" """Commissioning cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(DeviceTemperature.cluster_id)
general.DeviceTemperature.cluster_id class DeviceTemperatureClusterHandler(ClusterHandler):
)
class DeviceTemperature(ClusterHandler):
"""Device Temperature cluster handler.""" """Device Temperature cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
{ {
"attr": general.DeviceTemperature.AttributeDefs.current_temperature.name, "attr": DeviceTemperature.AttributeDefs.current_temperature.name,
"config": (REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 50), "config": (REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 50),
}, },
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.GreenPowerProxy.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(GreenPowerProxy.cluster_id)
class GreenPowerProxy(ClusterHandler): class GreenPowerProxyClusterHandler(ClusterHandler):
"""Green Power Proxy cluster handler.""" """Green Power Proxy cluster handler."""
BIND: bool = False BIND: bool = False
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.Groups.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Groups.cluster_id)
class Groups(ClusterHandler): class GroupsClusterHandler(ClusterHandler):
"""Groups cluster handler.""" """Groups cluster handler."""
BIND: bool = False BIND: bool = False
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.Identify.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Identify.cluster_id)
class Identify(ClusterHandler): class IdentifyClusterHandler(ClusterHandler):
"""Identify cluster handler.""" """Identify cluster handler."""
BIND: bool = False BIND: bool = False
@ -267,40 +282,40 @@ class Identify(ClusterHandler):
"""Handle commands received to this cluster.""" """Handle commands received to this cluster."""
cmd = parse_and_log_command(self, tsn, command_id, args) cmd = parse_and_log_command(self, tsn, command_id, args)
if cmd == general.Identify.ServerCommandDefs.trigger_effect.name: if cmd == Identify.ServerCommandDefs.trigger_effect.name:
self.async_send_signal(f"{self.unique_id}_{cmd}", args[0]) self.async_send_signal(f"{self.unique_id}_{cmd}", args[0])
@registries.CLIENT_CLUSTER_HANDLER_REGISTRY.register(general.LevelControl.cluster_id) @registries.CLIENT_CLUSTER_HANDLER_REGISTRY.register(LevelControl.cluster_id)
class LevelControlClientClusterHandler(ClientClusterHandler): class LevelControlClientClusterHandler(ClientClusterHandler):
"""LevelControl client cluster.""" """LevelControl client cluster."""
@registries.BINDABLE_CLUSTERS.register(general.LevelControl.cluster_id) @registries.BINDABLE_CLUSTERS.register(LevelControl.cluster_id)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.LevelControl.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(LevelControl.cluster_id)
class LevelControlClusterHandler(ClusterHandler): class LevelControlClusterHandler(ClusterHandler):
"""Cluster handler for the LevelControl Zigbee cluster.""" """Cluster handler for the LevelControl Zigbee cluster."""
CURRENT_LEVEL = 0 CURRENT_LEVEL = 0
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=general.LevelControl.AttributeDefs.current_level.name, attr=LevelControl.AttributeDefs.current_level.name,
config=REPORT_CONFIG_ASAP, config=REPORT_CONFIG_ASAP,
), ),
) )
ZCL_INIT_ATTRS = { ZCL_INIT_ATTRS = {
general.LevelControl.AttributeDefs.on_off_transition_time.name: True, LevelControl.AttributeDefs.on_off_transition_time.name: True,
general.LevelControl.AttributeDefs.on_level.name: True, LevelControl.AttributeDefs.on_level.name: True,
general.LevelControl.AttributeDefs.on_transition_time.name: True, LevelControl.AttributeDefs.on_transition_time.name: True,
general.LevelControl.AttributeDefs.off_transition_time.name: True, LevelControl.AttributeDefs.off_transition_time.name: True,
general.LevelControl.AttributeDefs.default_move_rate.name: True, LevelControl.AttributeDefs.default_move_rate.name: True,
general.LevelControl.AttributeDefs.start_up_current_level.name: True, LevelControl.AttributeDefs.start_up_current_level.name: True,
} }
@property @property
def current_level(self) -> int | None: def current_level(self) -> int | None:
"""Return cached value of the current_level attribute.""" """Return cached value of the current_level attribute."""
return self.cluster.get(general.LevelControl.AttributeDefs.current_level.name) return self.cluster.get(LevelControl.AttributeDefs.current_level.name)
@callback @callback
def cluster_command(self, tsn, command_id, args): def cluster_command(self, tsn, command_id, args):
@ -308,13 +323,13 @@ class LevelControlClusterHandler(ClusterHandler):
cmd = parse_and_log_command(self, tsn, command_id, args) cmd = parse_and_log_command(self, tsn, command_id, args)
if cmd in ( if cmd in (
general.LevelControl.ServerCommandDefs.move_to_level.name, LevelControl.ServerCommandDefs.move_to_level.name,
general.LevelControl.ServerCommandDefs.move_to_level_with_on_off.name, LevelControl.ServerCommandDefs.move_to_level_with_on_off.name,
): ):
self.dispatch_level_change(SIGNAL_SET_LEVEL, args[0]) self.dispatch_level_change(SIGNAL_SET_LEVEL, args[0])
elif cmd in ( elif cmd in (
general.LevelControl.ServerCommandDefs.move.name, LevelControl.ServerCommandDefs.move.name,
general.LevelControl.ServerCommandDefs.move_with_on_off.name, LevelControl.ServerCommandDefs.move_with_on_off.name,
): ):
# We should dim slowly -- for now, just step once # We should dim slowly -- for now, just step once
rate = args[1] rate = args[1]
@ -322,8 +337,8 @@ class LevelControlClusterHandler(ClusterHandler):
rate = 10 # Should read default move rate rate = 10 # Should read default move rate
self.dispatch_level_change(SIGNAL_MOVE_LEVEL, -rate if args[0] else rate) self.dispatch_level_change(SIGNAL_MOVE_LEVEL, -rate if args[0] else rate)
elif cmd in ( elif cmd in (
general.LevelControl.ServerCommandDefs.step.name, LevelControl.ServerCommandDefs.step.name,
general.LevelControl.ServerCommandDefs.step_with_on_off.name, LevelControl.ServerCommandDefs.step_with_on_off.name,
): ):
# Step (technically may change on/off) # Step (technically may change on/off)
self.dispatch_level_change( self.dispatch_level_change(
@ -342,61 +357,59 @@ class LevelControlClusterHandler(ClusterHandler):
self.async_send_signal(f"{self.unique_id}_{command}", level) self.async_send_signal(f"{self.unique_id}_{command}", level)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.MultistateInput.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(MultistateInput.cluster_id)
class MultistateInput(ClusterHandler): class MultistateInputClusterHandler(ClusterHandler):
"""Multistate Input cluster handler.""" """Multistate Input cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=general.MultistateInput.AttributeDefs.present_value.name, attr=MultistateInput.AttributeDefs.present_value.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(MultistateOutput.cluster_id)
general.MultistateOutput.cluster_id class MultistateOutputClusterHandler(ClusterHandler):
)
class MultistateOutput(ClusterHandler):
"""Multistate Output cluster handler.""" """Multistate Output cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=general.MultistateOutput.AttributeDefs.present_value.name, attr=MultistateOutput.AttributeDefs.present_value.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.MultistateValue.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(MultistateValue.cluster_id)
class MultistateValue(ClusterHandler): class MultistateValueClusterHandler(ClusterHandler):
"""Multistate Value cluster handler.""" """Multistate Value cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=general.MultistateValue.AttributeDefs.present_value.name, attr=MultistateValue.AttributeDefs.present_value.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
) )
@registries.CLIENT_CLUSTER_HANDLER_REGISTRY.register(general.OnOff.cluster_id) @registries.CLIENT_CLUSTER_HANDLER_REGISTRY.register(OnOff.cluster_id)
class OnOffClientClusterHandler(ClientClusterHandler): class OnOffClientClusterHandler(ClientClusterHandler):
"""OnOff client cluster handler.""" """OnOff client cluster handler."""
@registries.BINDABLE_CLUSTERS.register(general.OnOff.cluster_id) @registries.BINDABLE_CLUSTERS.register(OnOff.cluster_id)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.OnOff.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(OnOff.cluster_id)
class OnOffClusterHandler(ClusterHandler): class OnOffClusterHandler(ClusterHandler):
"""Cluster handler for the OnOff Zigbee cluster.""" """Cluster handler for the OnOff Zigbee cluster."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=general.OnOff.AttributeDefs.on_off.name, config=REPORT_CONFIG_IMMEDIATE attr=OnOff.AttributeDefs.on_off.name, config=REPORT_CONFIG_IMMEDIATE
), ),
) )
ZCL_INIT_ATTRS = { ZCL_INIT_ATTRS = {
general.OnOff.AttributeDefs.start_up_on_off.name: True, OnOff.AttributeDefs.start_up_on_off.name: True,
} }
def __init__(self, cluster: zigpy.zcl.Cluster, endpoint: Endpoint) -> None: def __init__(self, cluster: zigpy.zcl.Cluster, endpoint: Endpoint) -> None:
@ -422,25 +435,21 @@ class OnOffClusterHandler(ClusterHandler):
@property @property
def on_off(self) -> bool | None: def on_off(self) -> bool | None:
"""Return cached value of on/off attribute.""" """Return cached value of on/off attribute."""
return self.cluster.get(general.OnOff.AttributeDefs.on_off.name) return self.cluster.get(OnOff.AttributeDefs.on_off.name)
async def turn_on(self) -> None: async def turn_on(self) -> None:
"""Turn the on off cluster on.""" """Turn the on off cluster on."""
result = await self.on() result = await self.on()
if result[1] is not Status.SUCCESS: if result[1] is not Status.SUCCESS:
raise HomeAssistantError(f"Failed to turn on: {result[1]}") raise HomeAssistantError(f"Failed to turn on: {result[1]}")
self.cluster.update_attribute( self.cluster.update_attribute(OnOff.AttributeDefs.on_off.id, t.Bool.true)
general.OnOff.AttributeDefs.on_off.id, t.Bool.true
)
async def turn_off(self) -> None: async def turn_off(self) -> None:
"""Turn the on off cluster off.""" """Turn the on off cluster off."""
result = await self.off() result = await self.off()
if result[1] is not Status.SUCCESS: if result[1] is not Status.SUCCESS:
raise HomeAssistantError(f"Failed to turn off: {result[1]}") raise HomeAssistantError(f"Failed to turn off: {result[1]}")
self.cluster.update_attribute( self.cluster.update_attribute(OnOff.AttributeDefs.on_off.id, t.Bool.false)
general.OnOff.AttributeDefs.on_off.id, t.Bool.false
)
@callback @callback
def cluster_command(self, tsn, command_id, args): def cluster_command(self, tsn, command_id, args):
@ -448,20 +457,16 @@ class OnOffClusterHandler(ClusterHandler):
cmd = parse_and_log_command(self, tsn, command_id, args) cmd = parse_and_log_command(self, tsn, command_id, args)
if cmd in ( if cmd in (
general.OnOff.ServerCommandDefs.off.name, OnOff.ServerCommandDefs.off.name,
general.OnOff.ServerCommandDefs.off_with_effect.name, OnOff.ServerCommandDefs.off_with_effect.name,
): ):
self.cluster.update_attribute( self.cluster.update_attribute(OnOff.AttributeDefs.on_off.id, t.Bool.false)
general.OnOff.AttributeDefs.on_off.id, t.Bool.false
)
elif cmd in ( elif cmd in (
general.OnOff.ServerCommandDefs.on.name, OnOff.ServerCommandDefs.on.name,
general.OnOff.ServerCommandDefs.on_with_recall_global_scene.name, OnOff.ServerCommandDefs.on_with_recall_global_scene.name,
): ):
self.cluster.update_attribute( self.cluster.update_attribute(OnOff.AttributeDefs.on_off.id, t.Bool.true)
general.OnOff.AttributeDefs.on_off.id, t.Bool.true elif cmd == OnOff.ServerCommandDefs.on_with_timed_off.name:
)
elif cmd == general.OnOff.ServerCommandDefs.on_with_timed_off.name:
should_accept = args[0] should_accept = args[0]
on_time = args[1] on_time = args[1]
# 0 is always accept 1 is only accept when already on # 0 is always accept 1 is only accept when already on
@ -470,7 +475,7 @@ class OnOffClusterHandler(ClusterHandler):
self._off_listener() self._off_listener()
self._off_listener = None self._off_listener = None
self.cluster.update_attribute( self.cluster.update_attribute(
general.OnOff.AttributeDefs.on_off.id, t.Bool.true OnOff.AttributeDefs.on_off.id, t.Bool.true
) )
if on_time > 0: if on_time > 0:
self._off_listener = async_call_later( self._off_listener = async_call_later(
@ -480,25 +485,23 @@ class OnOffClusterHandler(ClusterHandler):
) )
elif cmd == "toggle": elif cmd == "toggle":
self.cluster.update_attribute( self.cluster.update_attribute(
general.OnOff.AttributeDefs.on_off.id, not bool(self.on_off) OnOff.AttributeDefs.on_off.id, not bool(self.on_off)
) )
@callback @callback
def set_to_off(self, *_): def set_to_off(self, *_):
"""Set the state to off.""" """Set the state to off."""
self._off_listener = None self._off_listener = None
self.cluster.update_attribute( self.cluster.update_attribute(OnOff.AttributeDefs.on_off.id, t.Bool.false)
general.OnOff.AttributeDefs.on_off.id, t.Bool.false
)
@callback @callback
def attribute_updated(self, attrid: int, value: Any, _: Any) -> None: def attribute_updated(self, attrid: int, value: Any, _: Any) -> None:
"""Handle attribute updates on this cluster.""" """Handle attribute updates on this cluster."""
if attrid == general.OnOff.AttributeDefs.on_off.id: if attrid == OnOff.AttributeDefs.on_off.id:
self.async_send_signal( self.async_send_signal(
f"{self.unique_id}_{SIGNAL_ATTR_UPDATED}", f"{self.unique_id}_{SIGNAL_ATTR_UPDATED}",
attrid, attrid,
general.OnOff.AttributeDefs.on_off.name, OnOff.AttributeDefs.on_off.name,
value, value,
) )
@ -509,21 +512,19 @@ class OnOffClusterHandler(ClusterHandler):
from_cache = not self._endpoint.device.is_mains_powered from_cache = not self._endpoint.device.is_mains_powered
self.debug("attempting to update onoff state - from cache: %s", from_cache) self.debug("attempting to update onoff state - from cache: %s", from_cache)
await self.get_attribute_value( await self.get_attribute_value(
general.OnOff.AttributeDefs.on_off.id, from_cache=from_cache OnOff.AttributeDefs.on_off.id, from_cache=from_cache
) )
await super().async_update() await super().async_update()
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(OnOffConfiguration.cluster_id)
general.OnOffConfiguration.cluster_id class OnOffConfigurationClusterHandler(ClusterHandler):
)
class OnOffConfiguration(ClusterHandler):
"""OnOff Configuration cluster handler.""" """OnOff Configuration cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.Ota.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Ota.cluster_id)
@registries.CLIENT_CLUSTER_HANDLER_REGISTRY.register(general.Ota.cluster_id) @registries.CLIENT_CLUSTER_HANDLER_REGISTRY.register(Ota.cluster_id)
class Ota(ClientClusterHandler): class OtaClientClusterHandler(ClientClusterHandler):
"""OTA cluster handler.""" """OTA cluster handler."""
BIND: bool = False BIND: bool = False
@ -544,14 +545,14 @@ class Ota(ClientClusterHandler):
self.async_send_signal(SIGNAL_UPDATE_DEVICE.format(signal_id), args[3]) self.async_send_signal(SIGNAL_UPDATE_DEVICE.format(signal_id), args[3])
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.Partition.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Partition.cluster_id)
class Partition(ClusterHandler): class PartitionClusterHandler(ClusterHandler):
"""Partition cluster handler.""" """Partition cluster handler."""
@registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(general.PollControl.cluster_id) @registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(PollControl.cluster_id)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.PollControl.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(PollControl.cluster_id)
class PollControl(ClusterHandler): class PollControlClusterHandler(ClusterHandler):
"""Poll Control cluster handler.""" """Poll Control cluster handler."""
CHECKIN_INTERVAL = 55 * 60 * 4 # 55min CHECKIN_INTERVAL = 55 * 60 * 4 # 55min
@ -564,9 +565,7 @@ class PollControl(ClusterHandler):
async def async_configure_cluster_handler_specific(self) -> None: async def async_configure_cluster_handler_specific(self) -> None:
"""Configure cluster handler: set check-in interval.""" """Configure cluster handler: set check-in interval."""
await self.write_attributes_safe( await self.write_attributes_safe(
{ {PollControl.AttributeDefs.checkin_interval.name: self.CHECKIN_INTERVAL}
general.PollControl.AttributeDefs.checkin_interval.name: self.CHECKIN_INTERVAL
}
) )
@callback @callback
@ -581,7 +580,7 @@ class PollControl(ClusterHandler):
self.debug("Received %s tsn command '%s': %s", tsn, cmd_name, args) self.debug("Received %s tsn command '%s': %s", tsn, cmd_name, args)
self.zha_send_event(cmd_name, args) self.zha_send_event(cmd_name, args)
if cmd_name == general.PollControl.ClientCommandDefs.checkin.name: if cmd_name == PollControl.ClientCommandDefs.checkin.name:
self.cluster.create_catching_task(self.check_in_response(tsn)) self.cluster.create_catching_task(self.check_in_response(tsn))
async def check_in_response(self, tsn: int) -> None: async def check_in_response(self, tsn: int) -> None:
@ -597,19 +596,17 @@ class PollControl(ClusterHandler):
self._IGNORED_MANUFACTURER_ID.add(manufacturer_code) self._IGNORED_MANUFACTURER_ID.add(manufacturer_code)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(PowerConfiguration.cluster_id)
general.PowerConfiguration.cluster_id
)
class PowerConfigurationClusterHandler(ClusterHandler): class PowerConfigurationClusterHandler(ClusterHandler):
"""Cluster handler for the zigbee power configuration cluster.""" """Cluster handler for the zigbee power configuration cluster."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=general.PowerConfiguration.AttributeDefs.battery_voltage.name, attr=PowerConfiguration.AttributeDefs.battery_voltage.name,
config=REPORT_CONFIG_BATTERY_SAVE, config=REPORT_CONFIG_BATTERY_SAVE,
), ),
AttrReportConfig( AttrReportConfig(
attr=general.PowerConfiguration.AttributeDefs.battery_percentage_remaining.name, attr=PowerConfiguration.AttributeDefs.battery_percentage_remaining.name,
config=REPORT_CONFIG_BATTERY_SAVE, config=REPORT_CONFIG_BATTERY_SAVE,
), ),
) )
@ -617,34 +614,34 @@ class PowerConfigurationClusterHandler(ClusterHandler):
def async_initialize_cluster_handler_specific(self, from_cache: bool) -> Coroutine: def async_initialize_cluster_handler_specific(self, from_cache: bool) -> Coroutine:
"""Initialize cluster handler specific attrs.""" """Initialize cluster handler specific attrs."""
attributes = [ attributes = [
general.PowerConfiguration.AttributeDefs.battery_size.name, PowerConfiguration.AttributeDefs.battery_size.name,
general.PowerConfiguration.AttributeDefs.battery_quantity.name, PowerConfiguration.AttributeDefs.battery_quantity.name,
] ]
return self.get_attributes( return self.get_attributes(
attributes, from_cache=from_cache, only_cache=from_cache attributes, from_cache=from_cache, only_cache=from_cache
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.PowerProfile.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(PowerProfile.cluster_id)
class PowerProfile(ClusterHandler): class PowerProfileClusterHandler(ClusterHandler):
"""Power Profile cluster handler.""" """Power Profile cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.RSSILocation.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(RSSILocation.cluster_id)
class RSSILocation(ClusterHandler): class RSSILocationClusterHandler(ClusterHandler):
"""RSSI Location cluster handler.""" """RSSI Location cluster handler."""
@registries.CLIENT_CLUSTER_HANDLER_REGISTRY.register(general.Scenes.cluster_id) @registries.CLIENT_CLUSTER_HANDLER_REGISTRY.register(Scenes.cluster_id)
class ScenesClientClusterHandler(ClientClusterHandler): class ScenesClientClusterHandler(ClientClusterHandler):
"""Scenes cluster handler.""" """Scenes cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.Scenes.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Scenes.cluster_id)
class Scenes(ClusterHandler): class ScenesClusterHandler(ClusterHandler):
"""Scenes cluster handler.""" """Scenes cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(general.Time.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Time.cluster_id)
class Time(ClusterHandler): class TimeClusterHandler(ClusterHandler):
"""Time cluster handler.""" """Time cluster handler."""

40
homeassistant/components/zha/core/cluster_handlers/homeautomation.py
View File

@ -3,8 +3,14 @@ from __future__ import annotations
import enum import enum
from zigpy.zcl.clusters import homeautomation from zigpy.zcl.clusters.homeautomation import (
from zigpy.zcl.clusters.homeautomation import ElectricalMeasurement ApplianceEventAlerts,
ApplianceIdentification,
ApplianceStatistics,
Diagnostic,
ElectricalMeasurement,
MeterIdentification,
)
from .. import registries from .. import registries
from ..const import ( from ..const import (
@ -16,31 +22,23 @@ from ..const import (
from . import AttrReportConfig, ClusterHandler from . import AttrReportConfig, ClusterHandler
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(ApplianceEventAlerts.cluster_id)
homeautomation.ApplianceEventAlerts.cluster_id class ApplianceEventAlertsClusterHandler(ClusterHandler):
)
class ApplianceEventAlerts(ClusterHandler):
"""Appliance Event Alerts cluster handler.""" """Appliance Event Alerts cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(ApplianceIdentification.cluster_id)
homeautomation.ApplianceIdentification.cluster_id class ApplianceIdentificationClusterHandler(ClusterHandler):
)
class ApplianceIdentification(ClusterHandler):
"""Appliance Identification cluster handler.""" """Appliance Identification cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(ApplianceStatistics.cluster_id)
homeautomation.ApplianceStatistics.cluster_id class ApplianceStatisticsClusterHandler(ClusterHandler):
)
class ApplianceStatistics(ClusterHandler):
"""Appliance Statistics cluster handler.""" """Appliance Statistics cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Diagnostic.cluster_id)
homeautomation.Diagnostic.cluster_id class DiagnosticClusterHandler(ClusterHandler):
)
class Diagnostic(ClusterHandler):
"""Diagnostic cluster handler.""" """Diagnostic cluster handler."""
@ -232,8 +230,6 @@ class ElectricalMeasurementClusterHandler(ClusterHandler):
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(MeterIdentification.cluster_id)
homeautomation.MeterIdentification.cluster_id class MeterIdentificationClusterHandler(ClusterHandler):
)
class MeterIdentification(ClusterHandler):
"""Metering Identification cluster handler.""" """Metering Identification cluster handler."""

21
homeassistant/components/zha/core/cluster_handlers/hvac.py
View File

@ -7,8 +7,13 @@ from __future__ import annotations
from typing import Any from typing import Any
from zigpy.zcl.clusters import hvac from zigpy.zcl.clusters.hvac import (
from zigpy.zcl.clusters.hvac import Fan, Thermostat Dehumidification,
Fan,
Pump,
Thermostat,
UserInterface,
)
from homeassistant.core import callback from homeassistant.core import callback
@ -26,8 +31,8 @@ REPORT_CONFIG_CLIMATE_DEMAND = (REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 5)
REPORT_CONFIG_CLIMATE_DISCRETE = (REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 1) REPORT_CONFIG_CLIMATE_DISCRETE = (REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 1)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(hvac.Dehumidification.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Dehumidification.cluster_id)
class Dehumidification(ClusterHandler): class DehumidificationClusterHandler(ClusterHandler):
"""Dehumidification cluster handler.""" """Dehumidification cluster handler."""
@ -75,8 +80,8 @@ class FanClusterHandler(ClusterHandler):
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(hvac.Pump.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Pump.cluster_id)
class Pump(ClusterHandler): class PumpClusterHandler(ClusterHandler):
"""Pump cluster handler.""" """Pump cluster handler."""
@ -333,6 +338,6 @@ class ThermostatClusterHandler(ClusterHandler):
return bool(self.occupancy) return bool(self.occupancy)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(hvac.UserInterface.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(UserInterface.cluster_id)
class UserInterface(ClusterHandler): class UserInterfaceClusterHandler(ClusterHandler):
"""User interface (thermostat) cluster handler.""" """User interface (thermostat) cluster handler."""

7
homeassistant/components/zha/core/cluster_handlers/lighting.py
View File

@ -1,8 +1,7 @@
"""Lighting cluster handlers module for Zigbee Home Automation.""" """Lighting cluster handlers module for Zigbee Home Automation."""
from __future__ import annotations from __future__ import annotations
from zigpy.zcl.clusters import lighting from zigpy.zcl.clusters.lighting import Ballast, Color
from zigpy.zcl.clusters.lighting import Color
from homeassistant.backports.functools import cached_property from homeassistant.backports.functools import cached_property
@ -11,8 +10,8 @@ from ..const import REPORT_CONFIG_DEFAULT
from . import AttrReportConfig, ClientClusterHandler, ClusterHandler from . import AttrReportConfig, ClientClusterHandler, ClusterHandler
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(lighting.Ballast.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Ballast.cluster_id)
class Ballast(ClusterHandler): class BallastClusterHandler(ClusterHandler):
"""Ballast cluster handler.""" """Ballast cluster handler."""

8
homeassistant/components/zha/core/cluster_handlers/lightlink.py
View File

@ -2,16 +2,16 @@
import asyncio import asyncio
import zigpy.exceptions import zigpy.exceptions
from zigpy.zcl.clusters import lightlink from zigpy.zcl.clusters.lightlink import LightLink
from zigpy.zcl.foundation import GENERAL_COMMANDS, GeneralCommand from zigpy.zcl.foundation import GENERAL_COMMANDS, GeneralCommand
from .. import registries from .. import registries
from . import ClusterHandler, ClusterHandlerStatus from . import ClusterHandler, ClusterHandlerStatus
@registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(lightlink.LightLink.cluster_id) @registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(LightLink.cluster_id)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(lightlink.LightLink.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(LightLink.cluster_id)
class LightLink(ClusterHandler): class LightLinkClusterHandler(ClusterHandler):
"""Lightlink cluster handler.""" """Lightlink cluster handler."""
BIND: bool = False BIND: bool = False

20
homeassistant/components/zha/core/cluster_handlers/manufacturerspecific.py
View File

@ -25,7 +25,7 @@ from ..const import (
UNKNOWN, UNKNOWN,
) )
from . import AttrReportConfig, ClientClusterHandler, ClusterHandler from . import AttrReportConfig, ClientClusterHandler, ClusterHandler
from .general import MultistateInput from .general import MultistateInputClusterHandler
if TYPE_CHECKING: if TYPE_CHECKING:
from ..endpoint import Endpoint from ..endpoint import Endpoint
@ -36,7 +36,7 @@ _LOGGER = logging.getLogger(__name__)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(
registries.SMARTTHINGS_HUMIDITY_CLUSTER registries.SMARTTHINGS_HUMIDITY_CLUSTER
) )
class SmartThingsHumidity(ClusterHandler): class SmartThingsHumidityClusterHandler(ClusterHandler):
"""Smart Things Humidity cluster handler.""" """Smart Things Humidity cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
@ -49,7 +49,7 @@ class SmartThingsHumidity(ClusterHandler):
@registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(0xFD00) @registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(0xFD00)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(0xFD00) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(0xFD00)
class OsramButton(ClusterHandler): class OsramButtonClusterHandler(ClusterHandler):
"""Osram button cluster handler.""" """Osram button cluster handler."""
REPORT_CONFIG = () REPORT_CONFIG = ()
@ -57,7 +57,7 @@ class OsramButton(ClusterHandler):
@registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(registries.PHILLIPS_REMOTE_CLUSTER) @registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(registries.PHILLIPS_REMOTE_CLUSTER)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(registries.PHILLIPS_REMOTE_CLUSTER) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(registries.PHILLIPS_REMOTE_CLUSTER)
class PhillipsRemote(ClusterHandler): class PhillipsRemoteClusterHandler(ClusterHandler):
"""Phillips remote cluster handler.""" """Phillips remote cluster handler."""
REPORT_CONFIG = () REPORT_CONFIG = ()
@ -84,7 +84,7 @@ class TuyaClusterHandler(ClusterHandler):
@registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(0xFCC0) @registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(0xFCC0)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(0xFCC0) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(0xFCC0)
class OppleRemote(ClusterHandler): class OppleRemoteClusterHandler(ClusterHandler):
"""Opple cluster handler.""" """Opple cluster handler."""
REPORT_CONFIG = () REPORT_CONFIG = ()
@ -173,7 +173,7 @@ class OppleRemote(ClusterHandler):
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(
registries.SMARTTHINGS_ACCELERATION_CLUSTER registries.SMARTTHINGS_ACCELERATION_CLUSTER
) )
class SmartThingsAcceleration(ClusterHandler): class SmartThingsAccelerationClusterHandler(ClusterHandler):
"""Smart Things Acceleration cluster handler.""" """Smart Things Acceleration cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
@ -220,7 +220,7 @@ class SmartThingsAcceleration(ClusterHandler):
@registries.CLIENT_CLUSTER_HANDLER_REGISTRY.register(0xFC31) @registries.CLIENT_CLUSTER_HANDLER_REGISTRY.register(0xFC31)
class InovelliNotificationClusterHandler(ClientClusterHandler): class InovelliNotificationClientClusterHandler(ClientClusterHandler):
"""Inovelli Notification cluster handler.""" """Inovelli Notification cluster handler."""
@callback @callback
@ -412,7 +412,7 @@ class IkeaAirPurifierClusterHandler(ClusterHandler):
@registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(0xFC80) @registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(0xFC80)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(0xFC80) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(0xFC80)
class IkeaRemote(ClusterHandler): class IkeaRemoteClusterHandler(ClusterHandler):
"""Ikea Matter remote cluster handler.""" """Ikea Matter remote cluster handler."""
REPORT_CONFIG = () REPORT_CONFIG = ()
@ -421,13 +421,13 @@ class IkeaRemote(ClusterHandler):
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(
DoorLock.cluster_id, XIAOMI_AQARA_VIBRATION_AQ1 DoorLock.cluster_id, XIAOMI_AQARA_VIBRATION_AQ1
) )
class XiaomiVibrationAQ1ClusterHandler(MultistateInput): class XiaomiVibrationAQ1ClusterHandler(MultistateInputClusterHandler):
"""Xiaomi DoorLock Cluster is in fact a MultiStateInput Cluster.""" """Xiaomi DoorLock Cluster is in fact a MultiStateInput Cluster."""
@registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(0xFC11) @registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(0xFC11)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(0xFC11) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(0xFC11)
class SonoffPresenceSenor(ClusterHandler): class SonoffPresenceSenorClusterHandler(ClusterHandler):
"""SonoffPresenceSensor cluster handler.""" """SonoffPresenceSensor cluster handler."""
ZCL_INIT_ATTRS = {"last_illumination_state": True} ZCL_INIT_ATTRS = {"last_illumination_state": True}

110
homeassistant/components/zha/core/cluster_handlers/measurement.py
View File

@ -4,7 +4,21 @@ from __future__ import annotations
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
import zigpy.zcl import zigpy.zcl
from zigpy.zcl.clusters import measurement from zigpy.zcl.clusters.measurement import (
PM25,
CarbonDioxideConcentration,
CarbonMonoxideConcentration,
FlowMeasurement,
FormaldehydeConcentration,
IlluminanceLevelSensing,
IlluminanceMeasurement,
LeafWetness,
OccupancySensing,
PressureMeasurement,
RelativeHumidity,
SoilMoisture,
TemperatureMeasurement,
)
from .. import registries from .. import registries
from ..const import ( from ..const import (
@ -20,57 +34,49 @@ if TYPE_CHECKING:
from ..endpoint import Endpoint from ..endpoint import Endpoint
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(FlowMeasurement.cluster_id)
measurement.FlowMeasurement.cluster_id class FlowMeasurementClusterHandler(ClusterHandler):
)
class FlowMeasurement(ClusterHandler):
"""Flow Measurement cluster handler.""" """Flow Measurement cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=measurement.FlowMeasurement.AttributeDefs.measured_value.name, attr=FlowMeasurement.AttributeDefs.measured_value.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(IlluminanceLevelSensing.cluster_id)
measurement.IlluminanceLevelSensing.cluster_id class IlluminanceLevelSensingClusterHandler(ClusterHandler):
)
class IlluminanceLevelSensing(ClusterHandler):
"""Illuminance Level Sensing cluster handler.""" """Illuminance Level Sensing cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=measurement.IlluminanceLevelSensing.AttributeDefs.level_status.name, attr=IlluminanceLevelSensing.AttributeDefs.level_status.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(IlluminanceMeasurement.cluster_id)
measurement.IlluminanceMeasurement.cluster_id class IlluminanceMeasurementClusterHandler(ClusterHandler):
)
class IlluminanceMeasurement(ClusterHandler):
"""Illuminance Measurement cluster handler.""" """Illuminance Measurement cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=measurement.IlluminanceMeasurement.AttributeDefs.measured_value.name, attr=IlluminanceMeasurement.AttributeDefs.measured_value.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(OccupancySensing.cluster_id)
measurement.OccupancySensing.cluster_id class OccupancySensingClusterHandler(ClusterHandler):
)
class OccupancySensing(ClusterHandler):
"""Occupancy Sensing cluster handler.""" """Occupancy Sensing cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=measurement.OccupancySensing.AttributeDefs.occupancy.name, attr=OccupancySensing.AttributeDefs.occupancy.name,
config=REPORT_CONFIG_IMMEDIATE, config=REPORT_CONFIG_IMMEDIATE,
), ),
) )
@ -87,123 +93,115 @@ class OccupancySensing(ClusterHandler):
self.ZCL_INIT_ATTRS["ultrasonic_u_to_o_threshold"] = True self.ZCL_INIT_ATTRS["ultrasonic_u_to_o_threshold"] = True
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(PressureMeasurement.cluster_id)
measurement.PressureMeasurement.cluster_id class PressureMeasurementClusterHandler(ClusterHandler):
)
class PressureMeasurement(ClusterHandler):
"""Pressure measurement cluster handler.""" """Pressure measurement cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=measurement.PressureMeasurement.AttributeDefs.measured_value.name, attr=PressureMeasurement.AttributeDefs.measured_value.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(RelativeHumidity.cluster_id)
measurement.RelativeHumidity.cluster_id class RelativeHumidityClusterHandler(ClusterHandler):
)
class RelativeHumidity(ClusterHandler):
"""Relative Humidity measurement cluster handler.""" """Relative Humidity measurement cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=measurement.RelativeHumidity.AttributeDefs.measured_value.name, attr=RelativeHumidity.AttributeDefs.measured_value.name,
config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 100), config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 100),
), ),
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(SoilMoisture.cluster_id)
measurement.SoilMoisture.cluster_id class SoilMoistureClusterHandler(ClusterHandler):
)
class SoilMoisture(ClusterHandler):
"""Soil Moisture measurement cluster handler.""" """Soil Moisture measurement cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=measurement.SoilMoisture.AttributeDefs.measured_value.name, attr=SoilMoisture.AttributeDefs.measured_value.name,
config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 100), config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 100),
), ),
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(measurement.LeafWetness.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(LeafWetness.cluster_id)
class LeafWetness(ClusterHandler): class LeafWetnessClusterHandler(ClusterHandler):
"""Leaf Wetness measurement cluster handler.""" """Leaf Wetness measurement cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=measurement.LeafWetness.AttributeDefs.measured_value.name, attr=LeafWetness.AttributeDefs.measured_value.name,
config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 100), config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 100),
), ),
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(TemperatureMeasurement.cluster_id)
measurement.TemperatureMeasurement.cluster_id class TemperatureMeasurementClusterHandler(ClusterHandler):
)
class TemperatureMeasurement(ClusterHandler):
"""Temperature measurement cluster handler.""" """Temperature measurement cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=measurement.TemperatureMeasurement.AttributeDefs.measured_value.name, attr=TemperatureMeasurement.AttributeDefs.measured_value.name,
config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 50), config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 50),
), ),
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(
measurement.CarbonMonoxideConcentration.cluster_id CarbonMonoxideConcentration.cluster_id
) )
class CarbonMonoxideConcentration(ClusterHandler): class CarbonMonoxideConcentrationClusterHandler(ClusterHandler):
"""Carbon Monoxide measurement cluster handler.""" """Carbon Monoxide measurement cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=measurement.CarbonMonoxideConcentration.AttributeDefs.measured_value.name, attr=CarbonMonoxideConcentration.AttributeDefs.measured_value.name,
config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 0.000001), config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 0.000001),
), ),
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(
measurement.CarbonDioxideConcentration.cluster_id CarbonDioxideConcentration.cluster_id
) )
class CarbonDioxideConcentration(ClusterHandler): class CarbonDioxideConcentrationClusterHandler(ClusterHandler):
"""Carbon Dioxide measurement cluster handler.""" """Carbon Dioxide measurement cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=measurement.CarbonDioxideConcentration.AttributeDefs.measured_value.name, attr=CarbonDioxideConcentration.AttributeDefs.measured_value.name,
config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 0.000001), config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 0.000001),
), ),
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(measurement.PM25.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(PM25.cluster_id)
class PM25(ClusterHandler): class PM25ClusterHandler(ClusterHandler):
"""Particulate Matter 2.5 microns or less measurement cluster handler.""" """Particulate Matter 2.5 microns or less measurement cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=measurement.PM25.AttributeDefs.measured_value.name, attr=PM25.AttributeDefs.measured_value.name,
config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 0.1), config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 0.1),
), ),
) )
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(
measurement.FormaldehydeConcentration.cluster_id FormaldehydeConcentration.cluster_id
) )
class FormaldehydeConcentration(ClusterHandler): class FormaldehydeConcentrationClusterHandler(ClusterHandler):
"""Formaldehyde measurement cluster handler.""" """Formaldehyde measurement cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=measurement.FormaldehydeConcentration.AttributeDefs.measured_value.name, attr=FormaldehydeConcentration.AttributeDefs.measured_value.name,
config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 0.000001), config=(REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 0.000001),
), ),
) )

139
homeassistant/components/zha/core/cluster_handlers/protocol.py
View File

@ -1,143 +1,128 @@
"""Protocol cluster handlers module for Zigbee Home Automation.""" """Protocol cluster handlers module for Zigbee Home Automation."""
from zigpy.zcl.clusters import protocol from zigpy.zcl.clusters.protocol import (
AnalogInputExtended,
AnalogInputRegular,
AnalogOutputExtended,
AnalogOutputRegular,
AnalogValueExtended,
AnalogValueRegular,
BacnetProtocolTunnel,
BinaryInputExtended,
BinaryInputRegular,
BinaryOutputExtended,
BinaryOutputRegular,
BinaryValueExtended,
BinaryValueRegular,
GenericTunnel,
MultistateInputExtended,
MultistateInputRegular,
MultistateOutputExtended,
MultistateOutputRegular,
MultistateValueExtended,
MultistateValueRegular,
)
from .. import registries from .. import registries
from . import ClusterHandler from . import ClusterHandler
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(AnalogInputExtended.cluster_id)
protocol.AnalogInputExtended.cluster_id class AnalogInputExtendedClusterHandler(ClusterHandler):
)
class AnalogInputExtended(ClusterHandler):
"""Analog Input Extended cluster handler.""" """Analog Input Extended cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(AnalogInputRegular.cluster_id)
protocol.AnalogInputRegular.cluster_id class AnalogInputRegularClusterHandler(ClusterHandler):
)
class AnalogInputRegular(ClusterHandler):
"""Analog Input Regular cluster handler.""" """Analog Input Regular cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(AnalogOutputExtended.cluster_id)
protocol.AnalogOutputExtended.cluster_id class AnalogOutputExtendedClusterHandler(ClusterHandler):
)
class AnalogOutputExtended(ClusterHandler):
"""Analog Output Regular cluster handler.""" """Analog Output Regular cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(AnalogOutputRegular.cluster_id)
protocol.AnalogOutputRegular.cluster_id class AnalogOutputRegularClusterHandler(ClusterHandler):
)
class AnalogOutputRegular(ClusterHandler):
"""Analog Output Regular cluster handler.""" """Analog Output Regular cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(AnalogValueExtended.cluster_id)
protocol.AnalogValueExtended.cluster_id class AnalogValueExtendedClusterHandler(ClusterHandler):
)
class AnalogValueExtended(ClusterHandler):
"""Analog Value Extended edition cluster handler.""" """Analog Value Extended edition cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(AnalogValueRegular.cluster_id)
protocol.AnalogValueRegular.cluster_id class AnalogValueRegularClusterHandler(ClusterHandler):
)
class AnalogValueRegular(ClusterHandler):
"""Analog Value Regular cluster handler.""" """Analog Value Regular cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(BacnetProtocolTunnel.cluster_id)
protocol.BacnetProtocolTunnel.cluster_id class BacnetProtocolTunnelClusterHandler(ClusterHandler):
)
class BacnetProtocolTunnel(ClusterHandler):
"""Bacnet Protocol Tunnel cluster handler.""" """Bacnet Protocol Tunnel cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(BinaryInputExtended.cluster_id)
protocol.BinaryInputExtended.cluster_id class BinaryInputExtendedClusterHandler(ClusterHandler):
)
class BinaryInputExtended(ClusterHandler):
"""Binary Input Extended cluster handler.""" """Binary Input Extended cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(BinaryInputRegular.cluster_id)
protocol.BinaryInputRegular.cluster_id class BinaryInputRegularClusterHandler(ClusterHandler):
)
class BinaryInputRegular(ClusterHandler):
"""Binary Input Regular cluster handler.""" """Binary Input Regular cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(BinaryOutputExtended.cluster_id)
protocol.BinaryOutputExtended.cluster_id class BinaryOutputExtendedClusterHandler(ClusterHandler):
)
class BinaryOutputExtended(ClusterHandler):
"""Binary Output Extended cluster handler.""" """Binary Output Extended cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(BinaryOutputRegular.cluster_id)
protocol.BinaryOutputRegular.cluster_id class BinaryOutputRegularClusterHandler(ClusterHandler):
)
class BinaryOutputRegular(ClusterHandler):
"""Binary Output Regular cluster handler.""" """Binary Output Regular cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(BinaryValueExtended.cluster_id)
protocol.BinaryValueExtended.cluster_id class BinaryValueExtendedClusterHandler(ClusterHandler):
)
class BinaryValueExtended(ClusterHandler):
"""Binary Value Extended cluster handler.""" """Binary Value Extended cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(BinaryValueRegular.cluster_id)
protocol.BinaryValueRegular.cluster_id class BinaryValueRegularClusterHandler(ClusterHandler):
)
class BinaryValueRegular(ClusterHandler):
"""Binary Value Regular cluster handler.""" """Binary Value Regular cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(protocol.GenericTunnel.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(GenericTunnel.cluster_id)
class GenericTunnel(ClusterHandler): class GenericTunnelClusterHandler(ClusterHandler):
"""Generic Tunnel cluster handler.""" """Generic Tunnel cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(MultistateInputExtended.cluster_id)
protocol.MultistateInputExtended.cluster_id class MultiStateInputExtendedClusterHandler(ClusterHandler):
)
class MultiStateInputExtended(ClusterHandler):
"""Multistate Input Extended cluster handler.""" """Multistate Input Extended cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(MultistateInputRegular.cluster_id)
protocol.MultistateInputRegular.cluster_id class MultiStateInputRegularClusterHandler(ClusterHandler):
)
class MultiStateInputRegular(ClusterHandler):
"""Multistate Input Regular cluster handler.""" """Multistate Input Regular cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(
protocol.MultistateOutputExtended.cluster_id MultistateOutputExtended.cluster_id
) )
class MultiStateOutputExtended(ClusterHandler): class MultiStateOutputExtendedClusterHandler(ClusterHandler):
"""Multistate Output Extended cluster handler.""" """Multistate Output Extended cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(MultistateOutputRegular.cluster_id)
protocol.MultistateOutputRegular.cluster_id class MultiStateOutputRegularClusterHandler(ClusterHandler):
)
class MultiStateOutputRegular(ClusterHandler):
"""Multistate Output Regular cluster handler.""" """Multistate Output Regular cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(MultistateValueExtended.cluster_id)
protocol.MultistateValueExtended.cluster_id class MultiStateValueExtendedClusterHandler(ClusterHandler):
)
class MultiStateValueExtended(ClusterHandler):
"""Multistate Value Extended cluster handler.""" """Multistate Value Extended cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(MultistateValueRegular.cluster_id)
protocol.MultistateValueRegular.cluster_id class MultiStateValueRegularClusterHandler(ClusterHandler):
)
class MultiStateValueRegular(ClusterHandler):
"""Multistate Value Regular cluster handler.""" """Multistate Value Regular cluster handler."""

41
homeassistant/components/zha/core/cluster_handlers/security.py
View File

@ -9,8 +9,7 @@ from collections.abc import Callable
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
import zigpy.zcl import zigpy.zcl
from zigpy.zcl.clusters import security from zigpy.zcl.clusters.security import IasAce as AceCluster, IasWd, IasZone
from zigpy.zcl.clusters.security import IasAce as AceCluster, IasZone
from homeassistant.core import callback from homeassistant.core import callback
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
@ -34,7 +33,7 @@ SIGNAL_ALARM_TRIGGERED = "zha_armed_triggered"
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(AceCluster.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(AceCluster.cluster_id)
class IasAce(ClusterHandler): class IasAceClusterHandler(ClusterHandler):
"""IAS Ancillary Control Equipment cluster handler.""" """IAS Ancillary Control Equipment cluster handler."""
def __init__(self, cluster: zigpy.zcl.Cluster, endpoint: Endpoint) -> None: def __init__(self, cluster: zigpy.zcl.Cluster, endpoint: Endpoint) -> None:
@ -236,16 +235,16 @@ class IasAce(ClusterHandler):
"""Handle the IAS ACE zone status command.""" """Handle the IAS ACE zone status command."""
@registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(security.IasWd.cluster_id) @registries.CLUSTER_HANDLER_ONLY_CLUSTERS.register(IasWd.cluster_id)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(security.IasWd.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(IasWd.cluster_id)
class IasWd(ClusterHandler): class IasWdClusterHandler(ClusterHandler):
"""IAS Warning Device cluster handler.""" """IAS Warning Device cluster handler."""
@staticmethod @staticmethod
def set_bit(destination_value, destination_bit, source_value, source_bit): def set_bit(destination_value, destination_bit, source_value, source_bit):
"""Set the specified bit in the value.""" """Set the specified bit in the value."""
if IasWd.get_bit(source_value, source_bit): if IasWdClusterHandler.get_bit(source_value, source_bit):
return destination_value | (1 << destination_bit) return destination_value | (1 << destination_bit)
return destination_value return destination_value
@ -267,15 +266,15 @@ class IasWd(ClusterHandler):
is currently active (warning in progress). is currently active (warning in progress).
""" """
value = 0 value = 0
value = IasWd.set_bit(value, 0, squawk_level, 0) value = IasWdClusterHandler.set_bit(value, 0, squawk_level, 0)
value = IasWd.set_bit(value, 1, squawk_level, 1) value = IasWdClusterHandler.set_bit(value, 1, squawk_level, 1)
value = IasWd.set_bit(value, 3, strobe, 0) value = IasWdClusterHandler.set_bit(value, 3, strobe, 0)
value = IasWd.set_bit(value, 4, mode, 0) value = IasWdClusterHandler.set_bit(value, 4, mode, 0)
value = IasWd.set_bit(value, 5, mode, 1) value = IasWdClusterHandler.set_bit(value, 5, mode, 1)
value = IasWd.set_bit(value, 6, mode, 2) value = IasWdClusterHandler.set_bit(value, 6, mode, 2)
value = IasWd.set_bit(value, 7, mode, 3) value = IasWdClusterHandler.set_bit(value, 7, mode, 3)
await self.squawk(value) await self.squawk(value)
@ -304,15 +303,15 @@ class IasWd(ClusterHandler):
and then turn OFF for 6/10ths of a second. and then turn OFF for 6/10ths of a second.
""" """
value = 0 value = 0
value = IasWd.set_bit(value, 0, siren_level, 0) value = IasWdClusterHandler.set_bit(value, 0, siren_level, 0)
value = IasWd.set_bit(value, 1, siren_level, 1) value = IasWdClusterHandler.set_bit(value, 1, siren_level, 1)
value = IasWd.set_bit(value, 2, strobe, 0) value = IasWdClusterHandler.set_bit(value, 2, strobe, 0)
value = IasWd.set_bit(value, 4, mode, 0) value = IasWdClusterHandler.set_bit(value, 4, mode, 0)
value = IasWd.set_bit(value, 5, mode, 1) value = IasWdClusterHandler.set_bit(value, 5, mode, 1)
value = IasWd.set_bit(value, 6, mode, 2) value = IasWdClusterHandler.set_bit(value, 6, mode, 2)
value = IasWd.set_bit(value, 7, mode, 3) value = IasWdClusterHandler.set_bit(value, 7, mode, 3)
await self.start_warning( await self.start_warning(
value, warning_duration, strobe_duty_cycle, strobe_intensity value, warning_duration, strobe_duty_cycle, strobe_intensity

120
homeassistant/components/zha/core/cluster_handlers/smartenergy.py
View File

@ -6,7 +6,20 @@ from functools import partialmethod
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
import zigpy.zcl import zigpy.zcl
from zigpy.zcl.clusters import smartenergy from zigpy.zcl.clusters.smartenergy import (
Calendar,
DeviceManagement,
Drlc,
EnergyManagement,
Events,
KeyEstablishment,
MduPairing,
Messaging,
Metering,
Prepayment,
Price,
Tunneling,
)
from .. import registries from .. import registries
from ..const import ( from ..const import (
@ -21,108 +34,99 @@ if TYPE_CHECKING:
from ..endpoint import Endpoint from ..endpoint import Endpoint
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(smartenergy.Calendar.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Calendar.cluster_id)
class Calendar(ClusterHandler): class CalendarClusterHandler(ClusterHandler):
"""Calendar cluster handler.""" """Calendar cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(DeviceManagement.cluster_id)
smartenergy.DeviceManagement.cluster_id class DeviceManagementClusterHandler(ClusterHandler):
)
class DeviceManagement(ClusterHandler):
"""Device Management cluster handler.""" """Device Management cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(smartenergy.Drlc.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Drlc.cluster_id)
class Drlc(ClusterHandler): class DrlcClusterHandler(ClusterHandler):
"""Demand Response and Load Control cluster handler.""" """Demand Response and Load Control cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(EnergyManagement.cluster_id)
smartenergy.EnergyManagement.cluster_id class EnergyManagementClusterHandler(ClusterHandler):
)
class EnergyManagement(ClusterHandler):
"""Energy Management cluster handler.""" """Energy Management cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(smartenergy.Events.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Events.cluster_id)
class Events(ClusterHandler): class EventsClusterHandler(ClusterHandler):
"""Event cluster handler.""" """Event cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register( @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(KeyEstablishment.cluster_id)
smartenergy.KeyEstablishment.cluster_id class KeyEstablishmentClusterHandler(ClusterHandler):
)
class KeyEstablishment(ClusterHandler):
"""Key Establishment cluster handler.""" """Key Establishment cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(smartenergy.MduPairing.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(MduPairing.cluster_id)
class MduPairing(ClusterHandler): class MduPairingClusterHandler(ClusterHandler):
"""Pairing cluster handler.""" """Pairing cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(smartenergy.Messaging.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Messaging.cluster_id)
class Messaging(ClusterHandler): class MessagingClusterHandler(ClusterHandler):
"""Messaging cluster handler.""" """Messaging cluster handler."""
SEAttrs = smartenergy.Metering.AttributeDefs @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Metering.cluster_id)
class MeteringClusterHandler(ClusterHandler):
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(smartenergy.Metering.cluster_id)
class Metering(ClusterHandler):
"""Metering cluster handler.""" """Metering cluster handler."""
REPORT_CONFIG = ( REPORT_CONFIG = (
AttrReportConfig( AttrReportConfig(
attr=SEAttrs.instantaneous_demand.name, attr=Metering.AttributeDefs.instantaneous_demand.name,
config=REPORT_CONFIG_OP, config=REPORT_CONFIG_OP,
), ),
AttrReportConfig( AttrReportConfig(
attr=SEAttrs.current_summ_delivered.name, attr=Metering.AttributeDefs.current_summ_delivered.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
AttrReportConfig( AttrReportConfig(
attr=SEAttrs.current_tier1_summ_delivered.name, attr=Metering.AttributeDefs.current_tier1_summ_delivered.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
AttrReportConfig( AttrReportConfig(
attr=SEAttrs.current_tier2_summ_delivered.name, attr=Metering.AttributeDefs.current_tier2_summ_delivered.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
AttrReportConfig( AttrReportConfig(
attr=SEAttrs.current_tier3_summ_delivered.name, attr=Metering.AttributeDefs.current_tier3_summ_delivered.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
AttrReportConfig( AttrReportConfig(
attr=SEAttrs.current_tier4_summ_delivered.name, attr=Metering.AttributeDefs.current_tier4_summ_delivered.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
AttrReportConfig( AttrReportConfig(
attr=SEAttrs.current_tier5_summ_delivered.name, attr=Metering.AttributeDefs.current_tier5_summ_delivered.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
AttrReportConfig( AttrReportConfig(
attr=SEAttrs.current_tier6_summ_delivered.name, attr=Metering.AttributeDefs.current_tier6_summ_delivered.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
AttrReportConfig( AttrReportConfig(
attr=SEAttrs.current_summ_received.name, attr=Metering.AttributeDefs.current_summ_received.name,
config=REPORT_CONFIG_DEFAULT, config=REPORT_CONFIG_DEFAULT,
), ),
AttrReportConfig( AttrReportConfig(
attr=SEAttrs.status.name, attr=Metering.AttributeDefs.status.name,
config=REPORT_CONFIG_ASAP, config=REPORT_CONFIG_ASAP,
), ),
) )
ZCL_INIT_ATTRS = { ZCL_INIT_ATTRS = {
SEAttrs.demand_formatting.name: True, Metering.AttributeDefs.demand_formatting.name: True,
SEAttrs.divisor.name: True, Metering.AttributeDefs.divisor.name: True,
SEAttrs.metering_device_type.name: True, Metering.AttributeDefs.metering_device_type.name: True,
SEAttrs.multiplier.name: True, Metering.AttributeDefs.multiplier.name: True,
SEAttrs.summation_formatting.name: True, Metering.AttributeDefs.summation_formatting.name: True,
SEAttrs.unit_of_measure.name: True, Metering.AttributeDefs.unit_of_measure.name: True,
} }
metering_device_type = { metering_device_type = {
@ -174,12 +178,12 @@ class Metering(ClusterHandler):
@property @property
def divisor(self) -> int: def divisor(self) -> int:
"""Return divisor for the value.""" """Return divisor for the value."""
return self.cluster.get(SEAttrs.divisor.name) or 1 return self.cluster.get(Metering.AttributeDefs.divisor.name) or 1
@property @property
def device_type(self) -> str | int | None: def device_type(self) -> str | int | None:
"""Return metering device type.""" """Return metering device type."""
dev_type = self.cluster.get(SEAttrs.metering_device_type.name) dev_type = self.cluster.get(Metering.AttributeDefs.metering_device_type.name)
if dev_type is None: if dev_type is None:
return None return None
return self.metering_device_type.get(dev_type, dev_type) return self.metering_device_type.get(dev_type, dev_type)
@ -187,14 +191,14 @@ class Metering(ClusterHandler):
@property @property
def multiplier(self) -> int: def multiplier(self) -> int:
"""Return multiplier for the value.""" """Return multiplier for the value."""
return self.cluster.get(SEAttrs.multiplier.name) or 1 return self.cluster.get(Metering.AttributeDefs.multiplier.name) or 1
@property @property
def status(self) -> int | None: def status(self) -> int | None:
"""Return metering device status.""" """Return metering device status."""
if (status := self.cluster.get(SEAttrs.status.name)) is None: if (status := self.cluster.get(Metering.AttributeDefs.status.name)) is None:
return None return None
if self.cluster.get(SEAttrs.metering_device_type.name) == 0: if self.cluster.get(Metering.AttributeDefs.metering_device_type.name) == 0:
# Electric metering device type # Electric metering device type
return self.DeviceStatusElectric(status) return self.DeviceStatusElectric(status)
return self.DeviceStatusDefault(status) return self.DeviceStatusDefault(status)
@ -202,18 +206,18 @@ class Metering(ClusterHandler):
@property @property
def unit_of_measurement(self) -> int: def unit_of_measurement(self) -> int:
"""Return unit of measurement.""" """Return unit of measurement."""
return self.cluster.get(SEAttrs.unit_of_measure.name) return self.cluster.get(Metering.AttributeDefs.unit_of_measure.name)
async def async_initialize_cluster_handler_specific(self, from_cache: bool) -> None: async def async_initialize_cluster_handler_specific(self, from_cache: bool) -> None:
"""Fetch config from device and updates format specifier.""" """Fetch config from device and updates format specifier."""
fmting = self.cluster.get( fmting = self.cluster.get(
SEAttrs.demand_formatting.name, 0xF9 Metering.AttributeDefs.demand_formatting.name, 0xF9
) # 1 digit to the right, 15 digits to the left ) # 1 digit to the right, 15 digits to the left
self._format_spec = self.get_formatting(fmting) self._format_spec = self.get_formatting(fmting)
fmting = self.cluster.get( fmting = self.cluster.get(
SEAttrs.summation_formatting.name, 0xF9 Metering.AttributeDefs.summation_formatting.name, 0xF9
) # 1 digit to the right, 15 digits to the left ) # 1 digit to the right, 15 digits to the left
self._summa_format = self.get_formatting(fmting) self._summa_format = self.get_formatting(fmting)
@ -277,16 +281,16 @@ class Metering(ClusterHandler):
summa_formatter = partialmethod(_formatter_function, FormatSelector.SUMMATION) summa_formatter = partialmethod(_formatter_function, FormatSelector.SUMMATION)
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(smartenergy.Prepayment.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Prepayment.cluster_id)
class Prepayment(ClusterHandler): class PrepaymentClusterHandler(ClusterHandler):
"""Prepayment cluster handler.""" """Prepayment cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(smartenergy.Price.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Price.cluster_id)
class Price(ClusterHandler): class PriceClusterHandler(ClusterHandler):
"""Price cluster handler.""" """Price cluster handler."""
@registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(smartenergy.Tunneling.cluster_id) @registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.register(Tunneling.cluster_id)
class Tunneling(ClusterHandler): class TunnelingClusterHandler(ClusterHandler):
"""Tunneling cluster handler.""" """Tunneling cluster handler."""

6
homeassistant/components/zha/siren.py
View File

@ -22,7 +22,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_call_later from homeassistant.helpers.event import async_call_later
from .core import discovery from .core import discovery
from .core.cluster_handlers.security import IasWd from .core.cluster_handlers.security import IasWdClusterHandler
from .core.const import ( from .core.const import (
CLUSTER_HANDLER_IAS_WD, CLUSTER_HANDLER_IAS_WD,
SIGNAL_ADD_ENTITIES, SIGNAL_ADD_ENTITIES,
@ -101,7 +101,9 @@ class ZHASiren(ZhaEntity, SirenEntity):
WARNING_DEVICE_MODE_EMERGENCY_PANIC: "Emergency Panic", WARNING_DEVICE_MODE_EMERGENCY_PANIC: "Emergency Panic",
} }
super().__init__(unique_id, zha_device, cluster_handlers, **kwargs) super().__init__(unique_id, zha_device, cluster_handlers, **kwargs)
self._cluster_handler: IasWd = cast(IasWd, cluster_handlers[0]) self._cluster_handler: IasWdClusterHandler = cast(
IasWdClusterHandler, cluster_handlers[0]
)
self._attr_is_on: bool = False self._attr_is_on: bool = False
self._off_listener: Callable[[], None] | None = None self._off_listener: Callable[[], None] | None = None

24
tests/components/zha/test_cluster_handlers.py
View File

@ -713,7 +713,9 @@ async def test_zll_device_groups(
"""Test adding coordinator to ZLL groups.""" """Test adding coordinator to ZLL groups."""
cluster = zigpy_zll_device.endpoints[1].lightlink cluster = zigpy_zll_device.endpoints[1].lightlink
cluster_handler = cluster_handlers.lightlink.LightLink(cluster, endpoint) cluster_handler = cluster_handlers.lightlink.LightLinkClusterHandler(
cluster, endpoint
)
get_group_identifiers_rsp = zigpy.zcl.clusters.lightlink.LightLink.commands_by_name[ get_group_identifiers_rsp = zigpy.zcl.clusters.lightlink.LightLink.commands_by_name[
"get_group_identifiers_rsp" "get_group_identifiers_rsp"
@ -980,3 +982,23 @@ async def test_retry_request(
assert func.await_count == 3 assert func.await_count == 3
assert isinstance(exc.value, HomeAssistantError) assert isinstance(exc.value, HomeAssistantError)
assert str(exc.value) == expected_error assert str(exc.value) == expected_error
async def test_cluster_handler_naming() -> None:
"""Test that all cluster handlers are named appropriately."""
for client_cluster_handler in registries.CLIENT_CLUSTER_HANDLER_REGISTRY.values():
assert issubclass(client_cluster_handler, cluster_handlers.ClientClusterHandler)
assert client_cluster_handler.__name__.endswith("ClientClusterHandler")
server_cluster_handlers = []
for cluster_handler_dict in registries.ZIGBEE_CLUSTER_HANDLER_REGISTRY.values():
# remove this filter in the update platform PR
server_cluster_handlers += [
cluster_handler
for cluster_handler in cluster_handler_dict.values()
if cluster_handler.__name__ != "OtaClientClusterHandler"
]
for cluster_handler in server_cluster_handlers:
assert not issubclass(cluster_handler, cluster_handlers.ClientClusterHandler)
assert cluster_handler.__name__.endswith("ClusterHandler")