Refactor ZHA Zigbee channel registry. (#25716)

* Update test to catch regression.

* Refactor ZHA Core channels.

Use channel decorator for Zigbee channel registry.

* Update tests.

* Pylint
This commit is contained in:
Alexei Chetroi 2019-08-05 17:40:29 -04:00 committed by Aaron Bach
parent 8241193fa8
commit 8dbac9176e
17 changed files with 738 additions and 174 deletions

View File

@ -11,7 +11,6 @@ from homeassistant.helpers.device_registry import CONNECTION_ZIGBEE
from . import config_flow # noqa # pylint: disable=unused-import
from . import api
from .core import ZHAGateway
from .core.channels.registry import populate_channel_registry
from .core.const import (
COMPONENTS,
CONF_BAUDRATE,
@ -90,7 +89,6 @@ async def async_setup_entry(hass, config_entry):
Will automatically load components to support devices found on the network.
"""
establish_device_mappings()
populate_channel_registry()
for component in COMPONENTS:
hass.data[DATA_ZHA][component] = hass.data[DATA_ZHA].get(component, {})

View File

@ -13,11 +13,13 @@ 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,
CHANNEL_EVENT_RELAY,
CHANNEL_ZDO,
REPORT_CONFIG_DEFAULT,
REPORT_CONFIG_MAX_INT,
REPORT_CONFIG_MIN_INT,
REPORT_CONFIG_RPT_CHANGE,
@ -28,6 +30,8 @@ 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."""
@ -282,6 +286,7 @@ class AttributeListeningChannel(ZigbeeChannel):
"""Channel for attribute reports from the cluster."""
CHANNEL_NAME = CHANNEL_ATTRIBUTE
REPORT_CONFIG = [{"attr": 0, "config": REPORT_CONFIG_DEFAULT}]
def __init__(self, cluster, device):
"""Initialize AttributeListeningChannel."""
@ -394,3 +399,17 @@ class EventRelayChannel(ZigbeeChannel):
self.zha_send_event(
self._cluster, self._cluster.server_commands.get(command_id)[0], args
)
# pylint: disable=wrong-import-position
from . import closures # noqa
from . import general # noqa
from . import homeautomation # noqa
from . import hvac # noqa
from . import lighting # noqa
from . import lightlink # noqa
from . import manufacturerspecific # noqa
from . import measurement # noqa
from . import protocol # noqa
from . import security # noqa
from . import smartenergy # noqa

View File

@ -6,15 +6,18 @@ https://home-assistant.io/components/zha/
"""
import logging
import zigpy.zcl.clusters.closures as closures
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_send
from . import ZigbeeChannel
from . import ZIGBEE_CHANNEL_REGISTRY, ZigbeeChannel
from ..const import REPORT_CONFIG_IMMEDIATE, SIGNAL_ATTR_UPDATED
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(closures.DoorLock.cluster_id)
class DoorLockChannel(ZigbeeChannel):
"""Door lock channel."""
@ -49,3 +52,17 @@ class DoorLockChannel(ZigbeeChannel):
"""Initialize channel."""
await self.get_attribute_value(self._value_attribute, from_cache=from_cache)
await super().async_initialize(from_cache)
@ZIGBEE_CHANNEL_REGISTRY.register(closures.Shade.cluster_id)
class Shade(ZigbeeChannel):
"""Shade channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(closures.WindowCovering.cluster_id)
class WindowCovering(ZigbeeChannel):
"""Window channel."""
pass

View File

@ -6,24 +6,229 @@ https://home-assistant.io/components/zha/
"""
import logging
import zigpy.zcl.clusters.general as general
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.event import async_call_later
from . import ZigbeeChannel, parse_and_log_command
from . import (
ZIGBEE_CHANNEL_REGISTRY,
AttributeListeningChannel,
ZigbeeChannel,
parse_and_log_command,
)
from ..const import (
REPORT_CONFIG_ASAP,
REPORT_CONFIG_BATTERY_SAVE,
REPORT_CONFIG_DEFAULT,
REPORT_CONFIG_IMMEDIATE,
SIGNAL_ATTR_UPDATED,
SIGNAL_MOVE_LEVEL,
SIGNAL_SET_LEVEL,
REPORT_CONFIG_ASAP,
REPORT_CONFIG_BATTERY_SAVE,
REPORT_CONFIG_IMMEDIATE,
)
from ..helpers import get_attr_id_by_name
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(general.Alarms.cluster_id)
class Alarms(ZigbeeChannel):
"""Alarms channel."""
pass
@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)
class AnalogOutput(AttributeListeningChannel):
"""Analog Output channel."""
REPORT_CONFIG = [{"attr": "present_value", "config": REPORT_CONFIG_DEFAULT}]
@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)
class ApplianceContorl(ZigbeeChannel):
"""Appliance Control channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.Basic.cluster_id)
class BasicChannel(ZigbeeChannel):
"""Channel to interact with the basic cluster."""
UNKNOWN = 0
BATTERY = 3
POWER_SOURCES = {
UNKNOWN: "Unknown",
1: "Mains (single phase)",
2: "Mains (3 phase)",
BATTERY: "Battery",
4: "DC source",
5: "Emergency mains constantly powered",
6: "Emergency mains and transfer switch",
}
def __init__(self, cluster, device):
"""Initialize BasicChannel."""
super().__init__(cluster, device)
self._power_source = None
async def async_configure(self):
"""Configure this channel."""
await super().async_configure()
await self.async_initialize(False)
async def async_initialize(self, from_cache):
"""Initialize channel."""
self._power_source = await self.get_attribute_value(
"power_source", from_cache=from_cache
)
await super().async_initialize(from_cache)
def get_power_source(self):
"""Get the power source."""
return self._power_source
@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)
class BinaryOutput(AttributeListeningChannel):
"""Binary Output channel."""
REPORT_CONFIG = [{"attr": "present_value", "config": REPORT_CONFIG_DEFAULT}]
@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)
class Commissioning(ZigbeeChannel):
"""Commissioning channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.DeviceTemperature.cluster_id)
class DeviceTemperature(ZigbeeChannel):
"""Device Temperatur channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.GreenPowerProxy.cluster_id)
class GreenPowerProxy(ZigbeeChannel):
"""Green Power Proxy channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.Groups.cluster_id)
class Groups(ZigbeeChannel):
"""Groups channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.Identify.cluster_id)
class Identify(ZigbeeChannel):
"""Identify channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.LevelControl.cluster_id)
class LevelControlChannel(ZigbeeChannel):
"""Channel for the LevelControl Zigbee cluster."""
CURRENT_LEVEL = 0
REPORT_CONFIG = ({"attr": "current_level", "config": REPORT_CONFIG_ASAP},)
@callback
def cluster_command(self, tsn, command_id, args):
"""Handle commands received to this cluster."""
cmd = parse_and_log_command(self, tsn, command_id, args)
if cmd in ("move_to_level", "move_to_level_with_on_off"):
self.dispatch_level_change(SIGNAL_SET_LEVEL, args[0])
elif cmd in ("move", "move_with_on_off"):
# We should dim slowly -- for now, just step once
rate = args[1]
if args[0] == 0xFF:
rate = 10 # Should read default move rate
self.dispatch_level_change(SIGNAL_MOVE_LEVEL, -rate if args[0] else rate)
elif cmd in ("step", "step_with_on_off"):
# Step (technically may change on/off)
self.dispatch_level_change(
SIGNAL_MOVE_LEVEL, -args[1] if args[0] else args[1]
)
@callback
def attribute_updated(self, attrid, value):
"""Handle attribute updates on this cluster."""
self.debug("received attribute: %s update with value: %s", attrid, value)
if attrid == self.CURRENT_LEVEL:
self.dispatch_level_change(SIGNAL_SET_LEVEL, value)
def dispatch_level_change(self, command, level):
"""Dispatch level change."""
async_dispatcher_send(
self._zha_device.hass, "{}_{}".format(self.unique_id, command), level
)
async def async_initialize(self, from_cache):
"""Initialize channel."""
await self.get_attribute_value(self.CURRENT_LEVEL, from_cache=from_cache)
await super().async_initialize(from_cache)
@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)
class MultistateOutput(AttributeListeningChannel):
"""Multistate Output channel."""
REPORT_CONFIG = [{"attr": "present_value", "config": REPORT_CONFIG_DEFAULT}]
@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)
class OnOffChannel(ZigbeeChannel):
"""Channel for the OnOff Zigbee cluster."""
@ -97,88 +302,35 @@ class OnOffChannel(ZigbeeChannel):
await super().async_update()
class LevelControlChannel(ZigbeeChannel):
"""Channel for the LevelControl Zigbee cluster."""
@ZIGBEE_CHANNEL_REGISTRY.register(general.OnOffConfiguration.cluster_id)
class OnOffConfiguration(ZigbeeChannel):
"""OnOff Configuration channel."""
CURRENT_LEVEL = 0
REPORT_CONFIG = ({"attr": "current_level", "config": REPORT_CONFIG_ASAP},)
@callback
def cluster_command(self, tsn, command_id, args):
"""Handle commands received to this cluster."""
cmd = parse_and_log_command(self, tsn, command_id, args)
if cmd in ("move_to_level", "move_to_level_with_on_off"):
self.dispatch_level_change(SIGNAL_SET_LEVEL, args[0])
elif cmd in ("move", "move_with_on_off"):
# We should dim slowly -- for now, just step once
rate = args[1]
if args[0] == 0xFF:
rate = 10 # Should read default move rate
self.dispatch_level_change(SIGNAL_MOVE_LEVEL, -rate if args[0] else rate)
elif cmd in ("step", "step_with_on_off"):
# Step (technically may change on/off)
self.dispatch_level_change(
SIGNAL_MOVE_LEVEL, -args[1] if args[0] else args[1]
)
@callback
def attribute_updated(self, attrid, value):
"""Handle attribute updates on this cluster."""
self.debug("received attribute: %s update with value: %s", attrid, value)
if attrid == self.CURRENT_LEVEL:
self.dispatch_level_change(SIGNAL_SET_LEVEL, value)
def dispatch_level_change(self, command, level):
"""Dispatch level change."""
async_dispatcher_send(
self._zha_device.hass, "{}_{}".format(self.unique_id, command), level
)
async def async_initialize(self, from_cache):
"""Initialize channel."""
await self.get_attribute_value(self.CURRENT_LEVEL, from_cache=from_cache)
await super().async_initialize(from_cache)
pass
class BasicChannel(ZigbeeChannel):
"""Channel to interact with the basic cluster."""
@ZIGBEE_CHANNEL_REGISTRY.register(general.Ota.cluster_id)
class Ota(ZigbeeChannel):
"""OTA Channel."""
UNKNOWN = 0
BATTERY = 3
POWER_SOURCES = {
UNKNOWN: "Unknown",
1: "Mains (single phase)",
2: "Mains (3 phase)",
BATTERY: "Battery",
4: "DC source",
5: "Emergency mains constantly powered",
6: "Emergency mains and transfer switch",
}
def __init__(self, cluster, device):
"""Initialize BasicChannel."""
super().__init__(cluster, device)
self._power_source = None
async def async_configure(self):
"""Configure this channel."""
await super().async_configure()
await self.async_initialize(False)
async def async_initialize(self, from_cache):
"""Initialize channel."""
self._power_source = await self.get_attribute_value(
"power_source", from_cache=from_cache
)
await super().async_initialize(from_cache)
def get_power_source(self):
"""Get the power source."""
return self._power_source
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.Partition.cluster_id)
class Partition(ZigbeeChannel):
"""Partition channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.PollControl.cluster_id)
class PollControl(ZigbeeChannel):
"""Poll Control channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.PowerConfiguration.cluster_id)
class PowerConfigurationChannel(ZigbeeChannel):
"""Channel for the zigbee power configuration cluster."""
@ -219,3 +371,31 @@ class PowerConfigurationChannel(ZigbeeChannel):
)
await self.get_attribute_value("battery_voltage", from_cache=from_cache)
await self.get_attribute_value("battery_quantity", from_cache=from_cache)
@ZIGBEE_CHANNEL_REGISTRY.register(general.PowerProfile.cluster_id)
class PowerProfile(ZigbeeChannel):
"""Power Profile channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.RSSILocation.cluster_id)
class RSSILocation(ZigbeeChannel):
"""RSSI Location channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.Scenes.cluster_id)
class Scenes(ZigbeeChannel):
"""Scenes channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(general.Time.cluster_id)
class Time(ZigbeeChannel):
"""Time channel."""
pass

