Huawei LTE misc improvements (#26203)
* Constant and whitespace cleanups * Upgrade huawei_lte_api to 1.3.0 https://github.com/Salamek/huawei-lte-api/releases * Hush traceback if device does not support logoutpull/26216/head
parent
bde572c91a
commit
0c49c82015
|
@ -1,4 +1,5 @@
|
|||
"""Support for Huawei LTE routers."""
|
||||
|
||||
from datetime import timedelta
|
||||
from functools import reduce
|
||||
from urllib.parse import urlparse
|
||||
|
@ -22,6 +23,14 @@ from homeassistant.const import (
|
|||
)
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.util import Throttle
|
||||
from .const import (
|
||||
DOMAIN,
|
||||
KEY_DEVICE_INFORMATION,
|
||||
KEY_DEVICE_SIGNAL,
|
||||
KEY_MONITORING_TRAFFIC_STATISTICS,
|
||||
KEY_WLAN_HOST_LIST,
|
||||
)
|
||||
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
@ -31,9 +40,6 @@ logging.getLogger("dicttoxml").setLevel(logging.WARNING)
|
|||
|
||||
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=10)
|
||||
|
||||
DOMAIN = "huawei_lte"
|
||||
DATA_KEY = "huawei_lte"
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema(
|
||||
{
|
||||
DOMAIN: vol.All(
|
||||
|
@ -107,12 +113,12 @@ class RouterData:
|
|||
finally:
|
||||
_LOGGER.debug("%s=%s", path, getattr(self, path))
|
||||
|
||||
get_data("device_information", self.client.device.information)
|
||||
get_data("device_signal", self.client.device.signal)
|
||||
get_data(KEY_DEVICE_INFORMATION, self.client.device.information)
|
||||
get_data(KEY_DEVICE_SIGNAL, self.client.device.signal)
|
||||
get_data(
|
||||
"monitoring_traffic_statistics", self.client.monitoring.traffic_statistics
|
||||
KEY_MONITORING_TRAFFIC_STATISTICS, self.client.monitoring.traffic_statistics
|
||||
)
|
||||
get_data("wlan_host_list", self.client.wlan.host_list)
|
||||
get_data(KEY_WLAN_HOST_LIST, self.client.wlan.host_list)
|
||||
|
||||
|
||||
@attr.s
|
||||
|
@ -133,8 +139,8 @@ class HuaweiLteData:
|
|||
|
||||
def setup(hass, config) -> bool:
|
||||
"""Set up Huawei LTE component."""
|
||||
if DATA_KEY not in hass.data:
|
||||
hass.data[DATA_KEY] = HuaweiLteData()
|
||||
if DOMAIN not in hass.data:
|
||||
hass.data[DOMAIN] = HuaweiLteData()
|
||||
for conf in config.get(DOMAIN, []):
|
||||
_setup_lte(hass, conf)
|
||||
return True
|
||||
|
@ -164,10 +170,13 @@ def _setup_lte(hass, lte_config) -> None:
|
|||
client = Client(connection)
|
||||
|
||||
data = RouterData(client, mac)
|
||||
hass.data[DATA_KEY].data[url] = data
|
||||
hass.data[DOMAIN].data[url] = data
|
||||
|
||||
def cleanup(event):
|
||||
"""Clean up resources."""
|
||||
client.user.logout()
|
||||
try:
|
||||
client.user.logout()
|
||||
except ResponseErrorNotSupportedException as ex:
|
||||
_LOGGER.debug("Logout not supported by device", exc_info=ex)
|
||||
|
||||
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, cleanup)
|
||||
|
|
|
@ -0,0 +1,8 @@
|
|||
"""Huawei LTE constants."""
|
||||
|
||||
DOMAIN = "huawei_lte"
|
||||
|
||||
KEY_DEVICE_INFORMATION = "device_information"
|
||||
KEY_DEVICE_SIGNAL = "device_signal"
|
||||
KEY_MONITORING_TRAFFIC_STATISTICS = "monitoring_traffic_statistics"
|
||||
KEY_WLAN_HOST_LIST = "wlan_host_list"
|
|
@ -1,4 +1,5 @@
|
|||
"""Support for device tracking of Huawei LTE routers."""
|
||||
|
||||
import logging
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
|
@ -8,19 +9,20 @@ import voluptuous as vol
|
|||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.components.device_tracker import PLATFORM_SCHEMA, DeviceScanner
|
||||
from homeassistant.const import CONF_URL
|
||||
from . import DATA_KEY, RouterData
|
||||
from . import RouterData
|
||||
from .const import DOMAIN, KEY_WLAN_HOST_LIST
|
||||
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({vol.Optional(CONF_URL): cv.url})
|
||||
|
||||
HOSTS_PATH = "wlan_host_list.Hosts.Host"
|
||||
HOSTS_PATH = f"{KEY_WLAN_HOST_LIST}.Hosts.Host"
|
||||
|
||||
|
||||
def get_scanner(hass, config):
|
||||
"""Get a Huawei LTE router scanner."""
|
||||
data = hass.data[DATA_KEY].get_data(config)
|
||||
data = hass.data[DOMAIN].get_data(config)
|
||||
data.subscribe(HOSTS_PATH)
|
||||
return HuaweiLteScanner(data)
|
||||
|
||||
|
|
|
@ -4,7 +4,7 @@
|
|||
"documentation": "https://www.home-assistant.io/components/huawei_lte",
|
||||
"requirements": [
|
||||
"getmac==0.8.1",
|
||||
"huawei-lte-api==1.2.0"
|
||||
"huawei-lte-api==1.3.0"
|
||||
],
|
||||
"dependencies": [],
|
||||
"codeowners": [
|
||||
|
|
|
@ -1,4 +1,5 @@
|
|||
"""Support for Huawei LTE router notifications."""
|
||||
|
||||
import logging
|
||||
|
||||
import voluptuous as vol
|
||||
|
@ -12,7 +13,8 @@ from homeassistant.components.notify import (
|
|||
from homeassistant.const import CONF_RECIPIENT, CONF_URL
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
||||
from . import DATA_KEY
|
||||
from .const import DOMAIN
|
||||
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
@ -44,7 +46,7 @@ class HuaweiLteSmsNotificationService(BaseNotificationService):
|
|||
if not targets or not message:
|
||||
return
|
||||
|
||||
data = self.hass.data[DATA_KEY].get_data(self.config)
|
||||
data = self.hass.data[DOMAIN].get_data(self.config)
|
||||
if not data:
|
||||
_LOGGER.error("Router not available")
|
||||
return
|
||||
|
|
|
@ -1,4 +1,5 @@
|
|||
"""Support for Huawei LTE sensors."""
|
||||
|
||||
import logging
|
||||
import re
|
||||
from typing import Optional
|
||||
|
@ -15,7 +16,14 @@ from homeassistant.helpers import entity_registry
|
|||
from homeassistant.helpers.entity import Entity
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
||||
from . import DATA_KEY, RouterData
|
||||
from . import RouterData
|
||||
from .const import (
|
||||
DOMAIN,
|
||||
KEY_DEVICE_INFORMATION,
|
||||
KEY_DEVICE_SIGNAL,
|
||||
KEY_MONITORING_TRAFFIC_STATISTICS,
|
||||
)
|
||||
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
@ -23,26 +31,30 @@ DEFAULT_NAME_TEMPLATE = "Huawei {} {}"
|
|||
DEFAULT_DEVICE_NAME = "LTE"
|
||||
|
||||
DEFAULT_SENSORS = [
|
||||
"device_information.WanIPAddress",
|
||||
"device_signal.rsrq",
|
||||
"device_signal.rsrp",
|
||||
"device_signal.rssi",
|
||||
"device_signal.sinr",
|
||||
f"{KEY_DEVICE_INFORMATION}.WanIPAddress",
|
||||
f"{KEY_DEVICE_SIGNAL}.rsrq",
|
||||
f"{KEY_DEVICE_SIGNAL}.rsrp",
|
||||
f"{KEY_DEVICE_SIGNAL}.rssi",
|
||||
f"{KEY_DEVICE_SIGNAL}.sinr",
|
||||
]
|
||||
|
||||
SENSOR_META = {
|
||||
"device_information.SoftwareVersion": dict(name="Software version"),
|
||||
"device_information.WanIPAddress": dict(name="WAN IP address", icon="mdi:ip"),
|
||||
"device_information.WanIPv6Address": dict(name="WAN IPv6 address", icon="mdi:ip"),
|
||||
"device_signal.band": dict(name="Band"),
|
||||
"device_signal.cell_id": dict(name="Cell ID"),
|
||||
"device_signal.lac": dict(name="LAC"),
|
||||
"device_signal.mode": dict(
|
||||
f"{KEY_DEVICE_INFORMATION}.SoftwareVersion": dict(name="Software version"),
|
||||
f"{KEY_DEVICE_INFORMATION}.WanIPAddress": dict(
|
||||
name="WAN IP address", icon="mdi:ip"
|
||||
),
|
||||
f"{KEY_DEVICE_INFORMATION}.WanIPv6Address": dict(
|
||||
name="WAN IPv6 address", icon="mdi:ip"
|
||||
),
|
||||
f"{KEY_DEVICE_SIGNAL}.band": dict(name="Band"),
|
||||
f"{KEY_DEVICE_SIGNAL}.cell_id": dict(name="Cell ID"),
|
||||
f"{KEY_DEVICE_SIGNAL}.lac": dict(name="LAC"),
|
||||
f"{KEY_DEVICE_SIGNAL}.mode": dict(
|
||||
name="Mode",
|
||||
formatter=lambda x: ({"0": "2G", "2": "3G", "7": "4G"}.get(x, "Unknown"), None),
|
||||
),
|
||||
"device_signal.pci": dict(name="PCI"),
|
||||
"device_signal.rsrq": dict(
|
||||
f"{KEY_DEVICE_SIGNAL}.pci": dict(name="PCI"),
|
||||
f"{KEY_DEVICE_SIGNAL}.rsrq": dict(
|
||||
name="RSRQ",
|
||||
device_class=DEVICE_CLASS_SIGNAL_STRENGTH,
|
||||
# http://www.lte-anbieter.info/technik/rsrq.php
|
||||
|
@ -54,7 +66,7 @@ SENSOR_META = {
|
|||
and "mdi:signal-cellular-2"
|
||||
or "mdi:signal-cellular-3",
|
||||
),
|
||||
"device_signal.rsrp": dict(
|
||||
f"{KEY_DEVICE_SIGNAL}.rsrp": dict(
|
||||
name="RSRP",
|
||||
device_class=DEVICE_CLASS_SIGNAL_STRENGTH,
|
||||
# http://www.lte-anbieter.info/technik/rsrp.php
|
||||
|
@ -66,7 +78,7 @@ SENSOR_META = {
|
|||
and "mdi:signal-cellular-2"
|
||||
or "mdi:signal-cellular-3",
|
||||
),
|
||||
"device_signal.rssi": dict(
|
||||
f"{KEY_DEVICE_SIGNAL}.rssi": dict(
|
||||
name="RSSI",
|
||||
device_class=DEVICE_CLASS_SIGNAL_STRENGTH,
|
||||
# https://eyesaas.com/wi-fi-signal-strength/
|
||||
|
@ -78,7 +90,7 @@ SENSOR_META = {
|
|||
and "mdi:signal-cellular-2"
|
||||
or "mdi:signal-cellular-3",
|
||||
),
|
||||
"device_signal.sinr": dict(
|
||||
f"{KEY_DEVICE_SIGNAL}.sinr": dict(
|
||||
name="SINR",
|
||||
device_class=DEVICE_CLASS_SIGNAL_STRENGTH,
|
||||
# http://www.lte-anbieter.info/technik/sinr.php
|
||||
|
@ -104,11 +116,11 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
|||
|
||||
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
|
||||
"""Set up Huawei LTE sensor devices."""
|
||||
data = hass.data[DATA_KEY].get_data(config)
|
||||
data = hass.data[DOMAIN].get_data(config)
|
||||
sensors = []
|
||||
for path in config.get(CONF_MONITORED_CONDITIONS):
|
||||
if path == "traffic_statistics": # backwards compatibility
|
||||
path = "monitoring_traffic_statistics"
|
||||
path = KEY_MONITORING_TRAFFIC_STATISTICS
|
||||
data.subscribe(path)
|
||||
sensors.append(HuaweiLteSensor(data, path, SENSOR_META.get(path, {})))
|
||||
|
||||
|
@ -119,7 +131,7 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
|
|||
# *_d.e.v.i.c.e._.s.i.g.n.a.l...s.i.n.r
|
||||
entreg = await entity_registry.async_get_registry(hass)
|
||||
for entid, ent in entreg.entities.items():
|
||||
if ent.platform != "huawei_lte":
|
||||
if ent.platform != DOMAIN:
|
||||
continue
|
||||
for sensor in sensors:
|
||||
oldsuf = ".".join(sensor.path)
|
||||
|
@ -169,7 +181,7 @@ class HuaweiLteSensor(Entity):
|
|||
def name(self) -> str:
|
||||
"""Return sensor name."""
|
||||
try:
|
||||
dname = self.data["device_information.DeviceName"]
|
||||
dname = self.data[f"{KEY_DEVICE_INFORMATION}.DeviceName"]
|
||||
except KeyError:
|
||||
dname = None
|
||||
vname = self.meta.get("name", self.path)
|
||||
|
|
|
@ -643,7 +643,7 @@ horimote==0.4.1
|
|||
httplib2==0.10.3
|
||||
|
||||
# homeassistant.components.huawei_lte
|
||||
huawei-lte-api==1.2.0
|
||||
huawei-lte-api==1.3.0
|
||||
|
||||
# homeassistant.components.hydrawise
|
||||
hydrawiser==0.1.1
|
||||
|
|
|
@ -189,7 +189,7 @@ homematicip==0.10.10
|
|||
httplib2==0.10.3
|
||||
|
||||
# homeassistant.components.huawei_lte
|
||||
huawei-lte-api==1.2.0
|
||||
huawei-lte-api==1.3.0
|
||||
|
||||
# homeassistant.components.influxdb
|
||||
influxdb==5.2.0
|
||||
|
|
|
@ -4,6 +4,7 @@ from unittest.mock import Mock
|
|||
import pytest
|
||||
|
||||
from homeassistant.components import huawei_lte
|
||||
from homeassistant.components.huawei_lte.const import KEY_DEVICE_INFORMATION
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
|
@ -23,25 +24,25 @@ async def test_routerdata_get_nonexistent_root(routerdata):
|
|||
async def test_routerdata_get_nonexistent_leaf(routerdata):
|
||||
"""Test that accessing a nonexistent leaf element raises KeyError."""
|
||||
with pytest.raises(KeyError):
|
||||
routerdata["device_information.foo"]
|
||||
routerdata[f"{KEY_DEVICE_INFORMATION}.foo"]
|
||||
|
||||
|
||||
async def test_routerdata_get_nonexistent_leaf_path(routerdata):
|
||||
"""Test that accessing a nonexistent long path raises KeyError."""
|
||||
with pytest.raises(KeyError):
|
||||
routerdata["device_information.long.path.foo"]
|
||||
routerdata[f"{KEY_DEVICE_INFORMATION}.long.path.foo"]
|
||||
|
||||
|
||||
async def test_routerdata_get_simple(routerdata):
|
||||
"""Test that accessing a short, simple path works."""
|
||||
assert routerdata["device_information.SoftwareVersion"] == "1.0"
|
||||
assert routerdata[f"{KEY_DEVICE_INFORMATION}.SoftwareVersion"] == "1.0"
|
||||
|
||||
|
||||
async def test_routerdata_get_longer(routerdata):
|
||||
"""Test that accessing a longer path works."""
|
||||
assert routerdata["device_information.nested.foo"] == "bar"
|
||||
assert routerdata[f"{KEY_DEVICE_INFORMATION}.nested.foo"] == "bar"
|
||||
|
||||
|
||||
async def test_routerdata_get_dict(routerdata):
|
||||
"""Test that returning an intermediate dict works."""
|
||||
assert routerdata["device_information.nested"] == {"foo": "bar"}
|
||||
assert routerdata[f"{KEY_DEVICE_INFORMATION}.nested"] == {"foo": "bar"}
|
||||
|
|
Loading…
Reference in New Issue