Add type hints to broadlink device/heartbeat

This commit is contained in:
epenet
2023-04-20 09:44:29 +00:00
parent 6342992791
commit ab6032d254
2 changed files with 33 additions and 24 deletions

View File

@@ -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

View File

@@ -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: