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.
This commit is contained in:
Alexei Chetroi 2019-12-21 16:26:58 -05:00 committed by GitHub
parent b41480ae46
commit fb3bb8220b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 415 additions and 192 deletions

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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