YoLink flexfob support (#90027)

This commit is contained in:
Matrix 2023-03-22 19:01:04 +08:00 committed by GitHub
parent 7efe058aa6
commit 87e6dd3949
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 308 additions and 6 deletions

View file

@ -7,6 +7,7 @@ from datetime import timedelta
from typing import Any
import async_timeout
from yolink.const import ATTR_DEVICE_SMART_REMOTER
from yolink.device import YoLinkDevice
from yolink.exception import YoLinkAuthFailError, YoLinkClientError
from yolink.home_manager import YoLinkHome
@ -16,11 +17,16 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import aiohttp_client, config_entry_oauth2_flow
from homeassistant.helpers import (
aiohttp_client,
config_entry_oauth2_flow,
device_registry as dr,
)
from . import api
from .const import DOMAIN
from .const import DOMAIN, YOLINK_EVENT
from .coordinator import YoLinkCoordinator
from .device_trigger import CONF_LONG_PRESS, CONF_SHORT_PRESS
SCAN_INTERVAL = timedelta(minutes=5)
@ -53,9 +59,32 @@ class YoLinkHomeMessageListener(MessageListener):
device_coordinators = entry_data.device_coordinators
if not device_coordinators:
return
device_coordiantor = device_coordinators.get(device.device_id)
if device_coordiantor is not None:
device_coordiantor.async_set_updated_data(msg_data)
device_coordinator = device_coordinators.get(device.device_id)
if device_coordinator is None:
return
device_coordinator.async_set_updated_data(msg_data)
# handling events
if (
device_coordinator.device.device_type == ATTR_DEVICE_SMART_REMOTER
and msg_data.get("event") is not None
):
device_registry = dr.async_get(self._hass)
device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, device_coordinator.device.device_id)}
)
if device_entry is None:
return
key_press_type = None
if msg_data["event"]["type"] == "Press":
key_press_type = CONF_SHORT_PRESS
else:
key_press_type = CONF_LONG_PRESS
button_idx = msg_data["event"]["keyMask"]
event_data = {
"type": f"button_{button_idx}_{key_press_type}",
"device_id": device_entry.id,
}
self._hass.bus.async_fire(YOLINK_EVENT, event_data)
@dataclass

View file

@ -7,3 +7,4 @@ ATTR_DEVICE_TYPE = "type"
ATTR_DEVICE_NAME = "name"
ATTR_DEVICE_STATE = "state"
ATTR_DEVICE_ID = "deviceId"
YOLINK_EVENT = f"{DOMAIN}_event"

View file

