Use more decorators for ZHA Core registries. (#25737)

* Move ZIGBEE_CHANNEL_REGISTRY to ZHA core registries.
* Refactor more ZHA Core registries to use decorator.
* Cleanup.
* Use relative imports for component.
* Flake8.
* Make pylint happy again.
This commit is contained in:
Alexei Chetroi 2019-08-06 20:01:14 -04:00 committed by GitHub
parent 98eb8efc6b
commit 152a9eb466
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 212 additions and 161 deletions

View File

@ -13,7 +13,6 @@ from random import uniform
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.util.decorator import Registry
from ..const import (
CHANNEL_ATTRIBUTE,
@ -30,8 +29,6 @@ from ..registries import CLUSTER_REPORT_CONFIGS
_LOGGER = logging.getLogger(__name__)
ZIGBEE_CHANNEL_REGISTRY = Registry()
def parse_and_log_command(channel, tsn, command_id, args):
"""Parse and log a zigbee cluster command."""

View File

@ -11,13 +11,14 @@ import zigpy.zcl.clusters.closures as closures
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_send
from . import ZIGBEE_CHANNEL_REGISTRY, ZigbeeChannel
from . import ZigbeeChannel
from .. import registries
from ..const import REPORT_CONFIG_IMMEDIATE, SIGNAL_ATTR_UPDATED
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(closures.DoorLock.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(closures.DoorLock.cluster_id)
class DoorLockChannel(ZigbeeChannel):
"""Door lock channel."""
@ -54,14 +55,14 @@ class DoorLockChannel(ZigbeeChannel):
await super().async_initialize(from_cache)
@ZIGBEE_CHANNEL_REGISTRY.register(closures.Shade.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(closures.Shade.cluster_id)
class Shade(ZigbeeChannel):
"""Shade channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(closures.WindowCovering.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(closures.WindowCovering.cluster_id)
class WindowCovering(ZigbeeChannel):
"""Window channel."""

View File

@ -12,12 +12,8 @@ from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.event import async_call_later
from . import (
ZIGBEE_CHANNEL_REGISTRY,
AttributeListeningChannel,
ZigbeeChannel,
parse_and_log_command,
)
from . import AttributeListeningChannel, ZigbeeChannel, parse_and_log_command
from .. import registries
from ..const import (
REPORT_CONFIG_ASAP,
REPORT_CONFIG_BATTERY_SAVE,
@ -32,42 +28,43 @@ from ..helpers import get_attr_id_by_name
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(general.Alarms.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.Alarms.cluster_id)
class Alarms(ZigbeeChannel):
"""Alarms channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.AnalogInput.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.AnalogInput.cluster_id)
class AnalogInput(AttributeListeningChannel):
"""Analog Input channel."""
REPORT_CONFIG = [{"attr": "present_value", "config": REPORT_CONFIG_DEFAULT}]
@ZIGBEE_CHANNEL_REGISTRY.register(general.AnalogOutput.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.AnalogOutput.cluster_id)
class AnalogOutput(AttributeListeningChannel):
"""Analog Output channel."""
REPORT_CONFIG = [{"attr": "present_value", "config": REPORT_CONFIG_DEFAULT}]
@ZIGBEE_CHANNEL_REGISTRY.register(general.AnalogValue.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.AnalogValue.cluster_id)
class AnalogValue(AttributeListeningChannel):
"""Analog Value channel."""
REPORT_CONFIG = [{"attr": "present_value", "config": REPORT_CONFIG_DEFAULT}]
@ZIGBEE_CHANNEL_REGISTRY.register(general.ApplianceControl.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.ApplianceControl.cluster_id)
class ApplianceContorl(ZigbeeChannel):
"""Appliance Control channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.Basic.cluster_id)
@registries.CHANNEL_ONLY_CLUSTERS.register(general.Basic.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.Basic.cluster_id)
class BasicChannel(ZigbeeChannel):
"""Channel to interact with the basic cluster."""
@ -106,63 +103,66 @@ class BasicChannel(ZigbeeChannel):
return self._power_source
@ZIGBEE_CHANNEL_REGISTRY.register(general.BinaryInput.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.BinaryInput.cluster_id)
class BinaryInput(AttributeListeningChannel):
"""Binary Input channel."""
REPORT_CONFIG = [{"attr": "present_value", "config": REPORT_CONFIG_DEFAULT}]
@ZIGBEE_CHANNEL_REGISTRY.register(general.BinaryOutput.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.BinaryOutput.cluster_id)
class BinaryOutput(AttributeListeningChannel):
"""Binary Output channel."""
REPORT_CONFIG = [{"attr": "present_value", "config": REPORT_CONFIG_DEFAULT}]
@ZIGBEE_CHANNEL_REGISTRY.register(general.BinaryValue.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.BinaryValue.cluster_id)
class BinaryValue(AttributeListeningChannel):
"""Binary Value channel."""
REPORT_CONFIG = [{"attr": "present_value", "config": REPORT_CONFIG_DEFAULT}]
@ZIGBEE_CHANNEL_REGISTRY.register(general.Commissioning.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.Commissioning.cluster_id)
class Commissioning(ZigbeeChannel):
"""Commissioning channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.DeviceTemperature.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.DeviceTemperature.cluster_id)
class DeviceTemperature(ZigbeeChannel):
"""Device Temperatur channel."""
"""Device Temperature channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.GreenPowerProxy.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.GreenPowerProxy.cluster_id)
class GreenPowerProxy(ZigbeeChannel):
"""Green Power Proxy channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.Groups.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.Groups.cluster_id)
class Groups(ZigbeeChannel):
"""Groups channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.Identify.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.Identify.cluster_id)
class Identify(ZigbeeChannel):
"""Identify channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.LevelControl.cluster_id)
@registries.BINDABLE_CLUSTERS.register(general.LevelControl.cluster_id)
@registries.EVENT_RELAY_CLUSTERS.register(general.LevelControl.cluster_id)
@registries.LIGHT_CLUSTERS.register(general.LevelControl.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.LevelControl.cluster_id)
class LevelControlChannel(ZigbeeChannel):
"""Channel for the LevelControl Zigbee cluster."""
@ -207,28 +207,33 @@ class LevelControlChannel(ZigbeeChannel):
await super().async_initialize(from_cache)
@ZIGBEE_CHANNEL_REGISTRY.register(general.MultistateInput.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.MultistateInput.cluster_id)
class MultistateInput(AttributeListeningChannel):
"""Multistate Input channel."""
REPORT_CONFIG = [{"attr": "present_value", "config": REPORT_CONFIG_DEFAULT}]
@ZIGBEE_CHANNEL_REGISTRY.register(general.MultistateOutput.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.MultistateOutput.cluster_id)
class MultistateOutput(AttributeListeningChannel):
"""Multistate Output channel."""
REPORT_CONFIG = [{"attr": "present_value", "config": REPORT_CONFIG_DEFAULT}]
@ZIGBEE_CHANNEL_REGISTRY.register(general.MultistateValue.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.MultistateValue.cluster_id)
class MultistateValue(AttributeListeningChannel):
"""Multistate Value channel."""
REPORT_CONFIG = [{"attr": "present_value", "config": REPORT_CONFIG_DEFAULT}]
@ZIGBEE_CHANNEL_REGISTRY.register(general.OnOff.cluster_id)
@registries.BINARY_SENSOR_CLUSTERS.register(general.OnOff.cluster_id)
@registries.BINDABLE_CLUSTERS.register(general.OnOff.cluster_id)
@registries.EVENT_RELAY_CLUSTERS.register(general.OnOff.cluster_id)
@registries.LIGHT_CLUSTERS.register(general.OnOff.cluster_id)
@registries.SWITCH_CLUSTERS.register(general.OnOff.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.OnOff.cluster_id)
class OnOffChannel(ZigbeeChannel):
"""Channel for the OnOff Zigbee cluster."""
@ -302,35 +307,36 @@ class OnOffChannel(ZigbeeChannel):
await super().async_update()
@ZIGBEE_CHANNEL_REGISTRY.register(general.OnOffConfiguration.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.OnOffConfiguration.cluster_id)
class OnOffConfiguration(ZigbeeChannel):
"""OnOff Configuration channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.Ota.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.Ota.cluster_id)
class Ota(ZigbeeChannel):
"""OTA Channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.Partition.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.Partition.cluster_id)
class Partition(ZigbeeChannel):
"""Partition channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.PollControl.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.PollControl.cluster_id)
class PollControl(ZigbeeChannel):
"""Poll Control channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.PowerConfiguration.cluster_id)
@registries.DEVICE_TRACKER_CLUSTERS.register(general.PowerConfiguration.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.PowerConfiguration.cluster_id)
class PowerConfigurationChannel(ZigbeeChannel):
"""Channel for the zigbee power configuration cluster."""
@ -373,28 +379,29 @@ class PowerConfigurationChannel(ZigbeeChannel):
await self.get_attribute_value("battery_quantity", from_cache=from_cache)
@ZIGBEE_CHANNEL_REGISTRY.register(general.PowerProfile.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.PowerProfile.cluster_id)
class PowerProfile(ZigbeeChannel):
"""Power Profile channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.RSSILocation.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.RSSILocation.cluster_id)
class RSSILocation(ZigbeeChannel):
"""RSSI Location channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.Scenes.cluster_id)
@registries.OUTPUT_CHANNEL_ONLY_CLUSTERS.register(general.Scenes.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.Scenes.cluster_id)
class Scenes(ZigbeeChannel):
"""Scenes channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.Time.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(general.Time.cluster_id)
class Time(ZigbeeChannel):
"""Time channel."""

View File

@ -10,7 +10,8 @@ import zigpy.zcl.clusters.homeautomation as homeautomation
from homeassistant.helpers.dispatcher import async_dispatcher_send
from . import ZIGBEE_CHANNEL_REGISTRY, AttributeListeningChannel, ZigbeeChannel
from . import AttributeListeningChannel, ZigbeeChannel
from .. import registries
from ..const import (
CHANNEL_ELECTRICAL_MEASUREMENT,
REPORT_CONFIG_DEFAULT,
@ -20,39 +21,48 @@ from ..const import (
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(homeautomation.ApplianceEventAlerts.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(
homeautomation.ApplianceEventAlerts.cluster_id
)
class ApplianceEventAlerts(ZigbeeChannel):
"""Appliance Event Alerts channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(homeautomation.ApplianceIdentification.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(
homeautomation.ApplianceIdentification.cluster_id
)
class ApplianceIdentification(ZigbeeChannel):
"""Appliance Identification channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(homeautomation.ApplianceStatistics.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(
homeautomation.ApplianceStatistics.cluster_id
)
class ApplianceStatistics(ZigbeeChannel):
"""Appliance Statistics channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(homeautomation.Diagnostic.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(homeautomation.Diagnostic.cluster_id)
class Diagnostic(ZigbeeChannel):
"""Diagnostic channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(homeautomation.ElectricalMeasurement.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(
homeautomation.ElectricalMeasurement.cluster_id
)
class ElectricalMeasurementChannel(AttributeListeningChannel):
"""Channel that polls active power level."""
CHANNEL_NAME = CHANNEL_ELECTRICAL_MEASUREMENT
REPORT_CONFIG = ({"attr": "active_power", "config": REPORT_CONFIG_DEFAULT},)
async def async_update(self):
@ -73,7 +83,9 @@ class ElectricalMeasurementChannel(AttributeListeningChannel):
await super().async_initialize(from_cache)
@ZIGBEE_CHANNEL_REGISTRY.register(homeautomation.MeterIdentification.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(
homeautomation.MeterIdentification.cluster_id
)
class MeterIdentification(ZigbeeChannel):
"""Metering Identification channel."""

View File

@ -11,24 +11,26 @@ import zigpy.zcl.clusters.hvac as hvac
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_send
from . import ZIGBEE_CHANNEL_REGISTRY, ZigbeeChannel
from . import ZigbeeChannel
from .. import registries
from ..const import REPORT_CONFIG_OP, SIGNAL_ATTR_UPDATED
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(hvac.Dehumidification.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(hvac.Dehumidification.cluster_id)
class Dehumidification(ZigbeeChannel):
"""Dehumidification channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(hvac.Fan.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(hvac.Fan.cluster_id)
class FanChannel(ZigbeeChannel):
"""Fan channel."""
_value_attribute = 0
REPORT_CONFIG = ({"attr": "fan_mode", "config": REPORT_CONFIG_OP},)
async def async_set_speed(self, value) -> None:
@ -71,21 +73,21 @@ class FanChannel(ZigbeeChannel):
await super().async_initialize(from_cache)
@ZIGBEE_CHANNEL_REGISTRY.register(hvac.Pump.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(hvac.Pump.cluster_id)
class Pump(ZigbeeChannel):
"""Pump channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(hvac.Thermostat.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(hvac.Thermostat.cluster_id)
class Thermostat(ZigbeeChannel):
"""Thermostat channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(hvac.UserInterface.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(hvac.UserInterface.cluster_id)
class UserInterface(ZigbeeChannel):
"""User interface (thermostat) channel."""

View File

@ -8,20 +8,24 @@ import logging
import zigpy.zcl.clusters.lighting as lighting
from . import ZIGBEE_CHANNEL_REGISTRY, ZigbeeChannel
from . import ZigbeeChannel
from .. import registries
from ..const import REPORT_CONFIG_DEFAULT
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(lighting.Ballast.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(lighting.Ballast.cluster_id)
class Ballast(ZigbeeChannel):
"""Ballast channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(lighting.Color.cluster_id)
@registries.BINDABLE_CLUSTERS.register(lighting.Color.cluster_id)
@registries.EVENT_RELAY_CLUSTERS.register(lighting.Color.cluster_id)
@registries.LIGHT_CLUSTERS.register(lighting.Color.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(lighting.Color.cluster_id)
class ColorChannel(ZigbeeChannel):
"""Color channel."""

View File

@ -8,12 +8,14 @@ import logging
import zigpy.zcl.clusters.lightlink as lightlink
from . import ZIGBEE_CHANNEL_REGISTRY, ZigbeeChannel
from . import ZigbeeChannel
from .. import registries
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(lightlink.LightLink.cluster_id)
@registries.CHANNEL_ONLY_CLUSTERS.register(lightlink.LightLink.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(lightlink.LightLink.cluster_id)
class LightLink(ZigbeeChannel):
"""Lightlink channel."""

View File

@ -8,7 +8,8 @@ import logging
import zigpy.zcl.clusters.measurement as measurement
from . import ZIGBEE_CHANNEL_REGISTRY, AttributeListeningChannel
from . import AttributeListeningChannel
from .. import registries
from ..const import (
REPORT_CONFIG_DEFAULT,
REPORT_CONFIG_IMMEDIATE,
@ -19,42 +20,47 @@ from ..const import (
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(measurement.FlowMeasurement.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(measurement.FlowMeasurement.cluster_id)
class FlowMeasurement(AttributeListeningChannel):
"""Flow Measurement channel."""
REPORT_CONFIG = [{"attr": "measured_value", "config": REPORT_CONFIG_DEFAULT}]
@ZIGBEE_CHANNEL_REGISTRY.register(measurement.IlluminanceLevelSensing.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(
measurement.IlluminanceLevelSensing.cluster_id
)
class IlluminanceLevelSensing(AttributeListeningChannel):
"""Illuminance Level Sensing channel."""
REPORT_CONFIG = [{"attr": "level_status", "config": REPORT_CONFIG_DEFAULT}]
@ZIGBEE_CHANNEL_REGISTRY.register(measurement.IlluminanceMeasurement.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(
measurement.IlluminanceMeasurement.cluster_id
)
class IlluminanceMeasurement(AttributeListeningChannel):
"""Illuminance Measurement channel."""
REPORT_CONFIG = [{"attr": "measured_value", "config": REPORT_CONFIG_DEFAULT}]
@ZIGBEE_CHANNEL_REGISTRY.register(measurement.OccupancySensing.cluster_id)
@registries.BINARY_SENSOR_CLUSTERS.register(measurement.OccupancySensing.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(measurement.OccupancySensing.cluster_id)
class OccupancySensing(AttributeListeningChannel):
"""Occupancy Sensing channel."""
REPORT_CONFIG = [{"attr": "occupancy", "config": REPORT_CONFIG_IMMEDIATE}]
@ZIGBEE_CHANNEL_REGISTRY.register(measurement.PressureMeasurement.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(measurement.PressureMeasurement.cluster_id)
class PressureMeasurement(AttributeListeningChannel):
"""Pressure measurement channel."""
REPORT_CONFIG = [{"attr": "measured_value", "config": REPORT_CONFIG_DEFAULT}]
@ZIGBEE_CHANNEL_REGISTRY.register(measurement.RelativeHumidity.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(measurement.RelativeHumidity.cluster_id)
class RelativeHumidity(AttributeListeningChannel):
"""Relative Humidity measurement channel."""
@ -66,7 +72,9 @@ class RelativeHumidity(AttributeListeningChannel):
]
@ZIGBEE_CHANNEL_REGISTRY.register(measurement.TemperatureMeasurement.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(
measurement.TemperatureMeasurement.cluster_id
)
class TemperatureMeasurement(AttributeListeningChannel):
"""Temperature measurement channel."""

View File

@ -8,145 +8,154 @@ import logging
import zigpy.zcl.clusters.protocol as protocol
from ..channels import ZIGBEE_CHANNEL_REGISTRY, ZigbeeChannel
from .. import registries
from ..channels import ZigbeeChannel
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.AnalogInputExtended.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(protocol.AnalogInputExtended.cluster_id)
class AnalogInputExtended(ZigbeeChannel):
"""Analog Input Extended channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.AnalogInputRegular.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(protocol.AnalogInputRegular.cluster_id)
class AnalogInputRegular(ZigbeeChannel):
"""Analog Input Regular channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.AnalogOutputExtended.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(protocol.AnalogOutputExtended.cluster_id)
class AnalogOutputExtended(ZigbeeChannel):
"""Analog Output Regular channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.AnalogOutputRegular.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(protocol.AnalogOutputRegular.cluster_id)
class AnalogOutputRegular(ZigbeeChannel):
"""Analog Output Regular channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.AnalogValueExtended.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(protocol.AnalogValueExtended.cluster_id)
class AnalogValueExtended(ZigbeeChannel):
"""Analog Value Extended edition channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.AnalogValueRegular.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(protocol.AnalogValueRegular.cluster_id)
class AnalogValueRegular(ZigbeeChannel):
"""Analog Value Regular channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.BacnetProtocolTunnel.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(protocol.BacnetProtocolTunnel.cluster_id)
class BacnetProtocolTunnel(ZigbeeChannel):
"""Bacnet Protocol Tunnel channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryInputExtended.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryInputExtended.cluster_id)
class BinaryInputExtended(ZigbeeChannel):
"""Binary Input Extended channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryInputRegular.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryInputRegular.cluster_id)
class BinaryInputRegular(ZigbeeChannel):
"""Binary Input Regular channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryOutputExtended.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryOutputExtended.cluster_id)
class BinaryOutputExtended(ZigbeeChannel):
"""Binary Output Extended channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryOutputRegular.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryOutputRegular.cluster_id)
class BinaryOutputRegular(ZigbeeChannel):
"""Binary Output Regular channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryValueExtended.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryValueExtended.cluster_id)
class BinaryValueExtended(ZigbeeChannel):
"""Binary Value Extended channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryValueRegular.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryValueRegular.cluster_id)
class BinaryValueRegular(ZigbeeChannel):
"""Binary Value Regular channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.GenericTunnel.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(protocol.GenericTunnel.cluster_id)
class GenericTunnel(ZigbeeChannel):
"""Generic Tunnel channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.MultistateInputExtended.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(
protocol.MultistateInputExtended.cluster_id
)
class MultiStateInputExtended(ZigbeeChannel):
"""Multistate Input Extended channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.MultistateInputRegular.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(protocol.MultistateInputRegular.cluster_id)
class MultiStateInputRegular(ZigbeeChannel):
"""Multistate Input Regular channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.MultistateOutputExtended.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(
protocol.MultistateOutputExtended.cluster_id
)
class MultiStateOutputExtended(ZigbeeChannel):
"""Multistate Output Extended channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.MultistateOutputRegular.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(
protocol.MultistateOutputRegular.cluster_id
)
class MultiStateOutputRegular(ZigbeeChannel):
"""Multistate Output Regular channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.MultistateValueExtended.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(
protocol.MultistateValueExtended.cluster_id
)
class MultiStateValueExtended(ZigbeeChannel):
"""Multistate Value Extended channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.MultistateValueRegular.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(protocol.MultistateValueRegular.cluster_id)
class MultiStateValueRegular(ZigbeeChannel):
"""Multistate Value Regular channel."""

View File

@ -11,27 +11,29 @@ import zigpy.zcl.clusters.security as security
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_send
from . import ZIGBEE_CHANNEL_REGISTRY, ZigbeeChannel
from . import ZigbeeChannel
from .. import registries
from ..const import SIGNAL_ATTR_UPDATED
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(security.IasAce.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(security.IasAce.cluster_id)
class IasAce(ZigbeeChannel):
"""IAS Ancillary Control Equipment channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(security.IasWd.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(security.IasWd.cluster_id)
class IasWd(ZigbeeChannel):
"""IAS Warning Device channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(security.IasZone.cluster_id)
@registries.BINARY_SENSOR_CLUSTERS.register(security.IasZone.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(security.IasZone.cluster_id)
class IASZoneChannel(ZigbeeChannel):
"""Channel for the IASZone Zigbee cluster."""

View File

@ -8,90 +8,91 @@ import logging
import zigpy.zcl.clusters.smartenergy as smartenergy
from ..channels import ZIGBEE_CHANNEL_REGISTRY, AttributeListeningChannel, ZigbeeChannel
from .. import registries
from ..channels import AttributeListeningChannel, ZigbeeChannel
from ..const import REPORT_CONFIG_DEFAULT
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Calendar.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Calendar.cluster_id)
class Calendar(ZigbeeChannel):
"""Calendar channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.DeviceManagement.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.DeviceManagement.cluster_id)
class DeviceManagement(ZigbeeChannel):
"""Device Management channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Drlc.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Drlc.cluster_id)
class Drlc(ZigbeeChannel):
"""Demand Response and Load Control channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.EnergyManagement.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.EnergyManagement.cluster_id)
class EnergyManagement(ZigbeeChannel):
"""Energy Management channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Events.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Events.cluster_id)
class Events(ZigbeeChannel):
"""Event channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.KeyEstablishment.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.KeyEstablishment.cluster_id)
class KeyEstablishment(ZigbeeChannel):
"""Key Establishment channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.MduPairing.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.MduPairing.cluster_id)
class MduPairing(ZigbeeChannel):
"""Pairing channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Messaging.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Messaging.cluster_id)
class Messaging(ZigbeeChannel):
"""Messaging channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Metering.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Metering.cluster_id)
class Metering(AttributeListeningChannel):
"""Metering channel."""
REPORT_CONFIG = [{"attr": "instantaneous_demand", "config": REPORT_CONFIG_DEFAULT}]
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Prepayment.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Prepayment.cluster_id)
class Prepayment(ZigbeeChannel):
"""Prepayment channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Price.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Price.cluster_id)
class Price(ZigbeeChannel):
"""Price channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Tunneling.cluster_id)
@registries.ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Tunneling.cluster_id)
class Tunneling(ZigbeeChannel):
"""Tunneling channel."""

