Add Xthings Cloud (home-assistant#167885)

Co-authored-by: Joostlek <joostlek@outlook.com>

Author: zhangluofeng

CODEOWNERS

@@ -0,0 +1,36 @@
+"""Xthings Cloud integration for Home Assistant."""
+
+from ha_xthings_cloud import XthingsCloudApiClient
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import CONF_TOKEN, PLATFORMS
+from .coordinator import XthingsCloudConfigEntry, XthingsCloudCoordinator
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: XthingsCloudConfigEntry
+) -> bool:
+    """Set up config entry."""
+    session = async_get_clientsession(hass)
+    client = XthingsCloudApiClient(session, token=entry.data[CONF_TOKEN])
+
+    coordinator = XthingsCloudCoordinator(hass, client, entry)
+    await coordinator.async_config_entry_first_refresh()
+
+    entry.runtime_data = coordinator
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    await coordinator.async_start_websocket()
+
+    return True
+
+
+async def async_unload_entry(
+    hass: HomeAssistant, entry: XthingsCloudConfigEntry
+) -> bool:
+    """Unload config entry."""
+    coordinator = entry.runtime_data
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        await coordinator.async_stop_websocket()
+    return unload_ok

homeassistant/components/xthings_cloud/__init__.py

@@ -0,0 +1,98 @@
+"""Config flow for Xthings Cloud."""
+
+from typing import Any
+
+from ha_xthings_cloud import (
+    XthingsCloudApiClient,
+    XthingsCloudApiError,
+    XthingsCloudAuthError,
+)
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.instance_id import async_get as async_get_instance_id
+
+from .const import (
+    CONF_EMAIL,
+    CONF_PASSWORD,
+    CONF_REFRESH_TOKEN,
+    CONF_TOKEN,
+    DOMAIN,
+    LOGGER,
+)
+
+ERROR_CODE_MAP: dict[int, str] = {
+    20001: "token_invalid",
+    21001: "email_empty",
+    21002: "email_invalid",
+    21004: "email_not_found",
+    21011: "password_empty",
+    21014: "password_wrong",
+    21021: "user_disabled",
+    21022: "user_not_logged_in",
+    21023: "user_not_activated",
+    20011: "token_invalid",
+    20012: "token_expired",
+    22001: "device_not_found",
+    22003: "device_offline",
+}
+
+
+def _error_from_exception(err: XthingsCloudApiError) -> str:
+    """Return translation key from error code."""
+    return ERROR_CODE_MAP.get(err.code, "unknown")
+
+
+class XthingsCloudConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Xthings Cloud config flow."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle user input step."""
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            instance_id = await async_get_instance_id(self.hass)
+            session = async_get_clientsession(self.hass)
+            client = XthingsCloudApiClient(session)
+            try:
+                token_data = await client.async_login(
+                    user_input[CONF_EMAIL],
+                    user_input[CONF_PASSWORD],
+                    client_id=instance_id,
+                )
+            except XthingsCloudAuthError as err:
+                errors["base"] = _error_from_exception(err)
+            except XthingsCloudApiError as err:
+                errors["base"] = (
+                    _error_from_exception(err) if err.code else "cannot_connect"
+                )
+            except Exception:  # noqa: BLE001
+                LOGGER.exception("Unexpected error during login")
+                errors["base"] = "unknown"
+            else:
+                await self.async_set_unique_id(token_data["user_id"])
+                self._abort_if_unique_id_configured()
+                return self.async_create_entry(
+                    title=user_input[CONF_EMAIL],
+                    data={
+                        CONF_EMAIL: user_input[CONF_EMAIL],
+                        CONF_TOKEN: token_data["token"],
+                        CONF_REFRESH_TOKEN: token_data["refresh_token"],
+                    },
+                )
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_EMAIL): str,
+                    vol.Required(CONF_PASSWORD): str,
+                }
+            ),
+            errors=errors,
+        )

homeassistant/components/xthings_cloud/config_flow.py

@@ -0,0 +1,20 @@
+"""Constants for Xthings Cloud integration."""
+
+import logging
+
+from homeassistant.const import Platform
+
+DOMAIN = "xthings_cloud"
+LOGGER = logging.getLogger(__package__)
+
+CONF_EMAIL = "email"
+CONF_PASSWORD = "password"
+CONF_TOKEN = "token"
+CONF_REFRESH_TOKEN = "refresh_token"
+CONF_CLIENT_ID = "client_id"
+CONF_INSTANCE_ID = "instance_id"
+
+# Polling interval (seconds)
+DEFAULT_SCAN_INTERVAL = 1800
+
+PLATFORMS: list[Platform] = [Platform.LIGHT]

homeassistant/components/xthings_cloud/const.py

