Add type hints to broadlink device/heartbeat (#91737)

* Add type hints to broadlink device/heartbeat

* Improve

* Force bool

* Revert "Force bool"

This reverts commit 65bce837d0e9beb4eb5aff8c190b412969684d4a.

homeassistant/components/broadlink/__init__.py

@@ -16,7 +16,7 @@ from .heartbeat import BroadlinkHeartbeat
 class BroadlinkData:
     """Class for sharing data within the Broadlink integration."""
 
-    devices: dict = field(default_factory=dict)
+    devices: dict[str, BroadlinkDevice] = field(default_factory=dict)
     platforms: dict = field(default_factory=dict)
     heartbeat: BroadlinkHeartbeat | None = None
 
@@ -29,7 +29,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up a Broadlink device from a config entry."""
-    data = hass.data[DOMAIN]
+    data: BroadlinkData = hass.data[DOMAIN]
 
     if data.heartbeat is None:
         data.heartbeat = BroadlinkHeartbeat(hass)
@@ -41,12 +41,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 
 async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Unload a config entry."""
-    data = hass.data[DOMAIN]
+    data: BroadlinkData = hass.data[DOMAIN]
 
     device = data.devices.pop(entry.entry_id)
     result = await device.async_unload()
 
-    if not data.devices:
+    if data.heartbeat and not data.devices:
         await data.heartbeat.async_unload()
         data.heartbeat = None
 

homeassistant/components/broadlink/device.py

@@ -13,8 +13,15 @@ from broadlink.exceptions import (
 )
 
 from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntry
-from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME, CONF_TIMEOUT, CONF_TYPE
-from homeassistant.core import HomeAssistant
+from homeassistant.const import (
+    CONF_HOST,
+    CONF_MAC,
+    CONF_NAME,
+    CONF_TIMEOUT,
+    CONF_TYPE,
+    Platform,
+)
+from homeassistant.core import CALLBACK_TYPE, HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import device_registry as dr
 
@@ -24,7 +31,7 @@ from .updater import get_update_manager
 _LOGGER = logging.getLogger(__name__)
 
 
-def get_domains(device_type):
+def get_domains(device_type: str) -> set[Platform]:
     """Return the domains available for a device type."""
     return {d for d, t in DOMAINS_AND_TYPES.items() if device_type in t}
 
@@ -32,33 +39,34 @@ def get_domains(device_type):
 class BroadlinkDevice:
     """Manages a Broadlink device."""
 
-    def __init__(self, hass, config):
+    api: blk.Device
+
+    def __init__(self, hass: HomeAssistant, config: ConfigEntry) -> None:
         """Initialize the device."""
         self.hass = hass
         self.config = config
-        self.api = None
         self.update_manager = None
-        self.fw_version = None
-        self.authorized = None
-        self.reset_jobs = []
+        self.fw_version: int | None = None
+        self.authorized: bool | None = None
+        self.reset_jobs: list[CALLBACK_TYPE] = []
 
     @property
-    def name(self):
+    def name(self) -> str:
         """Return the name of the device."""
         return self.config.title
 
     @property
-    def unique_id(self):
+    def unique_id(self) -> str | None:
         """Return the unique id of the device."""
         return self.config.unique_id
 
     @property
-    def mac_address(self):
+    def mac_address(self) -> str:
         """Return the mac address of the device."""
         return self.config.data[CONF_MAC]
 
     @property
-    def available(self):
+    def available(self) -> bool | None:
         """Return True if the device is available."""
         if self.update_manager is None:
             return False
@@ -77,14 +85,14 @@ class BroadlinkDevice:
         device_registry.async_update_device(device_entry.id, name=entry.title)
         await hass.config_entries.async_reload(entry.entry_id)
 
-    def _get_firmware_version(self):
+    def _get_firmware_version(self) -> int | None:
         """Get firmware version."""
         self.api.auth()
         with suppress(BroadlinkException, OSError):
             return self.api.get_fwversion()
         return None
 
-    async def async_setup(self):
+    async def async_setup(self) -> bool:
         """Set up the device and related entities."""
         config = self.config
 
@@ -132,7 +140,7 @@ class BroadlinkDevice:
 
         return True
 
-    async def async_unload(self):
+    async def async_unload(self) -> bool:
         """Unload the device and related entities."""
         if self.update_manager is None:
             return True
@@ -144,7 +152,7 @@ class BroadlinkDevice:
             self.config, get_domains(self.api.type)
         )
 
-    async def async_auth(self):
+    async def async_auth(self) -> bool:
         """Authenticate to the device."""
         try:
             await self.hass.async_add_executor_job(self.api.auth)
@@ -167,7 +175,7 @@ class BroadlinkDevice:
                 raise
             return await self.hass.async_add_executor_job(request)
 
-    async def _async_handle_auth_error(self):
+    async def _async_handle_auth_error(self) -> None:
         """Handle an authentication error."""
         if self.authorized is False:
             return

homeassistant/components/broadlink/heartbeat.py

@@ -5,6 +5,7 @@ import logging
 import broadlink as blk
 
 from homeassistant.const import CONF_HOST
+from homeassistant.core import CALLBACK_TYPE, HomeAssistant
 from homeassistant.helpers import event
 
 from .const import DOMAIN
@@ -21,12 +22,12 @@ class BroadlinkHeartbeat:
 
     HEARTBEAT_INTERVAL = dt.timedelta(minutes=2)
 
-    def __init__(self, hass):
+    def __init__(self, hass: HomeAssistant) -> None:
         """Initialize the heartbeat."""
         self._hass = hass
-        self._unsubscribe = None
+        self._unsubscribe: CALLBACK_TYPE | None = None
 
-    async def async_setup(self):
+    async def async_setup(self) -> None:
         """Set up the heartbeat."""
         if self._unsubscribe is None:
             await self.async_heartbeat(dt.datetime.now())
@@ -34,21 +35,21 @@ class BroadlinkHeartbeat:
                 self._hass, self.async_heartbeat, self.HEARTBEAT_INTERVAL
             )
 
-    async def async_unload(self):
+    async def async_unload(self) -> None:
         """Unload the heartbeat."""
         if self._unsubscribe is not None:
             self._unsubscribe()
             self._unsubscribe = None
 
-    async def async_heartbeat(self, now):
+    async def async_heartbeat(self, _: dt.datetime) -> None:
         """Send packets to feed watchdog timers."""
         hass = self._hass
         config_entries = hass.config_entries.async_entries(DOMAIN)
-        hosts = {entry.data[CONF_HOST] for entry in config_entries}
+        hosts: set[str] = {entry.data[CONF_HOST] for entry in config_entries}
         await hass.async_add_executor_job(self.heartbeat, hosts)
 
     @staticmethod
-    def heartbeat(hosts):
+    def heartbeat(hosts: set[str]) -> None:
         """Send packets to feed watchdog timers."""
         for host in hosts:
             try: