router.py

Go to the documentation of this file.

"""The Keenetic Client class."""
 
from __future__ import annotations
 
from collections.abc import Callable
from datetime import timedelta
import logging
 
from ndms2_client import Client, ConnectionException, Device, TelnetConnection
from ndms2_client.client import RouterInfo
 
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
    CONF_HOST,
    CONF_PASSWORD,
    CONF_PORT,
    CONF_SCAN_INTERVAL,
    CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.event import async_call_later
import homeassistant.util.dt as dt_util
 
from .const import (
    CONF_CONSIDER_HOME,
    CONF_INCLUDE_ARP,
    CONF_INCLUDE_ASSOCIATED,
    CONF_INTERFACES,
    CONF_TRY_HOTSPOT,
    DOMAIN,
)
 
_LOGGER = logging.getLogger(__name__)
 
 
class KeeneticRouter:
    """Keenetic client Object."""
 
    def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
        """Initialize the Client."""
        self.hasshass = hass
        self.config_entryconfig_entry = config_entry
        self._last_devices_last_devices: dict[str, Device] = {}
        self._router_info_router_info: RouterInfo | None = None
        self._connection_connection: TelnetConnection | None = None
        self._client_client: Client | None = None
        self._cancel_periodic_update_cancel_periodic_update: Callable | None = None
        self._available_available = False
        self._progress_progress = None
        self._tracked_interfaces_tracked_interfaces = set(config_entry.options[CONF_INTERFACES])
 
    @property
    def client(self):
        """Read-only accessor for the client connection."""
        return self._client_client
 
    @property
    def last_devices(self):
        """Read-only accessor for last_devices."""
        return self._last_devices_last_devices
 
    @property
    def host(self):
        """Return the host of this hub."""
        return self.config_entryconfig_entry.data[CONF_HOST]
 
    @property
    def device_info(self) -> DeviceInfo:
        """Return the host of this hub."""
        return DeviceInfo(
            identifiers={(DOMAIN, f"router-{self.config_entry.entry_id}")},
            manufacturer=self.manufacturermanufacturer,
            model=self.modelmodel,
            name=self.namename,
            sw_version=self.firmwarefirmware,
        )
 
    @property
    def name(self):
        """Return the name of the hub."""
        return self._router_info_router_info.name if self._router_info_router_info else self.hosthost
 
    @property
    def model(self):
        """Return the model of the hub."""
        return self._router_info_router_info.model if self._router_info_router_info else None
 
    @property
    def firmware(self):
        """Return the firmware of the hub."""
        return self._router_info_router_info.fw_version if self._router_info_router_info else None
 
    @property
    def manufacturer(self):
        """Return the firmware of the hub."""
        return self._router_info_router_info.manufacturer if self._router_info_router_info else None
 
    @property
    def available(self):
        """Return if the hub is connected."""
        return self._available_available
 
    @property
    def consider_home_interval(self):
        """Config entry option defining number of seconds from last seen to away."""
        return timedelta(seconds=self.config_entryconfig_entry.options[CONF_CONSIDER_HOME])
 
    @property
    def tracked_interfaces(self):
        """Tracked interfaces."""
        return self._tracked_interfaces_tracked_interfaces
 
    @property
    def signal_update(self):
        """Event specific per router entry to signal updates."""
        return f"keenetic-update-{self.config_entry.entry_id}"
 
    async def request_update(self):
        """Request an update."""
        if self._progress_progress is not None:
            await self._progress_progress
            return
 
        self._progress_progress = self.hasshass.async_create_task(self.async_updateasync_update())
        await self._progress_progress
 
        self._progress_progress = None
 
    async def async_update(self):
        """Update devices information."""
        await self.hasshass.async_add_executor_job(self._update_devices_update_devices)
        async_dispatcher_send(self.hasshass, self.signal_updatesignal_update)
 
    async def async_setup(self):
        """Set up the connection."""
        self._connection_connection = TelnetConnection(
            self.config_entryconfig_entry.data[CONF_HOST],
            self.config_entryconfig_entry.data[CONF_PORT],
            self.config_entryconfig_entry.data[CONF_USERNAME],
            self.config_entryconfig_entry.data[CONF_PASSWORD],
        )
        self._client_client = Client(self._connection_connection)
 
        try:
            await self.hasshass.async_add_executor_job(self._update_router_info_update_router_info)
        except ConnectionException as error:
            raise ConfigEntryNotReady from error
 
        async def async_update_data(_now):
            await self.request_updaterequest_update()
            self._cancel_periodic_update_cancel_periodic_update = async_call_later(
                self.hasshass,
                self.config_entryconfig_entry.options[CONF_SCAN_INTERVAL],
                async_update_data,
            )
 
        await async_update_data(dt_util.utcnow())
 
    async def async_teardown(self):
        """Teardown up the connection."""
        if self._cancel_periodic_update_cancel_periodic_update:
            self._cancel_periodic_update_cancel_periodic_update()
        self._connection_connection.disconnect()
 
    def _update_router_info(self):
        try:
            self._router_info_router_info = self._client_client.get_router_info()
            self._available_available = True
        except Exception:
            self._available_available = False
            raise
 
    def _update_devices(self):
        """Get ARP from keenetic router."""
        _LOGGER.debug("Fetching devices from router")
 
        try:
            _response = self._client_client.get_devices(
                try_hotspot=self.config_entryconfig_entry.options[CONF_TRY_HOTSPOT],
                include_arp=self.config_entryconfig_entry.options[CONF_INCLUDE_ARP],
                include_associated=self.config_entryconfig_entry.options[CONF_INCLUDE_ASSOCIATED],
            )
            self._last_devices_last_devices = {
                dev.mac: dev
                for dev in _response
                if dev.interface in self._tracked_interfaces_tracked_interfaces
            }
            _LOGGER.debug("Successfully fetched data from router: %s", str(_response))
            self._router_info_router_info = self._client_client.get_router_info()
            self._available_available = True
 
        except ConnectionException:
            _LOGGER.error("Error fetching data from router")
            self._available_available = False