214 lines
8.4 KiB
Python
214 lines
8.4 KiB
Python
"""Switch platform for AdGuard Control Hub integration."""
|
|
import logging
|
|
from typing import Any, Optional
|
|
|
|
from homeassistant.components.switch import SwitchEntity, SwitchDeviceClass
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers.entity import EntityCategory
|
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
|
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
|
|
|
from . import AdGuardControlHubCoordinator
|
|
from .api import AdGuardHomeAPI, AdGuardHomeError
|
|
from .const import DOMAIN, ICON_PROTECTION, ICON_PROTECTION_OFF, ICON_CLIENT, MANUFACTURER
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
config_entry: ConfigEntry,
|
|
async_add_entities: AddEntitiesCallback,
|
|
) -> None:
|
|
"""Set up AdGuard Control Hub switch platform."""
|
|
coordinator = hass.data[DOMAIN][config_entry.entry_id]["coordinator"]
|
|
api = hass.data[DOMAIN][config_entry.entry_id]["api"]
|
|
|
|
entities = [AdGuardProtectionSwitch(coordinator, api)]
|
|
|
|
# Add client switches if clients exist
|
|
for client_name in coordinator.clients.keys():
|
|
entities.append(AdGuardClientSwitch(coordinator, api, client_name))
|
|
|
|
async_add_entities(entities, update_before_add=True)
|
|
|
|
|
|
class AdGuardBaseSwitch(CoordinatorEntity, SwitchEntity):
|
|
"""Base class for AdGuard switches."""
|
|
|
|
def __init__(self, coordinator: AdGuardControlHubCoordinator, api: AdGuardHomeAPI) -> None:
|
|
"""Initialize the switch."""
|
|
super().__init__(coordinator)
|
|
self.api = api
|
|
self._attr_device_info = {
|
|
"identifiers": {(DOMAIN, f"{api.host}:{api.port}")},
|
|
"name": f"AdGuard Control Hub ({api.host})",
|
|
"manufacturer": MANUFACTURER,
|
|
"model": "AdGuard Home",
|
|
"configuration_url": f"{'https' if api.ssl else 'http'}://{api.host}:{api.port}",
|
|
}
|
|
|
|
@property
|
|
def available(self) -> bool:
|
|
"""Return if switch is available."""
|
|
return self.coordinator.last_update_success
|
|
|
|
|
|
class AdGuardProtectionSwitch(AdGuardBaseSwitch):
|
|
"""Switch to control global AdGuard protection."""
|
|
|
|
def __init__(self, coordinator: AdGuardControlHubCoordinator, api: AdGuardHomeAPI) -> None:
|
|
"""Initialize the switch."""
|
|
super().__init__(coordinator, api)
|
|
self._attr_unique_id = f"{api.host}_{api.port}_protection"
|
|
self._attr_name = "AdGuard Protection"
|
|
self._attr_device_class = SwitchDeviceClass.SWITCH
|
|
self._attr_entity_category = EntityCategory.CONFIG
|
|
|
|
@property
|
|
def is_on(self) -> Optional[bool]:
|
|
"""Return true if protection is enabled."""
|
|
return self.coordinator.protection_status.get("protection_enabled", False)
|
|
|
|
@property
|
|
def icon(self) -> str:
|
|
"""Return the icon for the switch."""
|
|
return ICON_PROTECTION if self.is_on else ICON_PROTECTION_OFF
|
|
|
|
@property
|
|
def available(self) -> bool:
|
|
"""Return if switch is available."""
|
|
return self.coordinator.last_update_success and bool(self.coordinator.protection_status)
|
|
|
|
@property
|
|
def extra_state_attributes(self) -> dict[str, Any]:
|
|
"""Return additional state attributes."""
|
|
status = self.coordinator.protection_status
|
|
return {
|
|
"dns_port": status.get("dns_port", "N/A"),
|
|
"version": status.get("version", "N/A"),
|
|
"running": status.get("running", False),
|
|
"dns_addresses": status.get("dns_addresses", []),
|
|
}
|
|
|
|
async def async_turn_on(self, **kwargs: Any) -> None:
|
|
"""Turn on AdGuard protection."""
|
|
try:
|
|
await self.api.set_protection(True)
|
|
await self.coordinator.async_request_refresh()
|
|
_LOGGER.info("AdGuard protection enabled")
|
|
except AdGuardHomeError as err:
|
|
_LOGGER.error("Failed to enable AdGuard protection: %s", err)
|
|
raise
|
|
except Exception as err:
|
|
_LOGGER.exception("Unexpected error enabling AdGuard protection")
|
|
raise
|
|
|
|
async def async_turn_off(self, **kwargs: Any) -> None:
|
|
"""Turn off AdGuard protection."""
|
|
try:
|
|
await self.api.set_protection(False)
|
|
await self.coordinator.async_request_refresh()
|
|
_LOGGER.warning("AdGuard protection disabled")
|
|
except AdGuardHomeError as err:
|
|
_LOGGER.error("Failed to disable AdGuard protection: %s", err)
|
|
raise
|
|
except Exception as err:
|
|
_LOGGER.exception("Unexpected error disabling AdGuard protection")
|
|
raise
|
|
|
|
|
|
class AdGuardClientSwitch(AdGuardBaseSwitch):
|
|
"""Switch to control client-specific protection."""
|
|
|
|
def __init__(
|
|
self,
|
|
coordinator: AdGuardControlHubCoordinator,
|
|
api: AdGuardHomeAPI,
|
|
client_name: str,
|
|
) -> None:
|
|
"""Initialize the switch."""
|
|
super().__init__(coordinator, api)
|
|
self.client_name = client_name
|
|
self._attr_unique_id = f"{api.host}_{api.port}_client_{client_name}"
|
|
self._attr_name = f"AdGuard {client_name}"
|
|
self._attr_icon = ICON_CLIENT
|
|
self._attr_device_class = SwitchDeviceClass.SWITCH
|
|
self._attr_entity_category = EntityCategory.CONFIG
|
|
|
|
@property
|
|
def is_on(self) -> Optional[bool]:
|
|
"""Return true if client protection is enabled."""
|
|
client = self.coordinator.clients.get(self.client_name, {})
|
|
return client.get("filtering_enabled", True)
|
|
|
|
@property
|
|
def available(self) -> bool:
|
|
"""Return if switch is available."""
|
|
return (
|
|
self.coordinator.last_update_success
|
|
and self.client_name in self.coordinator.clients
|
|
)
|
|
|
|
@property
|
|
def extra_state_attributes(self) -> dict[str, Any]:
|
|
"""Return additional state attributes."""
|
|
client = self.coordinator.clients.get(self.client_name, {})
|
|
blocked_services = client.get("blocked_services", {})
|
|
if isinstance(blocked_services, dict):
|
|
blocked_list = blocked_services.get("ids", [])
|
|
else:
|
|
blocked_list = blocked_services or []
|
|
|
|
return {
|
|
"client_ids": client.get("ids", []),
|
|
"safebrowsing_enabled": client.get("safebrowsing_enabled", False),
|
|
"parental_enabled": client.get("parental_enabled", False),
|
|
"safesearch_enabled": client.get("safesearch_enabled", False),
|
|
"blocked_services_count": len(blocked_list),
|
|
"blocked_services": blocked_list,
|
|
}
|
|
|
|
async def async_turn_on(self, **kwargs: Any) -> None:
|
|
"""Enable protection for this client."""
|
|
try:
|
|
client = await self.api.get_client_by_name(self.client_name)
|
|
if client:
|
|
update_data = {
|
|
"name": self.client_name,
|
|
"data": {**client, "filtering_enabled": True}
|
|
}
|
|
await self.api.update_client(update_data)
|
|
await self.coordinator.async_request_refresh()
|
|
_LOGGER.info("Enabled protection for client: %s", self.client_name)
|
|
else:
|
|
_LOGGER.error("Client not found: %s", self.client_name)
|
|
except AdGuardHomeError as err:
|
|
_LOGGER.error("Failed to enable protection for %s: %s", self.client_name, err)
|
|
raise
|
|
except Exception as err:
|
|
_LOGGER.exception("Unexpected error enabling protection for %s", self.client_name)
|
|
raise
|
|
|
|
async def async_turn_off(self, **kwargs: Any) -> None:
|
|
"""Disable protection for this client."""
|
|
try:
|
|
client = await self.api.get_client_by_name(self.client_name)
|
|
if client:
|
|
update_data = {
|
|
"name": self.client_name,
|
|
"data": {**client, "filtering_enabled": False}
|
|
}
|
|
await self.api.update_client(update_data)
|
|
await self.coordinator.async_request_refresh()
|
|
_LOGGER.info("Disabled protection for client: %s", self.client_name)
|
|
else:
|
|
_LOGGER.error("Client not found: %s", self.client_name)
|
|
except AdGuardHomeError as err:
|
|
_LOGGER.error("Failed to disable protection for %s: %s", self.client_name, err)
|
|
raise
|
|
except Exception as err:
|
|
_LOGGER.exception("Unexpected error disabling protection for %s", self.client_name)
|
|
raise
|