101 lines
3.7 KiB
Python
101 lines
3.7 KiB
Python
"""AdGuard Control Hub data update coordinator."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import timedelta
|
|
from typing import Any
|
|
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import ConfigEntryAuthFailed
|
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
|
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
|
|
|
from .api import AdGuardHomeAPI, AdGuardHomeApiError, AdGuardHomeAuthError, AdGuardHomeConnectionError
|
|
from .const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_SSL, CONF_USERNAME, CONF_VERIFY_SSL, DOMAIN
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class AdGuardDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
|
|
"""Class to manage fetching AdGuard Home data from the API."""
|
|
|
|
def __init__(
|
|
self,
|
|
hass: HomeAssistant,
|
|
session,
|
|
config: dict[str, Any],
|
|
update_interval: timedelta,
|
|
) -> None:
|
|
"""Initialize the data update coordinator."""
|
|
super().__init__(
|
|
hass,
|
|
_LOGGER,
|
|
name=DOMAIN,
|
|
update_interval=update_interval,
|
|
)
|
|
|
|
self.api = AdGuardHomeAPI(
|
|
session,
|
|
config[CONF_HOST],
|
|
config[CONF_PORT],
|
|
config.get(CONF_USERNAME),
|
|
config.get(CONF_PASSWORD),
|
|
config.get(CONF_SSL, False),
|
|
config.get(CONF_VERIFY_SSL, True),
|
|
)
|
|
|
|
self._config = config
|
|
|
|
async def _async_update_data(self) -> dict[str, Any]:
|
|
"""Update data via library."""
|
|
try:
|
|
# Get status and stats data
|
|
status_data = await self.api.get_status()
|
|
stats_data = await self.api.get_stats()
|
|
clients_data = await self.api.get_clients()
|
|
|
|
# Combine all data
|
|
data = {
|
|
"status": status_data,
|
|
"stats": stats_data,
|
|
"clients": clients_data,
|
|
}
|
|
|
|
# Calculate blocked percentage
|
|
if "stats" in data and "num_dns_queries" in data["stats"]:
|
|
queries = data["stats"].get("num_dns_queries", 0)
|
|
blocked = data["stats"].get("num_blocked_filtering", 0)
|
|
if queries > 0:
|
|
data["stats"]["blocked_percentage"] = round((blocked / queries) * 100, 2)
|
|
else:
|
|
data["stats"]["blocked_percentage"] = 0.0
|
|
|
|
return data
|
|
|
|
except AdGuardHomeAuthError as err:
|
|
raise ConfigEntryAuthFailed("Authentication failed") from err
|
|
except (AdGuardHomeConnectionError, AdGuardHomeApiError) as err:
|
|
raise UpdateFailed(f"Error communicating with AdGuard Home: {err}") from err
|
|
|
|
async def async_set_switch(self, switch_type: str, state: bool) -> None:
|
|
"""Set switch state."""
|
|
try:
|
|
if switch_type == "protection":
|
|
await self.api.set_protection(state)
|
|
elif switch_type == "filtering":
|
|
await self.api.set_filtering(state)
|
|
elif switch_type == "safebrowsing":
|
|
await self.api.set_safebrowsing(state)
|
|
elif switch_type == "parental":
|
|
await self.api.set_parental_control(state)
|
|
elif switch_type == "safesearch":
|
|
await self.api.set_safe_search(state)
|
|
elif switch_type == "query_log":
|
|
await self.api.set_query_log(state)
|
|
|
|
# Refresh data after changing state
|
|
await self.async_request_refresh()
|
|
|
|
except (AdGuardHomeConnectionError, AdGuardHomeApiError) as err:
|
|
raise UpdateFailed(f"Error setting switch {switch_type}: {err}") from err
|