mirror of
https://github.com/home-assistant/core.git
synced 2026-09-13 04:01:03 +01:00
114 lines
3.4 KiB
Python
114 lines
3.4 KiB
Python
"""Support for turning on and off Pi-hole system."""
|
|
|
|
import logging
|
|
from typing import Any, override
|
|
|
|
from hole.exceptions import HoleError
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.components.switch import SwitchEntity
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
from homeassistant.helpers import config_validation as cv, entity_platform
|
|
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
|
|
|
from .const import DOMAIN, SERVICE_DISABLE, SERVICE_DISABLE_ATTR_DURATION
|
|
from .coordinator import PiHoleConfigEntry
|
|
from .entity import PiHoleEntity
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
entry: PiHoleConfigEntry,
|
|
async_add_entities: AddConfigEntryEntitiesCallback,
|
|
) -> None:
|
|
"""Set up the Pi-hole switch."""
|
|
name = entry.title
|
|
hole_data = entry.runtime_data
|
|
switches = [
|
|
PiHoleSwitch(
|
|
hole_data.api,
|
|
hole_data.coordinator,
|
|
name,
|
|
entry.entry_id,
|
|
)
|
|
]
|
|
async_add_entities(switches, True)
|
|
|
|
# register service
|
|
platform = entity_platform.async_get_current_platform()
|
|
platform.async_register_entity_service(
|
|
SERVICE_DISABLE,
|
|
{
|
|
vol.Required(SERVICE_DISABLE_ATTR_DURATION): vol.All(
|
|
cv.time_period_str, cv.positive_timedelta
|
|
),
|
|
},
|
|
"async_disable",
|
|
)
|
|
|
|
|
|
class PiHoleSwitch(PiHoleEntity, SwitchEntity):
|
|
"""Representation of a Pi-hole switch."""
|
|
|
|
_attr_icon = "mdi:pi-hole"
|
|
|
|
@property
|
|
@override
|
|
def name(self) -> str:
|
|
"""Return the name of the switch."""
|
|
return self._name
|
|
|
|
@property
|
|
@override
|
|
def unique_id(self) -> str:
|
|
"""Return the unique id of the switch."""
|
|
return f"{self._server_unique_id}/Switch"
|
|
|
|
@property
|
|
@override
|
|
def is_on(self) -> bool:
|
|
"""Return if the service is on."""
|
|
return self.api.status == "enabled" # type: ignore[no-any-return]
|
|
|
|
@override
|
|
async def async_turn_on(self, **kwargs: Any) -> None:
|
|
"""Turn on the service."""
|
|
try:
|
|
await self.api.enable()
|
|
await self.async_update()
|
|
except HoleError as err:
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="enable_failed",
|
|
translation_placeholders={"error": str(err)},
|
|
) from err
|
|
|
|
@override
|
|
async def async_turn_off(self, **kwargs: Any) -> None:
|
|
"""Turn off the service."""
|
|
await self.async_disable()
|
|
|
|
async def async_disable(self, duration: Any = None) -> None:
|
|
"""Disable the service for a given duration."""
|
|
duration_seconds = True # Disable infinitely by default
|
|
if duration is not None:
|
|
duration_seconds = duration.total_seconds()
|
|
_LOGGER.debug(
|
|
"Disabling Pi-hole '%s' (%s) for %d seconds",
|
|
self.name,
|
|
self.api.host,
|
|
duration_seconds,
|
|
)
|
|
try:
|
|
await self.api.disable(duration_seconds)
|
|
await self.async_update()
|
|
except HoleError as err:
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="disable_failed",
|
|
translation_placeholders={"error": str(err)},
|
|
) from err
|