Add and fix some Huawei LTE type hints (#41196)

This commit is contained in:
Ville Skyttä 2020-10-05 12:21:48 +03:00 committed by GitHub
parent baef8b033a
commit 56f0a68c49
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 45 additions and 41 deletions

View file

@ -146,7 +146,7 @@ class Router:
suspended = attr.ib(init=False, default=False) suspended = attr.ib(init=False, default=False)
notify_last_attempt: float = attr.ib(init=False, default=-1) notify_last_attempt: float = attr.ib(init=False, default=-1)
def __attrs_post_init__(self): def __attrs_post_init__(self) -> None:
"""Set up internal state on init.""" """Set up internal state on init."""
self.client = Client(self.connection) self.client = Client(self.connection)
@ -176,7 +176,7 @@ class Router:
"""Get router connections for device registry.""" """Get router connections for device registry."""
return {(dr.CONNECTION_NETWORK_MAC, self.mac)} if self.mac else set() return {(dr.CONNECTION_NETWORK_MAC, self.mac)} if self.mac else set()
def _get_data(self, key: str, func: Callable[[None], Any]) -> None: def _get_data(self, key: str, func: Callable[[], Any]) -> None:
if not self.subscriptions.get(key): if not self.subscriptions.get(key):
return return
if key in self.inflight_gets: if key in self.inflight_gets:
@ -275,7 +275,7 @@ class Router:
except Exception: # pylint: disable=broad-except except Exception: # pylint: disable=broad-except
_LOGGER.warning("Logout error", exc_info=True) _LOGGER.warning("Logout error", exc_info=True)
def cleanup(self, *_) -> None: def cleanup(self, *_: Any) -> None:
"""Clean up resources.""" """Clean up resources."""
self.subscriptions.clear() self.subscriptions.clear()
@ -359,7 +359,7 @@ async def async_setup_entry(hass: HomeAssistantType, config_entry: ConfigEntry)
username = config_entry.data.get(CONF_USERNAME) username = config_entry.data.get(CONF_USERNAME)
password = config_entry.data.get(CONF_PASSWORD) password = config_entry.data.get(CONF_PASSWORD)
if username or password: if username or password:
connection = AuthorizedConnection( connection: Connection = AuthorizedConnection(
url, username=username, password=password, timeout=CONNECTION_TIMEOUT url, username=username, password=password, timeout=CONNECTION_TIMEOUT
) )
else: else:

View file

@ -1,7 +1,7 @@
"""Support for Huawei LTE binary sensors.""" """Support for Huawei LTE binary sensors."""
import logging import logging
from typing import Optional from typing import List, Optional
import attr import attr
from huawei_lte_api.enums.cradle import ConnectionStatusEnum from huawei_lte_api.enums.cradle import ConnectionStatusEnum
@ -11,6 +11,7 @@ from homeassistant.components.binary_sensor import (
BinarySensorEntity, BinarySensorEntity,
) )
from homeassistant.const import CONF_URL from homeassistant.const import CONF_URL
from homeassistant.helpers.entity import Entity
from . import HuaweiLteBaseEntity from . import HuaweiLteBaseEntity
from .const import ( from .const import (
@ -26,7 +27,7 @@ _LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, config_entry, async_add_entities): async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up from config entry.""" """Set up from config entry."""
router = hass.data[DOMAIN].routers[config_entry.data[CONF_URL]] router = hass.data[DOMAIN].routers[config_entry.data[CONF_URL]]
entities = [] entities: List[Entity] = []
if router.data.get(KEY_MONITORING_STATUS): if router.data.get(KEY_MONITORING_STATUS):
entities.append(HuaweiLteMobileConnectionBinarySensor(router)) entities.append(HuaweiLteMobileConnectionBinarySensor(router))
@ -57,19 +58,19 @@ class HuaweiLteBaseBinarySensor(HuaweiLteBaseEntity, BinarySensorEntity):
def _device_unique_id(self) -> str: def _device_unique_id(self) -> str:
return f"{self.key}.{self.item}" return f"{self.key}.{self.item}"
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Subscribe to needed data on add.""" """Subscribe to needed data on add."""
await super().async_added_to_hass() await super().async_added_to_hass()
self.router.subscriptions[self.key].add(f"{BINARY_SENSOR_DOMAIN}/{self.item}") self.router.subscriptions[self.key].add(f"{BINARY_SENSOR_DOMAIN}/{self.item}")
async def async_will_remove_from_hass(self): async def async_will_remove_from_hass(self) -> None:
"""Unsubscribe from needed data on remove.""" """Unsubscribe from needed data on remove."""
await super().async_will_remove_from_hass() await super().async_will_remove_from_hass()
self.router.subscriptions[self.key].remove( self.router.subscriptions[self.key].remove(
f"{BINARY_SENSOR_DOMAIN}/{self.item}" f"{BINARY_SENSOR_DOMAIN}/{self.item}"
) )
async def async_update(self): async def async_update(self) -> None:
"""Update state.""" """Update state."""
try: try:
value = self.router.data[self.key][self.item] value = self.router.data[self.key][self.item]
@ -94,7 +95,7 @@ CONNECTION_STATE_ATTRIBUTES = {
class HuaweiLteMobileConnectionBinarySensor(HuaweiLteBaseBinarySensor): class HuaweiLteMobileConnectionBinarySensor(HuaweiLteBaseBinarySensor):
"""Huawei LTE mobile connection binary sensor.""" """Huawei LTE mobile connection binary sensor."""
def __attrs_post_init__(self): def __attrs_post_init__(self) -> None:
"""Initialize identifiers.""" """Initialize identifiers."""
self.key = KEY_MONITORING_STATUS self.key = KEY_MONITORING_STATUS
self.item = "ConnectionStatus" self.item = "ConnectionStatus"
@ -121,7 +122,7 @@ class HuaweiLteMobileConnectionBinarySensor(HuaweiLteBaseBinarySensor):
) )
@property @property
def icon(self): def icon(self) -> str:
"""Return mobile connectivity sensor icon.""" """Return mobile connectivity sensor icon."""
return "mdi:signal" if self.is_on else "mdi:signal-off" return "mdi:signal" if self.is_on else "mdi:signal-off"
@ -157,7 +158,7 @@ class HuaweiLteBaseWifiStatusBinarySensor(HuaweiLteBaseBinarySensor):
return self._raw_state is None return self._raw_state is None
@property @property
def icon(self): def icon(self) -> str:
"""Return WiFi status sensor icon.""" """Return WiFi status sensor icon."""
return "mdi:wifi" if self.is_on else "mdi:wifi-off" return "mdi:wifi" if self.is_on else "mdi:wifi-off"
@ -166,7 +167,7 @@ class HuaweiLteBaseWifiStatusBinarySensor(HuaweiLteBaseBinarySensor):
class HuaweiLteWifiStatusBinarySensor(HuaweiLteBaseWifiStatusBinarySensor): class HuaweiLteWifiStatusBinarySensor(HuaweiLteBaseWifiStatusBinarySensor):
"""Huawei LTE WiFi status binary sensor.""" """Huawei LTE WiFi status binary sensor."""
def __attrs_post_init__(self): def __attrs_post_init__(self) -> None:
"""Initialize identifiers.""" """Initialize identifiers."""
self.key = KEY_MONITORING_STATUS self.key = KEY_MONITORING_STATUS
self.item = "WifiStatus" self.item = "WifiStatus"
@ -180,7 +181,7 @@ class HuaweiLteWifiStatusBinarySensor(HuaweiLteBaseWifiStatusBinarySensor):
class HuaweiLteWifi24ghzStatusBinarySensor(HuaweiLteBaseWifiStatusBinarySensor): class HuaweiLteWifi24ghzStatusBinarySensor(HuaweiLteBaseWifiStatusBinarySensor):
"""Huawei LTE 2.4GHz WiFi status binary sensor.""" """Huawei LTE 2.4GHz WiFi status binary sensor."""
def __attrs_post_init__(self): def __attrs_post_init__(self) -> None:
"""Initialize identifiers.""" """Initialize identifiers."""
self.key = KEY_WLAN_WIFI_FEATURE_SWITCH self.key = KEY_WLAN_WIFI_FEATURE_SWITCH
self.item = "wifi24g_switch_enable" self.item = "wifi24g_switch_enable"
@ -194,7 +195,7 @@ class HuaweiLteWifi24ghzStatusBinarySensor(HuaweiLteBaseWifiStatusBinarySensor):
class HuaweiLteWifi5ghzStatusBinarySensor(HuaweiLteBaseWifiStatusBinarySensor): class HuaweiLteWifi5ghzStatusBinarySensor(HuaweiLteBaseWifiStatusBinarySensor):
"""Huawei LTE 5GHz WiFi status binary sensor.""" """Huawei LTE 5GHz WiFi status binary sensor."""
def __attrs_post_init__(self): def __attrs_post_init__(self) -> None:
"""Initialize identifiers.""" """Initialize identifiers."""
self.key = KEY_WLAN_WIFI_FEATURE_SWITCH self.key = KEY_WLAN_WIFI_FEATURE_SWITCH
self.item = "wifi5g_enabled" self.item = "wifi5g_enabled"
@ -208,7 +209,7 @@ class HuaweiLteWifi5ghzStatusBinarySensor(HuaweiLteBaseWifiStatusBinarySensor):
class HuaweiLteSmsStorageFullBinarySensor(HuaweiLteBaseBinarySensor): class HuaweiLteSmsStorageFullBinarySensor(HuaweiLteBaseBinarySensor):
"""Huawei LTE SMS storage full binary sensor.""" """Huawei LTE SMS storage full binary sensor."""
def __attrs_post_init__(self): def __attrs_post_init__(self) -> None:
"""Initialize identifiers.""" """Initialize identifiers."""
self.key = KEY_MONITORING_CHECK_NOTIFICATIONS self.key = KEY_MONITORING_CHECK_NOTIFICATIONS
self.item = "SmsStorageFull" self.item = "SmsStorageFull"
@ -228,6 +229,6 @@ class HuaweiLteSmsStorageFullBinarySensor(HuaweiLteBaseBinarySensor):
return self._raw_state is None return self._raw_state is None
@property @property
def icon(self): def icon(self) -> str:
"""Return WiFi status sensor icon.""" """Return WiFi status sensor icon."""
return "mdi:email-alert" if self.is_on else "mdi:email-off" return "mdi:email-alert" if self.is_on else "mdi:email-off"

