Add HomeLink integration

homeassistant/components/gentex_homelink/__init__.py

@@ -0,0 +1,68 @@
+"""The homelink integration."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from homelink.provider import Provider
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import aiohttp_client, config_entry_oauth2_flow
+
+from . import api
+from .const import DOMAIN
+from .coordinator import HomelinkCoordinator
+
+PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR]
+
+type HomeLinkConfigEntry = ConfigEntry[dict[str, Any]]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: HomeLinkConfigEntry) -> bool:
+    """Set up homelink from a config entry."""
+    logging.debug("Starting config entry setup")
+    config_entry_oauth2_flow.async_register_implementation(
+        hass, DOMAIN, api.SRPAuthImplementation(hass, DOMAIN)
+    )
+
+    implementation = (
+        await config_entry_oauth2_flow.async_get_config_entry_implementation(
+            hass, entry
+        )
+    )
+
+    session = config_entry_oauth2_flow.OAuth2Session(hass, entry, implementation)
+    authenticated_session = api.AsyncConfigEntryAuth(
+        aiohttp_client.async_get_clientsession(hass), session
+    )
+
+    provider = Provider(authenticated_session)
+    coordinator = HomelinkCoordinator(hass, provider, entry)
+
+    entry.runtime_data = {
+        "provider": provider,
+        "coordinator": coordinator,
+        "last_update_id": None,
+    }
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    if update_listener not in entry.update_listeners:
+        entry.add_update_listener(update_listener)
+
+    if update_listener not in entry.update_listeners:
+        entry.add_update_listener(update_listener)
+    await coordinator.async_config_entry_first_refresh()
+    return True
+
+
+async def update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
+    """Update listener."""
+    await hass.config_entries.async_reload(entry.entry_id)
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: HomeLinkConfigEntry) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/gentex_homelink/api.py

@@ -0,0 +1,102 @@
+"""API for homelink bound to Home Assistant OAuth."""
+
+from json import JSONDecodeError
+import logging
+from typing import cast
+
+from aiohttp import ClientError, ClientSession
+from homelink.auth.abstract_auth import AbstractAuth
+from homelink.settings import COGNITO_CLIENT_ID
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import config_entry_oauth2_flow
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import OAUTH2_TOKEN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class SRPAuthImplementation(config_entry_oauth2_flow.AbstractOAuth2Implementation):
+    """Base class to abstract OAuth2 authentication."""
+
+    def __init__(self, hass: HomeAssistant, domain) -> None:
+        """Initialize the SRP Auth implementation."""
+        self.hass = hass
+        self._domain = domain
+        self.client_id = COGNITO_CLIENT_ID
+
+    @property
+    def name(self) -> str:
+        """Name of the implementation."""
+        return "SRPAuth"
+
+    @property
+    def domain(self) -> str:
+        """Domain that is providing the implementation."""
+        return self._domain
+
+    async def async_generate_authorize_url(self, flow_id: str) -> str:
+        """Left intentionally blank because the auth is handled by SRP."""
+        return ""
+
+    async def async_resolve_external_data(self, external_data) -> dict:
+        """No external data required."""
+        return {}
+
+    async def _token_request(self, data: dict) -> dict:
+        """Make a token request."""
+        session = async_get_clientsession(self.hass)
+
+        data["client_id"] = self.client_id
+
+        _LOGGER.debug("Sending token request to %s", OAUTH2_TOKEN)
+        resp = await session.post(OAUTH2_TOKEN, data=data)
+        if resp.status >= 400:
+            try:
+                error_response = await resp.json()
+            except (ClientError, JSONDecodeError):
+                error_response = {}
+                error_code = error_response.get("error", "unknown")
+                error_description = error_response.get(
+                    "error_description", "unknown error"
+                )
+                _LOGGER.error(
+                    "Token request for %s failed (%s): %s",
+                    self.domain,
+                    error_code,
+                    error_description,
+                )
+        resp.raise_for_status()
+        return cast(dict, await resp.json())
+
+    async def _async_refresh_token(self, token: dict) -> dict:
+        """Refresh tokens."""
+        new_token = await self._token_request(
+            {
+                "grant_type": "refresh_token",
+                "client_id": self.client_id,
+                "refresh_token": token["refresh_token"],
+            }
+        )
+        return {**token, **new_token}
+
+
+class AsyncConfigEntryAuth(AbstractAuth):
+    """Provide homelink authentication tied to an OAuth2 based config entry."""
+
+    def __init__(
+        self,
+        websession: ClientSession,
+        oauth_session: config_entry_oauth2_flow.OAuth2Session,
+    ) -> None:
+        """Initialize homelink auth."""
+        super().__init__(websession)
+        self._oauth_session = oauth_session
+
+    async def async_get_access_token(self) -> str:
+        """Return a valid access token."""
+        if not self._oauth_session.valid_token:
+            await self._oauth_session.async_ensure_token_valid()
+
+        return self._oauth_session.token["access_token"]

homeassistant/components/gentex_homelink/binary_sensor.py

@@ -0,0 +1,89 @@
+"""Platform for BinarySensor integration."""
+
+from __future__ import annotations
+
+from datetime import timedelta
+import logging
+import time
+
+from homeassistant.components.binary_sensor import BinarySensorEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+import homeassistant.helpers.device_registry as dr
+
+# Import the device class from the component that you want to support
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+SCAN_INTERVAL = timedelta(seconds=5)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up homelink from a config entry."""
+    coordinator = config_entry.runtime_data["coordinator"]
+    provider = config_entry.runtime_data["provider"]
+
+    await provider.enable()
+
+    device_data = await provider.discover()
+
+    logging.info(device_data)
+
+    for device in device_data:
+        device_info = DeviceInfo(
+            identifiers={
+                # Serial numbers are unique identifiers within a specific domain
+                (DOMAIN, device.id)
+            },
+            name=device.name,
+        )
+
+        buttons = [
+            HomelinkBinarySensor(b.id, b.name, device_info, coordinator)
+            for b in device.buttons
+        ]
+        async_add_entities(buttons)
+
+        if buttons[0].device_entry is not None:
+            registry = dr.async_get(hass)
+            registry.async_update_device(buttons[0].device_entry.id, name=device.name)
+
+
+class HomelinkBinarySensor(CoordinatorEntity, BinarySensorEntity):
+    """Binary sensor."""
+
+    def __init__(self, id, name, device_info, coordinator) -> None:
+        """Initialize the button."""
+        super().__init__(coordinator, context=id)
+        self.id = id
+        self.name = name
+        self.unique_id = f"{DOMAIN}.{id}"
+        self.device_info = device_info
+        self.on = False
+        self.last_request_id = None
+
+    @property
+    def is_on(self) -> bool:
+        """Return true if the binary sensor is on."""
+        return self.on
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Update this button."""
+        if not self.coordinator.data or self.id not in self.coordinator.data:
+            self.on = False
+        else:
+            latest_update = self.coordinator.data[self.id]
+            self.on = (time.time() - latest_update["timestamp"]) < 10 and latest_update[
+                "requestId"
+            ] != self.last_request_id
+            self.last_request_id = latest_update["requestId"]
+        self.async_write_ha_state()

homeassistant/components/gentex_homelink/config_flow.py

@@ -0,0 +1,71 @@
+"""Config flow for homelink."""
+
+import asyncio
+import time
+from typing import Any
+
+from homelink.auth.srp_auth import SRPAuth
+import voluptuous as vol
+
+from homeassistant import config_entries
+
+from .const import DOMAIN
+
+
+@config_entries.HANDLERS.register(DOMAIN)
+class SRPFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
+    """Config flow to handle homelink OAuth2 authentication."""
+
+    DOMAIN = DOMAIN
+    VERSION = 1
+    MINOR_VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> config_entries.ConfigFlowResult:
+        """Ask for username and password."""
+        errors = {}
+        if user_input is not None:
+            self._async_abort_entries_match({"email": user_input["email"]})
+
+            srp_auth = SRPAuth()
+            loop = asyncio.get_running_loop()
+            try:
+                tokens = await loop.run_in_executor(
+                    None,
+                    srp_auth.async_get_access_token,
+                    user_input["email"],
+                    user_input["password"],
+                )
+            except Exception:  # noqa: BLE001
+                errors["base"] = "Error authenticating HomeLink account"
+            else:
+                new_token = {}
+                new_token["access_token"] = tokens["AuthenticationResult"][
+                    "AccessToken"
+                ]
+                new_token["refresh_token"] = tokens["AuthenticationResult"][
+                    "RefreshToken"
+                ]
+                new_token["token_type"] = tokens["AuthenticationResult"]["TokenType"]
+                new_token["expires_in"] = tokens["AuthenticationResult"]["ExpiresIn"]
+                new_token["expires_at"] = (
+                    time.time() + tokens["AuthenticationResult"]["ExpiresIn"]
+                )
+
+                return self.async_create_entry(
+                    title="Token entry",
+                    data={
+                        "token": new_token,
+                        "auth_implementation": DOMAIN,
+                        "last_update_id": None,
+                        "email": user_input["email"],
+                    },
+                )
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {vol.Required("email"): str, vol.Required("password"): str}
+            ),
+            errors=errors,
+        )

homeassistant/components/gentex_homelink/const.py

@@ -0,0 +1,4 @@
+"""Constants for the homelink integration."""
+
+DOMAIN = "gentex_homelink"
+OAUTH2_TOKEN = "https://auth.homelinkcloud.com/oauth2/token"