Technical documentation for developers working with AdGuard Control Hub. This covers the internal API, entity structure, and integration architecture.

📡 AdGuard Home API Integration

Supported API Endpoints

The integration interacts with these AdGuard Home API endpoints:

Endpoint	Method	Purpose
`/control/status`	GET	Server status and configuration
`/control/clients`	GET	List all configured clients
`/control/clients/add`	POST	Add new client
`/control/clients/update`	POST	Update existing client
`/control/clients/delete`	POST	Remove client
`/control/stats`	GET	DNS query statistics
`/control/protection`	POST	Enable/disable protection
`/control/blocked_services/all`	GET	Available blocked services
`/control/blocked_services/get`	GET	Client blocked services
`/control/blocked_services/update`	POST	Update blocked services

API Authentication

# Basic authentication used for all API calls
from aiohttp import BasicAuth

auth = BasicAuth(username, password)
headers = {"Content-Type": "application/json"}

Error Handling

# API wrapper error handling pattern
async def _request(self, method: str, endpoint: str, data: dict = None) -> dict:
    try:
        async with self.session.request(method, url, json=data, headers=headers, auth=auth) as response:
            response.raise_for_status()
            if response.status == 204 or not response.content_length:
                return {}
            return await response.json()
    except aiohttp.ClientError as err:
        _LOGGER.error("API request failed: %s", err)
        raise
    except asyncio.TimeoutError:
        _LOGGER.error("API request timeout")
        raise

🏗️ Integration Architecture

Component Structure

adguard_hub/
├── __init__.py          # Main integration setup
├── api.py              # AdGuard Home API wrapper
├── config_flow.py      # Configuration UI flow
├── const.py            # Constants and configuration
├── switch.py           # Switch platform implementation
├── sensor.py           # Sensor platform (optional)
├── binary_sensor.py    # Binary sensor platform (optional)
├── services.py         # Custom services
└── strings.json        # UI strings and translations

Data Flow

graph TD
    A[Home Assistant] --> B[AdGuard Control Hub]
    B --> C[Update Coordinator]
    C --> D[AdGuard Home API]
    D --> E[AdGuard Home Server]

    C --> F[Switch Platform]
    C --> G[Sensor Platform]
    C --> H[Binary Sensor Platform]

    F --> I[Entity Registry]
    G --> I
    H --> I

Update Coordinator Pattern

class AdGuardControlHubCoordinator(DataUpdateCoordinator):
    """Manage data fetching from AdGuard Home."""

    def __init__(self, hass: HomeAssistant, api: AdGuardHomeAPI):
        super().__init__(
            hass,
            _LOGGER,
            name=f"{DOMAIN}_coordinator",
            update_interval=timedelta(seconds=SCAN_INTERVAL),
        )
        self.api = api

    async def _async_update_data(self):
        """Fetch data from AdGuard Home."""
        try:
            # Concurrent API calls for better performance
            results = await asyncio.gather(
                self.api.get_clients(),
                self.api.get_statistics(),
                self.api.get_status(),
                return_exceptions=True,
            )

            # Process and return structured data
            return self._process_api_results(results)
        except Exception as err:
            raise UpdateFailed(f"Error fetching data: {err}")

🔌 Entity System

Entity Types and Hierarchy

# Base entity class
class AdGuardBaseEntity(CoordinatorEntity):
    """Base class for AdGuard entities."""

    def __init__(self, coordinator: AdGuardControlHubCoordinator, api: AdGuardHomeAPI):
        super().__init__(coordinator)
        self.api = api
        self._attr_device_info = self._create_device_info()

    def _create_device_info(self) -> dict:
        """Create device info for entity grouping."""
        return {
            "identifiers": {(DOMAIN, f"{self.api.host}:{self.api.port}")},
            "name": f"AdGuard Control Hub ({self.api.host})",
            "manufacturer": MANUFACTURER,
            "model": "AdGuard Home",
            "sw_version": self.coordinator.protection_status.get("version"),
        }

Switch Entities

# Global protection switch
class AdGuardProtectionSwitch(AdGuardBaseEntity, SwitchEntity):
    """Control global AdGuard protection."""

    @property
    def unique_id(self) -> str:
        return f"{self.api.host}_{self.api.port}_protection"

    @property 
    def is_on(self) -> bool:
        return self.coordinator.protection_status.get("protection_enabled", False)

    async def async_turn_on(self, **kwargs):
        await self.api.set_protection(True)
        await self.coordinator.async_request_refresh()

