Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Huawei LTE simplifications (#62770) 

* Use enum types rather than strs in sensor type hints

* Name sensor meta fields same as in SensorEntityDescription

* Make integration shared state a NamedTuple

* Use dataclasses instead of attr
pull/63179/head
Ville Skyttä 2021-12-26 09:17:59 +02:00 committed by GitHub
parent e982e7403a
commit 3e3fb52dfa
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 101 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
homeassistant/components/huawei_lte/__init__.py
View File

@ -4,12 +4,12 @@ from __future__ import annotations
from collections import defaultdict
from collections.abc import Callable
from contextlib import suppress
from dataclasses import dataclass, field
from datetime import timedelta
import logging
import time
from typing import Any, cast
from typing import Any, NamedTuple, cast
import attr
from huawei_lte_api.AuthorizedConnection import AuthorizedConnection
from huawei_lte_api.Client import Client
from huawei_lte_api.Connection import Connection
@ -126,26 +126,28 @@ PLATFORMS = [
]
@attr.s
@dataclass
class Router:
"""Class for router state."""
hass: HomeAssistant = attr.ib()
config_entry: ConfigEntry = attr.ib()
connection: Connection = attr.ib()
url: str = attr.ib()
hass: HomeAssistant
config_entry: ConfigEntry
connection: Connection
url: str
data: dict[str, Any] = attr.ib(init=False, factory=dict)
subscriptions: dict[str, set[str]] = attr.ib(
data: dict[str, Any] = field(default_factory=dict, init=False)
subscriptions: dict[str, set[str]] = field(
default_factory=lambda: defaultdict(
set, ((x, {"initial_scan"}) for x in ALL_KEYS)
),
init=False,
factory=lambda: defaultdict(set, ((x, {"initial_scan"}) for x in ALL_KEYS)),
)
inflight_gets: set[str] = attr.ib(init=False, factory=set)
client: Client
suspended = attr.ib(init=False, default=False)
notify_last_attempt: float = attr.ib(init=False, default=-1)
inflight_gets: set[str] = field(default_factory=set, init=False)
client: Client = field(init=False)
suspended: bool = field(default=False, init=False)
notify_last_attempt: float = field(default=-1, init=False)
def __attrs_post_init__(self) -> None:
def __post_init__(self) -> None:
"""Set up internal state on init."""
self.client = Client(self.connection)
@ -293,14 +295,13 @@ class Router:
self.logout()
@attr.s
class HuaweiLteData:
class HuaweiLteData(NamedTuple):
"""Shared state."""
hass_config: ConfigType = attr.ib()
hass_config: ConfigType
# Our YAML config, keyed by router URL
config: dict[str, dict[str, Any]] = attr.ib()
routers: dict[str, Router] = attr.ib(init=False, factory=dict)
config: dict[str, dict[str, Any]]
routers: dict[str, Router]
async def async_setup_entry( # noqa: C901
@ -509,7 +510,9 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
# Arrange our YAML config to dict with normalized URLs as keys
domain_config: dict[str, dict[str, Any]] = {}
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, routers={}
)
for router_config in config.get(DOMAIN, []):
domain_config[url_normalize(router_config.pop(CONF_URL))] = router_config
@ -607,14 +610,14 @@ async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) ->
return True
@attr.s
@dataclass
class HuaweiLteBaseEntity(Entity):
"""Huawei LTE entity base class."""
router: Router = attr.ib()
router: Router
_available: bool = attr.ib(init=False, default=True)
_unsub_handlers: list[Callable] = attr.ib(init=False, factory=list)
_available: bool = field(default=True, init=False)
_unsub_handlers: list[Callable] = field(default_factory=list, init=False)
@property
def _entity_name(self) -> str:

30
homeassistant/components/huawei_lte/binary_sensor.py
View File

@ -1,10 +1,10 @@
"""Support for Huawei LTE binary sensors."""
from __future__ import annotations
from dataclasses import dataclass, field
import logging
from typing import Any
import attr
from huawei_lte_api.enums.cradle import ConnectionStatusEnum
from homeassistant.components.binary_sensor import (
@ -48,13 +48,13 @@ async def async_setup_entry(
async_add_entities(entities, True)
@attr.s
@dataclass
class HuaweiLteBaseBinarySensor(HuaweiLteBaseEntity, BinarySensorEntity):
"""Huawei LTE binary sensor device base class."""
key: str
item: str
_raw_state: str | None = attr.ib(init=False, default=None)
key: str = field(init=False)
item: str = field(init=False)
_raw_state: str | None = field(default=None, init=False)
@property
def entity_registry_enabled_default(self) -> bool:
@ -101,11 +101,11 @@ CONNECTION_STATE_ATTRIBUTES = {
}
@attr.s
@dataclass
class HuaweiLteMobileConnectionBinarySensor(HuaweiLteBaseBinarySensor):
"""Huawei LTE mobile connection binary sensor."""
def __attrs_post_init__(self) -> None:
def __post_init__(self) -> None:
"""Initialize identifiers."""
self.key = KEY_MONITORING_STATUS
self.item = "ConnectionStatus"
@ -172,11 +172,11 @@ class HuaweiLteBaseWifiStatusBinarySensor(HuaweiLteBaseBinarySensor):
return "mdi:wifi" if self.is_on else "mdi:wifi-off"
@attr.s
@dataclass
class HuaweiLteWifiStatusBinarySensor(HuaweiLteBaseWifiStatusBinarySensor):
"""Huawei LTE WiFi status binary sensor."""
def __attrs_post_init__(self) -> None:
def __post_init__(self) -> None:
"""Initialize identifiers."""
self.key = KEY_MONITORING_STATUS
self.item = "WifiStatus"
@ -186,11 +186,11 @@ class HuaweiLteWifiStatusBinarySensor(HuaweiLteBaseWifiStatusBinarySensor):
return "WiFi status"
@attr.s
@dataclass
class HuaweiLteWifi24ghzStatusBinarySensor(HuaweiLteBaseWifiStatusBinarySensor):
"""Huawei LTE 2.4GHz WiFi status binary sensor."""
def __attrs_post_init__(self) -> None:
def __post_init__(self) -> None:
"""Initialize identifiers."""
self.key = KEY_WLAN_WIFI_FEATURE_SWITCH
self.item = "wifi24g_switch_enable"
@ -200,11 +200,11 @@ class HuaweiLteWifi24ghzStatusBinarySensor(HuaweiLteBaseWifiStatusBinarySensor):
return "2.4GHz WiFi status"
@attr.s
@dataclass
class HuaweiLteWifi5ghzStatusBinarySensor(HuaweiLteBaseWifiStatusBinarySensor):
"""Huawei LTE 5GHz WiFi status binary sensor."""
def __attrs_post_init__(self) -> None:
def __post_init__(self) -> None:
"""Initialize identifiers."""
self.key = KEY_WLAN_WIFI_FEATURE_SWITCH
self.item = "wifi5g_enabled"
@ -214,11 +214,11 @@ class HuaweiLteWifi5ghzStatusBinarySensor(HuaweiLteBaseWifiStatusBinarySensor):
return "5GHz WiFi status"
@attr.s
@dataclass
class HuaweiLteSmsStorageFullBinarySensor(HuaweiLteBaseBinarySensor):
"""Huawei LTE SMS storage full binary sensor."""
def __attrs_post_init__(self) -> None:
def __post_init__(self) -> None:
"""Initialize identifiers."""
self.key = KEY_MONITORING_CHECK_NOTIFICATIONS
self.item = "SmsStorageFull"

14
homeassistant/components/huawei_lte/device_tracker.py
View File

@ -1,11 +1,11 @@
"""Support for device tracking of Huawei LTE routers."""
from __future__ import annotations
from dataclasses import dataclass, field
import logging
import re
from typing import Any, Dict, List, cast
import attr
from stringcase import snakecase
from homeassistant.components.device_tracker.config_entry import ScannerEntity
@ -173,16 +173,16 @@ def _better_snakecase(text: str) -> str:
return cast(str, snakecase(text))
@attr.s
@dataclass
class HuaweiLteScannerEntity(HuaweiLteBaseEntity, ScannerEntity):
"""Huawei LTE router scanner entity."""
_mac_address: str = attr.ib()
_mac_address: str
_ip_address: str | None = attr.ib(init=False, default=None)
_is_connected: bool = attr.ib(init=False, default=False)
_hostname: str | None = attr.ib(init=False, default=None)
_extra_state_attributes: dict[str, Any] = attr.ib(init=False, factory=dict)
_ip_address: str | None = field(default=None, init=False)
_is_connected: bool = field(default=False, init=False)
_hostname: str | None = field(default=None, init=False)
_extra_state_attributes: dict[str, Any] = field(default_factory=dict, init=False)
@property
def _entity_name(self) -> str:

8
homeassistant/components/huawei_lte/notify.py
View File

@ -1,11 +1,11 @@
"""Support for Huawei LTE router notifications."""
from __future__ import annotations
from dataclasses import dataclass
import logging
import time
from typing import Any
import attr
from huawei_lte_api.exceptions import ResponseErrorException
from homeassistant.components.notify import ATTR_TARGET, BaseNotificationService
@ -33,12 +33,12 @@ async def async_get_service(
return HuaweiLteSmsNotificationService(router, default_targets)
@attr.s
@dataclass
class HuaweiLteSmsNotificationService(BaseNotificationService):
"""Huawei LTE router SMS notification service."""
router: Router = attr.ib()
default_targets: list[str] = attr.ib()
router: Router
default_targets: list[str]
def send_message(self, message: str = "", **kwargs: Any) -> None:
"""Send message to target numbers."""

71
homeassistant/components/huawei_lte/sensor.py
View File

@ -3,12 +3,11 @@ from __future__ import annotations
from bisect import bisect
from collections.abc import Callable
from dataclasses import dataclass, field
import logging
import re
from typing import NamedTuple
import attr
from homeassistant.components.sensor import (
DOMAIN as SENSOR_DOMAIN,
SensorDeviceClass,
@ -51,12 +50,12 @@ class SensorMeta(NamedTuple):
"""Metadata for defining sensors."""
name: str | None = None
device_class: str | None = None
device_class: SensorDeviceClass | None = None
icon: str | Callable[[StateType], str] | None = None
unit: str | None = None
state_class: str | None = None
enabled_default: bool = False
entity_category: str | None = None
native_unit_of_measurement: str | None = None
state_class: SensorStateClass | None = None
entity_registry_enabled_default: bool = False
entity_category: EntityCategory | None = None
include: re.Pattern[str] | None = None
exclude: re.Pattern[str] | None = None
formatter: Callable[[str], tuple[StateType, str | None]] | None = None
@ -70,7 +69,7 @@ SENSOR_META: dict[str | tuple[str, str], SensorMeta] = {
name="WAN IP address",
icon="mdi:ip",
entity_category=EntityCategory.DIAGNOSTIC,
enabled_default=True,
entity_registry_enabled_default=True,
),
(KEY_DEVICE_INFORMATION, "WanIPv6Address"): SensorMeta(
name="WAN IPv6 address",
@ -80,7 +79,7 @@ SENSOR_META: dict[str | tuple[str, str], SensorMeta] = {
(KEY_DEVICE_INFORMATION, "uptime"): SensorMeta(
name="Uptime",
icon="mdi:timer-outline",
unit=TIME_SECONDS,
native_unit_of_measurement=TIME_SECONDS,
entity_category=EntityCategory.DIAGNOSTIC,
),
(KEY_DEVICE_SIGNAL, "band"): SensorMeta(
@ -181,7 +180,7 @@ SENSOR_META: dict[str | tuple[str, str], SensorMeta] = {
)[bisect((-11, -8, -5), x if x is not None else -1000)],
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
enabled_default=True,
entity_registry_enabled_default=True,
),
(KEY_DEVICE_SIGNAL, "rsrp"): SensorMeta(
name="RSRP",
@ -195,7 +194,7 @@ SENSOR_META: dict[str | tuple[str, str], SensorMeta] = {
)[bisect((-110, -95, -80), x if x is not None else -1000)],
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
enabled_default=True,
entity_registry_enabled_default=True,
),
(KEY_DEVICE_SIGNAL, "rssi"): SensorMeta(
name="RSSI",
@ -209,7 +208,7 @@ SENSOR_META: dict[str | tuple[str, str], SensorMeta] = {
)[bisect((-80, -70, -60), x if x is not None else -1000)],
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
enabled_default=True,
entity_registry_enabled_default=True,
),
(KEY_DEVICE_SIGNAL, "sinr"): SensorMeta(
name="SINR",
@ -223,7 +222,7 @@ SENSOR_META: dict[str | tuple[str, str], SensorMeta] = {
)[bisect((0, 5, 10), x if x is not None else -1000)],
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
enabled_default=True,
entity_registry_enabled_default=True,
),
(KEY_DEVICE_SIGNAL, "rscp"): SensorMeta(
name="RSCP",
@ -298,13 +297,13 @@ SENSOR_META: dict[str | tuple[str, str], SensorMeta] = {
),
(KEY_MONITORING_MONTH_STATISTICS, "CurrentMonthDownload"): SensorMeta(
name="Current month download",
unit=DATA_BYTES,
native_unit_of_measurement=DATA_BYTES,
icon="mdi:download",
state_class=SensorStateClass.TOTAL_INCREASING,
),
(KEY_MONITORING_MONTH_STATISTICS, "CurrentMonthUpload"): SensorMeta(
name="Current month upload",
unit=DATA_BYTES,
native_unit_of_measurement=DATA_BYTES,
icon="mdi:upload",
state_class=SensorStateClass.TOTAL_INCREASING,
),
@ -317,7 +316,7 @@ SENSOR_META: dict[str | tuple[str, str], SensorMeta] = {
(KEY_MONITORING_STATUS, "BatteryPercent"): SensorMeta(
name="Battery",
device_class=SensorDeviceClass.BATTERY,
unit=PERCENTAGE,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
),
@ -351,47 +350,49 @@ SENSOR_META: dict[str | tuple[str, str], SensorMeta] = {
exclude=re.compile(r"^showtraffic$", re.IGNORECASE)
),
(KEY_MONITORING_TRAFFIC_STATISTICS, "CurrentConnectTime"): SensorMeta(
name="Current connection duration", unit=TIME_SECONDS, icon="mdi:timer-outline"
name="Current connection duration",
native_unit_of_measurement=TIME_SECONDS,
icon="mdi:timer-outline",
),
(KEY_MONITORING_TRAFFIC_STATISTICS, "CurrentDownload"): SensorMeta(
name="Current connection download",
unit=DATA_BYTES,
native_unit_of_measurement=DATA_BYTES,
icon="mdi:download",
state_class=SensorStateClass.TOTAL_INCREASING,
),
(KEY_MONITORING_TRAFFIC_STATISTICS, "CurrentDownloadRate"): SensorMeta(
name="Current download rate",
unit=DATA_RATE_BYTES_PER_SECOND,
native_unit_of_measurement=DATA_RATE_BYTES_PER_SECOND,
icon="mdi:download",
state_class=SensorStateClass.MEASUREMENT,
),
(KEY_MONITORING_TRAFFIC_STATISTICS, "CurrentUpload"): SensorMeta(
name="Current connection upload",
unit=DATA_BYTES,
native_unit_of_measurement=DATA_BYTES,
icon="mdi:upload",
state_class=SensorStateClass.TOTAL_INCREASING,
),
(KEY_MONITORING_TRAFFIC_STATISTICS, "CurrentUploadRate"): SensorMeta(
name="Current upload rate",
unit=DATA_RATE_BYTES_PER_SECOND,
native_unit_of_measurement=DATA_RATE_BYTES_PER_SECOND,
icon="mdi:upload",
state_class=SensorStateClass.MEASUREMENT,
),
(KEY_MONITORING_TRAFFIC_STATISTICS, "TotalConnectTime"): SensorMeta(
name="Total connected duration",
unit=TIME_SECONDS,
native_unit_of_measurement=TIME_SECONDS,
icon="mdi:timer-outline",
state_class=SensorStateClass.TOTAL_INCREASING,
),
(KEY_MONITORING_TRAFFIC_STATISTICS, "TotalDownload"): SensorMeta(
name="Total download",
unit=DATA_BYTES,
native_unit_of_measurement=DATA_BYTES,
icon="mdi:download",
state_class=SensorStateClass.TOTAL_INCREASING,
),
(KEY_MONITORING_TRAFFIC_STATISTICS, "TotalUpload"): SensorMeta(
name="Total upload",
unit=DATA_BYTES,
native_unit_of_measurement=DATA_BYTES,
icon="mdi:upload",
state_class=SensorStateClass.TOTAL_INCREASING,
),
@ -521,16 +522,16 @@ def format_default(value: StateType) -> tuple[StateType, str | None]:
return value, unit
@attr.s
@dataclass
class HuaweiLteSensor(HuaweiLteBaseEntity, SensorEntity):
"""Huawei LTE sensor entity."""
key: str = attr.ib()
item: str = attr.ib()
meta: SensorMeta = attr.ib()
key: str
item: str
meta: SensorMeta
_state: StateType = attr.ib(init=False, default=STATE_UNKNOWN)
_unit: str | None = attr.ib(init=False)
_state: StateType = field(default=STATE_UNKNOWN, init=False)
_unit: str | None = field(default=None, init=False)
async def async_added_to_hass(self) -> None:
"""Subscribe to needed data on add."""
@ -556,14 +557,14 @@ class HuaweiLteSensor(HuaweiLteBaseEntity, SensorEntity):
return self._state
@property
def device_class(self) -> str | None:
def device_class(self) -> SensorDeviceClass | None:
"""Return sensor device class."""
return self.meta.device_class
@property
def native_unit_of_measurement(self) -> str | None:
"""Return sensor's unit of measurement."""
return self.meta.unit or self._unit
return self.meta.native_unit_of_measurement or self._unit
@property
def icon(self) -> str | None:
@ -574,14 +575,14 @@ class HuaweiLteSensor(HuaweiLteBaseEntity, SensorEntity):
return icon
@property
def state_class(self) -> str | None:
def state_class(self) -> SensorStateClass | None:
"""Return sensor state class."""
return self.meta.state_class
@property
def entity_registry_enabled_default(self) -> bool:
"""Return if the entity should be enabled when first added to the entity registry."""
return self.meta.enabled_default
return self.meta.entity_registry_enabled_default
async def async_update(self) -> None:
"""Update state."""
@ -599,6 +600,6 @@ class HuaweiLteSensor(HuaweiLteBaseEntity, SensorEntity):
self._available = value is not None
@property
def entity_category(self) -> str | None:
def entity_category(self) -> EntityCategory | None:
"""Return category of entity, if any."""
return self.meta.entity_category

20
homeassistant/components/huawei_lte/switch.py
View File

@ -1,11 +1,10 @@
"""Support for Huawei LTE switches."""
from __future__ import annotations
from dataclasses import dataclass, field
import logging
from typing import Any
import attr
from homeassistant.components.switch import (
DOMAIN as SWITCH_DOMAIN,
SwitchDeviceClass,
@ -37,14 +36,17 @@ async def async_setup_entry(
async_add_entities(switches, True)
@attr.s
@dataclass
class HuaweiLteBaseSwitch(HuaweiLteBaseEntity, SwitchEntity):
"""Huawei LTE switch device base class."""
key: str
item: str
_attr_device_class = SwitchDeviceClass.SWITCH
_raw_state: str | None = attr.ib(init=False, default=None)
key: str = field(init=False)
item: str = field(init=False)
_attr_device_class: SwitchDeviceClass = field(
default=SwitchDeviceClass.SWITCH, init=False
)
_raw_state: str | None = field(default=None, init=False)
def _turn(self, state: bool) -> None:
raise NotImplementedError
@ -79,11 +81,11 @@ class HuaweiLteBaseSwitch(HuaweiLteBaseEntity, SwitchEntity):
self._raw_state = str(value)
@attr.s
@dataclass
class HuaweiLteMobileDataSwitch(HuaweiLteBaseSwitch):
"""Huawei LTE mobile data switch device."""
def __attrs_post_init__(self) -> None:
def __post_init__(self) -> None:
"""Initialize identifiers."""
self.key = KEY_DIALUP_MOBILE_DATASWITCH
self.item = "dataswitch"