coordinator.py

Go to the documentation of this file.

"""Data update coordinator for the Webmin integration."""
 
from __future__ import annotations
 
from typing import Any
 
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_CONNECTIONS, ATTR_IDENTIFIERS, CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import (
    CONNECTION_NETWORK_MAC,
    DeviceInfo,
    format_mac,
)
from homeassistant.helpers.entity_component import DEFAULT_SCAN_INTERVAL
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
from .const import DOMAIN, LOGGER
from .helpers import get_instance_from_options, get_sorted_mac_addresses
 
 
class WebminUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
    """The Webmin data update coordinator."""
 
    mac_address: str
    unique_id: str
 
    def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
        """Initialize the Webmin data update coordinator."""
 
        super().__init__(
            hass, logger=LOGGER, name=DOMAIN, update_interval=DEFAULT_SCAN_INTERVAL
        )
 
        self.instance, base_url = get_instance_from_options(hass, config_entry.options)
 
        self.device_infodevice_info = DeviceInfo(
            configuration_url=base_url,
            name=config_entry.options[CONF_HOST],
        )
 
    async def async_setup(self) -> None:
        """Provide needed data to the device info."""
        mac_addresses = get_sorted_mac_addresses(self.datadata)
        if len(mac_addresses) > 0:
            self.mac_addressmac_address = mac_addresses[0]
            self.unique_idunique_id = self.mac_addressmac_address
            self.device_infodevice_info[ATTR_CONNECTIONS] = {
                (CONNECTION_NETWORK_MAC, format_mac(mac_address))
                for mac_address in mac_addresses
            }
            self.device_infodevice_info[ATTR_IDENTIFIERS] = {
                (DOMAIN, format_mac(mac_address)) for mac_address in mac_addresses
            }
        else:
            assert self.config_entryconfig_entry
            self.unique_idunique_id = self.config_entryconfig_entry.entry_id
 
    async def _async_update_data(self) -> dict[str, Any]:
        data = await self.instance.update()
        data["disk_fs"] = {item["dir"]: item for item in data["disk_fs"]}
        return data