# Client-specific switches
class AdGuardClientSwitch(AdGuardBaseEntity, SwitchEntity):
    """Control per-client protection."""

    def __init__(self, coordinator, api, client_name: str):
        super().__init__(coordinator, api)
        self.client_name = client_name

    @property
    def unique_id(self) -> str:
        return f"{self.api.host}_{self.api.port}_client_{self.client_name}"

Sensor Entities

class AdGuardStatsSensor(AdGuardBaseEntity, SensorEntity):
    """DNS statistics sensor."""

    @property
    def native_value(self) -> int:
        return self.coordinator.statistics.get(self.stat_key, 0)

    @property
    def device_class(self) -> str:
        return SensorDeviceClass.TIMESTAMP if "time" in self.stat_key else None

🔧 Custom Services

Service Registration

async def async_register_services(hass: HomeAssistant, api: AdGuardHomeAPI) -> None:
    """Register integration services."""

    async def emergency_unblock_service(call):
        """Handle emergency unblock service."""
        duration = call.data.get("duration", 300)
        clients = call.data.get("clients", ["all"])

        try:
            if "all" in clients:
                await api.set_protection(False)
                # Schedule re-enabling after duration
                hass.async_create_task(_restore_protection_after_delay(duration))
            else:
                # Handle specific clients
                for client_name in clients:
                    await api.temporary_disable_client(client_name, duration)
        except Exception as err:
            _LOGGER.error("Emergency unblock failed: %s", err)
            raise HomeAssistantError(f"Service call failed: {err}")

    # Register service with schema validation
    hass.services.async_register(
        DOMAIN,
        "emergency_unblock",
        emergency_unblock_service,
        schema=EMERGENCY_UNBLOCK_SCHEMA,
    )

Service Schemas

import voluptuous as vol
from homeassistant.helpers import config_validation as cv

# Service parameter validation schemas
EMERGENCY_UNBLOCK_SCHEMA = vol.Schema({
    vol.Optional("duration", default=300): cv.positive_int,
    vol.Optional("clients", default=["all"]): vol.All(cv.ensure_list, [cv.string]),
})

BLOCK_SERVICES_SCHEMA = vol.Schema({
    vol.Required("client_name"): cv.string,
    vol.Required("services"): vol.All(cv.ensure_list, [cv.string]),
    vol.Optional("schedule"): vol.Schema({
        vol.Required("enabled"): cv.boolean,
        vol.Optional("time_zone", default="Local"): cv.string,
        vol.Optional("mon"): TIME_RANGE_SCHEMA,
        vol.Optional("tue"): TIME_RANGE_SCHEMA,
        # ... other days
    }),
})

TIME_RANGE_SCHEMA = vol.Schema({
    vol.Required("start"): cv.time,
    vol.Required("end"): cv.time,
})

📊 Data Models

Client Data Structure

@dataclass
class AdGuardClient:
    """Represent an AdGuard Home client."""
    name: str
    ids: list[str]
    use_global_settings: bool = True
    use_global_blocked_services: bool = True
    blocked_services: list[str] = field(default_factory=list)
    filtering_enabled: bool = True
    parental_enabled: bool = False
    safebrowsing_enabled: bool = False
    safesearch_enabled: bool = False
    upstream_dns: list[str] = field(default_factory=list)

    @classmethod
    def from_api_response(cls, data: dict) -> "AdGuardClient":
        """Create client from API response."""
        return cls(
            name=data.get("name", ""),
            ids=data.get("ids", []),
            use_global_settings=data.get("use_global_settings", True),
            # ... map other fields
        )

    def to_api_request(self) -> dict:
        """Convert to API request format."""
        return {
            "name": self.name,
            "ids": self.ids,
            "use_global_settings": self.use_global_settings,
            # ... map all fields
        }

Statistics Data Structure

@dataclass
class AdGuardStatistics:
    """AdGuard Home DNS statistics."""
    dns_queries: int
    blocked_filtering: int  
    replaced_safebrowsing: int
    replaced_safesearch: int
    replaced_parental: int
    avg_processing_time: float

    @property
    def blocked_percentage(self) -> float:
        """Calculate blocking percentage."""
        if self.dns_queries == 0:
            return 0.0
        return (self.blocked_filtering / self.dns_queries) * 100

🔄 Configuration Flow

Configuration Steps