View File

@ -0,0 +1,37 @@
"""Decorators for ZHA core registries."""
from typing import Callable, TypeVar, Union
CALLABLE_T = TypeVar("CALLABLE_T", bound=Callable) # noqa pylint: disable=invalid-name
class DictRegistry(dict):
"""Dict Registry of items."""
def register(
self, name: Union[int, str], item: Union[str, CALLABLE_T] = None
) -> Callable[[CALLABLE_T], CALLABLE_T]:
"""Return decorator to register item with a specific name."""
def decorator(channel: CALLABLE_T) -> CALLABLE_T:
"""Register decorated channel or item."""
if item is None:
self[name] = channel
else:
self[name] = item
return channel
return decorator
class SetRegistry(set):
"""Set Registry of items."""
def register(self, name: Union[int, str]) -> Callable[[CALLABLE_T], CALLABLE_T]:
"""Return decorator to register item with a specific name."""
def decorator(channel: CALLABLE_T) -> CALLABLE_T:
"""Register decorated channel or item."""
self.add(name)
return channel
return decorator

View File

@ -13,12 +13,7 @@ from homeassistant.components.sensor import DOMAIN as SENSOR
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_send
from .channels import (
ZIGBEE_CHANNEL_REGISTRY,
AttributeListeningChannel,
EventRelayChannel,
ZDOChannel,
)
from .channels import AttributeListeningChannel, EventRelayChannel, ZDOChannel
from .const import (
COMPONENTS,
CONF_DEVICE_CONFIG,
@ -39,6 +34,7 @@ from .registries import (
SENSOR_TYPES,
SINGLE_INPUT_CLUSTER_DEVICE_CLASS,
SINGLE_OUTPUT_CLUSTER_DEVICE_CLASS,
ZIGBEE_CHANNEL_REGISTRY,
)
_LOGGER = logging.getLogger(__name__)

View File

@ -122,13 +122,12 @@ def async_is_bindable_target(source_zha_device, target_zha_device):
source_clusters = source_zha_device.async_get_std_clusters()
target_clusters = target_zha_device.async_get_std_clusters()
bindables = set(BINDABLE_CLUSTERS)
for endpoint_id in source_clusters:
for t_endpoint_id in target_clusters:
matches = set(
source_clusters[endpoint_id][CLUSTER_TYPE_OUT].keys()
).intersection(target_clusters[t_endpoint_id][CLUSTER_TYPE_IN].keys())
if any(bindable in bindables for bindable in matches):
if any(bindable in BINDABLE_CLUSTERS for bindable in matches):
return True
return False

View File

@ -4,6 +4,7 @@ Mapping registries for Zigbee Home Automation.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/zha/
"""
import collections
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR
from homeassistant.components.device_tracker import DOMAIN as DEVICE_TRACKER
@ -33,24 +34,25 @@ from .const import (
ZONE,
RadioType,
)
from .decorators import DictRegistry, SetRegistry
BINARY_SENSOR_CLUSTERS = set()
BINARY_SENSOR_CLUSTERS = SetRegistry()
BINARY_SENSOR_TYPES = {}
BINDABLE_CLUSTERS = []
CHANNEL_ONLY_CLUSTERS = []
BINDABLE_CLUSTERS = SetRegistry()
CHANNEL_ONLY_CLUSTERS = SetRegistry()
CLUSTER_REPORT_CONFIGS = {}
CUSTOM_CLUSTER_MAPPINGS = {}
DEVICE_CLASS = {}
DEVICE_TRACKER_CLUSTERS = set()
EVENT_RELAY_CLUSTERS = []
LIGHT_CLUSTERS = set()
OUTPUT_CHANNEL_ONLY_CLUSTERS = []
DEVICE_CLASS = collections.defaultdict(dict)
DEVICE_TRACKER_CLUSTERS = SetRegistry()
EVENT_RELAY_CLUSTERS = SetRegistry()
LIGHT_CLUSTERS = SetRegistry()
OUTPUT_CHANNEL_ONLY_CLUSTERS = SetRegistry()
RADIO_TYPES = {}
REMOTE_DEVICE_TYPES = {}
REMOTE_DEVICE_TYPES = collections.defaultdict(list)
SENSOR_TYPES = {}
SINGLE_INPUT_CLUSTER_DEVICE_CLASS = {}
SINGLE_OUTPUT_CLUSTER_DEVICE_CLASS = {}
SWITCH_CLUSTERS = set()
SWITCH_CLUSTERS = SetRegistry()
SMARTTHINGS_ACCELERATION_CLUSTER = 0xFC02
SMARTTHINGS_ARRIVAL_SENSOR_DEVICE_TYPE = 0x8000
SMARTTHINGS_HUMIDITY_CLUSTER = 0xFC45
@ -62,6 +64,11 @@ COMPONENT_CLUSTERS = {
SWITCH: SWITCH_CLUSTERS,
}
ZIGBEE_CHANNEL_REGISTRY = DictRegistry()
# importing channels updates registries
from . import channels # noqa pylint: disable=wrong-import-position,unused-import
def establish_device_mappings():
"""Establish mappings between ZCL objects and HA ZHA objects.
@ -72,16 +79,6 @@ def establish_device_mappings():
from zigpy import zcl
from zigpy.profiles import zha, zll
if zha.PROFILE_ID not in DEVICE_CLASS:
DEVICE_CLASS[zha.PROFILE_ID] = {}
if zll.PROFILE_ID not in DEVICE_CLASS:
DEVICE_CLASS[zll.PROFILE_ID] = {}
if zha.PROFILE_ID not in REMOTE_DEVICE_TYPES:
REMOTE_DEVICE_TYPES[zha.PROFILE_ID] = []
if zll.PROFILE_ID not in REMOTE_DEVICE_TYPES:
REMOTE_DEVICE_TYPES[zll.PROFILE_ID] = []
def get_ezsp_radio():
import bellows.ezsp
from bellows.zigbee.application import ControllerApplication
@ -133,9 +130,6 @@ def establish_device_mappings():
}
BINARY_SENSOR_CLUSTERS.add(SMARTTHINGS_ACCELERATION_CLUSTER)
BINARY_SENSOR_CLUSTERS.add(zcl.clusters.general.OnOff.cluster_id)
BINARY_SENSOR_CLUSTERS.add(zcl.clusters.measurement.OccupancySensing.cluster_id)
BINARY_SENSOR_CLUSTERS.add(zcl.clusters.security.IasZone.cluster_id)
BINARY_SENSOR_TYPES.update(
{
@ -146,13 +140,6 @@ def establish_device_mappings():
}
)
BINDABLE_CLUSTERS.append(zcl.clusters.general.LevelControl.cluster_id)
BINDABLE_CLUSTERS.append(zcl.clusters.general.OnOff.cluster_id)
BINDABLE_CLUSTERS.append(zcl.clusters.lighting.Color.cluster_id)
CHANNEL_ONLY_CLUSTERS.append(zcl.clusters.general.Basic.cluster_id)
CHANNEL_ONLY_CLUSTERS.append(zcl.clusters.lightlink.LightLink.cluster_id)
CLUSTER_REPORT_CONFIGS.update(
{
SMARTTHINGS_ACCELERATION_CLUSTER: [
@ -200,17 +187,6 @@ def establish_device_mappings():
}
)
DEVICE_TRACKER_CLUSTERS.add(zcl.clusters.general.PowerConfiguration.cluster_id)
EVENT_RELAY_CLUSTERS.append(zcl.clusters.general.LevelControl.cluster_id)
EVENT_RELAY_CLUSTERS.append(zcl.clusters.general.OnOff.cluster_id)
OUTPUT_CHANNEL_ONLY_CLUSTERS.append(zcl.clusters.general.Scenes.cluster_id)
LIGHT_CLUSTERS.add(zcl.clusters.general.LevelControl.cluster_id)
LIGHT_CLUSTERS.add(zcl.clusters.general.OnOff.cluster_id)
LIGHT_CLUSTERS.add(zcl.clusters.lighting.Color.cluster_id)
SINGLE_INPUT_CLUSTER_DEVICE_CLASS.update(
{
# this works for now but if we hit conflicts we can break it out to
@ -251,8 +227,6 @@ def establish_device_mappings():
}
)
SWITCH_CLUSTERS.add(zcl.clusters.general.OnOff.cluster_id)
zhap = zha.PROFILE_ID
REMOTE_DEVICE_TYPES[zhap].append(zha.DeviceType.COLOR_CONTROLLER)
REMOTE_DEVICE_TYPES[zhap].append(zha.DeviceType.COLOR_DIMMER_SWITCH)

View File

@ -1,10 +1,10 @@
"""Test ZHA Core channels."""
import homeassistant.components.zha.core.channels
import pytest
import zigpy.types as t
import homeassistant.components.zha.core.channels as channels
import homeassistant.components.zha.core.device as zha_device
import homeassistant.components.zha.core.registries as registries
from .common import make_device
@ -77,7 +77,7 @@ async def test_in_channel_config(cluster_id, bind_count, attrs, zha_gateway, has
zha_dev = zha_device.ZHADevice(hass, zigpy_dev, zha_gateway)
cluster = zigpy_dev.endpoints[1].in_clusters[cluster_id]
channel_class = channels.ZIGBEE_CHANNEL_REGISTRY.get(
channel_class = registries.ZIGBEE_CHANNEL_REGISTRY.get(
cluster_id, channels.AttributeListeningChannel
)
channel = channel_class(cluster, zha_dev)
@ -136,7 +136,7 @@ async def test_out_channel_config(cluster_id, bind_count, zha_gateway, hass):
cluster = zigpy_dev.endpoints[1].out_clusters[cluster_id]
cluster.bind_only = True
channel_class = homeassistant.components.zha.core.channels.ZIGBEE_CHANNEL_REGISTRY.get(
channel_class = registries.ZIGBEE_CHANNEL_REGISTRY.get(
cluster_id, channels.AttributeListeningChannel
)
channel = channel_class(cluster, zha_dev)
@ -149,7 +149,7 @@ async def test_out_channel_config(cluster_id, bind_count, zha_gateway, hass):
def test_channel_registry():
"""Test ZIGBEE Channel Registry."""
for cluster_id, channel in channels.ZIGBEE_CHANNEL_REGISTRY.items():
for (cluster_id, channel) in registries.ZIGBEE_CHANNEL_REGISTRY.items():
assert isinstance(cluster_id, int)
assert 0 <= cluster_id <= 0xFFFF
assert issubclass(channel, channels.ZigbeeChannel)