Add switchbot cloud integration

.coveragerc

@@ -1249,6 +1249,9 @@ omit =
     homeassistant/components/switchbot/sensor.py
     homeassistant/components/switchbot/switch.py
     homeassistant/components/switchbot/lock.py
+    homeassistant/components/switchbot_cloud/coordinator.py
+    homeassistant/components/switchbot_cloud/entity.py
+    homeassistant/components/switchbot_cloud/switch.py
     homeassistant/components/switchmate/switch.py
     homeassistant/components/syncthing/__init__.py
     homeassistant/components/syncthing/sensor.py

.strict-typing

@@ -318,6 +318,7 @@ homeassistant.components.sun.*
 homeassistant.components.surepetcare.*
 homeassistant.components.switch.*
 homeassistant.components.switchbee.*
+homeassistant.components.switchbot_cloud.*
 homeassistant.components.switcher_kis.*
 homeassistant.components.synology_dsm.*
 homeassistant.components.systemmonitor.*

CODEOWNERS

@@ -1236,6 +1236,8 @@ build.json @home-assistant/supervisor
 /tests/components/switchbee/ @jafar-atili
 /homeassistant/components/switchbot/ @danielhiversen @RenierM26 @murtas @Eloston @dsypniewski
 /tests/components/switchbot/ @danielhiversen @RenierM26 @murtas @Eloston @dsypniewski
+/homeassistant/components/switchbot_cloud/ @SeraphicRav
+/tests/components/switchbot_cloud/ @SeraphicRav
 /homeassistant/components/switcher_kis/ @thecode
 /tests/components/switcher_kis/ @thecode
 /homeassistant/components/switchmate/ @danielhiversen @qiz-li

homeassistant/brands/switchbot.json

@@ -0,0 +1,5 @@
+{
+  "domain": "switchbot",
+  "name": "SwitchBot",
+  "integrations": ["switchbot", "switchbot_cloud"]
+}

homeassistant/components/switchbot/manifest.json

@@ -1,6 +1,6 @@
 {
   "domain": "switchbot",
-  "name": "SwitchBot",
+  "name": "SwitchBot Bluetooth",
   "bluetooth": [
     {
       "service_data_uuid": "00000d00-0000-1000-8000-00805f9b34fb",

homeassistant/components/switchbot_cloud/__init__.py

@@ -0,0 +1,81 @@
+"""The SwitchBot via API integration."""
+from asyncio import gather
+from dataclasses import dataclass
+from logging import getLogger
+
+from switchbot_api import CannotConnect, Device, InvalidAuth, Remote, SwitchBotAPI
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_API_KEY, CONF_API_TOKEN, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+
+from .const import DOMAIN
+from .coordinator import SwitchBotCoordinator
+
+_LOGGER = getLogger(__name__)
+PLATFORMS: list[Platform] = [Platform.SWITCH]
+
+
+@dataclass
+class SwitchbotDevices:
+    """Switchbot devices data."""
+
+    switches: list[Device | Remote]
+
+
+@dataclass
+class SwitchbotCloudData:
+    """Data to use in platforms."""
+
+    api: SwitchBotAPI
+    devices: SwitchbotDevices
+
+
+async def async_setup_entry(hass: HomeAssistant, config: ConfigEntry) -> bool:
+    """Set up SwitchBot via API from a config entry."""
+    token = config.data[CONF_API_TOKEN]
+    secret = config.data[CONF_API_KEY]
+
+    api = SwitchBotAPI(token=token, secret=secret)
+    try:
+        devices = await api.list_devices()
+    except InvalidAuth as ex:
+        _LOGGER.error(
+            "Invalid authentication while connecting to SwitchBot API: %s", ex
+        )
+        return False
+    except CannotConnect as ex:
+        raise ConfigEntryNotReady from ex
+    _LOGGER.debug("Devices: %s", devices)
+    devices_and_coordinators = [
+        (device, SwitchBotCoordinator(hass, api, device)) for device in devices
+    ]
+    hass.data.setdefault(DOMAIN, {})
+    data = SwitchbotCloudData(
+        api=api,
+        devices=SwitchbotDevices(
+            switches=[
+                (device, coordinator)
+                for device, coordinator in devices_and_coordinators
+                if isinstance(device, Device)
+                and device.device_type.startswith("Plug")
+                or isinstance(device, Remote)
+            ],
+        ),
+    )
+    hass.data[DOMAIN][config.entry_id] = data
+    _LOGGER.debug("Switches: %s", data.devices.switches)
+    await hass.config_entries.async_forward_entry_setups(config, PLATFORMS)
+    await gather(
+        *[coordinator.async_refresh() for _, coordinator in devices_and_coordinators]
+    )
+    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):
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

homeassistant/components/switchbot_cloud/config_flow.py

@@ -0,0 +1,56 @@
+"""Config flow for SwitchBot via API integration."""
+
+from logging import getLogger
+from typing import Any
+
+from switchbot_api import CannotConnect, InvalidAuth, SwitchBotAPI
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_API_KEY, CONF_API_TOKEN
+from homeassistant.data_entry_flow import FlowResult
+
+from .const import DOMAIN, ENTRY_TITLE
+
+_LOGGER = getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_API_TOKEN): str,
+        vol.Required(CONF_API_KEY): str,
+    }
+)
+
+
+class SwitchBotCloudConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for SwitchBot via API."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            try:
+                await SwitchBotAPI(
+                    token=user_input[CONF_API_TOKEN], secret=user_input[CONF_API_KEY]
+                ).list_devices()
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+            except InvalidAuth:
+                errors["base"] = "invalid_auth"
+            except Exception:  # pylint: disable=broad-except
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                await self.async_set_unique_id(
+                    user_input[CONF_API_TOKEN], raise_on_progress=False
+                )
+                self._abort_if_unique_id_configured()
+                return self.async_create_entry(title=ENTRY_TITLE, data=user_input)
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )

homeassistant/components/switchbot_cloud/const.py

@@ -0,0 +1,7 @@
+"""Constants for the SwitchBot Cloud integration."""
+from datetime import timedelta
+from typing import Final
+
+DOMAIN: Final = "switchbot_cloud"
+ENTRY_TITLE = "SwitchBot Cloud"
+SCAN_INTERVAL = timedelta(seconds=600)

homeassistant/components/switchbot_cloud/coordinator.py

@@ -0,0 +1,50 @@
+"""SwitchBot Cloud coordinator."""
+from asyncio import timeout
+from logging import getLogger
+from typing import Any
+
+from switchbot_api import CannotConnect, Device, Remote, SwitchBotAPI
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DOMAIN, SCAN_INTERVAL
+
+_LOGGER = getLogger(__name__)
+
+Status = dict[str, Any] | None
+
+
+class SwitchBotCoordinator(DataUpdateCoordinator[Status]):
+    """SwitchBot Cloud coordinator."""
+
+    _api: SwitchBotAPI
+    _device_id: str
+    _should_poll = False
+
+    def __init__(
+        self, hass: HomeAssistant, api: SwitchBotAPI, device: Device | Remote
+    ) -> None:
+        """Initialize SwitchBot Cloud."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=DOMAIN,
+            update_interval=SCAN_INTERVAL,
+        )
+        self._api = api
+        self._device_id = device.device_id
+        self._should_poll = not isinstance(device, Remote)
+
+    async def _async_update_data(self) -> Status:
+        """Fetch data from API endpoint."""
+        if not self._should_poll:
+            return None
+        try:
+            _LOGGER.debug("Refreshing %s", self._device_id)
+            async with timeout(10):
+                status: Status = await self._api.get_status(self._device_id)
+                _LOGGER.debug("Refreshing %s with %s", self._device_id, status)
+                return status
+        except CannotConnect as err:
+            raise UpdateFailed(f"Error communicating with API: {err}") from err

homeassistant/components/switchbot_cloud/entity.py

@@ -0,0 +1,49 @@
+"""Base class for SwitchBot via API entities."""
+from typing import Any
+
+from switchbot_api import Commands, Device, Remote, SwitchBotAPI
+
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .coordinator import SwitchBotCoordinator
+
+
+class SwitchBotCloudEntity(CoordinatorEntity[SwitchBotCoordinator]):
+    """Representation of a SwitchBot Cloud entity."""
+
+    _api: SwitchBotAPI
+    _switchbot_state: dict[str, Any] | None = None
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        api: SwitchBotAPI,
+        device: Device | Remote,
+        coordinator: SwitchBotCoordinator,
+    ) -> None:
+        """Initialize the entity."""
+        super().__init__(coordinator)
+        self._api = api
+        self._attr_unique_id = device.device_id
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, device.device_id)},
+            name=device.device_name,
+            manufacturer="SwitchBot",
+            model=device.device_type,
+        )
+
+    async def send_command(
+        self,
+        command: Commands,
+        command_type: str = "command",
+        parameters: dict | str = "default",
+    ) -> None:
+        """Send command to device."""
+        await self._api.send_command(
+            self._attr_unique_id,
+            command,
+            command_type,
+            parameters,
+        )

homeassistant/components/switchbot_cloud/manifest.json

@@ -0,0 +1,10 @@
+{
+  "domain": "switchbot_cloud",
+  "name": "SwitchBot Cloud",
+  "codeowners": ["@SeraphicRav"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/switchbot_cloud",
+  "iot_class": "cloud_polling",
+  "loggers": ["switchbot-api"],
+  "requirements": ["switchbot-api==1.1.0"]
+}

homeassistant/components/switchbot_cloud/strings.json

@@ -0,0 +1,20 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "data": {
+          "api_token": "[%key:common::config_flow::data::api_token%]",
+          "api_key": "[%key:common::config_flow::data::api_key%]"
+        }
+      }
+    },
+    "error": {
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
+      "unknown": "[%key:common::config_flow::error::unknown%]"
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
+    }
+  }
+}