diff --git a/homeassistant/components/broadlink/__init__.py b/homeassistant/components/broadlink/__init__.py index 5d996c2ee1f..559aae25abf 100644 --- a/homeassistant/components/broadlink/__init__.py +++ b/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 diff --git a/homeassistant/components/broadlink/device.py b/homeassistant/components/broadlink/device.py index d6d064ea011..87d8cf398fb 100644 --- a/homeassistant/components/broadlink/device.py +++ b/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 diff --git a/homeassistant/components/broadlink/heartbeat.py b/homeassistant/components/broadlink/heartbeat.py index b4deffa5b81..70f6aec0d0f 100644 --- a/homeassistant/components/broadlink/heartbeat.py +++ b/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: