Enforce typing in bond (#47187)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
J. Nick Koston 2021-02-28 20:16:30 -06:00 committed by GitHub
parent 715a254913
commit 5784e14d0c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 63 additions and 57 deletions

View File

@ -19,13 +19,13 @@ PLATFORMS = ["cover", "fan", "light", "switch"]
_API_TIMEOUT = SLOW_UPDATE_WARNING - 1
async def async_setup(hass: HomeAssistant, config: dict):
async def async_setup(hass: HomeAssistant, config: dict) -> bool:
"""Set up the Bond component."""
hass.data.setdefault(DOMAIN, {})
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Bond from a config entry."""
host = entry.data[CONF_HOST]
token = entry.data[CONF_ACCESS_TOKEN]
@ -50,6 +50,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
if not entry.unique_id:
hass.config_entries.async_update_entry(entry, unique_id=hub.bond_id)
assert hub.bond_id is not None
hub_name = hub.name or hub.bond_id
device_registry = await dr.async_get_registry(hass)
device_registry.async_get_or_create(
@ -96,7 +97,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@callback
def _async_remove_old_device_identifiers(
config_entry_id: str, device_registry: dr.DeviceRegistry, hub: BondHub
):
) -> None:
"""Remove the non-unique device registry entries."""
for device in hub.devices:
dev = device_registry.async_get_device(identifiers={(DOMAIN, device.device_id)})

View File

@ -13,6 +13,7 @@ from homeassistant.const import (
CONF_NAME,
HTTP_UNAUTHORIZED,
)
from homeassistant.helpers.typing import DiscoveryInfoType
from .const import DOMAIN # pylint:disable=unused-import
from .utils import BondHub
@ -27,7 +28,7 @@ DISCOVERY_SCHEMA = vol.Schema({vol.Required(CONF_ACCESS_TOKEN): str})
TOKEN_SCHEMA = vol.Schema({})
async def _validate_input(data: Dict[str, Any]) -> Tuple[str, Optional[str]]:
async def _validate_input(data: Dict[str, Any]) -> Tuple[str, str]:
"""Validate the user input allows us to connect."""
bond = Bond(data[CONF_HOST], data[CONF_ACCESS_TOKEN])
@ -57,11 +58,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_PUSH
def __init__(self):
def __init__(self) -> None:
"""Initialize config flow."""
self._discovered: Optional[dict] = None
self._discovered: Dict[str, str] = {}
async def _async_try_automatic_configure(self):
async def _async_try_automatic_configure(self) -> None:
"""Try to auto configure the device.
Failure is acceptable here since the device may have been
@ -82,9 +83,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
_, hub_name = await _validate_input(self._discovered)
self._discovered[CONF_NAME] = hub_name
async def async_step_zeroconf(
self, discovery_info: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
async def async_step_zeroconf(self, discovery_info: DiscoveryInfoType) -> Dict[str, Any]: # type: ignore
"""Handle a flow initialized by zeroconf discovery."""
name: str = discovery_info[CONF_NAME]
host: str = discovery_info[CONF_HOST]
@ -107,7 +106,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_confirm()
async def async_step_confirm(
self, user_input: Dict[str, Any] = None
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Handle confirmation flow for discovered bond hub."""
errors = {}
@ -148,7 +147,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
)
async def async_step_user(
self, user_input: Dict[str, Any] = None
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Handle a flow initialized by the user."""
errors = {}

View File

@ -23,7 +23,7 @@ async def async_setup_entry(
hub: BondHub = data[HUB]
bpup_subs: BPUPSubscriptions = data[BPUP_SUBS]
covers = [
covers: List[Entity] = [
BondCover(hub, device, bpup_subs)
for device in hub.devices
if device.type == DeviceType.MOTORIZED_SHADES
@ -35,13 +35,15 @@ async def async_setup_entry(
class BondCover(BondEntity, CoverEntity):
"""Representation of a Bond cover."""
def __init__(self, hub: BondHub, device: BondDevice, bpup_subs: BPUPSubscriptions):
def __init__(
self, hub: BondHub, device: BondDevice, bpup_subs: BPUPSubscriptions
) -> None:
"""Create HA entity representing Bond cover."""
super().__init__(hub, device, bpup_subs)
self._closed: Optional[bool] = None
def _apply_state(self, state: dict):
def _apply_state(self, state: dict) -> None:
cover_open = state.get("open")
self._closed = True if cover_open == 0 else False if cover_open == 1 else None
@ -51,7 +53,7 @@ class BondCover(BondEntity, CoverEntity):
return DEVICE_CLASS_SHADE
@property
def is_closed(self):
def is_closed(self) -> Optional[bool]:
"""Return if the cover is closed or not."""
return self._closed
@ -63,6 +65,6 @@ class BondCover(BondEntity, CoverEntity):
"""Close cover."""
await self._hub.bond.action(self._device.device_id, Action.close())
async def async_stop_cover(self, **kwargs):
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Hold cover."""
await self._hub.bond.action(self._device.device_id, Action.hold())

View File

@ -38,7 +38,7 @@ class BondEntity(Entity):
self._sub_device = sub_device
self._available = True
self._bpup_subs = bpup_subs
self._update_lock = None
self._update_lock: Optional[Lock] = None
self._initialized = False
@property
@ -58,7 +58,7 @@ class BondEntity(Entity):
return self._device.name
@property
def should_poll(self):
def should_poll(self) -> bool:
"""No polling needed."""
return False
@ -96,15 +96,16 @@ class BondEntity(Entity):
"""Report availability of this entity based on last API call results."""
return self._available
async def async_update(self):
async def async_update(self) -> None:
"""Fetch assumed state of the cover from the hub using API."""
await self._async_update_from_api()
async def _async_update_if_bpup_not_alive(self, *_):
async def _async_update_if_bpup_not_alive(self, *_: Any) -> None:
"""Fetch via the API if BPUP is not alive."""
if self._bpup_subs.alive and self._initialized:
return
assert self._update_lock is not None
if self._update_lock.locked():
_LOGGER.warning(
"Updating %s took longer than the scheduled update interval %s",
@ -117,7 +118,7 @@ class BondEntity(Entity):
await self._async_update_from_api()
self.async_write_ha_state()
async def _async_update_from_api(self):
async def _async_update_from_api(self) -> None:
"""Fetch via the API."""
try:
state: dict = await self._hub.bond.device_state(self._device_id)
@ -131,11 +132,11 @@ class BondEntity(Entity):
self._async_state_callback(state)
@abstractmethod
def _apply_state(self, state: dict):
def _apply_state(self, state: dict) -> None:
raise NotImplementedError
@callback
def _async_state_callback(self, state):
def _async_state_callback(self, state: dict) -> None:
"""Process a state change."""
self._initialized = True
if not self._available:
@ -147,16 +148,17 @@ class BondEntity(Entity):
self._apply_state(state)
@callback
def _async_bpup_callback(self, state):
def _async_bpup_callback(self, state: dict) -> None:
"""Process a state change from BPUP."""
self._async_state_callback(state)
self.async_write_ha_state()
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Subscribe to BPUP and start polling."""
await super().async_added_to_hass()
self._update_lock = Lock()
self._bpup_subs.subscribe(self._device_id, self._async_bpup_callback)
assert self.hass is not None
self.async_on_remove(
async_track_time_interval(
self.hass, self._async_update_if_bpup_not_alive, _FALLBACK_SCAN_INTERVAL

View File

@ -38,7 +38,7 @@ async def async_setup_entry(
hub: BondHub = data[HUB]
bpup_subs: BPUPSubscriptions = data[BPUP_SUBS]
fans = [
fans: List[Entity] = [
BondFan(hub, device, bpup_subs)
for device in hub.devices
if DeviceType.is_fan(device.type)
@ -58,7 +58,7 @@ class BondFan(BondEntity, FanEntity):
self._speed: Optional[int] = None
self._direction: Optional[int] = None
def _apply_state(self, state: dict):
def _apply_state(self, state: dict) -> None:
self._power = state.get("power")
self._speed = state.get("speed")
self._direction = state.get("direction")
@ -80,7 +80,7 @@ class BondFan(BondEntity, FanEntity):
return (1, self._device.props.get("max_speed", 3))
@property
def percentage(self) -> Optional[str]:
def percentage(self) -> int:
"""Return the current speed percentage for the fan."""
if not self._speed or not self._power:
return 0
@ -128,7 +128,7 @@ class BondFan(BondEntity, FanEntity):
speed: Optional[str] = None,
percentage: Optional[int] = None,
preset_mode: Optional[str] = None,
**kwargs,
**kwargs: Any,
) -> None:
"""Turn on the fan."""
_LOGGER.debug("Fan async_turn_on called with percentage %s", percentage)
@ -142,7 +142,7 @@ class BondFan(BondEntity, FanEntity):
"""Turn the fan off."""
await self._hub.bond.action(self._device.device_id, Action.turn_off())
async def async_set_direction(self, direction: str):
async def async_set_direction(self, direction: str) -> None:
"""Set fan rotation direction."""
bond_direction = (
Direction.REVERSE if direction == DIRECTION_REVERSE else Direction.FORWARD

View File

@ -114,7 +114,7 @@ class BondLight(BondBaseLight, BondEntity, LightEntity):
super().__init__(hub, device, bpup_subs, sub_device)
self._brightness: Optional[int] = None
def _apply_state(self, state: dict):
def _apply_state(self, state: dict) -> None:
self._light = state.get("light")
self._brightness = state.get("brightness")
@ -126,7 +126,7 @@ class BondLight(BondBaseLight, BondEntity, LightEntity):
return 0
@property
def brightness(self) -> int:
def brightness(self) -> Optional[int]:
"""Return the brightness of this light between 1..255."""
brightness_value = (
round(self._brightness * 255 / 100) if self._brightness else None
@ -152,7 +152,7 @@ class BondLight(BondBaseLight, BondEntity, LightEntity):
class BondDownLight(BondBaseLight, BondEntity, LightEntity):
"""Representation of a Bond light."""
def _apply_state(self, state: dict):
def _apply_state(self, state: dict) -> None:
self._light = state.get("down_light") and state.get("light")
async def async_turn_on(self, **kwargs: Any) -> None:
@ -171,7 +171,7 @@ class BondDownLight(BondBaseLight, BondEntity, LightEntity):
class BondUpLight(BondBaseLight, BondEntity, LightEntity):
"""Representation of a Bond light."""
def _apply_state(self, state: dict):
def _apply_state(self, state: dict) -> None:
self._light = state.get("up_light") and state.get("light")
async def async_turn_on(self, **kwargs: Any) -> None:
@ -198,7 +198,7 @@ class BondFireplace(BondEntity, LightEntity):
# Bond flame level, 0-100
self._flame: Optional[int] = None
def _apply_state(self, state: dict):
def _apply_state(self, state: dict) -> None:
self._power = state.get("power")
self._flame = state.get("flame")
@ -230,7 +230,7 @@ class BondFireplace(BondEntity, LightEntity):
await self._hub.bond.action(self._device.device_id, Action.turn_off())
@property
def brightness(self):
def brightness(self) -> Optional[int]:
"""Return the flame of this fireplace converted to HA brightness between 0..255."""
return round(self._flame * 255 / 100) if self._flame else None

View File

@ -23,7 +23,7 @@ async def async_setup_entry(
hub: BondHub = data[HUB]
bpup_subs: BPUPSubscriptions = data[BPUP_SUBS]
switches = [
switches: List[Entity] = [
BondSwitch(hub, device, bpup_subs)
for device in hub.devices
if DeviceType.is_generic(device.type)
@ -41,7 +41,7 @@ class BondSwitch(BondEntity, SwitchEntity):
self._power: Optional[bool] = None
def _apply_state(self, state: dict):
def _apply_state(self, state: dict) -> None:
self._power = state.get("power")
@property

View File

@ -1,7 +1,7 @@
"""Reusable utilities for the Bond component."""
import asyncio
import logging
from typing import List, Optional, Set
from typing import Any, Dict, List, Optional, Set, cast
from aiohttp import ClientResponseError
from bond_api import Action, Bond
@ -14,13 +14,15 @@ _LOGGER = logging.getLogger(__name__)
class BondDevice:
"""Helper device class to hold ID and attributes together."""
def __init__(self, device_id: str, attrs: dict, props: dict):
def __init__(
self, device_id: str, attrs: Dict[str, Any], props: Dict[str, Any]
) -> None:
"""Create a helper device from ID and attributes returned by API."""
self.device_id = device_id
self.props = props
self._attrs = attrs
def __repr__(self):
def __repr__(self) -> str:
"""Return readable representation of a bond device."""
return {
"device_id": self.device_id,
@ -31,25 +33,25 @@ class BondDevice:
@property
def name(self) -> str:
"""Get the name of this device."""
return self._attrs["name"]
return cast(str, self._attrs["name"])
@property
def type(self) -> str:
"""Get the type of this device."""
return self._attrs["type"]
return cast(str, self._attrs["type"])
@property
def location(self) -> str:
def location(self) -> Optional[str]:
"""Get the location of this device."""
return self._attrs.get("location")
@property
def template(self) -> str:
def template(self) -> Optional[str]:
"""Return this model template."""
return self._attrs.get("template")
@property
def branding_profile(self) -> str:
def branding_profile(self) -> Optional[str]:
"""Return this branding profile."""
return self.props.get("branding_profile")
@ -58,7 +60,7 @@ class BondDevice:
"""Check if Trust State is turned on."""
return self.props.get("trust_state", False)
def _has_any_action(self, actions: Set[str]):
def _has_any_action(self, actions: Set[str]) -> bool:
"""Check to see if the device supports any of the actions."""
supported_actions: List[str] = self._attrs["actions"]
for action in supported_actions:
@ -99,11 +101,11 @@ class BondHub:
def __init__(self, bond: Bond):
"""Initialize Bond Hub."""
self.bond: Bond = bond
self._bridge: Optional[dict] = None
self._version: Optional[dict] = None
self._devices: Optional[List[BondDevice]] = None
self._bridge: Dict[str, Any] = {}
self._version: Dict[str, Any] = {}
self._devices: List[BondDevice] = []
async def setup(self, max_devices=None):
async def setup(self, max_devices: Optional[int] = None) -> None:
"""Read hub version information."""
self._version = await self.bond.version()
_LOGGER.debug("Bond reported the following version info: %s", self._version)
@ -135,12 +137,12 @@ class BondHub:
return self._version.get("bondid")
@property
def target(self) -> str:
def target(self) -> Optional[str]:
"""Return this hub target."""
return self._version.get("target")
@property
def model(self) -> str:
def model(self) -> Optional[str]:
"""Return this hub model."""
return self._version.get("model")
@ -154,7 +156,7 @@ class BondHub:
"""Get the name of this bridge."""
if not self.is_bridge and self._devices:
return self._devices[0].name
return self._bridge["name"]
return cast(str, self._bridge["name"])
@property
def location(self) -> Optional[str]:
@ -164,7 +166,7 @@ class BondHub:
return self._bridge.get("location")
@property
def fw_ver(self) -> str:
def fw_ver(self) -> Optional[str]:
"""Return this hub firmware version."""
return self._version.get("fw_ver")

View File

@ -42,7 +42,7 @@ warn_redundant_casts = true
warn_unused_configs = true
[mypy-homeassistant.block_async_io,homeassistant.bootstrap,homeassistant.components,homeassistant.config_entries,homeassistant.config,homeassistant.const,homeassistant.core,homeassistant.data_entry_flow,homeassistant.exceptions,homeassistant.__init__,homeassistant.loader,homeassistant.__main__,homeassistant.requirements,homeassistant.runner,homeassistant.setup,homeassistant.util,homeassistant.auth.*,homeassistant.components.automation.*,homeassistant.components.binary_sensor.*,homeassistant.components.calendar.*,homeassistant.components.cover.*,homeassistant.components.device_automation.*,homeassistant.components.frontend.*,homeassistant.components.geo_location.*,homeassistant.components.group.*,homeassistant.components.history.*,homeassistant.components.http.*,homeassistant.components.huawei_lte.*,homeassistant.components.hyperion.*,homeassistant.components.image_processing.*,homeassistant.components.integration.*,homeassistant.components.light.*,homeassistant.components.lock.*,homeassistant.components.mailbox.*,homeassistant.components.media_player.*,homeassistant.components.notify.*,homeassistant.components.number.*,homeassistant.components.persistent_notification.*,homeassistant.components.proximity.*,homeassistant.components.remote.*,homeassistant.components.scene.*,homeassistant.components.sensor.*,homeassistant.components.slack.*,homeassistant.components.sun.*,homeassistant.components.switch.*,homeassistant.components.systemmonitor.*,homeassistant.components.tts.*,homeassistant.components.vacuum.*,homeassistant.components.water_heater.*,homeassistant.components.weather.*,homeassistant.components.websocket_api.*,homeassistant.components.zone.*,homeassistant.components.zwave_js.*,homeassistant.helpers.*,homeassistant.scripts.*,homeassistant.util.*,tests.components.hyperion.*]
[mypy-homeassistant.block_async_io,homeassistant.bootstrap,homeassistant.components,homeassistant.config_entries,homeassistant.config,homeassistant.const,homeassistant.core,homeassistant.data_entry_flow,homeassistant.exceptions,homeassistant.__init__,homeassistant.loader,homeassistant.__main__,homeassistant.requirements,homeassistant.runner,homeassistant.setup,homeassistant.util,homeassistant.auth.*,homeassistant.components.automation.*,homeassistant.components.binary_sensor.*,homeassistant.components.bond.*,homeassistant.components.calendar.*,homeassistant.components.cover.*,homeassistant.components.device_automation.*,homeassistant.components.frontend.*,homeassistant.components.geo_location.*,homeassistant.components.group.*,homeassistant.components.history.*,homeassistant.components.http.*,homeassistant.components.huawei_lte.*,homeassistant.components.hyperion.*,homeassistant.components.image_processing.*,homeassistant.components.integration.*,homeassistant.components.light.*,homeassistant.components.lock.*,homeassistant.components.mailbox.*,homeassistant.components.media_player.*,homeassistant.components.notify.*,homeassistant.components.number.*,homeassistant.components.persistent_notification.*,homeassistant.components.proximity.*,homeassistant.components.remote.*,homeassistant.components.scene.*,homeassistant.components.sensor.*,homeassistant.components.slack.*,homeassistant.components.sun.*,homeassistant.components.switch.*,homeassistant.components.systemmonitor.*,homeassistant.components.tts.*,homeassistant.components.vacuum.*,homeassistant.components.water_heater.*,homeassistant.components.weather.*,homeassistant.components.websocket_api.*,homeassistant.components.zone.*,homeassistant.components.zwave_js.*,homeassistant.helpers.*,homeassistant.scripts.*,homeassistant.util.*,tests.components.hyperion.*]
strict = true
ignore_errors = false
warn_unreachable = true