@@ -0,0 +1,128 @@
+"""DataUpdateCoordinator for Xthings Cloud."""
+
+from datetime import timedelta
+from typing import Any
+
+from ha_xthings_cloud import (
+    XthingsCloudApiClient,
+    XthingsCloudApiError,
+    XthingsCloudAuthError,
+    XthingsCloudWebSocket,
+)
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import CONF_REFRESH_TOKEN, CONF_TOKEN, DEFAULT_SCAN_INTERVAL, DOMAIN, LOGGER
+
+type XthingsCloudConfigEntry = ConfigEntry["XthingsCloudCoordinator"]
+
+
+class XthingsCloudCoordinator(DataUpdateCoordinator[dict[str, Any]]):
+    """Xthings Cloud data update coordinator."""
+
+    config_entry: XthingsCloudConfigEntry
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        client: XthingsCloudApiClient,
+        entry: XthingsCloudConfigEntry,
+    ) -> None:
+        """Initialize the coordinator."""
+        super().__init__(
+            hass,
+            LOGGER,
+            name=DOMAIN,
+            update_interval=timedelta(seconds=DEFAULT_SCAN_INTERVAL),
+            config_entry=entry,
+        )
+        self.client = client
+        self.websocket: XthingsCloudWebSocket | None = None
+
+    async def _async_ensure_token_valid(self) -> None:
+        """Ensure the token is valid, refresh if expired.
+
+        Raises ConfigEntryAuthFailed if refresh fails.
+        """
+        if not self.client.is_token_expired():
+            return
+        try:
+            token_data = await self.client.async_refresh_token(
+                self.config_entry.data[CONF_REFRESH_TOKEN]
+            )
+        except XthingsCloudAuthError as err:
+            raise ConfigEntryAuthFailed(
+                "Token expired and refresh failed, re-authentication required"
+            ) from err
+        self.hass.config_entries.async_update_entry(
+            self.config_entry,
+            data={
+                **self.config_entry.data,
+                CONF_TOKEN: token_data["token"],
+                CONF_REFRESH_TOKEN: token_data["refresh_token"],
+            },
+        )
+
+    async def _async_update_data(self) -> dict[str, Any]:
+        """Fetch latest device data from cloud."""
+        await self._async_ensure_token_valid()
+        try:
+            devices = await self.client.async_get_devices()
+        except XthingsCloudAuthError as err:
+            raise ConfigEntryAuthFailed(
+                "Invalid token, re-authentication required"
+            ) from err
+        except XthingsCloudApiError as err:
+            raise UpdateFailed(f"Failed to fetch data: {err}") from err
+        return {device["id"]: device for device in devices}
+
+    async def async_start_websocket(self) -> None:
+        """Start WebSocket connection."""
+        if self.websocket:
+            return
+        session = async_get_clientsession(self.hass)
+        token = self.config_entry.data[CONF_TOKEN]
+        self.websocket = XthingsCloudWebSocket(
+            session=session,
+            token=token,
+            on_device_status=self._handle_ws_device_status,
+            on_token_expired=self._handle_ws_token_expired,
+        )
+        await self.websocket.async_start()
+
+    async def async_stop_websocket(self) -> None:
+        """Stop WebSocket connection."""
+        if self.websocket:
+            await self.websocket.async_stop()
+            self.websocket = None
+
+    def _handle_ws_device_status(
+        self, device_uuid: str, status: dict[str, Any]
+    ) -> None:
+        """Handle WebSocket device status update."""
+        if not self.data or device_uuid not in self.data:
+            LOGGER.debug(
+                "WebSocket received status for unknown device: %s", device_uuid
+            )
+            return
+        device_data = self.data[device_uuid]
+        device_data.setdefault("status", {}).update(status)
+        LOGGER.debug("WebSocket updated device status: %s", device_uuid)
+        self.async_set_updated_data(self.data)
+
+    async def _handle_ws_token_expired(self) -> None:
+        """Handle WebSocket auth expiry, refresh token."""
+        try:
+            await self._async_ensure_token_valid()
+        except ConfigEntryAuthFailed:
+            LOGGER.error("WebSocket token refresh failed")
+            return
+        new_token = self.config_entry.data[CONF_TOKEN]
+        self.client.token = new_token
+        if self.websocket:
+            self.websocket.token = new_token
+        LOGGER.info("WebSocket token refreshed successfully")

homeassistant/components/xthings_cloud/coordinator.py

@@ -0,0 +1,48 @@
+"""Base entity for Xthings Cloud."""
+
+from typing import Any
+
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .coordinator import XthingsCloudCoordinator
+
+
+class XthingsCloudEntity(CoordinatorEntity[XthingsCloudCoordinator]):
+    """Xthings Cloud base entity."""
+
+    _attr_has_entity_name = True
+    _attr_name = None
+
+    def __init__(
+        self,
+        coordinator: XthingsCloudCoordinator,
+        device_id: str,
+        device_data: dict[str, Any],
+    ) -> None:
+        """Initialize the entity."""
+        super().__init__(coordinator)
+        self._device_id = device_id
+        self._attr_unique_id = device_id
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, device_id)},
+            name=device_data["name"],
+            manufacturer="Xthings",
+            model=device_data["model"],
+            sw_version=device_data.get("version"),
+        )
+
+    @property
+    def device_data(self) -> dict[str, Any]:
+        """Return current device data."""
+        return self.coordinator.data[self._device_id]
+
+    @property
+    def available(self) -> bool:
+        """Return whether device is available (online)."""
+        return (
+            super().available
+            and self._device_id in self.coordinator.data
+            and self.device_data["online"]
+        )