Poll Hue lights in ZHA ()

* Weighted ZHA entity matching.
* Poll ZHA Hue lights in 5 min intervals.
* Add comment.
* Spelling.
pull/33031/head
Alexei Chetroi 2020-03-19 22:14:07 -04:00 committed by GitHub
parent f275b7e5ed
commit 8cb1d630c8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 101 additions and 4 deletions
homeassistant/components/zha
tests/components/zha

View File

@ -196,6 +196,30 @@ class MatchRule:
factory=frozenset, converter=set_or_callable
)
@property
def weight(self) -> int:
"""Return the weight of the matching rule.
Most specific matches should be preferred over less specific. Model matching
rules have a priority over manufacturer matching rules and rules matching a
single model/manufacturer get a better priority over rules matching multiple
models/manufacturers. And any model or manufacturers matching rules get better
priority over rules matching only channels.
But in case of a channel name/channel id matching, we give rules matching
multiple channels a better priority over rules matching a single channel.
"""
weight = 0
if self.models:
weight += 401 - len(self.models)
if self.manufacturers:
weight += 301 - len(self.manufacturers)
weight += 10 * len(self.channel_names)
weight += 5 * len(self.generic_ids)
weight += 1 * len(self.aux_channels)
return weight
def claim_channels(self, channel_pool: List[ChannelType]) -> List[ChannelType]:
"""Return a list of channels this rule matches + aux channels."""
claimed = []
@ -268,7 +292,8 @@ class ZHAEntityRegistry:
default: CALLABLE_T = None,
) -> Tuple[CALLABLE_T, List[ChannelType]]:
"""Match a ZHA Channels to a ZHA Entity class."""
for match in self._strict_registry[component]:
matches = self._strict_registry[component]
for match in sorted(matches, key=lambda x: x.weight, reverse=True):
if match.strict_matched(manufacturer, model, channels):
claimed = match.claim_channels(channels)
return self._strict_registry[component][match], claimed

View File

@ -47,7 +47,6 @@ FLASH_EFFECTS = {light.FLASH_SHORT: EFFECT_BLINK, light.FLASH_LONG: EFFECT_BREAT
UNSUPPORTED_ATTRIBUTE = 0x86
STRICT_MATCH = functools.partial(ZHA_ENTITIES.strict_match, light.DOMAIN)
PARALLEL_UPDATES = 0
_REFRESH_INTERVAL = (45, 75)
async def async_setup_entry(hass, config_entry, async_add_entities):
@ -68,6 +67,8 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
class Light(ZhaEntity, light.Light):
"""Representation of a ZHA or ZLL light."""
_REFRESH_INTERVAL = (45, 75)
def __init__(self, unique_id, zha_device: ZhaDeviceType, channels, **kwargs):
"""Initialize the ZHA light."""
super().__init__(unique_id, zha_device, channels, **kwargs)
@ -177,9 +178,9 @@ class Light(ZhaEntity, light.Light):
await self.async_accept_signal(
self._level_channel, SIGNAL_SET_LEVEL, self.set_level
)
refresh_interval = random.randint(*_REFRESH_INTERVAL)
refresh_interval = random.randint(*[x * 60 for x in self._REFRESH_INTERVAL])
self._cancel_refresh_handle = async_track_time_interval(
self.hass, self._refresh, timedelta(minutes=refresh_interval)
self.hass, self._refresh, timedelta(seconds=refresh_interval)
)
async def async_will_remove_from_hass(self) -> None:
@ -398,3 +399,14 @@ class Light(ZhaEntity, light.Light):
"""Call async_get_state at an interval."""
await self.async_get_state(from_cache=False)
self.async_write_ha_state()
@STRICT_MATCH(
channel_names=CHANNEL_ON_OFF,
aux_channels={CHANNEL_COLOR, CHANNEL_LEVEL},
manufacturers="Philips",
)
class HueLight(Light):
"""Representation of a HUE light which does not report attributes."""
_REFRESH_INTERVAL = (3, 5)

View File

@ -254,3 +254,63 @@ def test_match_rule_claim_channels(rule, match, channel, channels):
claimed = rule.claim_channels(channels)
assert match == set([ch.name for ch in claimed])
@pytest.fixture
def entity_registry():
"""Registry fixture."""
return registries.ZHAEntityRegistry()
@pytest.mark.parametrize(
"manufacturer, model, match_name",
(
("random manufacturer", "random model", "OnOff"),
("random manufacturer", MODEL, "OnOffModel"),
(MANUFACTURER, "random model", "OnOffManufacturer"),
(MANUFACTURER, MODEL, "OnOffModelManufacturer"),
(MANUFACTURER, "some model", "OnOffMultimodel"),
),
)
def test_weighted_match(channel, entity_registry, manufacturer, model, match_name):
"""Test weightedd match."""
s = mock.sentinel
@entity_registry.strict_match(
s.component,
channel_names="on_off",
models={MODEL, "another model", "some model"},
)
class OnOffMultimodel:
pass
@entity_registry.strict_match(s.component, channel_names="on_off")
class OnOff:
pass
@entity_registry.strict_match(
s.component, channel_names="on_off", manufacturers=MANUFACTURER
)
class OnOffManufacturer:
pass
@entity_registry.strict_match(s.component, channel_names="on_off", models=MODEL)
class OnOffModel:
pass
@entity_registry.strict_match(
s.component, channel_names="on_off", models=MODEL, manufacturers=MANUFACTURER
)
class OnOffModelManufacturer:
pass
ch_on_off = channel("on_off", 6)
ch_level = channel("level", 8)
match, claimed = entity_registry.get_entity(
s.component, manufacturer, model, [ch_on_off, ch_level]
)
assert match.__name__ == match_name
assert claimed == [ch_on_off]