@ -0,0 +1,88 @@
"""Provides device triggers for YoLink."""
from __future__ import annotations
from typing import Any
import voluptuous as vol
from yolink.const import ATTR_DEVICE_SMART_REMOTER
from homeassistant.components.device_automation import DEVICE_TRIGGER_BASE_SCHEMA
from homeassistant.components.homeassistant.triggers import event as event_trigger
from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF_TYPE
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo
from homeassistant.helpers.typing import ConfigType
from . import DOMAIN, YOLINK_EVENT
CONF_BUTTON_1 = "button_1"
CONF_BUTTON_2 = "button_2"
CONF_BUTTON_3 = "button_3"
CONF_BUTTON_4 = "button_4"
CONF_SHORT_PRESS = "short_press"
CONF_LONG_PRESS = "long_press"
REMOTE_TRIGGER_TYPES = {
f"{CONF_BUTTON_1}_{CONF_SHORT_PRESS}",
f"{CONF_BUTTON_1}_{CONF_LONG_PRESS}",
f"{CONF_BUTTON_2}_{CONF_SHORT_PRESS}",
f"{CONF_BUTTON_2}_{CONF_LONG_PRESS}",
f"{CONF_BUTTON_3}_{CONF_SHORT_PRESS}",
f"{CONF_BUTTON_3}_{CONF_LONG_PRESS}",
f"{CONF_BUTTON_4}_{CONF_SHORT_PRESS}",
f"{CONF_BUTTON_4}_{CONF_LONG_PRESS}",
}
TRIGGER_SCHEMA = DEVICE_TRIGGER_BASE_SCHEMA.extend(
{vol.Required(CONF_TYPE): vol.In(REMOTE_TRIGGER_TYPES)}
)
# YoLink Remotes YS3604/YS3605/YS3606/YS3607
DEVICE_TRIGGER_TYPES: dict[str, set[str]] = {
ATTR_DEVICE_SMART_REMOTER: REMOTE_TRIGGER_TYPES,
}
async def async_get_triggers(
hass: HomeAssistant, device_id: str
) -> list[dict[str, Any]]:
"""List device triggers for YoLink devices."""
device_registry = dr.async_get(hass)
registry_device = device_registry.async_get(device_id)
if not registry_device or registry_device.model != ATTR_DEVICE_SMART_REMOTER:
return []
triggers = []
for trigger in DEVICE_TRIGGER_TYPES[ATTR_DEVICE_SMART_REMOTER]:
triggers.append(
{
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_PLATFORM: "device",
CONF_TYPE: trigger,
}
)
return triggers
async def async_attach_trigger(
hass: HomeAssistant,
config: ConfigType,
action: TriggerActionType,
trigger_info: TriggerInfo,
) -> CALLBACK_TYPE:
"""Listen for state changes based on configuration."""
event_config = {
event_trigger.CONF_PLATFORM: "event",
event_trigger.CONF_EVENT_TYPE: YOLINK_EVENT,
event_trigger.CONF_EVENT_DATA: {
CONF_DEVICE_ID: config[CONF_DEVICE_ID],
CONF_TYPE: config[CONF_TYPE],
},
}
event_config = event_trigger.TRIGGER_SCHEMA(event_config)
return await event_trigger.async_attach_trigger(
hass, event_config, action, trigger_info, platform_type="device"
)

View file

