Add CatGenie integration

.strict-typing

@@ -139,6 +139,7 @@ homeassistant.components.cambridge_audio.*
 homeassistant.components.camera.*
 homeassistant.components.canary.*
 homeassistant.components.casper_glow.*
+homeassistant.components.catgenie.*
 homeassistant.components.cert_expiry.*
 homeassistant.components.clickatell.*
 homeassistant.components.clicksend.*

homeassistant/components/catgenie/__init__.py

@@ -0,0 +1,83 @@
+"""The CatGenie integration."""
+
+from __future__ import annotations
+
+from contextlib import AsyncExitStack
+
+from catgenie import CatGenieAuth, CatGenieClient, Credentials, Device
+from catgenie.exceptions import CatGenieAuthenticationError, CatGenieException
+
+from homeassistant.const import CONF_TOKEN, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
+
+from .coordinator import (
+    CatGenieConfigEntry,
+    CatGenieDeviceCoordinator,
+    CatGenieRuntimeData,
+)
+
+PLATFORMS: list[Platform] = [Platform.SENSOR]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: CatGenieConfigEntry) -> bool:
+    """Set up CatGenie from a config entry."""
+    credentials = Credentials(refresh_token=entry.data[CONF_TOKEN])
+    stack = AsyncExitStack()
+
+    auth = CatGenieAuth()
+    await stack.enter_async_context(auth)
+    auth.credentials = credentials
+
+    # Obtain a fresh access token using the stored refresh token
+    try:
+        credentials = await auth.refresh()
+    except CatGenieAuthenticationError as err:
+        await stack.aclose()
+        raise ConfigEntryAuthFailed(
+            translation_domain="catgenie",
+            translation_key="authentication_failed",
+        ) from err
+    except CatGenieException as err:
+        await stack.aclose()
+        raise ConfigEntryNotReady(
+            translation_domain="catgenie",
+            translation_key="communication_error",
+            translation_placeholders={"error": str(err)},
+        ) from err
+
+    client = CatGenieClient(credentials)
+    await stack.enter_async_context(client)
+    client.set_auth(auth)
+
+    # Fetch all devices and create a coordinator for each
+    try:
+        devices: list[Device] = await client.get_devices()
+    except Exception as err:
+        await stack.aclose()
+        raise ConfigEntryNotReady(
+            translation_domain="catgenie",
+            translation_key="communication_error",
+            translation_placeholders={"error": str(err)},
+        ) from err
+
+    device_coordinators: dict[str, CatGenieDeviceCoordinator] = {}
+    for device in devices:
+        coordinator = CatGenieDeviceCoordinator(hass, entry, client, auth, device)
+        await coordinator.async_config_entry_first_refresh()
+        device_coordinators[device.manufacturer_id] = coordinator
+
+    entry.runtime_data = CatGenieRuntimeData(
+        auth=auth,
+        client=client,
+        device_coordinators=device_coordinators,
+    )
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: CatGenieConfigEntry) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/catgenie/config_flow.py

@@ -0,0 +1,171 @@
+"""Config flow for the CatGenie integration."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from catgenie import CatGenieAuth
+from catgenie.exceptions import CatGenieAuthenticationError, CatGenieException
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_CODE, CONF_COUNTRY_CODE, CONF_TOKEN
+
+from .const import DOMAIN, LOGGER
+
+CONF_PHONE = "phone"
+
+STEP_PHONE_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_COUNTRY_CODE): int,
+        vol.Required(CONF_PHONE): str,
+    }
+)
+
+STEP_CODE_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_CODE): str,
+    }
+)
+
+
+class CatGenieConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for CatGenie."""
+
+    VERSION = 1
+
+    _country_code: int
+    _phone: str
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the phone number step."""
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            self._country_code = user_input[CONF_COUNTRY_CODE]
+            self._phone = user_input[CONF_PHONE]
+
+            try:
+                async with CatGenieAuth() as auth:
+                    await auth.request_login_code(
+                        country_code=self._country_code,
+                        phone=self._phone,
+                    )
+            except CatGenieAuthenticationError:
+                errors["base"] = "cannot_connect"
+            except CatGenieException:
+                LOGGER.exception("Unexpected exception requesting login code")
+                errors["base"] = "unknown"
+            else:
+                return await self.async_step_code()
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=STEP_PHONE_DATA_SCHEMA,
+            errors=errors,
+        )
+
+    async def async_step_code(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the SMS verification code step."""
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            try:
+                async with CatGenieAuth() as auth:
+                    credentials = await auth.login(
+                        country_code=self._country_code,
+                        phone=self._phone,
+                        code=user_input[CONF_CODE],
+                    )
+            except CatGenieAuthenticationError:
+                errors["base"] = "invalid_auth"
+            except CatGenieException:
+                LOGGER.exception("Unexpected exception during login")
+                errors["base"] = "unknown"
+            else:
+                await self.async_set_unique_id(credentials.user_id)
+                self._abort_if_unique_id_configured()
+
+                return self.async_create_entry(
+                    title=f"CatGenie ({self._phone})",
+                    data={CONF_TOKEN: credentials.refresh_token},
+                )
+
+        return self.async_show_form(
+            step_id=CONF_CODE,
+            data_schema=STEP_CODE_DATA_SCHEMA,
+            errors=errors,
+        )
+
+    async def async_step_reauth(
+        self, entry_data: Mapping[str, Any]
+    ) -> ConfigFlowResult:
+        """Handle re-authentication when the token is rejected."""
+        return await self.async_step_reauth_confirm()
+
+    async def async_step_reauth_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle re-auth: collect phone number and request a new SMS code."""
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            self._country_code = user_input[CONF_COUNTRY_CODE]
+            self._phone = user_input[CONF_PHONE]
+
+            try:
+                async with CatGenieAuth() as auth:
+                    await auth.request_login_code(
+                        country_code=self._country_code,
+                        phone=self._phone,
+                    )
+            except CatGenieAuthenticationError:
+                errors["base"] = "cannot_connect"
+            except CatGenieException:
+                LOGGER.exception("Unexpected exception requesting login code")
+                errors["base"] = "unknown"
+            else:
+                return await self.async_step_reauth_code()
+
+        return self.async_show_form(
+            step_id="reauth_confirm",
+            data_schema=STEP_PHONE_DATA_SCHEMA,
+            errors=errors,
+        )
+
+    async def async_step_reauth_code(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle re-auth: enter the new SMS code."""
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            try:
+                async with CatGenieAuth() as auth:
+                    credentials = await auth.login(
+                        country_code=self._country_code,
+                        phone=self._phone,
+                        code=user_input[CONF_CODE],
+                    )
+            except CatGenieAuthenticationError:
+                errors["base"] = "invalid_auth"
+            except CatGenieException:
+                LOGGER.exception("Unexpected exception during re-auth login")
+                errors["base"] = "unknown"
+            else:
+                reauth_entry = self._get_reauth_entry()
+                return self.async_update_reload_and_abort(
+                    reauth_entry,
+                    data={CONF_TOKEN: credentials.refresh_token},
+                )
+
+        return self.async_show_form(
+            step_id="reauth_code",
+            data_schema=STEP_CODE_DATA_SCHEMA,
+            errors=errors,
+        )