View File

@ -6,9 +6,11 @@ https://home-assistant.io/components/zha/
"""
import logging
import zigpy.zcl.clusters.homeautomation as homeautomation
from homeassistant.helpers.dispatcher import async_dispatcher_send
from . import AttributeListeningChannel
from . import ZIGBEE_CHANNEL_REGISTRY, AttributeListeningChannel, ZigbeeChannel
from ..const import (
CHANNEL_ELECTRICAL_MEASUREMENT,
REPORT_CONFIG_DEFAULT,
@ -18,6 +20,35 @@ from ..const import (
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(homeautomation.ApplianceEventAlerts.cluster_id)
class ApplianceEventAlerts(ZigbeeChannel):
"""Appliance Event Alerts channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(homeautomation.ApplianceIdentification.cluster_id)
class ApplianceIdentification(ZigbeeChannel):
"""Appliance Identification channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(homeautomation.ApplianceStatistics.cluster_id)
class ApplianceStatistics(ZigbeeChannel):
"""Appliance Statistics channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(homeautomation.Diagnostic.cluster_id)
class Diagnostic(ZigbeeChannel):
"""Diagnostic channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(homeautomation.ElectricalMeasurement.cluster_id)
class ElectricalMeasurementChannel(AttributeListeningChannel):
"""Channel that polls active power level."""
@ -40,3 +71,10 @@ class ElectricalMeasurementChannel(AttributeListeningChannel):
"""Initialize channel."""
await self.get_attribute_value("active_power", from_cache=from_cache)
await super().async_initialize(from_cache)
@ZIGBEE_CHANNEL_REGISTRY.register(homeautomation.MeterIdentification.cluster_id)
class MeterIdentification(ZigbeeChannel):
"""Metering Identification channel."""
pass

View File

@ -6,15 +6,25 @@ https://home-assistant.io/components/zha/
"""
import logging
import zigpy.zcl.clusters.hvac as hvac
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_send
from . import ZigbeeChannel
from . import ZIGBEE_CHANNEL_REGISTRY, ZigbeeChannel
from ..const import REPORT_CONFIG_OP, SIGNAL_ATTR_UPDATED
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(hvac.Dehumidification.cluster_id)
class Dehumidification(ZigbeeChannel):
"""Dehumidification channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(hvac.Fan.cluster_id)
class FanChannel(ZigbeeChannel):
"""Fan channel."""
@ -59,3 +69,24 @@ class FanChannel(ZigbeeChannel):
"""Initialize channel."""
await self.get_attribute_value(self._value_attribute, from_cache=from_cache)
await super().async_initialize(from_cache)
@ZIGBEE_CHANNEL_REGISTRY.register(hvac.Pump.cluster_id)
class Pump(ZigbeeChannel):
"""Pump channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(hvac.Thermostat.cluster_id)
class Thermostat(ZigbeeChannel):
"""Thermostat channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(hvac.UserInterface.cluster_id)
class UserInterface(ZigbeeChannel):
"""User interface (thermostat) channel."""
pass

View File

@ -6,12 +6,22 @@ https://home-assistant.io/components/zha/
"""
import logging
from . import ZigbeeChannel
import zigpy.zcl.clusters.lighting as lighting
from . import ZIGBEE_CHANNEL_REGISTRY, ZigbeeChannel
from ..const import REPORT_CONFIG_DEFAULT
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(lighting.Ballast.cluster_id)
class Ballast(ZigbeeChannel):
"""Ballast channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(lighting.Color.cluster_id)
class ColorChannel(ZigbeeChannel):
"""Color channel."""

View File

@ -6,4 +6,15 @@ https://home-assistant.io/components/zha/
"""
import logging
import zigpy.zcl.clusters.lightlink as lightlink
from . import ZIGBEE_CHANNEL_REGISTRY, ZigbeeChannel
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(lightlink.LightLink.cluster_id)
class LightLink(ZigbeeChannel):
"""Lightlink channel."""
pass

