Add Bridge module to AsusWRT (#84152)
* Add Bridge module to AsusWRT * Requested changes * Requested changes * Requested changes * Add check on router attributes value
This commit is contained in:
ollo69
GitHub
parent
c81b6255c2
commit
8108a0f947
7 changed files with 332 additions and 214 deletions
273
homeassistant/components/asuswrt/bridge.py
Normal file
273
homeassistant/components/asuswrt/bridge.py
Normal file
|
|
@ -0,0 +1,273 @@
|
|||
"""aioasuswrt and pyasuswrt bridge classes."""
|
||||
from __future__ import annotations
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from collections import namedtuple
|
||||
import logging
|
||||
from typing import Any, cast
|
||||
|
||||
from aioasuswrt.asuswrt import AsusWrt as AsusWrtLegacy
|
||||
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_MODE,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_PROTOCOL,
|
||||
CONF_USERNAME,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.device_registry import format_mac
|
||||
from homeassistant.helpers.update_coordinator import UpdateFailed
|
||||
|
||||
from .const import (
|
||||
CONF_DNSMASQ,
|
||||
CONF_INTERFACE,
|
||||
CONF_REQUIRE_IP,
|
||||
CONF_SSH_KEY,
|
||||
DEFAULT_DNSMASQ,
|
||||
DEFAULT_INTERFACE,
|
||||
KEY_METHOD,
|
||||
KEY_SENSORS,
|
||||
PROTOCOL_TELNET,
|
||||
SENSORS_BYTES,
|
||||
SENSORS_LOAD_AVG,
|
||||
SENSORS_RATES,
|
||||
SENSORS_TEMPERATURES,
|
||||
)
|
||||
|
||||
SENSORS_TYPE_BYTES = "sensors_bytes"
|
||||
SENSORS_TYPE_COUNT = "sensors_count"
|
||||
SENSORS_TYPE_LOAD_AVG = "sensors_load_avg"
|
||||
SENSORS_TYPE_RATES = "sensors_rates"
|
||||
SENSORS_TYPE_TEMPERATURES = "sensors_temperatures"
|
||||
|
||||
WrtDevice = namedtuple("WrtDevice", ["ip", "name", "connected_to"])
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _get_dict(keys: list, values: list) -> dict[str, Any]:
|
||||
"""Create a dict from a list of keys and values."""
|
||||
return dict(zip(keys, values))
|
||||
|
||||
|
||||
class AsusWrtBridge(ABC):
|
||||
"""The Base Bridge abstract class."""
|
||||
|
||||
@staticmethod
|
||||
def get_bridge(
|
||||
hass: HomeAssistant, conf: dict[str, Any], options: dict[str, Any] | None = None
|
||||
) -> AsusWrtBridge:
|
||||
"""Get Bridge instance."""
|
||||
return AsusWrtLegacyBridge(conf, options)
|
||||
|
||||
def __init__(self, host: str) -> None:
|
||||
"""Initialize Bridge."""
|
||||
self._host = host
|
||||
self._firmware: str | None = None
|
||||
self._label_mac: str | None = None
|
||||
self._model: str | None = None
|
||||
|
||||
@property
|
||||
def host(self) -> str:
|
||||
"""Return hostname."""
|
||||
return self._host
|
||||
|
||||
@property
|
||||
def firmware(self) -> str | None:
|
||||
"""Return firmware information."""
|
||||
return self._firmware
|
||||
|
||||
@property
|
||||
def label_mac(self) -> str | None:
|
||||
"""Return label mac information."""
|
||||
return self._label_mac
|
||||
|
||||
@property
|
||||
def model(self) -> str | None:
|
||||
"""Return model information."""
|
||||
return self._model
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def is_connected(self) -> bool:
|
||||
"""Get connected status."""
|
||||
|
||||
@abstractmethod
|
||||
async def async_connect(self) -> None:
|
||||
"""Connect to the device."""
|
||||
|
||||
@abstractmethod
|
||||
async def async_disconnect(self) -> None:
|
||||
"""Disconnect to the device."""
|
||||
|
||||
@abstractmethod
|
||||
async def async_get_connected_devices(self) -> dict[str, WrtDevice]:
|
||||
"""Get list of connected devices."""
|
||||
|
||||
@abstractmethod
|
||||
async def async_get_available_sensors(self) -> dict[str, dict[str, Any]]:
|
||||
"""Return a dictionary of available sensors for this bridge."""
|
||||
|
||||
|
||||
class AsusWrtLegacyBridge(AsusWrtBridge):
|
||||
"""The Bridge that use legacy library."""
|
||||
|
||||
def __init__(
|
||||
self, conf: dict[str, Any], options: dict[str, Any] | None = None
|
||||
) -> None:
|
||||
"""Initialize Bridge."""
|
||||
super().__init__(conf[CONF_HOST])
|
||||
self._protocol: str = conf[CONF_PROTOCOL]
|
||||
self._api: AsusWrtLegacy = self._get_api(conf, options)
|
||||
|
||||
@staticmethod
|
||||
def _get_api(
|
||||
conf: dict[str, Any], options: dict[str, Any] | None = None
|
||||
) -> AsusWrtLegacy:
|
||||
"""Get the AsusWrtLegacy API."""
|
||||
opt = options or {}
|
||||
|
||||
return AsusWrtLegacy(
|
||||
conf[CONF_HOST],
|
||||
conf.get(CONF_PORT),
|
||||
conf[CONF_PROTOCOL] == PROTOCOL_TELNET,
|
||||
conf[CONF_USERNAME],
|
||||
conf.get(CONF_PASSWORD, ""),
|
||||
conf.get(CONF_SSH_KEY, ""),
|
||||
conf[CONF_MODE],
|
||||
opt.get(CONF_REQUIRE_IP, True),
|
||||
interface=opt.get(CONF_INTERFACE, DEFAULT_INTERFACE),
|
||||
dnsmasq=opt.get(CONF_DNSMASQ, DEFAULT_DNSMASQ),
|
||||
)
|
||||
|
||||
@property
|
||||
def is_connected(self) -> bool:
|
||||
"""Get connected status."""
|
||||
return cast(bool, self._api.is_connected)
|
||||
|
||||
async def async_connect(self) -> None:
|
||||
"""Connect to the device."""
|
||||
await self._api.connection.async_connect()
|
||||
|
||||
# get main router properties
|
||||
if self._label_mac is None:
|
||||
await self._get_label_mac()
|
||||
if self._firmware is None:
|
||||
await self._get_firmware()
|
||||
if self._model is None:
|
||||
await self._get_model()
|
||||
|
||||
async def async_disconnect(self) -> None:
|
||||
"""Disconnect to the device."""
|
||||
if self._api is not None and self._protocol == PROTOCOL_TELNET:
|
||||
self._api.connection.disconnect()
|
||||
|
||||
async def async_get_connected_devices(self) -> dict[str, WrtDevice]:
|
||||
"""Get list of connected devices."""
|
||||
try:
|
||||
api_devices = await self._api.async_get_connected_devices()
|
||||
except OSError as exc:
|
||||
raise UpdateFailed(exc) from exc
|
||||
return {
|
||||
format_mac(mac): WrtDevice(dev.ip, dev.name, None)
|
||||
for mac, dev in api_devices.items()
|
||||
}
|
||||
|
||||
async def _get_nvram_info(self, info_type: str) -> dict[str, Any]:
|
||||
"""Get AsusWrt router info from nvram."""
|
||||
info = {}
|
||||
try:
|
||||
info = await self._api.async_get_nvram(info_type)
|
||||
except OSError as exc:
|
||||
_LOGGER.warning(
|
||||
"Error calling method async_get_nvram(%s): %s", info_type, exc
|
||||
)
|
||||
|
||||
return info
|
||||
|
||||
async def _get_label_mac(self) -> None:
|
||||
"""Get label mac information."""
|
||||
label_mac = await self._get_nvram_info("LABEL_MAC")
|
||||
if label_mac and "label_mac" in label_mac:
|
||||
self._label_mac = format_mac(label_mac["label_mac"])
|
||||
|
||||
async def _get_firmware(self) -> None:
|
||||
"""Get firmware information."""
|
||||
firmware = await self._get_nvram_info("FIRMWARE")
|
||||
if firmware and "firmver" in firmware:
|
||||
firmver: str = firmware["firmver"]
|
||||
if "buildno" in firmware:
|
||||
firmver += f" (build {firmware['buildno']})"
|
||||
self._firmware = firmver
|
||||
|
||||
async def _get_model(self) -> None:
|
||||
"""Get model information."""
|
||||
model = await self._get_nvram_info("MODEL")
|
||||
if model and "model" in model:
|
||||
self._model = model["model"]
|
||||
|
||||
async def async_get_available_sensors(self) -> dict[str, dict[str, Any]]:
|
||||
"""Return a dictionary of available sensors for this bridge."""
|
||||
sensors_temperatures = await self._get_available_temperature_sensors()
|
||||
sensors_types = {
|
||||
SENSORS_TYPE_BYTES: {
|
||||
KEY_SENSORS: SENSORS_BYTES,
|
||||
KEY_METHOD: self._get_bytes,
|
||||
},
|
||||
SENSORS_TYPE_LOAD_AVG: {
|
||||
KEY_SENSORS: SENSORS_LOAD_AVG,
|
||||
KEY_METHOD: self._get_load_avg,
|
||||
},
|
||||
SENSORS_TYPE_RATES: {
|
||||
KEY_SENSORS: SENSORS_RATES,
|
||||
KEY_METHOD: self._get_rates,
|
||||
},
|
||||
SENSORS_TYPE_TEMPERATURES: {
|
||||
KEY_SENSORS: sensors_temperatures,
|
||||
KEY_METHOD: self._get_temperatures,
|
||||
},
|
||||
}
|
||||
return sensors_types
|
||||
|
||||
async def _get_available_temperature_sensors(self) -> list[str]:
|
||||
"""Check which temperature information is available on the router."""
|
||||
availability = await self._api.async_find_temperature_commands()
|
||||
return [SENSORS_TEMPERATURES[i] for i in range(3) if availability[i]]
|
||||
|
||||
async def _get_bytes(self) -> dict[str, Any]:
|
||||
"""Fetch byte information from the router."""
|
||||
try:
|
||||
datas = await self._api.async_get_bytes_total()
|
||||
except (IndexError, OSError, ValueError) as exc:
|
||||
raise UpdateFailed(exc) from exc
|
||||
|
||||
return _get_dict(SENSORS_BYTES, datas)
|
||||
|
||||
async def _get_rates(self) -> dict[str, Any]:
|
||||
"""Fetch rates information from the router."""
|
||||
try:
|
||||
rates = await self._api.async_get_current_transfer_rates()
|
||||
except (IndexError, OSError, ValueError) as exc:
|
||||
raise UpdateFailed(exc) from exc
|
||||
|
||||
return _get_dict(SENSORS_RATES, rates)
|
||||
|
||||
async def _get_load_avg(self) -> dict[str, Any]:
|
||||
"""Fetch load average information from the router."""
|
||||
try:
|
||||
avg = await self._api.async_get_loadavg()
|
||||
except (IndexError, OSError, ValueError) as exc:
|
||||
raise UpdateFailed(exc) from exc
|
||||
|
||||
return _get_dict(SENSORS_LOAD_AVG, avg)
|
||||
|
||||
async def _get_temperatures(self) -> dict[str, Any]:
|
||||
"""Fetch temperatures information from the router."""
|
||||
try:
|
||||
temperatures: dict[str, Any] = await self._api.async_get_temperature()
|
||||
except (OSError, ValueError) as exc:
|
||||
raise UpdateFailed(exc) from exc
|
||||
|
||||
return temperatures
|
||||
|
|
@ -25,13 +25,13 @@ from homeassistant.const import (
|
|||
from homeassistant.core import callback
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.device_registry import format_mac
|
||||
from homeassistant.helpers.schema_config_entry_flow import (
|
||||
SchemaCommonFlowHandler,
|
||||
SchemaFlowFormStep,
|
||||
SchemaOptionsFlowHandler,
|
||||
)
|
||||
|
||||
from .bridge import AsusWrtBridge
|
||||
from .const import (
|
||||
CONF_DNSMASQ,
|
||||
CONF_INTERFACE,
|
||||
|
|
@ -47,7 +47,6 @@ from .const import (
|
|||
PROTOCOL_SSH,
|
||||
PROTOCOL_TELNET,
|
||||
)
|
||||
from .router import get_api, get_nvram_info
|
||||
|
||||
LABEL_MAC = "LABEL_MAC"
|
||||
|
||||
|
|
@ -143,16 +142,15 @@ class AsusWrtFlowHandler(ConfigFlow, domain=DOMAIN):
|
|||
errors=errors or {},
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
async def _async_check_connection(
|
||||
user_input: dict[str, Any]
|
||||
self, user_input: dict[str, Any]
|
||||
) -> tuple[str, str | None]:
|
||||
"""Attempt to connect the AsusWrt router."""
|
||||
|
||||
host: str = user_input[CONF_HOST]
|
||||
api = get_api(user_input)
|
||||
api = AsusWrtBridge.get_bridge(self.hass, user_input)
|
||||
try:
|
||||
await api.connection.async_connect()
|
||||
await api.async_connect()
|
||||
|
||||
except OSError:
|
||||
_LOGGER.error("Error connecting to the AsusWrt router at %s", host)
|
||||
|
|
@ -168,14 +166,9 @@ class AsusWrtFlowHandler(ConfigFlow, domain=DOMAIN):
|
|||
_LOGGER.error("Error connecting to the AsusWrt router at %s", host)
|
||||
return RESULT_CONN_ERROR, None
|
||||
|
||||
label_mac = await get_nvram_info(api, LABEL_MAC)
|
||||
conf_protocol = user_input[CONF_PROTOCOL]
|
||||
if conf_protocol == PROTOCOL_TELNET:
|
||||
api.connection.disconnect()
|
||||
unique_id = api.label_mac
|
||||
await api.async_disconnect()
|
||||
|
||||
unique_id = None
|
||||
if label_mac and "label_mac" in label_mac:
|
||||
unique_id = format_mac(label_mac["label_mac"])
|
||||
return RESULT_SUCCESS, unique_id
|
||||
|
||||
async def async_step_user(
|
||||
|
|
|
|||
|
|
@ -13,6 +13,10 @@ DEFAULT_DNSMASQ = "/var/lib/misc"
|
|||
DEFAULT_INTERFACE = "eth0"
|
||||
DEFAULT_TRACK_UNKNOWN = False
|
||||
|
||||
KEY_COORDINATOR = "coordinator"
|
||||
KEY_METHOD = "method"
|
||||
KEY_SENSORS = "sensors"
|
||||
|
||||
MODE_AP = "ap"
|
||||
MODE_ROUTER = "router"
|
||||
|
||||
|
|
|
|||
|
|
@ -6,22 +6,12 @@ from datetime import datetime, timedelta
|
|||
import logging
|
||||
from typing import Any
|
||||
|
||||
from aioasuswrt.asuswrt import AsusWrt, Device as WrtDevice
|
||||
|
||||
from homeassistant.components.device_tracker import (
|
||||
CONF_CONSIDER_HOME,
|
||||
DEFAULT_CONSIDER_HOME,
|
||||
DOMAIN as TRACKER_DOMAIN,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_MODE,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_PROTOCOL,
|
||||
CONF_USERNAME,
|
||||
)
|
||||
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
|
@ -32,55 +22,36 @@ from homeassistant.helpers.event import async_track_time_interval
|
|||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
from homeassistant.util import dt as dt_util
|
||||
|
||||
from .bridge import AsusWrtBridge, WrtDevice
|
||||
from .const import (
|
||||
CONF_DNSMASQ,
|
||||
CONF_INTERFACE,
|
||||
CONF_REQUIRE_IP,
|
||||
CONF_SSH_KEY,
|
||||
CONF_TRACK_UNKNOWN,
|
||||
DEFAULT_DNSMASQ,
|
||||
DEFAULT_INTERFACE,
|
||||
DEFAULT_TRACK_UNKNOWN,
|
||||
DOMAIN,
|
||||
PROTOCOL_TELNET,
|
||||
SENSORS_BYTES,
|
||||
KEY_COORDINATOR,
|
||||
KEY_METHOD,
|
||||
KEY_SENSORS,
|
||||
SENSORS_CONNECTED_DEVICE,
|
||||
SENSORS_LOAD_AVG,
|
||||
SENSORS_RATES,
|
||||
SENSORS_TEMPERATURES,
|
||||
)
|
||||
|
||||
CONF_REQ_RELOAD = [CONF_DNSMASQ, CONF_INTERFACE, CONF_REQUIRE_IP]
|
||||
DEFAULT_NAME = "Asuswrt"
|
||||
|
||||
KEY_COORDINATOR = "coordinator"
|
||||
KEY_SENSORS = "sensors"
|
||||
|
||||
SCAN_INTERVAL = timedelta(seconds=30)
|
||||
|
||||
SENSORS_TYPE_BYTES = "sensors_bytes"
|
||||
SENSORS_TYPE_COUNT = "sensors_count"
|
||||
SENSORS_TYPE_LOAD_AVG = "sensors_load_avg"
|
||||
SENSORS_TYPE_RATES = "sensors_rates"
|
||||
SENSORS_TYPE_TEMPERATURES = "sensors_temperatures"
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _get_dict(keys: list, values: list) -> dict[str, Any]:
|
||||
"""Create a dict from a list of keys and values."""
|
||||
ret_dict: dict[str, Any] = dict.fromkeys(keys)
|
||||
|
||||
for index, key in enumerate(ret_dict):
|
||||
ret_dict[key] = values[index]
|
||||
|
||||
return ret_dict
|
||||
|
||||
|
||||
class AsusWrtSensorDataHandler:
|
||||
"""Data handler for AsusWrt sensor."""
|
||||
|
||||
def __init__(self, hass: HomeAssistant, api: AsusWrt) -> None:
|
||||
def __init__(self, hass: HomeAssistant, api: AsusWrtBridge) -> None:
|
||||
"""Initialize a AsusWrt sensor data handler."""
|
||||
self._hass = hass
|
||||
self._api = api
|
||||
|
|
@ -90,42 +61,6 @@ class AsusWrtSensorDataHandler:
|
|||
"""Return number of connected devices."""
|
||||
return {SENSORS_CONNECTED_DEVICE[0]: self._connected_devices}
|
||||
|
||||
async def _get_bytes(self) -> dict[str, Any]:
|
||||
"""Fetch byte information from the router."""
|
||||
try:
|
||||
datas = await self._api.async_get_bytes_total()
|
||||
except (OSError, ValueError) as exc:
|
||||
raise UpdateFailed(exc) from exc
|
||||
|
||||
return _get_dict(SENSORS_BYTES, datas)
|
||||
|
||||
async def _get_rates(self) -> dict[str, Any]:
|
||||
"""Fetch rates information from the router."""
|
||||
try:
|
||||
rates = await self._api.async_get_current_transfer_rates()
|
||||
except (OSError, ValueError) as exc:
|
||||
raise UpdateFailed(exc) from exc
|
||||
|
||||
return _get_dict(SENSORS_RATES, rates)
|
||||
|
||||
async def _get_load_avg(self) -> dict[str, Any]:
|
||||
"""Fetch load average information from the router."""
|
||||
try:
|
||||
avg = await self._api.async_get_loadavg()
|
||||
except (OSError, ValueError) as exc:
|
||||
raise UpdateFailed(exc) from exc
|
||||
|
||||
return _get_dict(SENSORS_LOAD_AVG, avg)
|
||||
|
||||
async def _get_temperatures(self) -> dict[str, Any]:
|
||||
"""Fetch temperatures information from the router."""
|
||||
try:
|
||||
temperatures: dict[str, Any] = await self._api.async_get_temperature()
|
||||
except (OSError, ValueError) as exc:
|
||||
raise UpdateFailed(exc) from exc
|
||||
|
||||
return temperatures
|
||||
|
||||
def update_device_count(self, conn_devices: int) -> bool:
|
||||
"""Update connected devices attribute."""
|
||||
if self._connected_devices == conn_devices:
|
||||
|
|
@ -134,19 +69,17 @@ class AsusWrtSensorDataHandler:
|
|||
return True
|
||||
|
||||
async def get_coordinator(
|
||||
self, sensor_type: str, should_poll: bool = True
|
||||
self,
|
||||
sensor_type: str,
|
||||
update_method: Callable[[], Any] | None = None,
|
||||
) -> DataUpdateCoordinator:
|
||||
"""Get the coordinator for a specific sensor type."""
|
||||
should_poll = True
|
||||
if sensor_type == SENSORS_TYPE_COUNT:
|
||||
should_poll = False
|
||||
method = self._get_connected_devices
|
||||
elif sensor_type == SENSORS_TYPE_BYTES:
|
||||
method = self._get_bytes
|
||||
elif sensor_type == SENSORS_TYPE_LOAD_AVG:
|
||||
method = self._get_load_avg
|
||||
elif sensor_type == SENSORS_TYPE_RATES:
|
||||
method = self._get_rates
|
||||
elif sensor_type == SENSORS_TYPE_TEMPERATURES:
|
||||
method = self._get_temperatures
|
||||
elif update_method is not None:
|
||||
method = update_method
|
||||
else:
|
||||
raise RuntimeError(f"Invalid sensor type: {sensor_type}")
|
||||
|
||||
|
|
@ -226,12 +159,6 @@ class AsusWrtRouter:
|
|||
self.hass = hass
|
||||
self._entry = entry
|
||||
|
||||
self._api: AsusWrt = None
|
||||
self._protocol: str = entry.data[CONF_PROTOCOL]
|
||||
self._host: str = entry.data[CONF_HOST]
|
||||
self._model: str = "Asus Router"
|
||||
self._sw_v: str | None = None
|
||||
|
||||
self._devices: dict[str, AsusWrtDevInfo] = {}
|
||||
self._connected_devices: int = 0
|
||||
self._connect_error: bool = False
|
||||
|
|
@ -248,26 +175,19 @@ class AsusWrtRouter:
|
|||
}
|
||||
self._options.update(entry.options)
|
||||
|
||||
self._api: AsusWrtBridge = AsusWrtBridge.get_bridge(
|
||||
self.hass, dict(self._entry.data), self._options
|
||||
)
|
||||
|
||||
async def setup(self) -> None:
|
||||
"""Set up a AsusWrt router."""
|
||||
self._api = get_api(dict(self._entry.data), self._options)
|
||||
|
||||
try:
|
||||
await self._api.connection.async_connect()
|
||||
except OSError as exp:
|
||||
raise ConfigEntryNotReady from exp
|
||||
|
||||
await self._api.async_connect()
|
||||
except OSError as exc:
|
||||
raise ConfigEntryNotReady from exc
|
||||
if not self._api.is_connected:
|
||||
raise ConfigEntryNotReady
|
||||
|
||||
# System
|
||||
model = await get_nvram_info(self._api, "MODEL")
|
||||
if model and "model" in model:
|
||||
self._model = model["model"]
|
||||
firmware = await get_nvram_info(self._api, "FIRMWARE")
|
||||
if firmware and "firmver" in firmware and "buildno" in firmware:
|
||||
self._sw_v = f"{firmware['firmver']} (build {firmware['buildno']})"
|
||||
|
||||
# Load tracked entities from registry
|
||||
entity_reg = er.async_get(self.hass)
|
||||
track_entries = er.async_entries_for_config_entry(
|
||||
|
|
@ -312,24 +232,24 @@ class AsusWrtRouter:
|
|||
async def update_devices(self) -> None:
|
||||
"""Update AsusWrt devices tracker."""
|
||||
new_device = False
|
||||
_LOGGER.debug("Checking devices for ASUS router %s", self._host)
|
||||
_LOGGER.debug("Checking devices for ASUS router %s", self.host)
|
||||
try:
|
||||
api_devices = await self._api.async_get_connected_devices()
|
||||
except OSError as exc:
|
||||
wrt_devices = await self._api.async_get_connected_devices()
|
||||
except UpdateFailed as exc:
|
||||
if not self._connect_error:
|
||||
self._connect_error = True
|
||||
_LOGGER.error(
|
||||
"Error connecting to ASUS router %s for device update: %s",
|
||||
self._host,
|
||||
self.host,
|
||||
exc,
|
||||
)
|
||||
return
|
||||
|
||||
if self._connect_error:
|
||||
self._connect_error = False
|
||||
_LOGGER.info("Reconnected to ASUS router %s", self._host)
|
||||
_LOGGER.info("Reconnected to ASUS router %s", self.host)
|
||||
|
||||
self._connected_devices = len(api_devices)
|
||||
self._connected_devices = len(wrt_devices)
|
||||
consider_home: int = self._options.get(
|
||||
CONF_CONSIDER_HOME, DEFAULT_CONSIDER_HOME.total_seconds()
|
||||
)
|
||||
|
|
@ -337,7 +257,6 @@ class AsusWrtRouter:
|
|||
CONF_TRACK_UNKNOWN, DEFAULT_TRACK_UNKNOWN
|
||||
)
|
||||
|
||||
wrt_devices = {format_mac(mac): dev for mac, dev in api_devices.items()}
|
||||
for device_mac, device in self._devices.items():
|
||||
dev_info = wrt_devices.pop(device_mac, None)
|
||||
device.update(dev_info, consider_home)
|
||||
|
|
@ -363,19 +282,14 @@ class AsusWrtRouter:
|
|||
self._sensors_data_handler = AsusWrtSensorDataHandler(self.hass, self._api)
|
||||
self._sensors_data_handler.update_device_count(self._connected_devices)
|
||||
|
||||
sensors_types: dict[str, list[str]] = {
|
||||
SENSORS_TYPE_BYTES: SENSORS_BYTES,
|
||||
SENSORS_TYPE_COUNT: SENSORS_CONNECTED_DEVICE,
|
||||
SENSORS_TYPE_LOAD_AVG: SENSORS_LOAD_AVG,
|
||||
SENSORS_TYPE_RATES: SENSORS_RATES,
|
||||
SENSORS_TYPE_TEMPERATURES: await self._get_available_temperature_sensors(),
|
||||
}
|
||||
sensors_types = await self._api.async_get_available_sensors()
|
||||
sensors_types[SENSORS_TYPE_COUNT] = {KEY_SENSORS: SENSORS_CONNECTED_DEVICE}
|
||||
|
||||
for sensor_type, sensor_names in sensors_types.items():
|
||||
if not sensor_names:
|
||||
for sensor_type, sensor_def in sensors_types.items():
|
||||
if not (sensor_names := sensor_def.get(KEY_SENSORS)):
|
||||
continue
|
||||
coordinator = await self._sensors_data_handler.get_coordinator(
|
||||
sensor_type, sensor_type != SENSORS_TYPE_COUNT
|
||||
sensor_type, update_method=sensor_def.get(KEY_METHOD)
|
||||
)
|
||||
self._sensors_coordinator[sensor_type] = {
|
||||
KEY_COORDINATOR: coordinator,
|
||||
|
|
@ -392,31 +306,10 @@ class AsusWrtRouter:
|
|||
if self._sensors_data_handler.update_device_count(self._connected_devices):
|
||||
await coordinator.async_refresh()
|
||||
|
||||
async def _get_available_temperature_sensors(self) -> list[str]:
|
||||
"""Check which temperature information is available on the router."""
|
||||
try:
|
||||
availability = await self._api.async_find_temperature_commands()
|
||||
available_sensors = [
|
||||
SENSORS_TEMPERATURES[i] for i in range(3) if availability[i]
|
||||
]
|
||||
except Exception as exc: # pylint: disable=broad-except
|
||||
_LOGGER.debug(
|
||||
(
|
||||
"Failed checking temperature sensor availability for ASUS router"
|
||||
" %s. Exception: %s"
|
||||
),
|
||||
self._host,
|
||||
exc,
|
||||
)
|
||||
return []
|
||||
|
||||
return available_sensors
|
||||
|
||||
async def close(self) -> None:
|
||||
"""Close the connection."""
|
||||
if self._api is not None and self._protocol == PROTOCOL_TELNET:
|
||||
self._api.connection.disconnect()
|
||||
self._api = None
|
||||
if self._api is not None:
|
||||
await self._api.async_disconnect()
|
||||
|
||||
for func in self._on_close:
|
||||
func()
|
||||
|
|
@ -443,14 +336,17 @@ class AsusWrtRouter:
|
|||
@property
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Return the device information."""
|
||||
return DeviceInfo(
|
||||
info = DeviceInfo(
|
||||
identifiers={(DOMAIN, self.unique_id or "AsusWRT")},
|
||||
name=self._host,
|
||||
model=self._model,
|
||||
name=self.host,
|
||||
model=self._api.model or "Asus Router",
|
||||
manufacturer="Asus",
|
||||
sw_version=self._sw_v,
|
||||
configuration_url=f"http://{self._host}",
|
||||
configuration_url=f"http://{self.host}",
|
||||
)
|
||||
if self._api.firmware:
|
||||
info["sw_version"] = self._api.firmware
|
||||
|
||||
return info
|
||||
|
||||
@property
|
||||
def signal_device_new(self) -> str:
|
||||
|
|
@ -465,7 +361,7 @@ class AsusWrtRouter:
|
|||
@property
|
||||
def host(self) -> str:
|
||||
"""Return router hostname."""
|
||||
return self._host
|
||||
return self._api.host
|
||||
|
||||
@property
|
||||
def unique_id(self) -> str | None:
|
||||
|
|
@ -475,7 +371,7 @@ class AsusWrtRouter:
|
|||
@property
|
||||
def name(self) -> str:
|
||||
"""Return router name."""
|
||||
return self._host if self.unique_id else DEFAULT_NAME
|
||||
return self.host if self.unique_id else DEFAULT_NAME
|
||||
|
||||
@property
|
||||
def devices(self) -> dict[str, AsusWrtDevInfo]:
|
||||
|
|
@ -486,32 +382,3 @@ class AsusWrtRouter:
|
|||
def sensors_coordinator(self) -> dict[str, Any]:
|
||||
"""Return sensors coordinators."""
|
||||
return self._sensors_coordinator
|
||||
|
||||
|
||||
async def get_nvram_info(api: AsusWrt, info_type: str) -> dict[str, Any]:
|
||||
"""Get AsusWrt router info from nvram."""
|
||||
info = {}
|
||||
try:
|
||||
info = await api.async_get_nvram(info_type)
|
||||
except OSError as exc:
|
||||
_LOGGER.warning("Error calling method async_get_nvram(%s): %s", info_type, exc)
|
||||
|
||||
return info
|
||||
|
||||
|
||||
def get_api(conf: dict[str, Any], options: dict[str, Any] | None = None) -> AsusWrt:
|
||||
"""Get the AsusWrt API."""
|
||||
opt = options or {}
|
||||
|
||||
return AsusWrt(
|
||||
conf[CONF_HOST],
|
||||
conf.get(CONF_PORT),
|
||||
conf[CONF_PROTOCOL] == PROTOCOL_TELNET,
|
||||
conf[CONF_USERNAME],
|
||||
conf.get(CONF_PASSWORD, ""),
|
||||
conf.get(CONF_SSH_KEY, ""),
|
||||
conf[CONF_MODE],
|
||||
opt.get(CONF_REQUIRE_IP, True),
|
||||
interface=opt.get(CONF_INTERFACE, DEFAULT_INTERFACE),
|
||||
dnsmasq=opt.get(CONF_DNSMASQ, DEFAULT_DNSMASQ),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -26,13 +26,15 @@ from homeassistant.helpers.update_coordinator import (
|
|||
from .const import (
|
||||
DATA_ASUSWRT,
|
||||
DOMAIN,
|
||||
KEY_COORDINATOR,
|
||||
KEY_SENSORS,
|
||||
SENSORS_BYTES,
|
||||
SENSORS_CONNECTED_DEVICE,
|
||||
SENSORS_LOAD_AVG,
|
||||
SENSORS_RATES,
|
||||
SENSORS_TEMPERATURES,
|
||||
)
|
||||
from .router import KEY_COORDINATOR, KEY_SENSORS, AsusWrtRouter
|
||||
from .router import AsusWrtRouter
|
||||
|
||||
|
||||
@dataclass
|
||||
|
|
|
|||
|
|
@ -62,7 +62,7 @@ def mock_unique_id_fixture():
|
|||
@pytest.fixture(name="connect")
|
||||
def mock_controller_connect(mock_unique_id):
|
||||
"""Mock a successful connection."""
|
||||
with patch("homeassistant.components.asuswrt.router.AsusWrt") as service_mock:
|
||||
with patch("homeassistant.components.asuswrt.bridge.AsusWrtLegacy") as service_mock:
|
||||
service_mock.return_value.connection.async_connect = AsyncMock()
|
||||
service_mock.return_value.is_connected = True
|
||||
service_mock.return_value.connection.disconnect = Mock()
|
||||
|
|
@ -236,11 +236,12 @@ async def test_on_connect_failed(hass: HomeAssistant, side_effect, error) -> Non
|
|||
)
|
||||
|
||||
with PATCH_GET_HOST, patch(
|
||||
"homeassistant.components.asuswrt.router.AsusWrt"
|
||||
"homeassistant.components.asuswrt.bridge.AsusWrtLegacy"
|
||||
) as asus_wrt:
|
||||
asus_wrt.return_value.connection.async_connect = AsyncMock(
|
||||
side_effect=side_effect
|
||||
)
|
||||
asus_wrt.return_value.async_get_nvram = AsyncMock(return_value={})
|
||||
asus_wrt.return_value.is_connected = False
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
|
|
|
|||
|
|
@ -32,7 +32,7 @@ from homeassistant.util.dt import utcnow
|
|||
|
||||
from tests.common import MockConfigEntry, async_fire_time_changed
|
||||
|
||||
ASUSWRT_LIB = "homeassistant.components.asuswrt.router.AsusWrt"
|
||||
ASUSWRT_LIB = "homeassistant.components.asuswrt.bridge.AsusWrtLegacy"
|
||||
|
||||
HOST = "myrouter.asuswrt.com"
|
||||
IP_ADDRESS = "192.168.1.1"
|
||||
|
|
@ -311,28 +311,6 @@ async def test_loadavg_sensors(
|
|||
assert hass.states.get(f"{sensor_prefix}_load_avg_15m").state == "1.3"
|
||||
|
||||
|
||||
async def test_temperature_sensors_fail(
|
||||
hass: HomeAssistant,
|
||||
connect,
|
||||
mock_available_temps,
|
||||
) -> None:
|
||||
"""Test fail creating AsusWRT temperature sensors."""
|
||||
config_entry, sensor_prefix = _setup_entry(hass, CONFIG_DATA, SENSORS_TEMP)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
# Only length of 3 booleans is valid. Checking the exception handling.
|
||||
mock_available_temps.pop(2)
|
||||
|
||||
# initial devices setup
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# assert temperature availability exception is handled correctly
|
||||
assert not hass.states.get(f"{sensor_prefix}_2_4ghz_temperature")
|
||||
assert not hass.states.get(f"{sensor_prefix}_5ghz_temperature")
|
||||
assert not hass.states.get(f"{sensor_prefix}_cpu_temperature")
|
||||
|
||||
|
||||
async def test_temperature_sensors(
|
||||
hass: HomeAssistant,
|
||||
connect,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue