Cleanup homelink (#158477)

homeassistant/components/gentex_homelink/coordinator.py

@@ -6,7 +6,7 @@ from collections.abc import Callable
 from dataclasses import dataclass
 from functools import partial
 import logging
-from typing import TYPE_CHECKING, TypedDict
+from typing import TypedDict
 
 from homelink.model.device import Device
 from homelink.mqtt_provider import MQTTProvider
@@ -15,9 +15,6 @@ 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]
@@ -61,7 +58,6 @@ class HomeLinkCoordinator:
         self.config_entry = config_entry
         self.provider = provider
         self.device_data: list[Device] = []
-        self.buttons: list[HomeLinkEventEntity] = []
         self._listeners: dict[str, EventCallback] = {}
 
     @callback
@@ -76,7 +72,7 @@ class HomeLinkCoordinator:
         del self._listeners[listener_id]
 
     @callback
-    def async_handle_state_data(self, data: dict[str, HomeLinkEventData]):
+    def async_handle_state_data(self, data: dict[str, HomeLinkEventData]) -> None:
         """Notify listeners."""
         for button_id, event in data.items():
             if listener := self._listeners.get(button_id):
@@ -86,7 +82,7 @@ class HomeLinkCoordinator:
         """Refresh data for the first time when a config entry is setup."""
         await self._async_setup()
 
-    async def async_on_unload(self, _event):
+    async def async_on_unload(self, _event) -> None:
         """Disconnect and unregister when unloaded."""
         await self.provider.disable()
 
@@ -100,10 +96,8 @@ class HomeLinkCoordinator:
         """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."
+    def on_message(self, _topic: str, message: HomeLinkMQTTMessage) -> None:
+        """MQTT Callback function."""
         if message["type"] == "state":
             self.hass.add_job(self.async_handle_state_data, message["data"])
         if message["type"] == "requestSync":

homeassistant/components/gentex_homelink/event.py

@@ -19,14 +19,12 @@ async def async_setup_entry(
 ) -> 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)
+    async_add_entities(
+        HomeLinkEventEntity(button.id, button.name, device.id, device.name, coordinator)
+        for device in coordinator.device_data
+        for button in device.buttons
+    )
 
 
 # Updates are centralized by the coordinator.
@@ -42,7 +40,7 @@ class HomeLinkEventEntity(EventEntity):
 
     def __init__(
         self,
-        id: str,
+        button_id: str,
         param_name: str,
         device_id: str,
         device_name: str,
@@ -50,9 +48,9 @@ class HomeLinkEventEntity(EventEntity):
     ) -> None:
         """Initialize the event entity."""
 
-        self.id: str = id
-        self._attr_name: str = param_name
-        self._attr_unique_id: str = id
+        self.button_id = button_id
+        self._attr_name = param_name
+        self._attr_unique_id = button_id
         self._attr_device_info = DeviceInfo(
             identifiers={(DOMAIN, device_id)},
             name=device_name,
@@ -65,7 +63,7 @@ class HomeLinkEventEntity(EventEntity):
         await super().async_added_to_hass()
         self.async_on_remove(
             self.coordinator.async_add_event_listener(
-                self._handle_event_data_update, self.id
+                self._handle_event_data_update, self.button_id
             )
         )
 
@@ -78,6 +76,3 @@ class HomeLinkEventEntity(EventEntity):
             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/oauth2.py

@@ -21,7 +21,7 @@ _LOGGER = logging.getLogger(__name__)
 class SRPAuthImplementation(config_entry_oauth2_flow.AbstractOAuth2Implementation):
     """Base class to abstract OAuth2 authentication."""
 
-    def __init__(self, hass: HomeAssistant, domain) -> None:
+    def __init__(self, hass: HomeAssistant, domain: str) -> None:
         """Initialize the SRP Auth implementation."""
 
         self.hass = hass
@@ -45,16 +45,13 @@ class SRPAuthImplementation(config_entry_oauth2_flow.AbstractOAuth2Implementatio
     async def async_resolve_external_data(self, external_data) -> dict:
         """Format the token from the source appropriately for HomeAssistant."""
         tokens = external_data["tokens"]
-        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 new_token
+        return {
+            "access_token": tokens["AuthenticationResult"]["AccessToken"],
+            "refresh_token": tokens["AuthenticationResult"]["RefreshToken"],
+            "token_type": tokens["AuthenticationResult"]["TokenType"],
+            "expires_in": tokens["AuthenticationResult"]["ExpiresIn"],
+            "expires_at": (time.time() + tokens["AuthenticationResult"]["ExpiresIn"]),
+        }
 
     async def _token_request(self, data: dict) -> dict:
         """Make a token request."""