Add YoLink local integration

homeassistant/components/yolink_local/__init__.py

@@ -0,0 +1,91 @@
+"""The YoLink Local integration."""
+
+from __future__ import annotations
+
+import asyncio
+
+from yolink.device import YoLinkDevice
+from yolink.exception import YoLinkAuthFailError, YoLinkClientError
+from yolink.local_hub_client import YoLinkLocalHubClient
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET, CONF_HOST, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
+from homeassistant.helpers import aiohttp_client
+
+from .const import CONF_NET_ID
+from .coordinator import YoLinkLocalCoordinator
+from .message_listener import LocalHubMessageListener
+
+_PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR]
+
+
+type YoLinkLocalConfigEntry = ConfigEntry[
+    tuple[YoLinkLocalHubClient, dict[str, YoLinkLocalCoordinator]]
+]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: YoLinkLocalConfigEntry) -> bool:
+    """Set up YoLink Local from a config entry."""
+
+    session = aiohttp_client.async_create_clientsession(hass)
+
+    client = YoLinkLocalHubClient(
+        session,
+        entry.data[CONF_HOST],
+        entry.data[CONF_NET_ID],
+        entry.data[CONF_CLIENT_ID],
+        entry.data[CONF_CLIENT_SECRET],
+    )
+    try:
+        async with asyncio.timeout(10):
+            await client.async_setup(LocalHubMessageListener(hass, entry))
+    except YoLinkAuthFailError as yl_auth_err:
+        raise ConfigEntryAuthFailed from yl_auth_err
+    except (YoLinkClientError, TimeoutError) as err:
+        raise ConfigEntryNotReady from err
+
+    devices: list[YoLinkDevice] = client.get_devices()
+
+    device_pairing_mapping = {}
+    for device in devices:
+        if (parent_id := device.paired_device_id) is not None:
+            device_pairing_mapping[parent_id] = device.device_id
+
+    coordinators = {}
+
+    for device in devices:
+        paried_device: YoLinkDevice | None = None
+        if (
+            paried_device_id := device_pairing_mapping.get(device.device_id)
+        ) is not None:
+            paried_device = next(
+                (
+                    _device
+                    for _device in devices
+                    if _device.device_id == paried_device_id
+                ),
+                None,
+            )
+        coordinator = YoLinkLocalCoordinator(hass, entry, device, paried_device)
+        try:
+            await coordinator.async_config_entry_first_refresh()
+        except ConfigEntryNotReady:
+            # Not failure by fetching device state
+            coordinator.data = {}
+        coordinators[device.device_id] = coordinator
+
+    entry.runtime_data = (client, coordinators)
+    await hass.config_entries.async_forward_entry_setups(entry, _PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, _PLATFORMS):
+        if entry.runtime_data is not None:
+            client: YoLinkLocalHubClient = entry.runtime_data[0]
+            await client.async_unload()
+    return unload_ok

homeassistant/components/yolink_local/binary_sensor.py

@@ -0,0 +1,143 @@
+"""YoLink BinarySensor."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Any
+
+from yolink.const import (
+    ATTR_DEVICE_LEAK_SENSOR,
+    ATTR_DEVICE_MOTION_SENSOR,
+    ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR,
+)
+from yolink.device import YoLinkDevice
+
+from homeassistant.components.binary_sensor import (
+    BinarySensorDeviceClass,
+    BinarySensorEntity,
+    BinarySensorEntityDescription,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from .coordinator import YoLinkLocalCoordinator
+from .entity import YoLinkEntity
+
+
+@dataclass(frozen=True, kw_only=True)
+class YoLinkBinarySensorEntityDescription(BinarySensorEntityDescription):
+    """YoLink BinarySensorEntityDescription."""
+
+    exists_fn: Callable[[YoLinkDevice], bool] = lambda _: True
+    should_update_entity: Callable = lambda state: True
+    value: Callable[[YoLinkDevice, dict], bool | None]
+    is_available: Callable[[YoLinkDevice, dict], bool] = lambda _, __: True
+
+
+def parse_data_leak_sensor_state(device: YoLinkDevice, data: dict) -> bool | None:
+    """Parse leak sensor state."""
+    if device.device_type == ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR:
+        if (state := data.get("state")) is None or (
+            value := state.get("waterDetection")
+        ) is None:
+            return None
+        return (
+            value == "normal"
+            if data.get("waterDetectionMode") == "WaterPeak"
+            else value == "leak"
+        )
+    return data.get("state") in ["alert", "full"]
+
+
+def is_leak_sensor_state_available(device: YoLinkDevice, data: dict) -> bool:
+    """Check leak sensor state availability."""
+    if device.device_type == ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR:
+        if (
+            (alarms := data.get("alarm")) is not None
+            and isinstance(alarms, list)
+            and "Alert.DetectorError" in alarms
+        ):
+            return False
+    if (alarms := data.get("alarmState")) is not None and alarms.get(
+        "detectorError"
+    ) is True:
+        return False
+    return True
+
+
+BINARY_SENSOR_DESCRIPTIONS: tuple[YoLinkBinarySensorEntityDescription, ...] = (
+    YoLinkBinarySensorEntityDescription(
+        key="leak_state",
+        device_class=BinarySensorDeviceClass.MOISTURE,
+        exists_fn=lambda device: (
+            device.device_type
+            in [ATTR_DEVICE_LEAK_SENSOR, ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR]
+        ),
+        value=parse_data_leak_sensor_state,
+        is_available=is_leak_sensor_state_available,
+    ),
+    YoLinkBinarySensorEntityDescription(
+        key="motion_state",
+        device_class=BinarySensorDeviceClass.MOTION,
+        exists_fn=lambda device: device.device_type == ATTR_DEVICE_MOTION_SENSOR,
+        value=lambda device, data: (
+            value == "alert" if (value := data.get("state")) is not None else None
+        ),
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up Sensor from a config entry."""
+    coordinators: dict[str, YoLinkLocalCoordinator] = config_entry.runtime_data[1]
+    binary_sensor_coordinators = [
+        coordinator
+        for coordinator in coordinators.values()
+        if coordinator.device.device_type
+        in [ATTR_DEVICE_LEAK_SENSOR, ATTR_DEVICE_MOTION_SENSOR]
+    ]
+    async_add_entities(
+        YoLinkBinarySensorEntity(config_entry, binary_sensor_coordinator, description)
+        for binary_sensor_coordinator in binary_sensor_coordinators
+        for description in BINARY_SENSOR_DESCRIPTIONS
+        if description.exists_fn(binary_sensor_coordinator.device)
+    )
+
+
+class YoLinkBinarySensorEntity(YoLinkEntity, BinarySensorEntity):
+    """YoLink Sensor Entity."""
+
+    entity_description: YoLinkBinarySensorEntityDescription
+
+    def __init__(
+        self,
+        config_entry: ConfigEntry,
+        coordinator: YoLinkLocalCoordinator,
+        description: YoLinkBinarySensorEntityDescription,
+    ) -> None:
+        """Init YoLink Sensor."""
+        super().__init__(config_entry, coordinator)
+        self.entity_description = description
+        self._attr_unique_id = (
+            f"{coordinator.device.device_id} {self.entity_description.key}"
+        )
+
+    @callback
+    def update_entity_state(self, state: dict[str, Any]) -> None:
+        """Update HA Entity State."""
+        if (
+            _attr_val := self.entity_description.value(self.coordinator.device, state)
+        ) is None or self.entity_description.should_update_entity(_attr_val) is False:
+            return
+        _is_attr_available = self.entity_description.is_available(
+            self.coordinator.device, state
+        )
+        self._attr_available = _is_attr_available
+        self._attr_is_on = _attr_val if _is_attr_available else None
+        self.async_write_ha_state()

homeassistant/components/yolink_local/config_flow.py

@@ -0,0 +1,88 @@
+"""Config flow for the YoLink Local integration."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from aiohttp import ClientError
+import voluptuous as vol
+from yolink.local_hub_client import YoLinkLocalHubClient
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET, CONF_HOST
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import aiohttp_client
+
+from .const import CONF_NET_ID, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_HOST): str,
+        vol.Required(CONF_NET_ID): str,
+        vol.Required(CONF_CLIENT_ID): str,
+        vol.Required(CONF_CLIENT_SECRET): str,
+    }
+)
+
+
+async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, Any]:
+    """Validate the user input allows us to connect.
+
+    Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
+    """
+    session = aiohttp_client.async_create_clientsession(hass)
+    localHubClient = YoLinkLocalHubClient(
+        session,
+        data[CONF_HOST],
+        data[CONF_NET_ID],
+        data[CONF_CLIENT_ID],
+        data[CONF_CLIENT_SECRET],
+    )
+    try:
+        if not await localHubClient.authenticate():
+            raise InvalidAuth
+    except ClientError as err:
+        raise CannotConnect from err
+    return {"title": "YoLink Local Hub"}
+
+
+class YoLinkLocalHubConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for YoLink Local Hub."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            await self.async_set_unique_id(f"yolink_local_{user_input[CONF_NET_ID]}")
+            self._abort_if_unique_id_configured()
+            try:
+                info = await validate_input(self.hass, user_input)
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+            except InvalidAuth:
+                errors["base"] = "invalid_auth"
+            except Exception:
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                return self.async_create_entry(title=info["title"], data=user_input)
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )
+
+
+class CannotConnect(HomeAssistantError):
+    """Error to indicate we cannot connect."""
+
+
+class InvalidAuth(HomeAssistantError):
+    """Error to indicate there is invalid auth."""

homeassistant/components/yolink_local/const.py

@@ -0,0 +1,20 @@
+"""Constants for the YoLink Local integration."""
+
+DOMAIN = "yolink_local"
+MANUFACTURER = "YoLink"
+CONF_NET_ID = "net_id"
+
+ATTR_DEVICE_STATE = "state"
+ATTR_LORA_INFO = "loraInfo"
+
+
+SUPPORTED_BINARY_SENSORS_MODELS = [
+    # Leak Sensor
+    "YS7903",
+    "YS7904",
+    "YS7905",
+    "YS7906",
+    "YS7916",
+    # Motion Sensor
+    "YS7805",
+]