View file

@ -121,7 +121,7 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
if self._already_configured(user_input): if self._already_configured(user_input):
return self.async_abort(reason="already_configured") return self.async_abort(reason="already_configured")
conn = None conn: Optional[Connection] = None
def logout(): def logout():
if hasattr(conn, "user"): if hasattr(conn, "user"):

View file

@ -16,6 +16,7 @@ from homeassistant.const import CONF_URL
from homeassistant.core import callback from homeassistant.core import callback
from homeassistant.helpers import entity_registry from homeassistant.helpers import entity_registry
from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import Entity
from . import HuaweiLteBaseEntity from . import HuaweiLteBaseEntity
from .const import DOMAIN, KEY_WLAN_HOST_LIST, UPDATE_SIGNAL from .const import DOMAIN, KEY_WLAN_HOST_LIST, UPDATE_SIGNAL
@ -81,7 +82,7 @@ def async_add_new_entities(hass, router_url, async_add_entities, tracked):
_LOGGER.debug("%s[%s][%s] not in data", KEY_WLAN_HOST_LIST, "Hosts", "Host") _LOGGER.debug("%s[%s][%s] not in data", KEY_WLAN_HOST_LIST, "Hosts", "Host")
return return
new_entities = [] new_entities: List[Entity] = []
for host in (x for x in hosts if x.get("MacAddress")): for host in (x for x in hosts if x.get("MacAddress")):
entity = HuaweiLteScannerEntity(router, host["MacAddress"]) entity = HuaweiLteScannerEntity(router, host["MacAddress"])
if entity.unique_id in tracked: if entity.unique_id in tracked:
@ -116,7 +117,7 @@ class HuaweiLteScannerEntity(HuaweiLteBaseEntity, ScannerEntity):
_hostname: Optional[str] = attr.ib(init=False, default=None) _hostname: Optional[str] = attr.ib(init=False, default=None)
_device_state_attributes: Dict[str, Any] = attr.ib(init=False, factory=dict) _device_state_attributes: Dict[str, Any] = attr.ib(init=False, factory=dict)
def __attrs_post_init__(self): def __attrs_post_init__(self) -> None:
"""Initialize internal state.""" """Initialize internal state."""
self._device_state_attributes["mac_address"] = self.mac self._device_state_attributes["mac_address"] = self.mac
@ -148,7 +149,7 @@ class HuaweiLteScannerEntity(HuaweiLteBaseEntity, ScannerEntity):
hosts = self.router.data[KEY_WLAN_HOST_LIST]["Hosts"]["Host"] hosts = self.router.data[KEY_WLAN_HOST_LIST]["Hosts"]["Host"]
host = next((x for x in hosts if x.get("MacAddress") == self.mac), None) host = next((x for x in hosts if x.get("MacAddress") == self.mac), None)
self._is_connected = host is not None self._is_connected = host is not None
if self._is_connected: if host is not None:
self._hostname = host.get("HostName") self._hostname = host.get("HostName")
self._device_state_attributes = { self._device_state_attributes = {
_better_snakecase(k): v for k, v in host.items() if k != "HostName" _better_snakecase(k): v for k, v in host.items() if k != "HostName"

View file

@ -2,7 +2,7 @@
import logging import logging
import re import re
from typing import Callable, Dict, NamedTuple, Optional, Pattern, Tuple, Union from typing import Callable, Dict, List, NamedTuple, Optional, Pattern, Tuple, Union
import attr import attr
@ -17,6 +17,7 @@ from homeassistant.const import (
STATE_UNKNOWN, STATE_UNKNOWN,
TIME_SECONDS, TIME_SECONDS,
) )
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.typing import StateType from homeassistant.helpers.typing import StateType
from . import HuaweiLteBaseEntity from . import HuaweiLteBaseEntity
@ -324,7 +325,7 @@ SENSOR_META: Dict[Union[str, Tuple[str, str]], SensorMeta] = {
async def async_setup_entry(hass, config_entry, async_add_entities): async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up from config entry.""" """Set up from config entry."""
router = hass.data[DOMAIN].routers[config_entry.data[CONF_URL]] router = hass.data[DOMAIN].routers[config_entry.data[CONF_URL]]
sensors = [] sensors: List[Entity] = []
for key in SENSOR_KEYS: for key in SENSOR_KEYS:
items = router.data.get(key) items = router.data.get(key)
if not items: if not items:
@ -370,15 +371,15 @@ class HuaweiLteSensor(HuaweiLteBaseEntity):
item: str = attr.ib() item: str = attr.ib()
meta: SensorMeta = attr.ib() meta: SensorMeta = attr.ib()
_state = attr.ib(init=False, default=STATE_UNKNOWN) _state: StateType = attr.ib(init=False, default=STATE_UNKNOWN)
_unit: str = attr.ib(init=False) _unit: Optional[str] = attr.ib(init=False)
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Subscribe to needed data on add.""" """Subscribe to needed data on add."""
await super().async_added_to_hass() await super().async_added_to_hass()
self.router.subscriptions[self.key].add(f"{SENSOR_DOMAIN}/{self.item}") self.router.subscriptions[self.key].add(f"{SENSOR_DOMAIN}/{self.item}")
async def async_will_remove_from_hass(self): async def async_will_remove_from_hass(self) -> None:
"""Unsubscribe from needed data on remove.""" """Unsubscribe from needed data on remove."""
await super().async_will_remove_from_hass() await super().async_will_remove_from_hass()
self.router.subscriptions[self.key].remove(f"{SENSOR_DOMAIN}/{self.item}") self.router.subscriptions[self.key].remove(f"{SENSOR_DOMAIN}/{self.item}")
@ -392,7 +393,7 @@ class HuaweiLteSensor(HuaweiLteBaseEntity):
return f"{self.key}.{self.item}" return f"{self.key}.{self.item}"
@property @property
def state(self): def state(self) -> StateType:
"""Return sensor state.""" """Return sensor state."""
return self._state return self._state
@ -402,12 +403,12 @@ class HuaweiLteSensor(HuaweiLteBaseEntity):
return self.meta.device_class return self.meta.device_class
@property @property
def unit_of_measurement(self): def unit_of_measurement(self) -> Optional[str]:
"""Return sensor's unit of measurement.""" """Return sensor's unit of measurement."""
return self.meta.unit or self._unit return self.meta.unit or self._unit
@property @property
def icon(self): def icon(self) -> Optional[str]:
"""Return icon for sensor.""" """Return icon for sensor."""
icon = self.meta.icon icon = self.meta.icon
if callable(icon): if callable(icon):
@ -419,7 +420,7 @@ class HuaweiLteSensor(HuaweiLteBaseEntity):
"""Return if the entity should be enabled when first added to the entity registry.""" """Return if the entity should be enabled when first added to the entity registry."""
return self.meta.enabled_default return self.meta.enabled_default
async def async_update(self): async def async_update(self) -> None:
"""Update state.""" """Update state."""
try: try:
value = self.router.data[self.key][self.item] value = self.router.data[self.key][self.item]

View file

@ -1,7 +1,7 @@
"""Support for Huawei LTE switches.""" """Support for Huawei LTE switches."""
import logging import logging
from typing import Optional from typing import Any, List, Optional
import attr import attr
@ -11,6 +11,7 @@ from homeassistant.components.switch import (
SwitchEntity, SwitchEntity,
) )
from homeassistant.const import CONF_URL from homeassistant.const import CONF_URL
from homeassistant.helpers.entity import Entity
from . import HuaweiLteBaseEntity from . import HuaweiLteBaseEntity
from .const import DOMAIN, KEY_DIALUP_MOBILE_DATASWITCH from .const import DOMAIN, KEY_DIALUP_MOBILE_DATASWITCH
@ -21,7 +22,7 @@ _LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, config_entry, async_add_entities): async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up from config entry.""" """Set up from config entry."""
router = hass.data[DOMAIN].routers[config_entry.data[CONF_URL]] router = hass.data[DOMAIN].routers[config_entry.data[CONF_URL]]
switches = [] switches: List[Entity] = []
if router.data.get(KEY_DIALUP_MOBILE_DATASWITCH): if router.data.get(KEY_DIALUP_MOBILE_DATASWITCH):
switches.append(HuaweiLteMobileDataSwitch(router)) switches.append(HuaweiLteMobileDataSwitch(router))
@ -40,30 +41,30 @@ class HuaweiLteBaseSwitch(HuaweiLteBaseEntity, SwitchEntity):
def _turn(self, state: bool) -> None: def _turn(self, state: bool) -> None:
raise NotImplementedError raise NotImplementedError
def turn_on(self, **kwargs): def turn_on(self, **kwargs: Any) -> None:
"""Turn switch on.""" """Turn switch on."""
self._turn(state=True) self._turn(state=True)
def turn_off(self, **kwargs): def turn_off(self, **kwargs: Any) -> None:
"""Turn switch off.""" """Turn switch off."""
self._turn(state=False) self._turn(state=False)
@property @property
def device_class(self): def device_class(self) -> str:
"""Return device class.""" """Return device class."""
return DEVICE_CLASS_SWITCH return DEVICE_CLASS_SWITCH
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Subscribe to needed data on add.""" """Subscribe to needed data on add."""
await super().async_added_to_hass() await super().async_added_to_hass()
self.router.subscriptions[self.key].add(f"{SWITCH_DOMAIN}/{self.item}") self.router.subscriptions[self.key].add(f"{SWITCH_DOMAIN}/{self.item}")
async def async_will_remove_from_hass(self): async def async_will_remove_from_hass(self) -> None:
"""Unsubscribe from needed data on remove.""" """Unsubscribe from needed data on remove."""
await super().async_will_remove_from_hass() await super().async_will_remove_from_hass()
self.router.subscriptions[self.key].remove(f"{SWITCH_DOMAIN}/{self.item}") self.router.subscriptions[self.key].remove(f"{SWITCH_DOMAIN}/{self.item}")
async def async_update(self): async def async_update(self) -> None:
"""Update state.""" """Update state."""
try: try:
value = self.router.data[self.key][self.item] value = self.router.data[self.key][self.item]
@ -79,7 +80,7 @@ class HuaweiLteBaseSwitch(HuaweiLteBaseEntity, SwitchEntity):
class HuaweiLteMobileDataSwitch(HuaweiLteBaseSwitch): class HuaweiLteMobileDataSwitch(HuaweiLteBaseSwitch):
"""Huawei LTE mobile data switch device.""" """Huawei LTE mobile data switch device."""
def __attrs_post_init__(self): def __attrs_post_init__(self) -> None:
"""Initialize identifiers.""" """Initialize identifiers."""
self.key = KEY_DIALUP_MOBILE_DATASWITCH self.key = KEY_DIALUP_MOBILE_DATASWITCH
self.item = "dataswitch" self.item = "dataswitch"
@ -104,6 +105,6 @@ class HuaweiLteMobileDataSwitch(HuaweiLteBaseSwitch):
self.schedule_update_ha_state() self.schedule_update_ha_state()
@property @property
def icon(self): def icon(self) -> str:
"""Return switch icon.""" """Return switch icon."""
return "mdi:signal" if self.is_on else "mdi:signal-off" return "mdi:signal" if self.is_on else "mdi:signal-off"