Add Huawei LTE router platform, device tracker, and sensor (#16498)

* Add Huawei LTE router platform, device tracker, and sensor

* Add myself to CODEOWNERS for huawei_lte

.coveragerc

@@ -141,6 +141,9 @@ omit =
     homeassistant/components/homematicip_cloud.py
     homeassistant/components/*/homematicip_cloud.py
 
+    homeassistant/components/huawei_lte.py
+    homeassistant/components/*/huawei_lte.py
+
     homeassistant/components/hydrawise.py
     homeassistant/components/*/hydrawise.py
 

CODEOWNERS

@@ -97,6 +97,8 @@ homeassistant/components/*/eight_sleep.py @mezz64
 homeassistant/components/hive.py @Rendili @KJonline
 homeassistant/components/*/hive.py @Rendili @KJonline
 homeassistant/components/homekit/* @cdce8p
+homeassistant/components/huawei_lte.py @scop
+homeassistant/components/*/huawei_lte.py @scop
 homeassistant/components/knx.py @Julius2342
 homeassistant/components/*/knx.py @Julius2342
 homeassistant/components/konnected.py @heythisisnate

homeassistant/components/device_tracker/huawei_lte.py

@@ -0,0 +1,65 @@
+"""
+Support for Huawei LTE routers.
+
+For more details about this platform, please refer to the documentation at
+https://home-assistant.io/components/device_tracker.huawei_lte/
+"""
+from typing import Any, Dict, List, Optional
+
+import attr
+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 ..huawei_lte import DATA_KEY, RouterData
+
+
+DEPENDENCIES = ['huawei_lte']
+
+PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
+    vol.Optional(CONF_URL): cv.url,
+})
+
+
+def get_scanner(hass, config):
+    """Get a Huawei LTE router scanner."""
+    data = hass.data[DATA_KEY].get_data(config)
+    return HuaweiLteScanner(data)
+
+
+@attr.s
+class HuaweiLteScanner(DeviceScanner):
+    """Huawei LTE router scanner."""
+
+    data = attr.ib(type=RouterData)
+
+    _hosts = attr.ib(init=False, factory=dict)
+
+    def scan_devices(self) -> List[str]:
+        """Scan for devices."""
+        self.data.update()
+        self._hosts = {
+            x["MacAddress"]: x
+            for x in self.data["wlan_host_list.Hosts.Host"]
+            if x.get("MacAddress")
+        }
+        return list(self._hosts)
+
+    def get_device_name(self, device: str) -> Optional[str]:
+        """Get name for a device."""
+        host = self._hosts.get(device)
+        return host.get("HostName") or None if host else None
+
+    def get_extra_attributes(self, device: str) -> Dict[str, Any]:
+        """
+        Get extra attributes of a device.
+
+        Some known extra attributes that may be returned in the dict
+        include MacAddress (MAC address), ID (client ID), IpAddress
+        (IP address), AssociatedSsid (associated SSID), AssociatedTime
+        (associated time in seconds), and HostName (host name).
+        """
+        return self._hosts.get(device) or {}

homeassistant/components/huawei_lte.py

@@ -0,0 +1,123 @@
+"""
+Support for Huawei LTE routers.
+
+For more details about this component, please refer to the documentation at
+https://home-assistant.io/components/huawei_lte/
+"""
+from datetime import timedelta
+from functools import reduce
+import logging
+import operator
+
+import voluptuous as vol
+import attr
+
+from homeassistant.const import (
+    CONF_URL, CONF_USERNAME, CONF_PASSWORD, EVENT_HOMEASSISTANT_STOP,
+)
+from homeassistant.helpers import config_validation as cv
+from homeassistant.util import Throttle
+
+
+_LOGGER = logging.getLogger(__name__)
+
+REQUIREMENTS = ['huawei-lte-api==1.0.12']
+
+MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=10)
+
+DOMAIN = 'huawei_lte'
+DATA_KEY = 'huawei_lte'
+
+CONFIG_SCHEMA = vol.Schema({
+    DOMAIN: vol.All(cv.ensure_list, [vol.Schema({
+        vol.Required(CONF_URL): cv.url,
+        vol.Required(CONF_USERNAME): cv.string,
+        vol.Required(CONF_PASSWORD): cv.string,
+    })])
+}, extra=vol.ALLOW_EXTRA)
+
+
+@attr.s
+class RouterData:
+    """Class for router state."""
+
+    client = attr.ib()
+    device_information = attr.ib(init=False, factory=dict)
+    device_signal = attr.ib(init=False, factory=dict)
+    traffic_statistics = attr.ib(init=False, factory=dict)
+    wlan_host_list = attr.ib(init=False, factory=dict)
+
+    def __getitem__(self, path: str):
+        """
+        Get value corresponding to a dotted path.
+
+        The first path component designates a member of this class
+        such as device_information, device_signal etc, and the remaining
+        path points to a value in the member's data structure.
+        """
+        cat, *path_ = path.split(".")
+        return reduce(operator.getitem, path_, getattr(self, cat))
+
+    @Throttle(MIN_TIME_BETWEEN_UPDATES)
+    def update(self) -> None:
+        """Call API to update data."""
+        self.device_information = self.client.device.information()
+        _LOGGER.debug("device_information=%s", self.device_information)
+        self.device_signal = self.client.device.signal()
+        _LOGGER.debug("device_signal=%s", self.device_signal)
+        self.traffic_statistics = self.client.monitoring.traffic_statistics()
+        _LOGGER.debug("traffic_statistics=%s", self.traffic_statistics)
+        self.wlan_host_list = self.client.wlan.host_list()
+        _LOGGER.debug("wlan_host_list=%s", self.wlan_host_list)
+
+
+@attr.s
+class HuaweiLteData:
+    """Shared state."""
+
+    data = attr.ib(init=False, factory=dict)
+
+    def get_data(self, config):
+        """Get the requested or the only data value."""
+        if CONF_URL in config:
+            return self.data.get(config[CONF_URL])
+        if len(self.data) == 1:
+            return next(iter(self.data.values()))
+
+        return None
+
+
+def setup(hass, config) -> bool:
+    """Set up Huawei LTE component."""
+    if DATA_KEY not in hass.data:
+        hass.data[DATA_KEY] = HuaweiLteData()
+    for conf in config.get(DOMAIN, []):
+        _setup_lte(hass, conf)
+    return True
+
+
+def _setup_lte(hass, lte_config) -> None:
+    """Set up Huawei LTE router."""
+    from huawei_lte_api.AuthorizedConnection import AuthorizedConnection
+    from huawei_lte_api.Client import Client
+
+    url = lte_config[CONF_URL]
+    username = lte_config[CONF_USERNAME]
+    password = lte_config[CONF_PASSWORD]
+
+    connection = AuthorizedConnection(
+        url,
+        username=username,
+        password=password,
+    )
+    client = Client(connection)
+
+    data = RouterData(client)
+    data.update()
+    hass.data[DATA_KEY].data[url] = data
+
+    def cleanup(event):
+        """Clean up resources."""
+        client.user.logout()
+
+    hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, cleanup)

homeassistant/components/sensor/huawei_lte.py

@@ -0,0 +1,169 @@
+"""Huawei LTE sensors.
+
+For more details about this platform, please refer to the documentation at
+https://home-assistant.io/components/sensor.huawei_lte/
+"""
+
+import logging
+import re
+
+import attr
+import voluptuous as vol
+
+from homeassistant.const import (
+    CONF_URL, CONF_MONITORED_CONDITIONS, STATE_UNKNOWN,
+)
+from homeassistant.components.sensor import PLATFORM_SCHEMA
+from homeassistant.helpers.entity import Entity
+import homeassistant.helpers.config_validation as cv
+
+from ..huawei_lte import DATA_KEY, RouterData
+
+
+_LOGGER = logging.getLogger(__name__)
+
+DEPENDENCIES = ['huawei_lte']
+
+DEFAULT_NAME_TEMPLATE = 'Huawei {}: {}'
+
+DEFAULT_SENSORS = [
+    "device_information.WanIPAddress",
+    "device_signal.rssi",
+]
+
+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.rsrq": dict(
+        name="RSRQ",
+        # http://www.lte-anbieter.info/technik/rsrq.php
+        icon=lambda x:
+        x >= -5 and "mdi:signal-cellular-3"
+        or x >= -8 and "mdi:signal-cellular-2"
+        or x >= -11 and "mdi:signal-cellular-1"
+        or "mdi:signal-cellular-outline"
+    ),
+    "device_signal.rsrp": dict(
+        name="RSRP",
+        # http://www.lte-anbieter.info/technik/rsrp.php
+        icon=lambda x:
+        x >= -80 and "mdi:signal-cellular-3"
+        or x >= -95 and "mdi:signal-cellular-2"
+        or x >= -110 and "mdi:signal-cellular-1"
+        or "mdi:signal-cellular-outline"
+    ),
+    "device_signal.rssi": dict(
+        name="RSSI",
+        # https://eyesaas.com/wi-fi-signal-strength/
+        icon=lambda x:
+        x >= -60 and "mdi:signal-cellular-3"
+        or x >= -70 and "mdi:signal-cellular-2"
+        or x >= -80 and "mdi:signal-cellular-1"
+        or "mdi:signal-cellular-outline"
+    ),
+    "device_signal.sinr": dict(
+        name="SINR",
+        # http://www.lte-anbieter.info/technik/sinr.php
+        icon=lambda x:
+        x >= 10 and "mdi:signal-cellular-3"
+        or x >= 5 and "mdi:signal-cellular-2"
+        or x >= 0 and "mdi:signal-cellular-1"
+        or "mdi:signal-cellular-outline"
+    ),
+}
+
+PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
+    vol.Optional(CONF_URL): cv.url,
+    vol.Optional(
+        CONF_MONITORED_CONDITIONS, default=DEFAULT_SENSORS): cv.ensure_list,
+})
+
+
+def setup_platform(
+        hass, config, add_entities, discovery_info):
+    """Set up Huawei LTE sensor devices."""
+    data = hass.data[DATA_KEY].get_data(config)
+    sensors = []
+    for path in config.get(CONF_MONITORED_CONDITIONS):
+        sensors.append(HuaweiLteSensor(
+            data, path, SENSOR_META.get(path, {})))
+    add_entities(sensors, True)
+
+
+@attr.s
+class HuaweiLteSensor(Entity):
+    """Huawei LTE sensor entity."""
+
+    data = attr.ib(type=RouterData)
+    path = attr.ib(type=list)
+    meta = attr.ib(type=dict)
+
+    _state = attr.ib(init=False, default=STATE_UNKNOWN)
+    _unit = attr.ib(init=False, type=str)
+
+    @property
+    def unique_id(self) -> str:
+        """Return unique ID for sensor."""
+        return "%s_%s" % (
+            self.path,
+            self.data["device_information.SerialNumber"],
+        )
+
+    @property
+    def name(self) -> str:
+        """Return sensor name."""
+        dname = self.data["device_information.DeviceName"]
+        vname = self.meta.get("name", self.path)
+        return DEFAULT_NAME_TEMPLATE.format(dname, vname)
+
+    @property
+    def state(self):
+        """Return sensor state."""
+        return self._state
+
+    @property
+    def unit_of_measurement(self):
+        """Return sensor's unit of measurement."""
+        return self.meta.get("unit", self._unit)
+
+    @property
+    def icon(self):
+        """Return icon for sensor."""
+        icon = self.meta.get("icon")
+        if callable(icon):
+            return icon(self.state)
+        return icon
+
+    def update(self):
+        """Update state."""
+        self.data.update()
+
+        unit = None
+        try:
+            value = self.data[self.path]
+        except KeyError:
+            _LOGGER.warning("%s not in data", self.path)
+            value = None
+
+        if value is not None:
+            # Clean up value and infer unit, e.g. -71dBm, 15 dB
+            match = re.match(
+                r"(?P<value>.+?)\s*(?P<unit>[a-zA-Z]+)\s*$", str(value))
+            if match:
+                try:
+                    value = float(match.group("value"))
+                    unit = match.group("unit")
+                except ValueError:
+                    pass
+
+        self._state = value
+        self._unit = unit

requirements_all.txt

@@ -463,6 +463,9 @@ homematicip==0.9.8
 # homeassistant.components.remember_the_milk
 httplib2==0.10.3
 
+# homeassistant.components.huawei_lte
+huawei-lte-api==1.0.12
+
 # homeassistant.components.hydrawise
 hydrawiser==0.1.1