From fb3bb8220b99ad0b3a7ad23accf7c7a31478c52f Mon Sep 17 00:00:00 2001 From: Alexei Chetroi Date: Sat, 21 Dec 2019 16:26:58 -0500 Subject: [PATCH] Implement ZHA entity classes registry (#30108) * ZHA Entity registry. Match a zha_device and channels to a ZHA entity. * Refactor ZHA sensor to use registry. * Remove sensor_types registry. * Fix ZHA device tracker battery remaining. * Remove should_poll/force_update attributes. * Fix binary_sensor regression. * isort. * Pylint. * Don't access protected members. * Address comments and fix spelling. * Make pylint happy again. --- homeassistant/components/zha/binary_sensor.py | 4 +- .../components/zha/core/channels/__init__.py | 2 - homeassistant/components/zha/core/const.py | 18 +- .../components/zha/core/discovery.py | 7 - .../components/zha/core/registries.py | 118 +++++-- .../components/zha/device_tracker.py | 4 +- homeassistant/components/zha/sensor.py | 289 +++++++++--------- tests/components/zha/test_registries.py | 165 ++++++++++ 8 files changed, 415 insertions(+), 192 deletions(-) create mode 100644 tests/components/zha/test_registries.py diff --git a/homeassistant/components/zha/binary_sensor.py b/homeassistant/components/zha/binary_sensor.py index 24c2b92e739..e6176fe9da3 100644 --- a/homeassistant/components/zha/binary_sensor.py +++ b/homeassistant/components/zha/binary_sensor.py @@ -18,7 +18,7 @@ from homeassistant.core import callback from homeassistant.helpers.dispatcher import async_dispatcher_connect from .core.const import ( - CHANNEL_ATTRIBUTE, + CHANNEL_OCCUPANCY, CHANNEL_ON_OFF, CHANNEL_ZONE, DATA_ZHA, @@ -111,7 +111,7 @@ class BinarySensor(ZhaEntity, BinarySensorDevice): self._device_state_attributes = {} self._zone_channel = self.cluster_channels.get(CHANNEL_ZONE) self._on_off_channel = self.cluster_channels.get(CHANNEL_ON_OFF) - self._attr_channel = self.cluster_channels.get(CHANNEL_ATTRIBUTE) + self._attr_channel = self.cluster_channels.get(CHANNEL_OCCUPANCY) self._zha_sensor_type = kwargs[SENSOR_TYPE] async def _determine_device_class(self): diff --git a/homeassistant/components/zha/core/channels/__init__.py b/homeassistant/components/zha/core/channels/__init__.py index 715156c43ed..4013f05e0b6 100644 --- a/homeassistant/components/zha/core/channels/__init__.py +++ b/homeassistant/components/zha/core/channels/__init__.py @@ -17,7 +17,6 @@ from homeassistant.core import callback from homeassistant.helpers.dispatcher import async_dispatcher_send from ..const import ( - CHANNEL_ATTRIBUTE, CHANNEL_EVENT_RELAY, CHANNEL_ZDO, REPORT_CONFIG_DEFAULT, @@ -280,7 +279,6 @@ class ZigbeeChannel(LogMixin): 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): diff --git a/homeassistant/components/zha/core/const.py b/homeassistant/components/zha/core/const.py index 24c0126ba60..6c991a319ac 100644 --- a/homeassistant/components/zha/core/const.py +++ b/homeassistant/components/zha/core/const.py @@ -50,10 +50,16 @@ CHANNEL_DOORLOCK = "door_lock" CHANNEL_ELECTRICAL_MEASUREMENT = "electrical_measurement" CHANNEL_EVENT_RELAY = "event_relay" CHANNEL_FAN = "fan" +CHANNEL_HUMIDITY = "humidity" CHANNEL_IAS_WD = "ias_wd" +CHANNEL_ILLUMINANCE = "illuminance" CHANNEL_LEVEL = ATTR_LEVEL +CHANNEL_OCCUPANCY = "occupancy" CHANNEL_ON_OFF = "on_off" CHANNEL_POWER_CONFIGURATION = "power" +CHANNEL_PRESSURE = "pressure" +CHANNEL_SMARTENERGY_METERING = "smartenergy_metering" +CHANNEL_TEMPERATURE = "temperature" CHANNEL_ZDO = "zdo" CHANNEL_ZONE = ZONE = "ias_zone" @@ -166,15 +172,15 @@ REPORT_CONFIG_OP = ( SENSOR_ACCELERATION = "acceleration" SENSOR_BATTERY = "battery" -SENSOR_ELECTRICAL_MEASUREMENT = "electrical_measurement" +SENSOR_ELECTRICAL_MEASUREMENT = CHANNEL_ELECTRICAL_MEASUREMENT SENSOR_GENERIC = "generic" -SENSOR_HUMIDITY = "humidity" -SENSOR_ILLUMINANCE = "illuminance" +SENSOR_HUMIDITY = CHANNEL_HUMIDITY +SENSOR_ILLUMINANCE = CHANNEL_ILLUMINANCE SENSOR_METERING = "metering" -SENSOR_OCCUPANCY = "occupancy" +SENSOR_OCCUPANCY = CHANNEL_OCCUPANCY SENSOR_OPENING = "opening" -SENSOR_PRESSURE = "pressure" -SENSOR_TEMPERATURE = "temperature" +SENSOR_PRESSURE = CHANNEL_PRESSURE +SENSOR_TEMPERATURE = CHANNEL_TEMPERATURE SENSOR_TYPE = "sensor_type" SIGNAL_ATTR_UPDATED = "attribute_updated" diff --git a/homeassistant/components/zha/core/discovery.py b/homeassistant/components/zha/core/discovery.py index e23862a7d3e..108bd841252 100644 --- a/homeassistant/components/zha/core/discovery.py +++ b/homeassistant/components/zha/core/discovery.py @@ -12,7 +12,6 @@ from zigpy.zcl.clusters.general import OnOff, PowerConfiguration from homeassistant import const as ha_const from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR -from homeassistant.components.sensor import DOMAIN as SENSOR from homeassistant.core import callback from homeassistant.helpers.dispatcher import async_dispatcher_send @@ -21,7 +20,6 @@ from .const import ( COMPONENTS, CONF_DEVICE_CONFIG, DATA_ZHA, - SENSOR_GENERIC, SENSOR_TYPE, UNKNOWN, ZHA_DISCOVERY_NEW, @@ -34,7 +32,6 @@ from .registries import ( EVENT_RELAY_CLUSTERS, OUTPUT_CHANNEL_ONLY_CLUSTERS, REMOTE_DEVICE_TYPES, - SENSOR_TYPES, SINGLE_INPUT_CLUSTER_DEVICE_CLASS, SINGLE_OUTPUT_CLUSTER_DEVICE_CLASS, ZIGBEE_CHANNEL_REGISTRY, @@ -291,10 +288,6 @@ def _async_handle_single_cluster_match( "component": component, } - if component == SENSOR: - discovery_info.update( - {SENSOR_TYPE: SENSOR_TYPES.get(cluster.cluster_id, SENSOR_GENERIC)} - ) if component == BINARY_SENSOR: discovery_info.update( {SENSOR_TYPE: BINARY_SENSOR_TYPES.get(cluster.cluster_id, UNKNOWN)} diff --git a/homeassistant/components/zha/core/registries.py b/homeassistant/components/zha/core/registries.py index 13688a6c420..c2d3b13e375 100644 --- a/homeassistant/components/zha/core/registries.py +++ b/homeassistant/components/zha/core/registries.py @@ -5,7 +5,9 @@ For more details about this component, please refer to the documentation at https://home-assistant.io/integrations/zha/ """ import collections +from typing import Callable, Set +import attr import bellows.ezsp import bellows.zigbee.application import zigpy.profiles.zha @@ -31,21 +33,14 @@ from . import channels # noqa: F401 pylint: disable=unused-import from .const import ( CONTROLLER, SENSOR_ACCELERATION, - SENSOR_BATTERY, - SENSOR_ELECTRICAL_MEASUREMENT, - SENSOR_HUMIDITY, - SENSOR_ILLUMINANCE, - SENSOR_METERING, SENSOR_OCCUPANCY, SENSOR_OPENING, - SENSOR_PRESSURE, - SENSOR_TEMPERATURE, ZHA_GW_RADIO, ZHA_GW_RADIO_DESCRIPTION, ZONE, RadioType, ) -from .decorators import DictRegistry, SetRegistry +from .decorators import CALLABLE_T, DictRegistry, SetRegistry BINARY_SENSOR_CLUSTERS = SetRegistry() BINARY_SENSOR_TYPES = {} @@ -60,7 +55,6 @@ LIGHT_CLUSTERS = SetRegistry() OUTPUT_CHANNEL_ONLY_CLUSTERS = SetRegistry() RADIO_TYPES = {} REMOTE_DEVICE_TYPES = collections.defaultdict(list) -SENSOR_TYPES = {} SINGLE_INPUT_CLUSTER_DEVICE_CLASS = {} SINGLE_OUTPUT_CLUSTER_DEVICE_CLASS = {} SWITCH_CLUSTERS = SetRegistry() @@ -176,19 +170,6 @@ def establish_device_mappings(): {zcl.clusters.general.OnOff: BINARY_SENSOR} ) - SENSOR_TYPES.update( - { - SMARTTHINGS_HUMIDITY_CLUSTER: SENSOR_HUMIDITY, - zcl.clusters.general.PowerConfiguration.cluster_id: SENSOR_BATTERY, - zcl.clusters.homeautomation.ElectricalMeasurement.cluster_id: SENSOR_ELECTRICAL_MEASUREMENT, - zcl.clusters.measurement.IlluminanceMeasurement.cluster_id: SENSOR_ILLUMINANCE, - zcl.clusters.measurement.PressureMeasurement.cluster_id: SENSOR_PRESSURE, - zcl.clusters.measurement.RelativeHumidity.cluster_id: SENSOR_HUMIDITY, - zcl.clusters.measurement.TemperatureMeasurement.cluster_id: SENSOR_TEMPERATURE, - zcl.clusters.smartenergy.Metering.cluster_id: SENSOR_METERING, - } - ) - zha = zigpy.profiles.zha REMOTE_DEVICE_TYPES[zha.PROFILE_ID].append(zha.DeviceType.COLOR_CONTROLLER) REMOTE_DEVICE_TYPES[zha.PROFILE_ID].append(zha.DeviceType.COLOR_DIMMER_SWITCH) @@ -207,3 +188,96 @@ def establish_device_mappings(): REMOTE_DEVICE_TYPES[zll.PROFILE_ID].append(zll.DeviceType.CONTROL_BRIDGE) REMOTE_DEVICE_TYPES[zll.PROFILE_ID].append(zll.DeviceType.CONTROLLER) REMOTE_DEVICE_TYPES[zll.PROFILE_ID].append(zll.DeviceType.SCENE_CONTROLLER) + + +@attr.s(frozen=True) +class MatchRule: + """Match a ZHA Entity to a channel name or generic id.""" + + channel_names: Set[str] = attr.ib(factory=frozenset, converter=frozenset) + generic_ids: Set[str] = attr.ib(factory=frozenset, converter=frozenset) + manufacturer: str = attr.ib(default=None) + model: str = attr.ib(default=None) + + +class ZHAEntityRegistry: + """Channel to ZHA Entity mapping.""" + + def __init__(self): + """Initialize Registry instance.""" + self._strict_registry = collections.defaultdict(dict) + self._loose_registry = collections.defaultdict(dict) + + def get_entity( + self, component: str, zha_device, chnls: list, default: CALLABLE_T = None + ) -> CALLABLE_T: + """Match a ZHA Channels to a ZHA Entity class.""" + for match in self._strict_registry[component]: + if self._strict_matched(zha_device, chnls, match): + return self._strict_registry[component][match] + + return default + + def strict_match( + self, component: str, rule: MatchRule + ) -> Callable[[CALLABLE_T], CALLABLE_T]: + """Decorate a strict match rule.""" + + def decorator(zha_ent: CALLABLE_T) -> CALLABLE_T: + """Register a strict match rule. + + All non empty fields of a match rule must match. + """ + self._strict_registry[component][rule] = zha_ent + return zha_ent + + return decorator + + def loose_match( + self, component: str, rule: MatchRule + ) -> Callable[[CALLABLE_T], CALLABLE_T]: + """Decorate a loose match rule.""" + + def decorator(zha_entity: CALLABLE_T) -> CALLABLE_T: + """Register a loose match rule. + + All non empty fields of a match rule must match. + """ + self._loose_registry[component][rule] = zha_entity + return zha_entity + + return decorator + + def _strict_matched(self, zha_device, chnls: dict, rule: MatchRule) -> bool: + """Return True if this device matches the criteria.""" + return all(self._matched(zha_device, chnls, rule)) + + def _loose_matched(self, zha_device, chnls: dict, rule: MatchRule) -> bool: + """Return True if this device matches the criteria.""" + return any(self._matched(zha_device, chnls, rule)) + + @staticmethod + def _matched(zha_device, chnls: list, rule: MatchRule) -> bool: + """Return a list of field matches.""" + if not any(attr.asdict(rule).values()): + return [False] + + matches = [] + if rule.channel_names: + channel_names = {ch.name for ch in chnls} + matches.append(rule.channel_names.issubset(channel_names)) + + if rule.generic_ids: + all_generic_ids = {ch.generic_id for ch in chnls} + matches.append(rule.generic_ids.issubset(all_generic_ids)) + + if rule.manufacturer: + matches.append(zha_device.manufacturer == rule.manufacturer) + + if rule.model: + matches.append(zha_device.model == rule.model) + + return matches + + +ZHA_ENTITIES = ZHAEntityRegistry() diff --git a/homeassistant/components/zha/device_tracker.py b/homeassistant/components/zha/device_tracker.py index 60a1f6c3c40..e7663b35686 100644 --- a/homeassistant/components/zha/device_tracker.py +++ b/homeassistant/components/zha/device_tracker.py @@ -15,7 +15,7 @@ from .core.const import ( ZHA_DISCOVERY_NEW, ) from .entity import ZhaEntity -from .sensor import battery_percentage_remaining_formatter +from .sensor import Battery _LOGGER = logging.getLogger(__name__) @@ -100,7 +100,7 @@ class ZHADeviceScannerEntity(ScannerEntity, ZhaEntity): """Handle tracking.""" self.debug("battery_percentage_remaining updated: %s", value) self._connected = True - self._battery_level = battery_percentage_remaining_formatter(value) + self._battery_level = Battery.formatter(value) self.async_schedule_update_ha_state() @property diff --git a/homeassistant/components/zha/sensor.py b/homeassistant/components/zha/sensor.py index b260dfc5459..133e82e6914 100644 --- a/homeassistant/components/zha/sensor.py +++ b/homeassistant/components/zha/sensor.py @@ -1,4 +1,5 @@ """Sensors on Zigbee Home Automation networks.""" +import functools import logging import numbers @@ -16,25 +17,20 @@ from homeassistant.core import callback from homeassistant.helpers.dispatcher import async_dispatcher_connect from .core.const import ( - CHANNEL_ATTRIBUTE, CHANNEL_ELECTRICAL_MEASUREMENT, + CHANNEL_HUMIDITY, + CHANNEL_ILLUMINANCE, CHANNEL_POWER_CONFIGURATION, + CHANNEL_PRESSURE, + CHANNEL_SMARTENERGY_METERING, + CHANNEL_TEMPERATURE, DATA_ZHA, DATA_ZHA_DISPATCHERS, - SENSOR_BATTERY, - SENSOR_ELECTRICAL_MEASUREMENT, - SENSOR_GENERIC, - SENSOR_HUMIDITY, - SENSOR_ILLUMINANCE, - SENSOR_METERING, - SENSOR_PRESSURE, - SENSOR_TEMPERATURE, - SENSOR_TYPE, SIGNAL_ATTR_UPDATED, SIGNAL_STATE_ATTR, - UNKNOWN, ZHA_DISCOVERY_NEW, ) +from .core.registries import SMARTTHINGS_HUMIDITY_CLUSTER, ZHA_ENTITIES, MatchRule from .entity import ZhaEntity PARALLEL_UPDATES = 5 @@ -56,115 +52,8 @@ BATTERY_SIZES = { 255: "Unknown", } - -# Formatter functions -def pass_through_formatter(value): - """No op update function.""" - return value - - -def illuminance_formatter(value): - """Convert Illimination data.""" - if value is None: - return None - return round(pow(10, ((value - 1) / 10000)), 1) - - -def temperature_formatter(value): - """Convert temperature data.""" - if value is None: - return None - return round(value / 100, 1) - - -def humidity_formatter(value): - """Return the state of the entity.""" - if value is None: - return None - return round(float(value) / 100, 1) - - -def active_power_formatter(value): - """Return the state of the entity.""" - if value is None: - return None - return round(float(value) / 10, 1) - - -def pressure_formatter(value): - """Return the state of the entity.""" - if value is None: - return None - - return round(float(value)) - - -def battery_percentage_remaining_formatter(value): - """Return the state of the entity.""" - # per zcl specs battery percent is reported at 200% ¯\_(ツ)_/¯ - if not isinstance(value, numbers.Number) or value == -1: - return value - value = value / 2 - value = int(round(value)) - return value - - -async def async_battery_device_state_attr_provider(channel): - """Return device statr attrs for battery sensors.""" - state_attrs = {} - battery_size = await channel.get_attribute_value("battery_size") - if battery_size is not None: - state_attrs["battery_size"] = BATTERY_SIZES.get(battery_size, "Unknown") - battery_quantity = await channel.get_attribute_value("battery_quantity") - if battery_quantity is not None: - state_attrs["battery_quantity"] = battery_quantity - return state_attrs - - -FORMATTER_FUNC_REGISTRY = { - SENSOR_HUMIDITY: humidity_formatter, - SENSOR_TEMPERATURE: temperature_formatter, - SENSOR_PRESSURE: pressure_formatter, - SENSOR_ELECTRICAL_MEASUREMENT: active_power_formatter, - SENSOR_ILLUMINANCE: illuminance_formatter, - SENSOR_GENERIC: pass_through_formatter, - SENSOR_BATTERY: battery_percentage_remaining_formatter, -} - -UNIT_REGISTRY = { - SENSOR_HUMIDITY: "%", - SENSOR_TEMPERATURE: TEMP_CELSIUS, - SENSOR_PRESSURE: "hPa", - SENSOR_ILLUMINANCE: "lx", - SENSOR_ELECTRICAL_MEASUREMENT: POWER_WATT, - SENSOR_GENERIC: None, - SENSOR_BATTERY: "%", -} - -CHANNEL_REGISTRY = { - SENSOR_ELECTRICAL_MEASUREMENT: CHANNEL_ELECTRICAL_MEASUREMENT, - SENSOR_BATTERY: CHANNEL_POWER_CONFIGURATION, -} - -POLLING_REGISTRY = {SENSOR_ELECTRICAL_MEASUREMENT: True} - -FORCE_UPDATE_REGISTRY = {SENSOR_ELECTRICAL_MEASUREMENT: False} - -DEVICE_CLASS_REGISTRY = { - UNKNOWN: None, - SENSOR_HUMIDITY: DEVICE_CLASS_HUMIDITY, - SENSOR_TEMPERATURE: DEVICE_CLASS_TEMPERATURE, - SENSOR_PRESSURE: DEVICE_CLASS_PRESSURE, - SENSOR_ILLUMINANCE: DEVICE_CLASS_ILLUMINANCE, - SENSOR_METERING: DEVICE_CLASS_POWER, - SENSOR_ELECTRICAL_MEASUREMENT: DEVICE_CLASS_POWER, - SENSOR_BATTERY: DEVICE_CLASS_BATTERY, -} - - -DEVICE_STATE_ATTR_PROVIDER_REGISTRY = { - SENSOR_BATTERY: async_battery_device_state_attr_provider -} +CHANNEL_ST_HUMIDITY_CLUSTER = f"channel_0x{SMARTTHINGS_HUMIDITY_CLUSTER:04x}" +STRICT_MATCH = functools.partial(ZHA_ENTITIES.strict_match, DOMAIN) async def async_setup_platform(hass, config, async_add_entities, discovery_info=None): @@ -206,43 +95,34 @@ async def _async_setup_entities( async def make_sensor(discovery_info): """Create ZHA sensors factory.""" - return Sensor(**discovery_info) + + zha_dev = discovery_info["zha_device"] + channels = discovery_info["channels"] + + entity = ZHA_ENTITIES.get_entity(DOMAIN, zha_dev, channels, Sensor) + return entity(**discovery_info) class Sensor(ZhaEntity): """Base ZHA sensor.""" + _decimals = 1 + _device_class = None + _divisor = 1 _domain = DOMAIN + _multiplier = 1 + _unit = None def __init__(self, unique_id, zha_device, channels, **kwargs): """Init this sensor.""" super().__init__(unique_id, zha_device, channels, **kwargs) - self._sensor_type = kwargs.get(SENSOR_TYPE, SENSOR_GENERIC) - self._channel = self.cluster_channels.get( - CHANNEL_REGISTRY.get(self._sensor_type, CHANNEL_ATTRIBUTE) - ) - if self._sensor_type == SENSOR_METERING: - self._unit = self._channel.unit_of_measurement - self._formatter_function = self._channel.formatter_function - else: - self._unit = UNIT_REGISTRY.get(self._sensor_type) - self._formatter_function = FORMATTER_FUNC_REGISTRY.get( - self._sensor_type, pass_through_formatter - ) - self._force_update = FORCE_UPDATE_REGISTRY.get(self._sensor_type, False) - self._should_poll = POLLING_REGISTRY.get(self._sensor_type, False) - self._device_class = DEVICE_CLASS_REGISTRY.get(self._sensor_type, None) - self.state_attr_provider = DEVICE_STATE_ATTR_PROVIDER_REGISTRY.get( - self._sensor_type, None - ) + self._channel = channels[0] async def async_added_to_hass(self): """Run when about to be added to hass.""" await super().async_added_to_hass() - if self.state_attr_provider is not None: - self._device_state_attributes = await self.state_attr_provider( - self._channel - ) + self._device_state_attributes = await self.async_state_attr_provider() + await self.async_accept_signal( self._channel, SIGNAL_ATTR_UPDATED, self.async_set_state ) @@ -271,14 +151,9 @@ class Sensor(ZhaEntity): def async_set_state(self, state): """Handle state update from channel.""" - # this is necessary because HA saves the unit based on what shows in - # the UI and not based on what the sensor has configured so we need - # to flip it back after state restoration - if self._sensor_type == SENSOR_METERING: - self._unit = self._channel.unit_of_measurement - else: - self._unit = UNIT_REGISTRY.get(self._sensor_type) - self._state = self._formatter_function(state) + if state is not None: + state = self.formatter(state) + self._state = state self.async_schedule_update_ha_state() @callback @@ -286,3 +161,115 @@ class Sensor(ZhaEntity): """Restore previous state.""" self._state = last_state.state self._unit = last_state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) + + @callback + async def async_state_attr_provider(self): + """Initialize device state attributes.""" + return {} + + def formatter(self, value): + """Numeric pass-through formatter.""" + if self._decimals > 0: + return round( + float(value * self._multiplier) / self._divisor, self._decimals + ) + return round(float(value * self._multiplier) / self._divisor) + + +@STRICT_MATCH(MatchRule(channel_names={CHANNEL_POWER_CONFIGURATION})) +class Battery(Sensor): + """Battery sensor of power configuration cluster.""" + + _device_class = DEVICE_CLASS_BATTERY + _unit = "%" + + @staticmethod + def formatter(value): + """Return the state of the entity.""" + # per zcl specs battery percent is reported at 200% ¯\_(ツ)_/¯ + if not isinstance(value, numbers.Number) or value == -1: + return value + value = round(value / 2) + return value + + async def async_state_attr_provider(self): + """Return device state attrs for battery sensors.""" + state_attrs = {} + battery_size = await self._channel.get_attribute_value("battery_size") + if battery_size is not None: + state_attrs["battery_size"] = BATTERY_SIZES.get(battery_size, "Unknown") + battery_quantity = await self._channel.get_attribute_value("battery_quantity") + if battery_quantity is not None: + state_attrs["battery_quantity"] = battery_quantity + return state_attrs + + +@STRICT_MATCH(MatchRule(channel_names={CHANNEL_ELECTRICAL_MEASUREMENT})) +class ElectricalMeasurement(Sensor): + """Active power measurement.""" + + _device_class = DEVICE_CLASS_POWER + _divisor = 10 + _unit = POWER_WATT + + @property + def should_poll(self) -> bool: + """Return True if HA needs to poll for state changes.""" + return True + + +@STRICT_MATCH(MatchRule(generic_ids={CHANNEL_ST_HUMIDITY_CLUSTER})) +@STRICT_MATCH(MatchRule(channel_names={CHANNEL_HUMIDITY})) +class Humidity(Sensor): + """Humidity sensor.""" + + _device_class = DEVICE_CLASS_HUMIDITY + _divisor = 100 + _unit = "%" + + +@STRICT_MATCH(MatchRule(channel_names={CHANNEL_ILLUMINANCE})) +class Illuminance(Sensor): + """Illuminance Sensor.""" + + _device_class = DEVICE_CLASS_ILLUMINANCE + _unit = "lx" + + @staticmethod + def formatter(value): + """Convert illumination data.""" + return round(pow(10, ((value - 1) / 10000)), 1) + + +@STRICT_MATCH(MatchRule(channel_names={CHANNEL_SMARTENERGY_METERING})) +class SmartEnergyMetering(Sensor): + """Metering sensor.""" + + _device_class = DEVICE_CLASS_POWER + + def formatter(self, value): + """Pass through channel formatter.""" + return self._channel.formatter_function(value) + + @property + def unit_of_measurement(self): + """Return Unit of measurement.""" + return self._channel.unit_of_measurement + + +@STRICT_MATCH(MatchRule(channel_names={CHANNEL_PRESSURE})) +class Pressure(Sensor): + """Pressure sensor.""" + + _device_class = DEVICE_CLASS_PRESSURE + _decimals = 0 + _unit = "hPa" + + +@STRICT_MATCH(MatchRule(channel_names={CHANNEL_TEMPERATURE})) +class Temperature(Sensor): + """Temperature Sensor.""" + + _device_class = DEVICE_CLASS_TEMPERATURE + _divisor = 100 + _unit = TEMP_CELSIUS diff --git a/tests/components/zha/test_registries.py b/tests/components/zha/test_registries.py new file mode 100644 index 00000000000..a0eef355229 --- /dev/null +++ b/tests/components/zha/test_registries.py @@ -0,0 +1,165 @@ +"""Test ZHA registries.""" +from unittest import mock + +import pytest + +import homeassistant.components.zha.core.registries as registries + +MANUFACTURER = "mock manufacturer" +MODEL = "mock model" + + +@pytest.fixture +def zha_device(): + """Return a mock of ZHA device.""" + dev = mock.MagicMock() + dev.manufacturer = MANUFACTURER + dev.model = MODEL + return dev + + +@pytest.fixture +def channels(): + """Return a mock of channels.""" + + def channel(name, chan_id): + ch = mock.MagicMock() + ch.name = name + ch.generic_id = chan_id + return ch + + return [channel("level", "channel_0x0008"), channel("on_off", "channel_0x0006")] + + +@pytest.mark.parametrize( + "rule, matched", + [ + (registries.MatchRule(), False), + (registries.MatchRule(channel_names={"level"}), True), + (registries.MatchRule(channel_names={"level", "no match"}), False), + (registries.MatchRule(channel_names={"on_off"}), True), + (registries.MatchRule(channel_names={"on_off", "no match"}), False), + (registries.MatchRule(channel_names={"on_off", "level"}), True), + (registries.MatchRule(channel_names={"on_off", "level", "no match"}), False), + # test generic_id matching + (registries.MatchRule(generic_ids={"channel_0x0006"}), True), + (registries.MatchRule(generic_ids={"channel_0x0008"}), True), + (registries.MatchRule(generic_ids={"channel_0x0006", "channel_0x0008"}), True), + ( + registries.MatchRule( + generic_ids={"channel_0x0006", "channel_0x0008", "channel_0x0009"} + ), + False, + ), + ( + registries.MatchRule( + generic_ids={"channel_0x0006", "channel_0x0008"}, + channel_names={"on_off", "level"}, + ), + True, + ), + # manufacturer matching + (registries.MatchRule(manufacturer="no match"), False), + (registries.MatchRule(manufacturer=MANUFACTURER), True), + (registries.MatchRule(model=MODEL), True), + (registries.MatchRule(model="no match"), False), + # match everything + ( + registries.MatchRule( + generic_ids={"channel_0x0006", "channel_0x0008"}, + channel_names={"on_off", "level"}, + manufacturer=MANUFACTURER, + model=MODEL, + ), + True, + ), + ], +) +def test_registry_matching(rule, matched, zha_device, channels): + """Test empty rule matching.""" + reg = registries.ZHAEntityRegistry() + assert reg._strict_matched(zha_device, channels, rule) is matched + + +@pytest.mark.parametrize( + "rule, matched", + [ + (registries.MatchRule(), False), + (registries.MatchRule(channel_names={"level"}), True), + (registries.MatchRule(channel_names={"level", "no match"}), False), + (registries.MatchRule(channel_names={"on_off"}), True), + (registries.MatchRule(channel_names={"on_off", "no match"}), False), + (registries.MatchRule(channel_names={"on_off", "level"}), True), + (registries.MatchRule(channel_names={"on_off", "level", "no match"}), False), + ( + registries.MatchRule(channel_names={"on_off", "level"}, model="no match"), + True, + ), + ( + registries.MatchRule( + channel_names={"on_off", "level"}, + model="no match", + manufacturer="no match", + ), + True, + ), + ( + registries.MatchRule( + channel_names={"on_off", "level"}, + model="no match", + manufacturer=MANUFACTURER, + ), + True, + ), + # test generic_id matching + (registries.MatchRule(generic_ids={"channel_0x0006"}), True), + (registries.MatchRule(generic_ids={"channel_0x0008"}), True), + (registries.MatchRule(generic_ids={"channel_0x0006", "channel_0x0008"}), True), + ( + registries.MatchRule( + generic_ids={"channel_0x0006", "channel_0x0008", "channel_0x0009"} + ), + False, + ), + ( + registries.MatchRule( + generic_ids={"channel_0x0006", "channel_0x0008", "channel_0x0009"}, + model="mo match", + ), + False, + ), + ( + registries.MatchRule( + generic_ids={"channel_0x0006", "channel_0x0008", "channel_0x0009"}, + model=MODEL, + ), + True, + ), + ( + registries.MatchRule( + generic_ids={"channel_0x0006", "channel_0x0008"}, + channel_names={"on_off", "level"}, + ), + True, + ), + # manufacturer matching + (registries.MatchRule(manufacturer="no match"), False), + (registries.MatchRule(manufacturer=MANUFACTURER), True), + (registries.MatchRule(model=MODEL), True), + (registries.MatchRule(model="no match"), False), + # match everything + ( + registries.MatchRule( + generic_ids={"channel_0x0006", "channel_0x0008"}, + channel_names={"on_off", "level"}, + manufacturer=MANUFACTURER, + model=MODEL, + ), + True, + ), + ], +) +def test_registry_loose_matching(rule, matched, zha_device, channels): + """Test loose rule matching.""" + reg = registries.ZHAEntityRegistry() + assert reg._loose_matched(zha_device, channels, rule) is matched