View File

@ -6,4 +6,73 @@ https://home-assistant.io/components/zha/
"""
import logging
import zigpy.zcl.clusters.measurement as measurement
from . import ZIGBEE_CHANNEL_REGISTRY, AttributeListeningChannel
from ..const import (
REPORT_CONFIG_DEFAULT,
REPORT_CONFIG_IMMEDIATE,
REPORT_CONFIG_MAX_INT,
REPORT_CONFIG_MIN_INT,
)
_LOGGER = logging.getLogger(__name__)
@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)
class IlluminanceLevelSensing(AttributeListeningChannel):
"""Illuminance Level Sensing channel."""
REPORT_CONFIG = [{"attr": "level_status", "config": REPORT_CONFIG_DEFAULT}]
@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)
class OccupancySensing(AttributeListeningChannel):
"""Occupancy Sensing channel."""
REPORT_CONFIG = [{"attr": "occupancy", "config": REPORT_CONFIG_IMMEDIATE}]
@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)
class RelativeHumidity(AttributeListeningChannel):
"""Relative Humidity measurement channel."""
REPORT_CONFIG = [
{
"attr": "measured_value",
"config": (REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 50),
}
]
@ZIGBEE_CHANNEL_REGISTRY.register(measurement.TemperatureMeasurement.cluster_id)
class TemperatureMeasurement(AttributeListeningChannel):
"""Temperature measurement channel."""
REPORT_CONFIG = [
{
"attr": "measured_value",
"config": (REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 50),
}
]

View File

@ -6,4 +6,148 @@ https://home-assistant.io/components/zha/
"""
import logging
import zigpy.zcl.clusters.protocol as protocol
from ..channels import ZIGBEE_CHANNEL_REGISTRY, ZigbeeChannel
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.AnalogInputExtended.cluster_id)
class AnalogInputExtended(ZigbeeChannel):
"""Analog Input Extended channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.AnalogInputRegular.cluster_id)
class AnalogInputRegular(ZigbeeChannel):
"""Analog Input Regular channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.AnalogOutputExtended.cluster_id)
class AnalogOutputExtended(ZigbeeChannel):
"""Analog Output Regular channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.AnalogOutputRegular.cluster_id)
class AnalogOutputRegular(ZigbeeChannel):
"""Analog Output Regular channel."""
pass
@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)
class AnalogValueRegular(ZigbeeChannel):
"""Analog Value Regular channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.BacnetProtocolTunnel.cluster_id)
class BacnetProtocolTunnel(ZigbeeChannel):
"""Bacnet Protocol Tunnel channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryInputExtended.cluster_id)
class BinaryInputExtended(ZigbeeChannel):
"""Binary Input Extended channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryInputRegular.cluster_id)
class BinaryInputRegular(ZigbeeChannel):
"""Binary Input Regular channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryOutputExtended.cluster_id)
class BinaryOutputExtended(ZigbeeChannel):
"""Binary Output Extended channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryOutputRegular.cluster_id)
class BinaryOutputRegular(ZigbeeChannel):
"""Binary Output Regular channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryValueExtended.cluster_id)
class BinaryValueExtended(ZigbeeChannel):
"""Binary Value Extended channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.BinaryValueRegular.cluster_id)
class BinaryValueRegular(ZigbeeChannel):
"""Binary Value Regular channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.GenericTunnel.cluster_id)
class GenericTunnel(ZigbeeChannel):
"""Generic Tunnel channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.MultistateInputExtended.cluster_id)
class MultiStateInputExtended(ZigbeeChannel):
"""Multistate Input Extended channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.MultistateInputRegular.cluster_id)
class MultiStateInputRegular(ZigbeeChannel):
"""Multistate Input Regular channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.MultistateOutputExtended.cluster_id)
class MultiStateOutputExtended(ZigbeeChannel):
"""Multistate Output Extended channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.MultistateOutputRegular.cluster_id)
class MultiStateOutputRegular(ZigbeeChannel):
"""Multistate Output Regular channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.MultistateValueExtended.cluster_id)
class MultiStateValueExtended(ZigbeeChannel):
"""Multistate Value Extended channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(protocol.MultistateValueRegular.cluster_id)
class MultiStateValueRegular(ZigbeeChannel):
"""Multistate Value Regular channel."""
pass

View File

@ -1,52 +0,0 @@
"""
Channel registry module for Zigbee Home Automation.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/zha/
"""
from . import ZigbeeChannel
from .closures import DoorLockChannel
from .general import (
BasicChannel,
LevelControlChannel,
OnOffChannel,
PowerConfigurationChannel,
)
from .homeautomation import ElectricalMeasurementChannel
from .hvac import FanChannel
from .lighting import ColorChannel
from .security import IASZoneChannel
ZIGBEE_CHANNEL_REGISTRY = {}
def populate_channel_registry():
"""Populate the channel registry."""
from zigpy import zcl
ZIGBEE_CHANNEL_REGISTRY.update(
{
zcl.clusters.closures.DoorLock.cluster_id: DoorLockChannel,
zcl.clusters.general.Alarms.cluster_id: ZigbeeChannel,
zcl.clusters.general.ApplianceControl.cluster_id: ZigbeeChannel,
zcl.clusters.general.Basic.cluster_id: BasicChannel,
zcl.clusters.general.Commissioning.cluster_id: ZigbeeChannel,
zcl.clusters.general.GreenPowerProxy.cluster_id: ZigbeeChannel,
zcl.clusters.general.Groups.cluster_id: ZigbeeChannel,
zcl.clusters.general.Identify.cluster_id: ZigbeeChannel,
zcl.clusters.general.LevelControl.cluster_id: LevelControlChannel,
zcl.clusters.general.OnOff.cluster_id: OnOffChannel,
zcl.clusters.general.OnOffConfiguration.cluster_id: ZigbeeChannel,
zcl.clusters.general.Ota.cluster_id: ZigbeeChannel,
zcl.clusters.general.Partition.cluster_id: ZigbeeChannel,
zcl.clusters.general.PollControl.cluster_id: ZigbeeChannel,
zcl.clusters.general.PowerConfiguration.cluster_id: PowerConfigurationChannel,
zcl.clusters.general.PowerProfile.cluster_id: ZigbeeChannel,
zcl.clusters.general.Scenes.cluster_id: ZigbeeChannel,
zcl.clusters.homeautomation.ElectricalMeasurement.cluster_id: ElectricalMeasurementChannel,
zcl.clusters.hvac.Fan.cluster_id: FanChannel,
zcl.clusters.lighting.Color.cluster_id: ColorChannel,
zcl.clusters.lightlink.LightLink.cluster_id: ZigbeeChannel,
zcl.clusters.security.IasZone.cluster_id: IASZoneChannel,
}
)

View File

@ -6,15 +6,32 @@ https://home-assistant.io/components/zha/
"""
import logging
import zigpy.zcl.clusters.security as security
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_send
from . import ZigbeeChannel
from . import ZIGBEE_CHANNEL_REGISTRY, ZigbeeChannel
from ..const import SIGNAL_ATTR_UPDATED
_LOGGER = logging.getLogger(__name__)
@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)
class IasWd(ZigbeeChannel):
"""IAS Warning Device channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(security.IasZone.cluster_id)
class IASZoneChannel(ZigbeeChannel):
"""Channel for the IASZone Zigbee cluster."""

View File

@ -6,4 +6,93 @@ https://home-assistant.io/components/zha/
"""
import logging
import zigpy.zcl.clusters.smartenergy as smartenergy
from ..channels import ZIGBEE_CHANNEL_REGISTRY, AttributeListeningChannel, ZigbeeChannel
from ..const import REPORT_CONFIG_DEFAULT
_LOGGER = logging.getLogger(__name__)
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Calendar.cluster_id)
class Calendar(ZigbeeChannel):
"""Calendar channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.DeviceManagement.cluster_id)
class DeviceManagement(ZigbeeChannel):
"""Device Management channel."""
pass
@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)
class EnergyManagement(ZigbeeChannel):
"""Energy Management channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Events.cluster_id)
class Events(ZigbeeChannel):
"""Event channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.KeyEstablishment.cluster_id)
class KeyEstablishment(ZigbeeChannel):
"""Key Establishment channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.MduPairing.cluster_id)
class MduPairing(ZigbeeChannel):
"""Pairing channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Messaging.cluster_id)
class Messaging(ZigbeeChannel):
"""Messaging channel."""
pass
@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)
class Prepayment(ZigbeeChannel):
"""Prepayment channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Price.cluster_id)
class Price(ZigbeeChannel):
"""Price channel."""
pass
@ZIGBEE_CHANNEL_REGISTRY.register(smartenergy.Tunneling.cluster_id)
class Tunneling(ZigbeeChannel):
"""Tunneling channel."""
pass

