mirror of
https://github.com/home-assistant/core.git
synced 2025-07-14 08:47:10 +00:00
Complete Huawei LTE type hints, make mypy check them (#41503)
This commit is contained in:
parent
61a5bf5645
commit
cf83c6bf00
@ -6,7 +6,7 @@ from functools import partial
|
|||||||
import ipaddress
|
import ipaddress
|
||||||
import logging
|
import logging
|
||||||
import time
|
import time
|
||||||
from typing import Any, Callable, Dict, List, Set, Tuple
|
from typing import Any, Callable, Dict, List, Set, Tuple, cast
|
||||||
from urllib.parse import urlparse
|
from urllib.parse import urlparse
|
||||||
|
|
||||||
import attr
|
import attr
|
||||||
@ -23,7 +23,9 @@ from url_normalize import url_normalize
|
|||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
|
|
||||||
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
|
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
|
||||||
from homeassistant.components.device_tracker import DOMAIN as DEVICE_TRACKER_DOMAIN
|
from homeassistant.components.device_tracker.const import (
|
||||||
|
DOMAIN as DEVICE_TRACKER_DOMAIN,
|
||||||
|
)
|
||||||
from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
|
from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
|
||||||
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
|
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
|
||||||
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
|
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
|
||||||
@ -36,7 +38,7 @@ from homeassistant.const import (
|
|||||||
CONF_USERNAME,
|
CONF_USERNAME,
|
||||||
EVENT_HOMEASSISTANT_STOP,
|
EVENT_HOMEASSISTANT_STOP,
|
||||||
)
|
)
|
||||||
from homeassistant.core import CALLBACK_TYPE
|
from homeassistant.core import CALLBACK_TYPE, ServiceCall
|
||||||
from homeassistant.exceptions import ConfigEntryNotReady
|
from homeassistant.exceptions import ConfigEntryNotReady
|
||||||
from homeassistant.helpers import (
|
from homeassistant.helpers import (
|
||||||
config_validation as cv,
|
config_validation as cv,
|
||||||
@ -50,7 +52,7 @@ from homeassistant.helpers.dispatcher import (
|
|||||||
)
|
)
|
||||||
from homeassistant.helpers.entity import Entity
|
from homeassistant.helpers.entity import Entity
|
||||||
from homeassistant.helpers.event import async_track_time_interval
|
from homeassistant.helpers.event import async_track_time_interval
|
||||||
from homeassistant.helpers.typing import HomeAssistantType
|
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
|
||||||
|
|
||||||
from .const import (
|
from .const import (
|
||||||
ADMIN_SERVICES,
|
ADMIN_SERVICES,
|
||||||
@ -158,7 +160,7 @@ class Router:
|
|||||||
(KEY_DEVICE_INFORMATION, "DeviceName"),
|
(KEY_DEVICE_INFORMATION, "DeviceName"),
|
||||||
):
|
):
|
||||||
try:
|
try:
|
||||||
return self.data[key][item]
|
return cast(str, self.data[key][item])
|
||||||
except (KeyError, TypeError):
|
except (KeyError, TypeError):
|
||||||
pass
|
pass
|
||||||
return DEFAULT_DEVICE_NAME
|
return DEFAULT_DEVICE_NAME
|
||||||
@ -465,7 +467,7 @@ async def async_unload_entry(
|
|||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
async def async_setup(hass: HomeAssistantType, config) -> bool:
|
async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
|
||||||
"""Set up Huawei LTE component."""
|
"""Set up Huawei LTE component."""
|
||||||
|
|
||||||
# dicttoxml (used by huawei-lte-api) has uselessly verbose INFO level.
|
# dicttoxml (used by huawei-lte-api) has uselessly verbose INFO level.
|
||||||
@ -473,13 +475,13 @@ async def async_setup(hass: HomeAssistantType, config) -> bool:
|
|||||||
logging.getLogger("dicttoxml").setLevel(logging.WARNING)
|
logging.getLogger("dicttoxml").setLevel(logging.WARNING)
|
||||||
|
|
||||||
# Arrange our YAML config to dict with normalized URLs as keys
|
# Arrange our YAML config to dict with normalized URLs as keys
|
||||||
domain_config = {}
|
domain_config: Dict[str, Dict[str, Any]] = {}
|
||||||
if DOMAIN not in hass.data:
|
if DOMAIN not in hass.data:
|
||||||
hass.data[DOMAIN] = HuaweiLteData(hass_config=config, config=domain_config)
|
hass.data[DOMAIN] = HuaweiLteData(hass_config=config, config=domain_config)
|
||||||
for router_config in config.get(DOMAIN, []):
|
for router_config in config.get(DOMAIN, []):
|
||||||
domain_config[url_normalize(router_config.pop(CONF_URL))] = router_config
|
domain_config[url_normalize(router_config.pop(CONF_URL))] = router_config
|
||||||
|
|
||||||
def service_handler(service) -> None:
|
def service_handler(service: ServiceCall) -> None:
|
||||||
"""Apply a service."""
|
"""Apply a service."""
|
||||||
url = service.data.get(CONF_URL)
|
url = service.data.get(CONF_URL)
|
||||||
routers = hass.data[DOMAIN].routers
|
routers = hass.data[DOMAIN].routers
|
||||||
@ -555,10 +557,12 @@ async def async_signal_options_update(
|
|||||||
async_dispatcher_send(hass, UPDATE_OPTIONS_SIGNAL, config_entry)
|
async_dispatcher_send(hass, UPDATE_OPTIONS_SIGNAL, config_entry)
|
||||||
|
|
||||||
|
|
||||||
async def async_migrate_entry(hass: HomeAssistantType, config_entry: ConfigEntry):
|
async def async_migrate_entry(
|
||||||
|
hass: HomeAssistantType, config_entry: ConfigEntry
|
||||||
|
) -> bool:
|
||||||
"""Migrate config entry to new version."""
|
"""Migrate config entry to new version."""
|
||||||
if config_entry.version == 1:
|
if config_entry.version == 1:
|
||||||
options = config_entry.options
|
options = dict(config_entry.options)
|
||||||
recipient = options.get(CONF_RECIPIENT)
|
recipient = options.get(CONF_RECIPIENT)
|
||||||
if isinstance(recipient, str):
|
if isinstance(recipient, str):
|
||||||
options[CONF_RECIPIENT] = [x.strip() for x in recipient.split(",")]
|
options[CONF_RECIPIENT] = [x.strip() for x in recipient.split(",")]
|
||||||
@ -623,6 +627,7 @@ class HuaweiLteBaseEntity(Entity):
|
|||||||
|
|
||||||
async def async_added_to_hass(self) -> None:
|
async def async_added_to_hass(self) -> None:
|
||||||
"""Connect to update signals."""
|
"""Connect to update signals."""
|
||||||
|
assert self.hass is not None
|
||||||
self._unsub_handlers.append(
|
self._unsub_handlers.append(
|
||||||
async_dispatcher_connect(self.hass, UPDATE_SIGNAL, self._async_maybe_update)
|
async_dispatcher_connect(self.hass, UPDATE_SIGNAL, self._async_maybe_update)
|
||||||
)
|
)
|
||||||
|
@ -1,7 +1,7 @@
|
|||||||
"""Support for Huawei LTE binary sensors."""
|
"""Support for Huawei LTE binary sensors."""
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
from typing import List, Optional
|
from typing import Any, Callable, Dict, List, Optional
|
||||||
|
|
||||||
import attr
|
import attr
|
||||||
from huawei_lte_api.enums.cradle import ConnectionStatusEnum
|
from huawei_lte_api.enums.cradle import ConnectionStatusEnum
|
||||||
@ -10,8 +10,10 @@ from homeassistant.components.binary_sensor import (
|
|||||||
DOMAIN as BINARY_SENSOR_DOMAIN,
|
DOMAIN as BINARY_SENSOR_DOMAIN,
|
||||||
BinarySensorEntity,
|
BinarySensorEntity,
|
||||||
)
|
)
|
||||||
|
from homeassistant.config_entries import ConfigEntry
|
||||||
from homeassistant.const import CONF_URL
|
from homeassistant.const import CONF_URL
|
||||||
from homeassistant.helpers.entity import Entity
|
from homeassistant.helpers.entity import Entity
|
||||||
|
from homeassistant.helpers.typing import HomeAssistantType
|
||||||
|
|
||||||
from . import HuaweiLteBaseEntity
|
from . import HuaweiLteBaseEntity
|
||||||
from .const import (
|
from .const import (
|
||||||
@ -24,7 +26,11 @@ from .const import (
|
|||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
async def async_setup_entry(
|
||||||
|
hass: HomeAssistantType,
|
||||||
|
config_entry: ConfigEntry,
|
||||||
|
async_add_entities: Callable[[List[Entity], bool], None],
|
||||||
|
) -> None:
|
||||||
"""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: List[Entity] = []
|
entities: List[Entity] = []
|
||||||
@ -107,9 +113,10 @@ class HuaweiLteMobileConnectionBinarySensor(HuaweiLteBaseBinarySensor):
|
|||||||
@property
|
@property
|
||||||
def is_on(self) -> bool:
|
def is_on(self) -> bool:
|
||||||
"""Return whether the binary sensor is on."""
|
"""Return whether the binary sensor is on."""
|
||||||
return self._raw_state and int(self._raw_state) in (
|
return bool(
|
||||||
ConnectionStatusEnum.CONNECTED,
|
self._raw_state
|
||||||
ConnectionStatusEnum.DISCONNECTING,
|
and int(self._raw_state)
|
||||||
|
in (ConnectionStatusEnum.CONNECTED, ConnectionStatusEnum.DISCONNECTING)
|
||||||
)
|
)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
@ -132,7 +139,7 @@ class HuaweiLteMobileConnectionBinarySensor(HuaweiLteBaseBinarySensor):
|
|||||||
return True
|
return True
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def device_state_attributes(self):
|
def device_state_attributes(self) -> Optional[Dict[str, Any]]:
|
||||||
"""Get additional attributes related to connection status."""
|
"""Get additional attributes related to connection status."""
|
||||||
attributes = super().device_state_attributes
|
attributes = super().device_state_attributes
|
||||||
if self._raw_state in CONNECTION_STATE_ATTRIBUTES:
|
if self._raw_state in CONNECTION_STATE_ATTRIBUTES:
|
||||||
|
@ -2,7 +2,7 @@
|
|||||||
|
|
||||||
from collections import OrderedDict
|
from collections import OrderedDict
|
||||||
import logging
|
import logging
|
||||||
from typing import Optional
|
from typing import Any, Dict, Optional
|
||||||
from urllib.parse import urlparse
|
from urllib.parse import urlparse
|
||||||
|
|
||||||
from huawei_lte_api.AuthorizedConnection import AuthorizedConnection
|
from huawei_lte_api.AuthorizedConnection import AuthorizedConnection
|
||||||
@ -46,11 +46,17 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
|||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
@callback
|
@callback
|
||||||
def async_get_options_flow(config_entry):
|
def async_get_options_flow(
|
||||||
|
config_entry: config_entries.ConfigEntry,
|
||||||
|
) -> "OptionsFlowHandler":
|
||||||
"""Get options flow."""
|
"""Get options flow."""
|
||||||
return OptionsFlowHandler(config_entry)
|
return OptionsFlowHandler(config_entry)
|
||||||
|
|
||||||
async def _async_show_user_form(self, user_input=None, errors=None):
|
async def _async_show_user_form(
|
||||||
|
self,
|
||||||
|
user_input: Optional[Dict[str, Any]] = None,
|
||||||
|
errors: Optional[Dict[str, str]] = None,
|
||||||
|
) -> Dict[str, Any]:
|
||||||
if user_input is None:
|
if user_input is None:
|
||||||
user_input = {}
|
user_input = {}
|
||||||
return self.async_show_form(
|
return self.async_show_form(
|
||||||
@ -89,11 +95,13 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
|||||||
errors=errors or {},
|
errors=errors or {},
|
||||||
)
|
)
|
||||||
|
|
||||||
async def async_step_import(self, user_input=None):
|
async def async_step_import(
|
||||||
|
self, user_input: Optional[Dict[str, Any]] = None
|
||||||
|
) -> Dict[str, Any]:
|
||||||
"""Handle import initiated config flow."""
|
"""Handle import initiated config flow."""
|
||||||
return await self.async_step_user(user_input)
|
return await self.async_step_user(user_input)
|
||||||
|
|
||||||
def _already_configured(self, user_input):
|
def _already_configured(self, user_input: Dict[str, Any]) -> bool:
|
||||||
"""See if we already have a router matching user input configured."""
|
"""See if we already have a router matching user input configured."""
|
||||||
existing_urls = {
|
existing_urls = {
|
||||||
url_normalize(entry.data[CONF_URL], default_scheme="http")
|
url_normalize(entry.data[CONF_URL], default_scheme="http")
|
||||||
@ -101,7 +109,9 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
|||||||
}
|
}
|
||||||
return user_input[CONF_URL] in existing_urls
|
return user_input[CONF_URL] in existing_urls
|
||||||
|
|
||||||
async def async_step_user(self, user_input=None):
|
async def async_step_user(
|
||||||
|
self, user_input: Optional[Dict[str, Any]] = None
|
||||||
|
) -> Dict[str, Any]:
|
||||||
"""Handle user initiated config flow."""
|
"""Handle user initiated config flow."""
|
||||||
if user_input is None:
|
if user_input is None:
|
||||||
return await self._async_show_user_form()
|
return await self._async_show_user_form()
|
||||||
@ -123,15 +133,18 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
|||||||
|
|
||||||
conn: Optional[Connection] = None
|
conn: Optional[Connection] = None
|
||||||
|
|
||||||
def logout():
|
def logout() -> None:
|
||||||
if hasattr(conn, "user"):
|
if isinstance(conn, AuthorizedConnection):
|
||||||
try:
|
try:
|
||||||
conn.user.logout()
|
conn.user.logout()
|
||||||
except Exception: # pylint: disable=broad-except
|
except Exception: # pylint: disable=broad-except
|
||||||
_LOGGER.debug("Could not logout", exc_info=True)
|
_LOGGER.debug("Could not logout", exc_info=True)
|
||||||
|
|
||||||
def try_connect(username: Optional[str], password: Optional[str]) -> Connection:
|
def try_connect(user_input: Dict[str, Any]) -> Connection:
|
||||||
"""Try connecting with given credentials."""
|
"""Try connecting with given credentials."""
|
||||||
|
username = user_input.get(CONF_USERNAME)
|
||||||
|
password = user_input.get(CONF_PASSWORD)
|
||||||
|
conn: Connection
|
||||||
if username or password:
|
if username or password:
|
||||||
conn = AuthorizedConnection(
|
conn = AuthorizedConnection(
|
||||||
user_input[CONF_URL],
|
user_input[CONF_URL],
|
||||||
@ -178,12 +191,9 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
|||||||
title = info.get("DeviceName")
|
title = info.get("DeviceName")
|
||||||
return title or DEFAULT_DEVICE_NAME
|
return title or DEFAULT_DEVICE_NAME
|
||||||
|
|
||||||
username = user_input.get(CONF_USERNAME)
|
assert self.hass is not None
|
||||||
password = user_input.get(CONF_PASSWORD)
|
|
||||||
try:
|
try:
|
||||||
conn = await self.hass.async_add_executor_job(
|
conn = await self.hass.async_add_executor_job(try_connect, user_input)
|
||||||
try_connect, username, password
|
|
||||||
)
|
|
||||||
except LoginErrorUsernameWrongException:
|
except LoginErrorUsernameWrongException:
|
||||||
errors[CONF_USERNAME] = "incorrect_username"
|
errors[CONF_USERNAME] = "incorrect_username"
|
||||||
except LoginErrorPasswordWrongException:
|
except LoginErrorPasswordWrongException:
|
||||||
@ -215,7 +225,9 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
|||||||
|
|
||||||
return self.async_create_entry(title=title, data=user_input)
|
return self.async_create_entry(title=title, data=user_input)
|
||||||
|
|
||||||
async def async_step_ssdp(self, discovery_info):
|
async def async_step_ssdp( # type: ignore # mypy says signature incompatible with supertype, but it's the same?
|
||||||
|
self, discovery_info: Dict[str, Any]
|
||||||
|
) -> Dict[str, Any]:
|
||||||
"""Handle SSDP initiated config flow."""
|
"""Handle SSDP initiated config flow."""
|
||||||
await self.async_set_unique_id(discovery_info[ssdp.ATTR_UPNP_UDN])
|
await self.async_set_unique_id(discovery_info[ssdp.ATTR_UPNP_UDN])
|
||||||
self._abort_if_unique_id_configured()
|
self._abort_if_unique_id_configured()
|
||||||
@ -256,7 +268,9 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
|
|||||||
"""Initialize options flow."""
|
"""Initialize options flow."""
|
||||||
self.config_entry = config_entry
|
self.config_entry = config_entry
|
||||||
|
|
||||||
async def async_step_init(self, user_input=None):
|
async def async_step_init(
|
||||||
|
self, user_input: Optional[Dict[str, Any]] = None
|
||||||
|
) -> Dict[str, Any]:
|
||||||
"""Handle options flow."""
|
"""Handle options flow."""
|
||||||
|
|
||||||
# Recipients are persisted as a list, but handled as comma separated string in UI
|
# Recipients are persisted as a list, but handled as comma separated string in UI
|
||||||
|
@ -2,21 +2,23 @@
|
|||||||
|
|
||||||
import logging
|
import logging
|
||||||
import re
|
import re
|
||||||
from typing import Any, Dict, List, Optional, Set
|
from typing import Any, Callable, Dict, List, Optional, Set, cast
|
||||||
|
|
||||||
import attr
|
import attr
|
||||||
from stringcase import snakecase
|
from stringcase import snakecase
|
||||||
|
|
||||||
from homeassistant.components.device_tracker import (
|
from homeassistant.components.device_tracker.config_entry import ScannerEntity
|
||||||
|
from homeassistant.components.device_tracker.const import (
|
||||||
DOMAIN as DEVICE_TRACKER_DOMAIN,
|
DOMAIN as DEVICE_TRACKER_DOMAIN,
|
||||||
SOURCE_TYPE_ROUTER,
|
SOURCE_TYPE_ROUTER,
|
||||||
)
|
)
|
||||||
from homeassistant.components.device_tracker.config_entry import ScannerEntity
|
from homeassistant.config_entries import ConfigEntry
|
||||||
from homeassistant.const import CONF_URL
|
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 homeassistant.helpers.entity import Entity
|
||||||
|
from homeassistant.helpers.typing import HomeAssistantType
|
||||||
|
|
||||||
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
|
||||||
@ -26,7 +28,11 @@ _LOGGER = logging.getLogger(__name__)
|
|||||||
_DEVICE_SCAN = f"{DEVICE_TRACKER_DOMAIN}/device_scan"
|
_DEVICE_SCAN = f"{DEVICE_TRACKER_DOMAIN}/device_scan"
|
||||||
|
|
||||||
|
|
||||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
async def async_setup_entry(
|
||||||
|
hass: HomeAssistantType,
|
||||||
|
config_entry: ConfigEntry,
|
||||||
|
async_add_entities: Callable[[List[Entity], bool], None],
|
||||||
|
) -> None:
|
||||||
"""Set up from config entry."""
|
"""Set up from config entry."""
|
||||||
|
|
||||||
# Grab hosts list once to examine whether the initial fetch has got some data for
|
# Grab hosts list once to examine whether the initial fetch has got some data for
|
||||||
@ -42,7 +48,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
|
|||||||
# Initialize already tracked entities
|
# Initialize already tracked entities
|
||||||
tracked: Set[str] = set()
|
tracked: Set[str] = set()
|
||||||
registry = await entity_registry.async_get_registry(hass)
|
registry = await entity_registry.async_get_registry(hass)
|
||||||
known_entities: List[HuaweiLteScannerEntity] = []
|
known_entities: List[Entity] = []
|
||||||
for entity in registry.entities.values():
|
for entity in registry.entities.values():
|
||||||
if (
|
if (
|
||||||
entity.domain == DEVICE_TRACKER_DOMAIN
|
entity.domain == DEVICE_TRACKER_DOMAIN
|
||||||
@ -73,7 +79,12 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
|
|||||||
|
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def async_add_new_entities(hass, router_url, async_add_entities, tracked):
|
def async_add_new_entities(
|
||||||
|
hass: HomeAssistantType,
|
||||||
|
router_url: str,
|
||||||
|
async_add_entities: Callable[[List[Entity], bool], None],
|
||||||
|
tracked: Set[str],
|
||||||
|
) -> None:
|
||||||
"""Add new entities that are not already being tracked."""
|
"""Add new entities that are not already being tracked."""
|
||||||
router = hass.data[DOMAIN].routers[router_url]
|
router = hass.data[DOMAIN].routers[router_url]
|
||||||
try:
|
try:
|
||||||
@ -104,7 +115,7 @@ def _better_snakecase(text: str) -> str:
|
|||||||
lambda match: f"{match.group(1)}{match.group(2).lower()}{match.group(3)}",
|
lambda match: f"{match.group(1)}{match.group(2).lower()}{match.group(3)}",
|
||||||
text,
|
text,
|
||||||
)
|
)
|
||||||
return snakecase(text)
|
return cast(str, snakecase(text))
|
||||||
|
|
||||||
|
|
||||||
@attr.s
|
@attr.s
|
||||||
|
@ -2,13 +2,14 @@
|
|||||||
|
|
||||||
import logging
|
import logging
|
||||||
import time
|
import time
|
||||||
from typing import Any, List
|
from typing import Any, Dict, List, Optional
|
||||||
|
|
||||||
import attr
|
import attr
|
||||||
from huawei_lte_api.exceptions import ResponseErrorException
|
from huawei_lte_api.exceptions import ResponseErrorException
|
||||||
|
|
||||||
from homeassistant.components.notify import ATTR_TARGET, BaseNotificationService
|
from homeassistant.components.notify import ATTR_TARGET, BaseNotificationService
|
||||||
from homeassistant.const import CONF_RECIPIENT, CONF_URL
|
from homeassistant.const import CONF_RECIPIENT, CONF_URL
|
||||||
|
from homeassistant.helpers.typing import HomeAssistantType
|
||||||
|
|
||||||
from . import Router
|
from . import Router
|
||||||
from .const import DOMAIN
|
from .const import DOMAIN
|
||||||
@ -16,7 +17,11 @@ from .const import DOMAIN
|
|||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
async def async_get_service(hass, config, discovery_info=None):
|
async def async_get_service(
|
||||||
|
hass: HomeAssistantType,
|
||||||
|
config: Dict[str, Any],
|
||||||
|
discovery_info: Optional[Dict[str, Any]] = None,
|
||||||
|
) -> Optional["HuaweiLteSmsNotificationService"]:
|
||||||
"""Get the notification service."""
|
"""Get the notification service."""
|
||||||
if discovery_info is None:
|
if discovery_info is None:
|
||||||
return None
|
return None
|
||||||
|
@ -1,5 +1,6 @@
|
|||||||
"""Support for Huawei LTE sensors."""
|
"""Support for Huawei LTE sensors."""
|
||||||
|
|
||||||
|
from bisect import bisect
|
||||||
import logging
|
import logging
|
||||||
import re
|
import re
|
||||||
from typing import Callable, Dict, List, NamedTuple, Optional, Pattern, Tuple, Union
|
from typing import Callable, Dict, List, NamedTuple, Optional, Pattern, Tuple, Union
|
||||||
@ -10,6 +11,7 @@ from homeassistant.components.sensor import (
|
|||||||
DEVICE_CLASS_SIGNAL_STRENGTH,
|
DEVICE_CLASS_SIGNAL_STRENGTH,
|
||||||
DOMAIN as SENSOR_DOMAIN,
|
DOMAIN as SENSOR_DOMAIN,
|
||||||
)
|
)
|
||||||
|
from homeassistant.config_entries import ConfigEntry
|
||||||
from homeassistant.const import (
|
from homeassistant.const import (
|
||||||
CONF_URL,
|
CONF_URL,
|
||||||
DATA_BYTES,
|
DATA_BYTES,
|
||||||
@ -18,7 +20,7 @@ from homeassistant.const import (
|
|||||||
TIME_SECONDS,
|
TIME_SECONDS,
|
||||||
)
|
)
|
||||||
from homeassistant.helpers.entity import Entity
|
from homeassistant.helpers.entity import Entity
|
||||||
from homeassistant.helpers.typing import StateType
|
from homeassistant.helpers.typing import HomeAssistantType, StateType
|
||||||
|
|
||||||
from . import HuaweiLteBaseEntity
|
from . import HuaweiLteBaseEntity
|
||||||
from .const import (
|
from .const import (
|
||||||
@ -66,11 +68,11 @@ SENSOR_META: Dict[Union[str, Tuple[str, str]], SensorMeta] = {
|
|||||||
(KEY_DEVICE_SIGNAL, "dl_mcs"): SensorMeta(name="Downlink MCS"),
|
(KEY_DEVICE_SIGNAL, "dl_mcs"): SensorMeta(name="Downlink MCS"),
|
||||||
(KEY_DEVICE_SIGNAL, "dlbandwidth"): SensorMeta(
|
(KEY_DEVICE_SIGNAL, "dlbandwidth"): SensorMeta(
|
||||||
name="Downlink bandwidth",
|
name="Downlink bandwidth",
|
||||||
icon=lambda x: (x is None or x < 8)
|
icon=lambda x: (
|
||||||
and "mdi:speedometer-slow"
|
"mdi:speedometer-slow",
|
||||||
or x < 15
|
"mdi:speedometer-medium",
|
||||||
and "mdi:speedometer-medium"
|
"mdi:speedometer",
|
||||||
or "mdi:speedometer",
|
)[bisect((8, 15), x if x is not None else -1000)],
|
||||||
),
|
),
|
||||||
(KEY_DEVICE_SIGNAL, "earfcn"): SensorMeta(name="EARFCN"),
|
(KEY_DEVICE_SIGNAL, "earfcn"): SensorMeta(name="EARFCN"),
|
||||||
(KEY_DEVICE_SIGNAL, "lac"): SensorMeta(name="LAC", icon="mdi:map-marker"),
|
(KEY_DEVICE_SIGNAL, "lac"): SensorMeta(name="LAC", icon="mdi:map-marker"),
|
||||||
@ -86,11 +88,11 @@ SENSOR_META: Dict[Union[str, Tuple[str, str]], SensorMeta] = {
|
|||||||
(KEY_DEVICE_SIGNAL, "ul_mcs"): SensorMeta(name="Uplink MCS"),
|
(KEY_DEVICE_SIGNAL, "ul_mcs"): SensorMeta(name="Uplink MCS"),
|
||||||
(KEY_DEVICE_SIGNAL, "ulbandwidth"): SensorMeta(
|
(KEY_DEVICE_SIGNAL, "ulbandwidth"): SensorMeta(
|
||||||
name="Uplink bandwidth",
|
name="Uplink bandwidth",
|
||||||
icon=lambda x: (x is None or x < 8)
|
icon=lambda x: (
|
||||||
and "mdi:speedometer-slow"
|
"mdi:speedometer-slow",
|
||||||
or x < 15
|
"mdi:speedometer-medium",
|
||||||
and "mdi:speedometer-medium"
|
"mdi:speedometer",
|
||||||
or "mdi:speedometer",
|
)[bisect((8, 15), x if x is not None else -1000)],
|
||||||
),
|
),
|
||||||
(KEY_DEVICE_SIGNAL, "mode"): SensorMeta(
|
(KEY_DEVICE_SIGNAL, "mode"): SensorMeta(
|
||||||
name="Mode",
|
name="Mode",
|
||||||
@ -101,77 +103,71 @@ SENSOR_META: Dict[Union[str, Tuple[str, str]], SensorMeta] = {
|
|||||||
name="RSRQ",
|
name="RSRQ",
|
||||||
device_class=DEVICE_CLASS_SIGNAL_STRENGTH,
|
device_class=DEVICE_CLASS_SIGNAL_STRENGTH,
|
||||||
# http://www.lte-anbieter.info/technik/rsrq.php
|
# http://www.lte-anbieter.info/technik/rsrq.php
|
||||||
icon=lambda x: (x is None or x < -11)
|
icon=lambda x: (
|
||||||
and "mdi:signal-cellular-outline"
|
"mdi:signal-cellular-outline",
|
||||||
or x < -8
|
"mdi:signal-cellular-1",
|
||||||
and "mdi:signal-cellular-1"
|
"mdi:signal-cellular-2",
|
||||||
or x < -5
|
"mdi:signal-cellular-3",
|
||||||
and "mdi:signal-cellular-2"
|
)[bisect((-11, -8, -5), x if x is not None else -1000)],
|
||||||
or "mdi:signal-cellular-3",
|
|
||||||
enabled_default=True,
|
enabled_default=True,
|
||||||
),
|
),
|
||||||
(KEY_DEVICE_SIGNAL, "rsrp"): SensorMeta(
|
(KEY_DEVICE_SIGNAL, "rsrp"): SensorMeta(
|
||||||
name="RSRP",
|
name="RSRP",
|
||||||
device_class=DEVICE_CLASS_SIGNAL_STRENGTH,
|
device_class=DEVICE_CLASS_SIGNAL_STRENGTH,
|
||||||
# http://www.lte-anbieter.info/technik/rsrp.php
|
# http://www.lte-anbieter.info/technik/rsrp.php
|
||||||
icon=lambda x: (x is None or x < -110)
|
icon=lambda x: (
|
||||||
and "mdi:signal-cellular-outline"
|
"mdi:signal-cellular-outline",
|
||||||
or x < -95
|
"mdi:signal-cellular-1",
|
||||||
and "mdi:signal-cellular-1"
|
"mdi:signal-cellular-2",
|
||||||
or x < -80
|
"mdi:signal-cellular-3",
|
||||||
and "mdi:signal-cellular-2"
|
)[bisect((-110, -95, -80), x if x is not None else -1000)],
|
||||||
or "mdi:signal-cellular-3",
|
|
||||||
enabled_default=True,
|
enabled_default=True,
|
||||||
),
|
),
|
||||||
(KEY_DEVICE_SIGNAL, "rssi"): SensorMeta(
|
(KEY_DEVICE_SIGNAL, "rssi"): SensorMeta(
|
||||||
name="RSSI",
|
name="RSSI",
|
||||||
device_class=DEVICE_CLASS_SIGNAL_STRENGTH,
|
device_class=DEVICE_CLASS_SIGNAL_STRENGTH,
|
||||||
# https://eyesaas.com/wi-fi-signal-strength/
|
# https://eyesaas.com/wi-fi-signal-strength/
|
||||||
icon=lambda x: (x is None or x < -80)
|
icon=lambda x: (
|
||||||
and "mdi:signal-cellular-outline"
|
"mdi:signal-cellular-outline",
|
||||||
or x < -70
|
"mdi:signal-cellular-1",
|
||||||
and "mdi:signal-cellular-1"
|
"mdi:signal-cellular-2",
|
||||||
or x < -60
|
"mdi:signal-cellular-3",
|
||||||
and "mdi:signal-cellular-2"
|
)[bisect((-80, -70, -60), x if x is not None else -1000)],
|
||||||
or "mdi:signal-cellular-3",
|
|
||||||
enabled_default=True,
|
enabled_default=True,
|
||||||
),
|
),
|
||||||
(KEY_DEVICE_SIGNAL, "sinr"): SensorMeta(
|
(KEY_DEVICE_SIGNAL, "sinr"): SensorMeta(
|
||||||
name="SINR",
|
name="SINR",
|
||||||
device_class=DEVICE_CLASS_SIGNAL_STRENGTH,
|
device_class=DEVICE_CLASS_SIGNAL_STRENGTH,
|
||||||
# http://www.lte-anbieter.info/technik/sinr.php
|
# http://www.lte-anbieter.info/technik/sinr.php
|
||||||
icon=lambda x: (x is None or x < 0)
|
icon=lambda x: (
|
||||||
and "mdi:signal-cellular-outline"
|
"mdi:signal-cellular-outline",
|
||||||
or x < 5
|
"mdi:signal-cellular-1",
|
||||||
and "mdi:signal-cellular-1"
|
"mdi:signal-cellular-2",
|
||||||
or x < 10
|
"mdi:signal-cellular-3",
|
||||||
and "mdi:signal-cellular-2"
|
)[bisect((0, 5, 10), x if x is not None else -1000)],
|
||||||
or "mdi:signal-cellular-3",
|
|
||||||
enabled_default=True,
|
enabled_default=True,
|
||||||
),
|
),
|
||||||
(KEY_DEVICE_SIGNAL, "rscp"): SensorMeta(
|
(KEY_DEVICE_SIGNAL, "rscp"): SensorMeta(
|
||||||
name="RSCP",
|
name="RSCP",
|
||||||
device_class=DEVICE_CLASS_SIGNAL_STRENGTH,
|
device_class=DEVICE_CLASS_SIGNAL_STRENGTH,
|
||||||
# https://wiki.teltonika.lt/view/RSCP
|
# https://wiki.teltonika.lt/view/RSCP
|
||||||
icon=lambda x: (x is None or x < -95)
|
icon=lambda x: (
|
||||||
and "mdi:signal-cellular-outline"
|
"mdi:signal-cellular-outline",
|
||||||
or x < -85
|
"mdi:signal-cellular-1",
|
||||||
and "mdi:signal-cellular-1"
|
"mdi:signal-cellular-2",
|
||||||
or x < -75
|
"mdi:signal-cellular-3",
|
||||||
and "mdi:signal-cellular-2"
|
)[bisect((-95, -85, -75), x if x is not None else -1000)],
|
||||||
or "mdi:signal-cellular-3",
|
|
||||||
),
|
),
|
||||||
(KEY_DEVICE_SIGNAL, "ecio"): SensorMeta(
|
(KEY_DEVICE_SIGNAL, "ecio"): SensorMeta(
|
||||||
name="EC/IO",
|
name="EC/IO",
|
||||||
device_class=DEVICE_CLASS_SIGNAL_STRENGTH,
|
device_class=DEVICE_CLASS_SIGNAL_STRENGTH,
|
||||||
# https://wiki.teltonika.lt/view/EC/IO
|
# https://wiki.teltonika.lt/view/EC/IO
|
||||||
icon=lambda x: (x is None or x < -20)
|
icon=lambda x: (
|
||||||
and "mdi:signal-cellular-outline"
|
"mdi:signal-cellular-outline",
|
||||||
or x < -10
|
"mdi:signal-cellular-1",
|
||||||
and "mdi:signal-cellular-1"
|
"mdi:signal-cellular-2",
|
||||||
or x < -6
|
"mdi:signal-cellular-3",
|
||||||
and "mdi:signal-cellular-2"
|
)[bisect((-20, -10, -6), x if x is not None else -1000)],
|
||||||
or "mdi:signal-cellular-3",
|
|
||||||
),
|
),
|
||||||
KEY_MONITORING_CHECK_NOTIFICATIONS: SensorMeta(
|
KEY_MONITORING_CHECK_NOTIFICATIONS: SensorMeta(
|
||||||
exclude=re.compile(
|
exclude=re.compile(
|
||||||
@ -322,7 +318,11 @@ 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: HomeAssistantType,
|
||||||
|
config_entry: ConfigEntry,
|
||||||
|
async_add_entities: Callable[[List[Entity], bool], None],
|
||||||
|
) -> None:
|
||||||
"""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: List[Entity] = []
|
sensors: List[Entity] = []
|
||||||
@ -346,7 +346,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
|
|||||||
async_add_entities(sensors, True)
|
async_add_entities(sensors, True)
|
||||||
|
|
||||||
|
|
||||||
def format_default(value):
|
def format_default(value: StateType) -> Tuple[StateType, Optional[str]]:
|
||||||
"""Format value."""
|
"""Format value."""
|
||||||
unit = None
|
unit = None
|
||||||
if value is not None:
|
if value is not None:
|
||||||
|
@ -1,7 +1,7 @@
|
|||||||
"""Support for Huawei LTE switches."""
|
"""Support for Huawei LTE switches."""
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
from typing import Any, List, Optional
|
from typing import Any, Callable, List, Optional
|
||||||
|
|
||||||
import attr
|
import attr
|
||||||
|
|
||||||
@ -10,8 +10,10 @@ from homeassistant.components.switch import (
|
|||||||
DOMAIN as SWITCH_DOMAIN,
|
DOMAIN as SWITCH_DOMAIN,
|
||||||
SwitchEntity,
|
SwitchEntity,
|
||||||
)
|
)
|
||||||
|
from homeassistant.config_entries import ConfigEntry
|
||||||
from homeassistant.const import CONF_URL
|
from homeassistant.const import CONF_URL
|
||||||
from homeassistant.helpers.entity import Entity
|
from homeassistant.helpers.entity import Entity
|
||||||
|
from homeassistant.helpers.typing import HomeAssistantType
|
||||||
|
|
||||||
from . import HuaweiLteBaseEntity
|
from . import HuaweiLteBaseEntity
|
||||||
from .const import DOMAIN, KEY_DIALUP_MOBILE_DATASWITCH
|
from .const import DOMAIN, KEY_DIALUP_MOBILE_DATASWITCH
|
||||||
@ -19,7 +21,11 @@ from .const import DOMAIN, KEY_DIALUP_MOBILE_DATASWITCH
|
|||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
async def async_setup_entry(
|
||||||
|
hass: HomeAssistantType,
|
||||||
|
config_entry: ConfigEntry,
|
||||||
|
async_add_entities: Callable[[List[Entity], bool], None],
|
||||||
|
) -> None:
|
||||||
"""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: List[Entity] = []
|
switches: List[Entity] = []
|
||||||
|
@ -40,7 +40,7 @@ warn_incomplete_stub = true
|
|||||||
warn_redundant_casts = true
|
warn_redundant_casts = true
|
||||||
warn_unused_configs = 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.image_processing.*,homeassistant.components.integration.*,homeassistant.components.light.*,homeassistant.components.lock.*,homeassistant.components.mailbox.*,homeassistant.components.media_player.*,homeassistant.components.notify.*,homeassistant.components.persistent_notification.*,homeassistant.components.proximity.*,homeassistant.components.remote.*,homeassistant.components.scene.*,homeassistant.components.sensor.*,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.helpers.*,homeassistant.scripts.*,homeassistant.util.*]
|
[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.image_processing.*,homeassistant.components.integration.*,homeassistant.components.light.*,homeassistant.components.lock.*,homeassistant.components.mailbox.*,homeassistant.components.media_player.*,homeassistant.components.notify.*,homeassistant.components.persistent_notification.*,homeassistant.components.proximity.*,homeassistant.components.remote.*,homeassistant.components.scene.*,homeassistant.components.sensor.*,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.helpers.*,homeassistant.scripts.*,homeassistant.util.*]
|
||||||
strict = true
|
strict = true
|
||||||
ignore_errors = false
|
ignore_errors = false
|
||||||
warn_unreachable = true
|
warn_unreachable = true
|
||||||
|
Loading…
x
Reference in New Issue
Block a user