__init__.py

Go to the documentation of this file.

"""The Vilfo Router integration."""
 
from datetime import timedelta
import logging
 
from vilfo import Client as VilfoClient
from vilfo.exceptions import VilfoException
 
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_HOST, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.util import Throttle
 
from .const import ATTR_BOOT_TIME, ATTR_LOAD, DOMAIN, ROUTER_DEFAULT_HOST
 
PLATFORMS = [Platform.SENSOR]
 
DEFAULT_SCAN_INTERVAL = timedelta(seconds=30)
 
_LOGGER = logging.getLogger(__name__)
 
 
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
    """Set up Vilfo Router from a config entry."""
    host = entry.data[CONF_HOST]
    access_token = entry.data[CONF_ACCESS_TOKEN]
 
    vilfo_router = VilfoRouterData(hass, host, access_token)
 
    await vilfo_router.async_update()
 
    if not vilfo_router.available:
        raise ConfigEntryNotReady
 
    hass.data.setdefault(DOMAIN, {})
    hass.data[DOMAIN][entry.entry_id] = vilfo_router
 
    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
 
    return True
 
 
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
    """Unload a config entry."""
    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
    if unload_ok:
        hass.data[DOMAIN].pop(entry.entry_id)
 
    return unload_ok
 
 
class VilfoRouterData:
    """Define an object to hold sensor data."""
 
    def __init__(self, hass, host, access_token):
        """Initialize."""
        self._vilfo_vilfo = VilfoClient(host, access_token)
        self.hasshass = hass
        self.hosthost = host
        self.availableavailable = False
        self.firmware_versionfirmware_version = None
        self.mac_addressmac_address = self._vilfo_vilfo.mac
        self.datadata = {}
        self._unavailable_logged_unavailable_logged = False
 
    @property
    def unique_id(self):
        """Get the unique_id for the Vilfo Router."""
        if self.mac_addressmac_address:
            return self.mac_addressmac_address
 
        if self.hosthost == ROUTER_DEFAULT_HOST:
            return self.hosthost
 
        return self.hosthost
 
    def _fetch_data(self):
        board_information = self._vilfo_vilfo.get_board_information()
        load = self._vilfo_vilfo.get_load()
 
        return {
            "board_information": board_information,
            "load": load,
        }
 
    @Throttle(DEFAULT_SCAN_INTERVAL)
    async def async_update(self):
        """Update data using calls to VilfoClient library."""
        try:
            data = await self.hasshass.async_add_executor_job(self._fetch_data_fetch_data)
 
            self.firmware_versionfirmware_version = data["board_information"]["version"]
            self.datadata[ATTR_BOOT_TIME] = data["board_information"]["bootTime"]
            self.datadata[ATTR_LOAD] = data["load"]
 
            self.availableavailable = True
        except VilfoException as error:
            if not self._unavailable_logged_unavailable_logged:
                _LOGGER.error(
                    "Could not fetch data from %s, error: %s", self.hosthost, error
                )
                self._unavailable_logged_unavailable_logged = True
            self.availableavailable = False
            return
 
        if self.availableavailable and self._unavailable_logged_unavailable_logged:
            _LOGGER.warning("Vilfo Router %s is available again", self.hosthost)
            self._unavailable_logged_unavailable_logged = False