Add HomeLink integration (home-assistant#136460)

Co-authored-by: Nicholas Aelick <[email protected]>

Author: ryanjones-gentex

CODEOWNERS

@@ -0,0 +1,58 @@
+"""The homelink integration."""
+
+from __future__ import annotations
+
+from homelink.mqtt_provider import MQTTProvider
+
+from homeassistant.const import EVENT_HOMEASSISTANT_STOP, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import aiohttp_client, config_entry_oauth2_flow
+
+from . import oauth2
+from .const import DOMAIN
+from .coordinator import HomeLinkConfigEntry, HomeLinkCoordinator, HomeLinkData
+
+PLATFORMS: list[Platform] = [Platform.EVENT]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: HomeLinkConfigEntry) -> bool:
+    """Set up homelink from a config entry."""
+    auth_implementation = oauth2.SRPAuthImplementation(hass, DOMAIN)
+
+    config_entry_oauth2_flow.async_register_implementation(
+        hass, DOMAIN, auth_implementation
+    )
+
+    implementation = (
+        await config_entry_oauth2_flow.async_get_config_entry_implementation(
+            hass, entry
+        )
+    )
+
+    session = config_entry_oauth2_flow.OAuth2Session(hass, entry, implementation)
+    authenticated_session = oauth2.AsyncConfigEntryAuth(
+        aiohttp_client.async_get_clientsession(hass), session
+    )
+
+    provider = MQTTProvider(authenticated_session)
+    coordinator = HomeLinkCoordinator(hass, provider, entry)
+
+    entry.async_on_unload(
+        hass.bus.async_listen_once(
+            EVENT_HOMEASSISTANT_STOP, coordinator.async_on_unload
+        )
+    )
+
+    await coordinator.async_config_entry_first_refresh()
+    entry.runtime_data = HomeLinkData(
+        provider=provider, coordinator=coordinator, last_update_id=None
+    )
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: HomeLinkConfigEntry) -> bool:
+    """Unload a config entry."""
+    await entry.runtime_data.coordinator.async_on_unload(None)
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/gentex_homelink/__init__.py

@@ -0,0 +1,14 @@
+"""application_credentials platform for the gentex homelink integration."""
+
+from homeassistant.components.application_credentials import ClientCredential
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import config_entry_oauth2_flow
+
+from . import oauth2
+
+
+async def async_get_auth_implementation(
+    hass: HomeAssistant, auth_domain: str, _credential: ClientCredential
+) -> config_entry_oauth2_flow.AbstractOAuth2Implementation:
+    """Return custom SRPAuth implementation."""
+    return oauth2.SRPAuthImplementation(hass, auth_domain)

homeassistant/components/gentex_homelink/application_credentials.py

@@ -0,0 +1,66 @@
+"""Config flow for homelink."""
+
+import logging
+from typing import Any
+
+import botocore.exceptions
+from homelink.auth.srp_auth import SRPAuth
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
+from homeassistant.helpers.config_entry_oauth2_flow import AbstractOAuth2FlowHandler
+
+from .const import DOMAIN
+from .oauth2 import SRPAuthImplementation
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class SRPFlowHandler(AbstractOAuth2FlowHandler, domain=DOMAIN):
+    """Config flow to handle homelink OAuth2 authentication."""
+
+    DOMAIN = DOMAIN
+
+    def __init__(self) -> None:
+        """Set up the flow handler."""
+        super().__init__()
+        self.flow_impl = SRPAuthImplementation(self.hass, DOMAIN)
+
+    @property
+    def logger(self):
+        """Get the logger."""
+        return _LOGGER
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> config_entries.ConfigFlowResult:
+        """Ask for username and password."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            self._async_abort_entries_match({CONF_EMAIL: user_input[CONF_EMAIL]})
+
+            srp_auth = SRPAuth()
+            try:
+                tokens = await self.hass.async_add_executor_job(
+                    srp_auth.async_get_access_token,
+                    user_input[CONF_EMAIL],
+                    user_input[CONF_PASSWORD],
+                )
+            except botocore.exceptions.ClientError:
+                _LOGGER.exception("Error authenticating homelink account")
+                errors["base"] = "srp_auth_failed"
+            except Exception:
+                _LOGGER.exception("An unexpected error occurred")
+                errors["base"] = "unknown"
+            else:
+                self.external_data = {"tokens": tokens}
+                return await self.async_step_creation()
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {vol.Required(CONF_EMAIL): str, vol.Required(CONF_PASSWORD): str}
+            ),
+            errors=errors,
+        )

homeassistant/components/gentex_homelink/config_flow.py

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

homeassistant/components/gentex_homelink/const.py

@@ -0,0 +1,113 @@
+"""Makes requests to the state server and stores the resulting data so that the buttons can access it."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from functools import partial
+import logging
+from typing import TYPE_CHECKING, TypedDict
+
+from homelink.model.device import Device
+from homelink.mqtt_provider import MQTTProvider
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.util.ssl import get_default_context
+
+if TYPE_CHECKING:
+    from .event import HomeLinkEventEntity
+
+_LOGGER = logging.getLogger(__name__)
+
+type HomeLinkConfigEntry = ConfigEntry[HomeLinkData]
+type EventCallback = Callable[[HomeLinkEventData], None]
+
+
+@dataclass
+class HomeLinkData:
+    """Class for HomeLink integration runtime data."""
+
+    provider: MQTTProvider
+    coordinator: HomeLinkCoordinator
+    last_update_id: str | None
+
+
+class HomeLinkEventData(TypedDict):
+    """Data for a single event."""
+
+    requestId: str
+    timestamp: int
+
+
+class HomeLinkMQTTMessage(TypedDict):
+    """HomeLink MQTT Event message."""
+
+    type: str
+    data: dict[str, HomeLinkEventData]  # Each key is a button id
+
+
+class HomeLinkCoordinator:
+    """HomeLink integration coordinator."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        provider: MQTTProvider,
+        config_entry: HomeLinkConfigEntry,
+    ) -> None:
+        """Initialize my coordinator."""
+        self.hass = hass
+        self.config_entry = config_entry
+        self.provider = provider
+        self.device_data: list[Device] = []
+        self.buttons: list[HomeLinkEventEntity] = []
+        self._listeners: dict[str, EventCallback] = {}
+
+    @callback
+    def async_add_event_listener(
+        self, update_callback: EventCallback, target_event_id: str
+    ) -> Callable[[], None]:
+        """Listen for updates."""
+        self._listeners[target_event_id] = update_callback
+        return partial(self.__async_remove_listener_internal, target_event_id)
+
+    def __async_remove_listener_internal(self, listener_id: str):
+        del self._listeners[listener_id]
+
+    @callback
+    def async_handle_state_data(self, data: dict[str, HomeLinkEventData]):
+        """Notify listeners."""
+        for button_id, event in data.items():
+            if listener := self._listeners.get(button_id):
+                listener(event)
+
+    async def async_config_entry_first_refresh(self) -> None:
+        """Refresh data for the first time when a config entry is setup."""
+        await self._async_setup()
+
+    async def async_on_unload(self, _event):
+        """Disconnect and unregister when unloaded."""
+        await self.provider.disable()
+
+    async def _async_setup(self) -> None:
+        """Set up the coordinator."""
+        await self.provider.enable(get_default_context())
+        await self.discover_devices()
+        self.provider.listen(self.on_message)
+
+    async def discover_devices(self):
+        """Discover devices and build the Entities."""
+        self.device_data = await self.provider.discover()
+
+    def on_message(
+        self: HomeLinkCoordinator, _topic: str, message: HomeLinkMQTTMessage
+    ):
+        "MQTT Callback function."
+        if message["type"] == "state":
+            self.hass.add_job(self.async_handle_state_data, message["data"])
+        if message["type"] == "requestSync":
+            self.hass.add_job(
+                self.hass.config_entries.async_reload,
+                self.config_entry.entry_id,
+            )

homeassistant/components/gentex_homelink/coordinator.py

@@ -0,0 +1,83 @@
+"""Platform for Event integration."""
+
+from __future__ import annotations
+
+from homeassistant.components.event import EventDeviceClass, EventEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from .const import DOMAIN, EVENT_PRESSED
+from .coordinator import HomeLinkCoordinator, HomeLinkEventData
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Add the entities for the binary sensor."""
+    coordinator = config_entry.runtime_data.coordinator
+    for device in coordinator.device_data:
+        buttons = [
+            HomeLinkEventEntity(b.id, b.name, device.id, device.name, coordinator)
+            for b in device.buttons
+        ]
+        coordinator.buttons.extend(buttons)
+
+    async_add_entities(coordinator.buttons)
+
+
+# Updates are centralized by the coordinator.
+PARALLEL_UPDATES = 0
+
+
+class HomeLinkEventEntity(EventEntity):
+    """Event Entity."""
+
+    _attr_has_entity_name = True
+    _attr_event_types = [EVENT_PRESSED]
+    _attr_device_class = EventDeviceClass.BUTTON
+
+    def __init__(
+        self,
+        id: str,
+        param_name: str,
+        device_id: str,
+        device_name: str,
+        coordinator: HomeLinkCoordinator,
+    ) -> None:
+        """Initialize the event entity."""
+
+        self.id: str = id
+        self._attr_name: str = param_name
+        self._attr_unique_id: str = id
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, device_id)},
+            name=device_name,
+        )
+        self.coordinator = coordinator
+        self.last_request_id: str | None = None
+
+    async def async_added_to_hass(self) -> None:
+        """When entity is added to hass."""
+        await super().async_added_to_hass()
+        self.async_on_remove(
+            self.coordinator.async_add_event_listener(
+                self._handle_event_data_update, self.id
+            )
+        )
+
+    @callback
+    def _handle_event_data_update(self, update_data: HomeLinkEventData) -> None:
+        """Update this button."""
+
+        if update_data["requestId"] != self.last_request_id:
+            self._trigger_event(EVENT_PRESSED)
+            self.last_request_id = update_data["requestId"]
+
+        self.async_write_ha_state()
+
+    async def async_update(self):
+        """Request early polling. Left intentionally blank because it's not possible in this implementation."""

homeassistant/components/gentex_homelink/event.py

@@ -0,0 +1,11 @@
+{
+  "domain": "gentex_homelink",
+  "name": "HomeLink",
+  "codeowners": ["@niaexa", "@ryanjones-gentex"],
+  "config_flow": true,
+  "dependencies": ["application_credentials"],
+  "documentation": "https://www.home-assistant.io/integrations/gentex_homelink",
+  "iot_class": "cloud_push",
+  "quality_scale": "bronze",
+  "requirements": ["homelink-integration-api==0.0.1"]
+}