Add Linear Garage Door integration

.strict-typing

@@ -202,6 +202,7 @@ homeassistant.components.ld2410_ble.*
 homeassistant.components.lidarr.*
 homeassistant.components.lifx.*
 homeassistant.components.light.*
+homeassistant.components.linear_garage_door.*
 homeassistant.components.litejet.*
 homeassistant.components.litterrobot.*
 homeassistant.components.local_ip.*

CODEOWNERS

@@ -700,6 +700,8 @@ build.json @home-assistant/supervisor
 /tests/components/life360/ @pnbruckner
 /homeassistant/components/light/ @home-assistant/core
 /tests/components/light/ @home-assistant/core
+/homeassistant/components/linear_garage_door/ @IceBotYT
+/tests/components/linear_garage_door/ @IceBotYT
 /homeassistant/components/linux_battery/ @fabaff
 /homeassistant/components/litejet/ @joncar
 /tests/components/litejet/ @joncar

homeassistant/components/linear_garage_door/__init__.py

@@ -0,0 +1,32 @@
+"""The Linear Garage Door integration."""
+from __future__ import annotations
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .const import DOMAIN
+from .coordinator import LinearUpdateCoordinator
+
+PLATFORMS: list[Platform] = [Platform.COVER]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Linear Garage Door from a config entry."""
+
+    coordinator = LinearUpdateCoordinator(hass, entry)
+
+    await coordinator.async_config_entry_first_refresh()
+
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
+    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):
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

homeassistant/components/linear_garage_door/config_flow.py

@@ -0,0 +1,166 @@
+"""Config flow for Linear Garage Door integration."""
+from __future__ import annotations
+
+from collections.abc import Collection, Mapping, Sequence
+import logging
+from typing import Any
+import uuid
+
+from linear_garage_door import Linear
+from linear_garage_door.errors import InvalidLoginError
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = {
+    vol.Required(CONF_EMAIL): str,
+    vol.Required(CONF_PASSWORD): str,
+}
+
+
+async def validate_input(
+    hass: HomeAssistant,
+    data: dict[str, str],
+) -> dict[str, Sequence[Collection[str]]]:
+    """Validate the user input allows us to connect.
+
+    Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
+    """
+
+    hub = Linear()
+
+    device_id = str(uuid.uuid4())
+    try:
+        await hub.login(
+            data["email"],
+            data["password"],
+            device_id=device_id,
+            client_session=async_get_clientsession(hass),
+        )
+
+        sites = await hub.get_sites()
+    except InvalidLoginError as err:
+        raise InvalidAuth from err
+    finally:
+        await hub.close()
+
+    info = {
+        "email": data["email"],
+        "password": data["password"],
+        "sites": sites,
+        "device_id": device_id,
+    }
+
+    return info
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Linear Garage Door."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Initialize the config flow."""
+        self.data: dict[str, Sequence[Collection[str]]] = {}
+        self._reauth_entry: config_entries.ConfigEntry | None = None
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the initial step."""
+        data_schema = STEP_USER_DATA_SCHEMA
+
+        data_schema = vol.Schema(data_schema)
+
+        if user_input is None:
+            return self.async_show_form(step_id="user", data_schema=data_schema)
+
+        errors = {}
+
+        try:
+            info = await validate_input(self.hass, user_input)
+        except InvalidAuth:
+            errors["base"] = "invalid_auth"
+        except Exception:  # pylint: disable=broad-except
+            _LOGGER.exception("Unexpected exception")
+            errors["base"] = "unknown"
+        else:
+            self.data = info
+
+            # Check if we are reauthenticating
+            if self._reauth_entry is not None:
+                self.hass.config_entries.async_update_entry(
+                    self._reauth_entry,
+                    data=self._reauth_entry.data
+                    | {"email": self.data["email"], "password": self.data["password"]},
+                )
+                await self.hass.config_entries.async_reload(self._reauth_entry.entry_id)
+                return self.async_abort(reason="reauth_successful")
+
+            return await self.async_step_site()
+
+        return self.async_show_form(
+            step_id="user", data_schema=data_schema, errors=errors
+        )
+
+    async def async_step_site(
+        self,
+        user_input: dict[str, Any] | None = None,
+    ) -> FlowResult:
+        """Handle the site step."""
+
+        if isinstance(self.data["sites"], list):
+            sites: list[dict[str, str]] = self.data["sites"]
+
+        if not user_input:
+            return self.async_show_form(
+                step_id="site",
+                data_schema=vol.Schema(
+                    {
+                        vol.Required("site"): vol.In(
+                            {site["id"]: site["name"] for site in sites}
+                        )
+                    }
+                ),
+            )
+
+        site_id = user_input["site"]
+
+        site_name = next(site["name"] for site in sites if site["id"] == site_id)
+
+        await self.async_set_unique_id(site_id)
+        self._abort_if_unique_id_configured()
+
+        return self.async_create_entry(
+            title=site_name,
+            data={
+                "site_id": site_id,
+                "email": self.data["email"],
+                "password": self.data["password"],
+                "device_id": self.data["device_id"],
+            },
+        )
+
+    async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
+        """Reauth in case of a password change or other error."""
+        self._reauth_entry = self.hass.config_entries.async_get_entry(
+            self.context["entry_id"]
+        )
+        return await self.async_step_user()
+
+
+class InvalidAuth(HomeAssistantError):
+    """Error to indicate there is invalid auth."""
+
+
+class InvalidDeviceID(HomeAssistantError):
+    """Error to indicate there is invalid device ID."""

homeassistant/components/linear_garage_door/const.py

@@ -0,0 +1,3 @@
+"""Constants for the Linear Garage Door integration."""
+
+DOMAIN = "linear_garage_door"

homeassistant/components/linear_garage_door/coordinator.py

@@ -0,0 +1,81 @@
+"""DataUpdateCoordinator for Linear."""
+from __future__ import annotations
+
+from datetime import timedelta
+import logging
+from typing import Any
+
+from linear_garage_door import Linear
+from linear_garage_door.errors import InvalidLoginError, ResponseError
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class LinearUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
+    """DataUpdateCoordinator for Linear."""
+
+    _email: str
+    _password: str
+    _device_id: str
+    _site_id: str
+    _devices: list[dict[str, list[str] | str]] | None
+    _linear: Linear
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        entry: ConfigEntry,
+    ) -> None:
+        """Initialize DataUpdateCoordinator for Linear."""
+        self._email = entry.data["email"]
+        self._password = entry.data["password"]
+        self._device_id = entry.data["device_id"]
+        self._site_id = entry.data["site_id"]
+        self._devices = None
+
+        super().__init__(
+            hass,
+            _LOGGER,
+            name="Linear Garage Door",
+            update_interval=timedelta(seconds=60),
+        )
+
+    async def _async_update_data(self) -> dict[str, Any]:
+        """Get the data for Linear."""
+
+        linear = Linear()
+
+        try:
+            await linear.login(
+                email=self._email,
+                password=self._password,
+                device_id=self._device_id,
+            )
+        except InvalidLoginError as err:
+            if (
+                str(err)
+                == "Login error: Login provided is invalid, please check the email and password"
+            ):
+                raise ConfigEntryAuthFailed from err
+            raise ConfigEntryNotReady from err
+        except ResponseError as err:
+            raise ConfigEntryNotReady from err
+
+        if not self._devices:
+            self._devices = await linear.get_devices(self._site_id)
+
+        data = {}
+
+        for device in self._devices:
+            device_id = str(device["id"])
+            state = await linear.get_device_state(device_id)
+            data[device_id] = {"name": device["name"], "subdevices": state}
+
+        await linear.close()
+
+        return data