homeassistant/components/catgenie/const.py

@@ -0,0 +1,8 @@
+"""Constants for the CatGenie integration."""
+
+from __future__ import annotations
+
+import logging
+
+DOMAIN = "catgenie"
+LOGGER = logging.getLogger(__package__)

homeassistant/components/catgenie/coordinator.py

@@ -0,0 +1,95 @@
+"""Data update coordinator for the CatGenie integration."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import timedelta
+
+from catgenie import CatGenieAuth, CatGenieClient, Credentials, Device
+from catgenie.exceptions import CatGenieAPIError, CatGenieAuthenticationError
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_TOKEN
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DOMAIN, LOGGER
+
+SCAN_INTERVAL = timedelta(seconds=60)
+
+
+@dataclass
+class CatGenieRuntimeData:
+    """Runtime data for CatGenie."""
+
+    auth: CatGenieAuth
+    client: CatGenieClient
+    device_coordinators: dict[str, CatGenieDeviceCoordinator]
+
+
+type CatGenieConfigEntry = ConfigEntry[CatGenieRuntimeData]
+
+
+class CatGenieDeviceCoordinator(DataUpdateCoordinator[Device]):
+    """Coordinator for a single CatGenie device."""
+
+    config_entry: CatGenieConfigEntry
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        config_entry: CatGenieConfigEntry,
+        client: CatGenieClient,
+        auth: CatGenieAuth,
+        device: Device,
+    ) -> None:
+        """Initialize the coordinator."""
+        super().__init__(
+            hass,
+            LOGGER,
+            config_entry=config_entry,
+            name=f"{DOMAIN}_{device.manufacturer_id}",
+            update_interval=SCAN_INTERVAL,
+        )
+        self.client = client
+        self.auth = auth
+        self.device_id = device.manufacturer_id
+
+    def _update_entry_tokens(self, credentials: Credentials) -> None:
+        """Persist refreshed refresh token back to the config entry."""
+        self.hass.config_entries.async_update_entry(
+            self.config_entry,
+            data={CONF_TOKEN: credentials.refresh_token},
+        )
+
+    async def _async_update_data(self) -> Device:
+        """Fetch data for this device from the CatGenie API."""
+        try:
+            devices = await self.client.get_devices()
+        except CatGenieAuthenticationError:
+            try:
+                credentials = await self.auth.refresh()
+                self._update_entry_tokens(credentials)
+                devices = await self.client.get_devices()
+            except CatGenieAuthenticationError as refresh_err:
+                raise ConfigEntryAuthFailed(
+                    translation_domain=DOMAIN,
+                    translation_key="authentication_failed",
+                ) from refresh_err
+        except CatGenieAPIError as err:
+            raise UpdateFailed(
+                translation_domain=DOMAIN,
+                translation_key="communication_error",
+                translation_placeholders={"error": str(err)},
+            ) from err
+
+        for device in devices:
+            if device.manufacturer_id == self.device_id:
+                return device
+
+        raise UpdateFailed(
+            translation_domain=DOMAIN,
+            translation_key="communication_error",
+            translation_placeholders={"error": f"Device {self.device_id} not found"},
+        )