class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
    """Handle configuration flow."""

    VERSION = 1

    async def async_step_user(self, user_input=None):
        """Handle user input."""
        errors = {}

        if user_input is not None:
            try:
                # Validate connection
                info = await self._validate_input(user_input)
            except CannotConnect:
                errors["base"] = "cannot_connect"
            except InvalidAuth:
                errors["base"] = "invalid_auth"
            else:
                # Create config entry
                return self.async_create_entry(
                    title=info["title"],
                    data=user_input
                )

        return self.async_show_form(
            step_id="user",
            data_schema=self._get_schema(user_input),
            errors=errors
        )

    async def _validate_input(self, data: dict) -> dict:
        """Validate user input and test connection."""
        api = AdGuardHomeAPI(
            host=data[CONF_HOST],
            port=data[CONF_PORT],
            username=data.get(CONF_USERNAME),
            password=data.get(CONF_PASSWORD),
            ssl=data.get(CONF_SSL, False),
            session=async_get_clientsession(self.hass),
        )

        # Test connection
        if not await api.test_connection():
            raise CannotConnect

        # Get server info for title
        status = await api.get_status()
        return {"title": f"AdGuard Control Hub ({data[CONF_HOST]})"}

🧪 Testing Framework

Unit Test Structure

import pytest
from unittest.mock import AsyncMock, MagicMock
from homeassistant.core import HomeAssistant
from custom_components.adguard_hub.api import AdGuardHomeAPI

@pytest.fixture
def mock_api():
    """Mock API for testing."""
    api = MagicMock(spec=AdGuardHomeAPI)
    api.get_status = AsyncMock(return_value={"protection_enabled": True})
    api.get_clients = AsyncMock(return_value={"clients": []})
    return api

async def test_protection_switch_on(hass: HomeAssistant, mock_api):
    """Test protection switch turns on correctly."""
    coordinator = AdGuardControlHubCoordinator(hass, mock_api)
    switch = AdGuardProtectionSwitch(coordinator, mock_api)

    await switch.async_turn_on()

    mock_api.set_protection.assert_called_once_with(True)
    coordinator.async_request_refresh.assert_called_once()

Integration Testing

async def test_integration_setup(hass: HomeAssistant):
    """Test integration setup."""
    config_entry = MockConfigEntry(
        domain=DOMAIN,
        data={
            CONF_HOST: "192.168.1.100",
            CONF_PORT: 3000,
            CONF_USERNAME: "admin",
            CONF_PASSWORD: "password",
        },
    )

    with patch("custom_components.adguard_hub.api.AdGuardHomeAPI.test_connection", return_value=True):
        config_entry.add_to_hass(hass)
        await hass.config_entries.async_setup(config_entry.entry_id)

    assert config_entry.state == config_entries.ConfigEntryState.LOADED

🔍 Debugging and Logging

Logging Configuration

import logging
_LOGGER = logging.getLogger(__name__)

# Log levels for different scenarios
_LOGGER.debug("Detailed debugging information")
_LOGGER.info("General information")
_LOGGER.warning("Warning about potential issues")
_LOGGER.error("Error occurred but recoverable")
_LOGGER.exception("Exception occurred with traceback")

Performance Monitoring

import time
from functools import wraps

def monitor_performance(func):
    """Decorator to monitor function performance."""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start_time = time.time()
        try:
            result = await func(*args, **kwargs)
            duration = time.time() - start_time
            _LOGGER.debug("%s completed in %.2fs", func.__name__, duration)
            return result
        except Exception as err:
            duration = time.time() - start_time
            _LOGGER.error("%s failed after %.2fs: %s", func.__name__, duration, err)
            raise
    return wrapper

📈 Metrics and Monitoring

Integration Health Metrics

class HealthMetrics:
    """Track integration health metrics."""

    def __init__(self):
        self.api_calls_total = 0
        self.api_calls_failed = 0
        self.last_successful_update = None
        self.last_error = None

    def record_api_call(self, success: bool, error: Exception = None):
        """Record API call metrics."""
        self.api_calls_total += 1
        if success:
            self.last_successful_update = datetime.now()
        else:
            self.api_calls_failed += 1
            self.last_error = str(error) if error else "Unknown error"

    @property
    def success_rate(self) -> float:
        """Calculate API call success rate."""
        if self.api_calls_total == 0:
            return 0.0
        return ((self.api_calls_total - self.api_calls_failed) / self.api_calls_total) * 100

For more technical details, see the source code in the repository or the Development Guide for setting up a development environment.