View File

@ -13,8 +13,12 @@ from homeassistant.components.sensor import DOMAIN as SENSOR
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_send
from .channels import AttributeListeningChannel, EventRelayChannel, ZDOChannel
from .channels.registry import ZIGBEE_CHANNEL_REGISTRY
from .channels import (
ZIGBEE_CHANNEL_REGISTRY,
AttributeListeningChannel,
EventRelayChannel,
ZDOChannel,
)
from .const import (
COMPONENTS,
CONF_DEVICE_CONFIG,

View File

@ -16,8 +16,6 @@ from homeassistant.components.switch import DOMAIN as SWITCH
from .const import (
CONTROLLER,
REPORT_CONFIG_ASAP,
REPORT_CONFIG_DEFAULT,
REPORT_CONFIG_IMMEDIATE,
REPORT_CONFIG_MAX_INT,
REPORT_CONFIG_MIN_INT,
SENSOR_ACCELERATION,
@ -143,18 +141,6 @@ def establish_device_mappings():
CLUSTER_REPORT_CONFIGS.update(
{
zcl.clusters.measurement.RelativeHumidity.cluster_id: [
{
"attr": "measured_value",
"config": (REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 50),
}
],
zcl.clusters.measurement.TemperatureMeasurement.cluster_id: [
{
"attr": "measured_value",
"config": (REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 50),
}
],
SMARTTHINGS_ACCELERATION_CLUSTER: [
{"attr": "acceleration", "config": REPORT_CONFIG_ASAP},
{"attr": "x_axis", "config": REPORT_CONFIG_ASAP},
@ -167,18 +153,6 @@ def establish_device_mappings():
"config": (REPORT_CONFIG_MIN_INT, REPORT_CONFIG_MAX_INT, 50),
}
],
zcl.clusters.measurement.PressureMeasurement.cluster_id: [
{"attr": "measured_value", "config": REPORT_CONFIG_DEFAULT}
],
zcl.clusters.measurement.IlluminanceMeasurement.cluster_id: [
{"attr": "measured_value", "config": REPORT_CONFIG_DEFAULT}
],
zcl.clusters.smartenergy.Metering.cluster_id: [
{"attr": "instantaneous_demand", "config": REPORT_CONFIG_DEFAULT}
],
zcl.clusters.measurement.OccupancySensing.cluster_id: [
{"attr": "occupancy", "config": REPORT_CONFIG_IMMEDIATE}
],
}
)

View File

@ -6,9 +6,6 @@ from homeassistant.components.zha.core.const import DOMAIN, DATA_ZHA, COMPONENTS
from homeassistant.helpers.device_registry import async_get_registry as get_dev_reg
from homeassistant.components.zha.core.gateway import ZHAGateway
from homeassistant.components.zha.core.registries import establish_device_mappings
from homeassistant.components.zha.core.channels.registry import (
populate_channel_registry,
)
from .common import async_setup_entry
from homeassistant.components.zha.core.store import async_get_registry
@ -29,7 +26,6 @@ async def zha_gateway_fixture(hass, config_entry):
Create a ZHAGateway object that can be used to interact with as if we
had a real zigbee network running.
"""
populate_channel_registry()
establish_device_mappings()
for component in COMPONENTS:
hass.data[DATA_ZHA][component] = hass.data[DATA_ZHA].get(component, {})

View File

@ -1,9 +1,9 @@
"""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.channels.registry as channel_reg
import homeassistant.components.zha.core.device as zha_device
from .common import make_device
@ -33,6 +33,15 @@ def nwk():
(0x0007, 1, {}),
(0x0008, 1, {"current_level"}),
(0x0009, 1, {}),
(0x000C, 1, {"present_value"}),
(0x000D, 1, {"present_value"}),
(0x000E, 1, {"present_value"}),
(0x000D, 1, {"present_value"}),
(0x0010, 1, {"present_value"}),
(0x0011, 1, {"present_value"}),
(0x0012, 1, {"present_value"}),
(0x0013, 1, {"present_value"}),
(0x0014, 1, {"present_value"}),
(0x0015, 1, {}),
(0x0016, 1, {}),
(0x0019, 1, {}),
@ -44,8 +53,10 @@ def nwk():
(0x0202, 1, {"fan_mode"}),
(0x0300, 1, {"current_x", "current_y", "color_temperature"}),
(0x0400, 1, {"measured_value"}),
(0x0401, 1, {"level_status"}),
(0x0402, 1, {"measured_value"}),
(0x0403, 1, {"measured_value"}),
(0x0404, 1, {"measured_value"}),
(0x0405, 1, {"measured_value"}),
(0x0406, 1, {"occupancy"}),
(0x0702, 1, {"instantaneous_demand"}),
@ -66,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 = channel_reg.ZIGBEE_CHANNEL_REGISTRY.get(
channel_class = channels.ZIGBEE_CHANNEL_REGISTRY.get(
cluster_id, channels.AttributeListeningChannel
)
channel = channel_class(cluster, zha_dev)
@ -125,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 = channel_reg.ZIGBEE_CHANNEL_REGISTRY.get(
channel_class = homeassistant.components.zha.core.channels.ZIGBEE_CHANNEL_REGISTRY.get(
cluster_id, channels.AttributeListeningChannel
)
channel = channel_class(cluster, zha_dev)
@ -134,3 +145,11 @@ async def test_out_channel_config(cluster_id, bind_count, zha_gateway, hass):
assert cluster.bind.call_count == bind_count
assert cluster.configure_reporting.call_count == 0
def test_channel_registry():
"""Test ZIGBEE Channel Registry."""
for cluster_id, channel in channels.ZIGBEE_CHANNEL_REGISTRY.items():
assert isinstance(cluster_id, int)
assert 0 <= cluster_id <= 0xFFFF
assert issubclass(channel, channels.ZigbeeChannel)