New integration for Gaposa blinds

homeassistant/components/gaposa/__init__.py

@@ -0,0 +1,42 @@
+"""The Gaposa integration."""
+
+from __future__ import annotations
+
+from datetime import timedelta
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .const import UPDATE_INTERVAL
+from .coordinator import DataUpdateCoordinatorGaposa
+
+PLATFORMS: list[Platform] = [Platform.COVER]
+
+type GaposaConfigEntry = ConfigEntry[DataUpdateCoordinatorGaposa]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: GaposaConfigEntry) -> bool:
+    """Set up Gaposa from a config entry."""
+    coordinator = DataUpdateCoordinatorGaposa(
+        hass,
+        entry,
+        name=entry.title,
+        update_interval=timedelta(seconds=UPDATE_INTERVAL),
+    )
+    try:
+        await coordinator.async_config_entry_first_refresh()
+    except Exception:
+        await coordinator.async_shutdown()
+        raise
+
+    entry.runtime_data = coordinator
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: GaposaConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        await entry.runtime_data.async_shutdown()
+    return unload_ok

homeassistant/components/gaposa/config_flow.py

@@ -0,0 +1,84 @@
+"""Config flow for Gaposa integration."""
+
+from __future__ import annotations
+
+from asyncio import timeout
+from collections.abc import Mapping
+import logging
+from typing import Any
+
+from aiohttp import ClientError
+from pygaposa import FirebaseAuthException, Gaposa, GaposaAuthException
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_API_KEY, CONF_PASSWORD, CONF_USERNAME
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import DEFAULT_GATEWAY_NAME, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_API_KEY): str,
+        vol.Required(CONF_USERNAME): str,
+        vol.Required(CONF_PASSWORD): str,
+    }
+)
+
+
+class GaposaConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Gaposa."""
+
+    async def _async_validate_credentials(
+        self, data: Mapping[str, Any]
+    ) -> tuple[str | None, str]:
+        """Attempt to authenticate against the Gaposa cloud.
+
+        Returns a ``(client_id, error)`` tuple. ``client_id`` is ``None``
+        on any failure and ``error`` is ``""`` on success.
+        """
+        gaposa = Gaposa(
+            data[CONF_API_KEY],
+            websession=async_get_clientsession(self.hass),
+        )
+        try:
+            async with timeout(10):
+                await gaposa.login(data[CONF_USERNAME], data[CONF_PASSWORD])
+        except (GaposaAuthException, FirebaseAuthException) as exc:
+            _LOGGER.debug("Gaposa authentication failed: %s", exc)
+            return None, "invalid_auth"
+        except (ClientError, TimeoutError, OSError) as exc:
+            _LOGGER.debug("Gaposa connection failed: %s", exc)
+            return None, "cannot_connect"
+        except Exception:
+            _LOGGER.exception("Unexpected exception during Gaposa login")
+            return None, "unknown"
+        finally:
+            await gaposa.close()
+
+        if not gaposa.clients:
+            return None, "unknown"
+        return gaposa.clients[0][0].id, ""
+
+    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:
+            client_id, error = await self._async_validate_credentials(user_input)
+            if error:
+                errors["base"] = error
+            else:
+                await self.async_set_unique_id(client_id)
+                self._abort_if_unique_id_configured()
+                return self.async_create_entry(
+                    title=DEFAULT_GATEWAY_NAME, data=user_input
+                )
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )

homeassistant/components/gaposa/const.py

@@ -0,0 +1,28 @@
+"""Constants for the Gaposa integration."""
+
+DOMAIN = "gaposa"
+DEFAULT_GATEWAY_NAME = "Gaposa Gateway"
+
+# Motor state strings returned by pygaposa's Motor.state attribute.
+# These map directly onto the values the Gaposa cloud emits.
+STATE_UP = "UP"
+STATE_DOWN = "DOWN"
+
+# Command strings recorded on a cover entity to remember what the
+# last user-initiated action was. They are compared in is_opening /
+# is_closing to decide whether the cover should report as moving.
+COMMAND_UP = "UP"
+COMMAND_DOWN = "DOWN"
+COMMAND_STOP = "STOP"
+
+# Seconds between coordinator refreshes during normal operation and
+# after a transient failure, respectively. The fast interval lets the
+# integration recover quickly from a blip without hammering the API.
+UPDATE_INTERVAL = 600
+UPDATE_INTERVAL_FAST = 60
+
+# Seconds a cover entity reports as "opening"/"closing" after an
+# open/close command is issued. Gaposa's cloud API does not report
+# motion state directly, so we approximate it from the time the
+# command was sent.
+MOTION_DELAY = 60

homeassistant/components/gaposa/coordinator.py

@@ -0,0 +1,148 @@
+"""Data update coordinator for the Gaposa integration."""
+
+from __future__ import annotations
+
+from asyncio import timeout
+from collections.abc import Callable
+from datetime import timedelta
+import logging
+from typing import TYPE_CHECKING
+
+from aiohttp import ClientError
+from pygaposa import Device, FirebaseAuthException, Gaposa, GaposaAuthException, Motor
+
+from homeassistant.const import CONF_API_KEY, CONF_PASSWORD, CONF_USERNAME
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import UPDATE_INTERVAL, UPDATE_INTERVAL_FAST
+
+if TYPE_CHECKING:
+    from . import GaposaConfigEntry
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class DataUpdateCoordinatorGaposa(DataUpdateCoordinator[dict[str, Motor]]):
+    """Fetch state for every Gaposa motor on the account."""
+
+    config_entry: GaposaConfigEntry
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        config_entry: GaposaConfigEntry,
+        *,
+        name: str,
+        update_interval: timedelta,
+    ) -> None:
+        """Initialize the coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            config_entry=config_entry,
+            name=name,
+            update_interval=update_interval,
+        )
+        self.gaposa: Gaposa | None = None
+        self.devices: list[Device] = []
+        self._listener: Callable[[], None] | None = None
+
+    async def _async_setup(self) -> None:
+        """Log in to the Gaposa API once, before the first refresh."""
+        websession = async_get_clientsession(self.hass)
+        gaposa = Gaposa(self.config_entry.data[CONF_API_KEY], websession=websession)
+        try:
+            async with timeout(10):
+                await gaposa.login(
+                    self.config_entry.data[CONF_USERNAME],
+                    self.config_entry.data[CONF_PASSWORD],
+                )
+        except (GaposaAuthException, FirebaseAuthException) as exc:
+            await gaposa.close()
+            raise ConfigEntryNotReady("Gaposa authentication failed") from exc
+        except (ClientError, TimeoutError, OSError) as exc:
+            await gaposa.close()
+            raise ConfigEntryNotReady(f"Error connecting to Gaposa: {exc}") from exc
+        self.gaposa = gaposa
+
+    async def _async_update_data(self) -> dict[str, Motor]:
+        """Refresh motor state from the Gaposa cloud."""
+        assert self.gaposa is not None  # set in _async_setup
+
+        try:
+            async with timeout(10):
+                await self.gaposa.update()
+        except (
+            GaposaAuthException,
+            FirebaseAuthException,
+            ClientError,
+            TimeoutError,
+            OSError,
+        ) as exc:
+            self.update_interval = timedelta(seconds=UPDATE_INTERVAL_FAST)
+            raise UpdateFailed(f"Error talking to Gaposa: {exc}") from exc
+
+        # pygaposa polls the Firestore REST API internally after commands
+        # (every 2 s for ~20 s). Register a listener on each device so
+        # those rapid post-command polls push fresh data to our entities
+        # via async_set_updated_data, rather than waiting for the next
+        # coordinator poll (600 s).
+        if self._listener is None:
+            self._listener = self._on_device_polled
+
+        current_devices: list[Device] = []
+        for client, _user in self.gaposa.clients:
+            for device in client.devices:
+                current_devices.append(device)
+                if device not in self.devices:
+                    device.addListener(self._listener)
+
+        for device in self.devices:
+            if device not in current_devices:
+                device.removeListener(self._listener)
+
+        self.devices = current_devices
+
+        self.update_interval = timedelta(seconds=UPDATE_INTERVAL)
+
+        return self._get_data_from_devices()
+
+    def _get_data_from_devices(self) -> dict[str, Motor]:
+        """Flatten all motors across all devices into a single dict.
+
+        The dictionary key is a unique id for the motor of the form
+        ``<device serial number>.motors.<motor.id>``.
+        """
+        data: dict[str, Motor] = {}
+        if self.gaposa is None:
+            return data
+        for client, _user in self.gaposa.clients:
+            for device in client.devices:
+                for motor in device.motors:
+                    data[f"{device.serial}.motors.{motor.id}"] = motor
+        return data
+
+    def _on_device_polled(self) -> None:
+        """Called by pygaposa after each internal poll of a device document.
+
+        This fires during pygaposa's rapid post-command polling (every ~2 s)
+        and pushes the latest motor state to all coordinator subscribers
+        without waiting for the next scheduled coordinator refresh.
+        """
+        _LOGGER.debug("Gaposa device polled, pushing new data")
+        self.async_set_updated_data(self._get_data_from_devices())
+
+    async def async_shutdown(self) -> None:
+        """Detach listeners and close the Gaposa session."""
+        await super().async_shutdown()
+        if self._listener is not None:
+            for device in self.devices:
+                device.removeListener(self._listener)
+            self._listener = None
+        self.devices = []
+        if self.gaposa is not None:
+            await self.gaposa.close()
+            self.gaposa = None