@ -1,4 +1,4 @@
"""YoLink Binary Sensor."""
"""YoLink Sensor."""
from __future__ import annotations
from collections.abc import Callable
@ -15,6 +15,7 @@ from yolink.const import (
ATTR_DEVICE_MULTI_OUTLET,
ATTR_DEVICE_OUTLET,
ATTR_DEVICE_SIREN,
ATTR_DEVICE_SMART_REMOTER,
ATTR_DEVICE_SWITCH,
ATTR_DEVICE_TH_SENSOR,
ATTR_DEVICE_THERMOSTAT,
@ -68,6 +69,7 @@ SENSOR_DEVICE_TYPE = [
ATTR_DEVICE_LEAK_SENSOR,
ATTR_DEVICE_MOTION_SENSOR,
ATTR_DEVICE_MULTI_OUTLET,
ATTR_DEVICE_SMART_REMOTER,
ATTR_DEVICE_OUTLET,
ATTR_DEVICE_SIREN,
ATTR_DEVICE_SWITCH,
@ -84,6 +86,7 @@ BATTERY_POWER_SENSOR = [
ATTR_DEVICE_DOOR_SENSOR,
ATTR_DEVICE_LEAK_SENSOR,
ATTR_DEVICE_MOTION_SENSOR,
ATTR_DEVICE_SMART_REMOTER,
ATTR_DEVICE_TH_SENSOR,
ATTR_DEVICE_VIBRATION_SENSOR,
ATTR_DEVICE_LOCK,

View file

@ -21,5 +21,17 @@
"create_entry": {
"default": "[%key:common::config_flow::create_entry::authenticated%]"
}
},
"device_automation": {
"trigger_type": {
"button_1_short_press": "Button_1 (short press)",
"button_1_long_press": "Button_1 (long press)",
"button_2_short_press": "Button_2 (short press)",
"button_2_long_press": "Button_2 (long press)",
"button_3_short_press": "Button_3 (short press)",
"button_3_long_press": "Button_3 (long press)",
"button_4_short_press": "Button_4 (short press)",
"button_4_long_press": "Button_4 (long press)"
}
}
}

View file

@ -0,0 +1,169 @@
"""The tests for YoLink device triggers."""
import pytest
from yolink.const import ATTR_DEVICE_DIMMER, ATTR_DEVICE_SMART_REMOTER
from homeassistant.components import automation
from homeassistant.components.device_automation import DeviceAutomationType
from homeassistant.components.yolink import DOMAIN, YOLINK_EVENT
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.setup import async_setup_component
from tests.common import (
MockConfigEntry,
assert_lists_same,
async_get_device_automations,
async_mock_service,
)
@pytest.fixture
def calls(hass: HomeAssistant):
"""Track calls to a mock service."""
return async_mock_service(hass, "yolink", "automation")
async def test_get_triggers(
hass: HomeAssistant, device_registry: dr.DeviceRegistry
) -> None:
"""Test we get the expected triggers from a yolink flexfob."""
config_entry = MockConfigEntry(domain="yolink", data={})
config_entry.add_to_hass(hass)
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
model=ATTR_DEVICE_SMART_REMOTER,
)
expected_triggers = [
{
"platform": "device",
"domain": DOMAIN,
"type": "button_1_short_press",
"device_id": device_entry.id,
"metadata": {},
},
{
"platform": "device",
"domain": DOMAIN,
"type": "button_1_long_press",
"device_id": device_entry.id,
"metadata": {},
},
{
"platform": "device",
"domain": DOMAIN,
"type": "button_2_short_press",
"device_id": device_entry.id,
"metadata": {},
},
{
"platform": "device",
"domain": DOMAIN,
"type": "button_2_long_press",
"device_id": device_entry.id,
"metadata": {},
},
{
"platform": "device",
"domain": DOMAIN,
"type": "button_3_short_press",
"device_id": device_entry.id,
"metadata": {},
},
{
"platform": "device",
"domain": DOMAIN,
"type": "button_3_long_press",
"device_id": device_entry.id,
"metadata": {},
},
{
"platform": "device",
"domain": DOMAIN,
"type": "button_4_short_press",
"device_id": device_entry.id,
"metadata": {},
},
{
"platform": "device",
"domain": DOMAIN,
"type": "button_4_long_press",
"device_id": device_entry.id,
"metadata": {},
},
]
triggers = await async_get_device_automations(
hass, DeviceAutomationType.TRIGGER, device_entry.id
)
assert_lists_same(triggers, expected_triggers)
async def test_get_triggers_exception(
hass: HomeAssistant, device_registry: dr.DeviceRegistry
) -> None:
"""Test get triggers when device type not flexfob."""
config_entry = MockConfigEntry(domain="yolink", data={})
config_entry.add_to_hass(hass)
device_entity = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
model=ATTR_DEVICE_DIMMER,
)
expected_triggers = []
triggers = await async_get_device_automations(
hass, DeviceAutomationType.TRIGGER, device_entity.id
)
assert_lists_same(triggers, expected_triggers)
async def test_if_fires_on_event(
hass: HomeAssistant, calls, device_registry: dr.DeviceRegistry
) -> None:
"""Test for event triggers firing."""
mac_address = "12:34:56:AB:CD:EF"
connection = (dr.CONNECTION_NETWORK_MAC, mac_address)
config_entry = MockConfigEntry(domain=DOMAIN, data={})
config_entry.add_to_hass(hass)
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={connection},
identifiers={(DOMAIN, mac_address)},
model=ATTR_DEVICE_SMART_REMOTER,
)
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
"platform": "device",
"domain": DOMAIN,
"device_id": device_entry.id,
"type": "button_1_long_press",
},
"action": {
"service": "yolink.automation",
"data": {"message": "service called"},
},
},
]
},
)
device = device_registry.async_get_device(set(), {connection})
assert device is not None
# Fake remote button long press.
hass.bus.async_fire(
event_type=YOLINK_EVENT,
event_data={
"type": "button_1_long_press",
"device_id": device.id,
},
)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["message"] == "service called"